Fixes problem with discarding big streams

If user hasn't consumed the result and the stream contains more than
1000 records (default fetch size) the discard could hang.

Author: Peter Wilhelmsson

neo4j/internal/bolt/bolt4.go

@@ -407,9 +407,18 @@ func (b *bolt4) discardStream() {
 		return
 	}
 
+	discarded := false
 	for {
 		_, batch, sum := b.receiveNext()
 		if batch {
+			if discarded {
+				// Response to discard, see below
+				return
+			}
+			// Discard all! After this the next receive will get another batch
+			// as a response to the discard, we need to keep track of that we
+			// already sent a discard.
+			discarded = true
 			stream.fetchSize = -1
 			b.out.appendDiscardNQid(stream.fetchSize, stream.qid)
 			b.out.send(b.conn)

neo4j/internal/bolt/bolt4_test.go

@@ -284,6 +284,42 @@ func TestBolt4(ot *testing.T) {
 		AssertStringEqual(t, committedBookmark, bolt.Bookmark())
 	})
 
+	// Verifies that current stream is discarded correctly even if it is larger
+	// than what is served by a single pull.
+	ot.Run("Commit while streaming", func(t *testing.T) {
+		qid := int64(2)
+		bolt, cleanup := connectToServer(t, func(srv *bolt4server) {
+			srv.accept(4)
+			srv.waitForTxBegin()
+			srv.send(msgSuccess, map[string]interface{}{})
+			srv.waitForRun()
+			srv.waitForPullN(1)
+			// Send Pull response
+			srv.send(msgSuccess, map[string]interface{}{"fields": []interface{}{"k"}, "t_first": int64(1), "qid": qid})
+			// ... and the record
+			srv.send(msgRecord, []interface{}{"v1"})
+			// ... and the batch summary
+			srv.send(msgSuccess, map[string]interface{}{"has_more": true})
+			// Wait for the discard message
+			srv.waitForDiscardN(-1, int(qid))
+			// Respond to discard with has more to indicate that there are more records
+			srv.send(msgSuccess, map[string]interface{}{"has_more": true})
+			// Wait for the commit
+			srv.waitForTxCommit()
+			srv.send(msgSuccess, map[string]interface{}{"bookmark": "x"})
+		})
+		defer cleanup()
+		defer bolt.Close()
+
+		tx, err := bolt.TxBegin(db.TxConfig{Mode: db.ReadMode})
+		AssertNoError(t, err)
+		_, err = bolt.RunTx(tx, db.Command{Cypher: "Whatever", FetchSize: 1})
+		AssertNoError(t, err)
+
+		err = bolt.TxCommit(tx)
+		AssertNoError(t, err)
+	})
+
 	ot.Run("Begin transaction with bookmark success", func(t *testing.T) {
 		committedBookmark := "cbm"
 		bolt, cleanup := connectToServer(t, func(srv *bolt4server) {