Remove discarded stream

The stream will not be readable after it has been discarded.

This could cause applications to get an error that they didn't
previously got IF they use results outside of transaction boundary
(which is not allowed) BUT even if they previously didn't see an error
they actually didn't get the correct result.

Author: Peter Wilhelmsson

neo4j/internal/bolt/bolt4.go

@@ -413,6 +413,7 @@ func (b *bolt4) discardStream() {
 		if batch {
 			if discarded {
 				// Response to discard, see below
+				b.streams.remove(stream)
 				return
 			}
 			// Discard all! After this the next receive will get another batch
@@ -423,6 +424,7 @@ func (b *bolt4) discardStream() {
 			b.out.appendDiscardNQid(stream.fetchSize, stream.qid)
 			b.out.send(b.conn)
 		} else if sum != nil || b.err != nil {
+			b.streams.remove(stream)
 			return
 		}
 	}

neo4j/internal/bolt/bolt4_test.go

@@ -318,6 +318,7 @@ func TestBolt4(ot *testing.T) {
 
 		err = bolt.TxCommit(tx)
 		AssertNoError(t, err)
+		assertBoltState(t, bolt4_ready, bolt)
 	})
 
 	ot.Run("Begin transaction with bookmark success", func(t *testing.T) {

neo4j/internal/bolt/stream.go

@@ -115,13 +115,9 @@ func (o *openstreams) detach(sum *db.Summary, err error) {
 	}
 
 	o.remove(o.curr)
-	o.curr = nil
-	if o.num <= 0 {
-		o.onEmpty()
-	}
 }
 
-// Streams can be paused when they have received a 'has_more' response from server
+// Streams can be paused when they have received a "has_more" response from server
 // Pauses the current stream
 func (o *openstreams) pause() {
 	o.curr = nil
@@ -136,10 +132,17 @@ func (o *openstreams) resume(s *stream) {
 	o.curr = s
 }
 
-// Internal, "removes" the stream by setting it's corresponding entry to nil.
+// Removes the stream by disabling its key and removing it from the count of streams.
+// If the stream is current the current is set to nil.
 func (o *openstreams) remove(s *stream) {
 	o.num--
 	s.key = 0
+	if o.curr == s {
+		o.curr = nil
+	}
+	if o.num == 0 {
+		o.onEmpty()
+	}
 }
 
 func (o *openstreams) reset() {