Merge pull request #1686 from ydb-platform/close-stream-result

Close stream result

Author: asmyasnikov

internal/query/result.go

@@ -143,10 +143,9 @@ func newResult(
 	}
 
 	r.closeOnce = sync.OnceFunc(func() {
-		for _, onClose := range r.onClose {
-			onClose()
+		for i := range r.onClose { // descending calls for LIFO
+			r.onClose[len(r.onClose)-i-1]()
 		}
-		r.stream = nil
 	})
 
 	if r.trace != nil {
@@ -239,6 +238,8 @@ func (r *streamResult) Close(ctx context.Context) (finalErr error) {
 
 	for {
 		select {
+		case <-ctx.Done():
+			return xerrors.WithStackTrace(ctx.Err())
 		case <-r.closed:
 			return nil
 		default:
@@ -280,6 +281,7 @@ func (r *streamResult) nextResultSet(ctx context.Context) (_ *resultSet, err err
 			}
 			if part.GetResultSetIndex() < r.resultSetIndex {
 				r.closeOnce()
+
 				if part.GetResultSetIndex() <= 0 && r.resultSetIndex > 0 {
 					return nil, xerrors.WithStackTrace(io.EOF)
 				}

internal/query/transaction_test.go

@@ -274,6 +274,10 @@ func TestTxOnCompleted(t *testing.T) {
 
 			res, err := tx.Query(sf.Context(e), "")
 			require.NoError(t, err)
+			_, err = res.NextResultSet(sf.Context(e))
+			require.NoError(t, err)
+			_, err = res.NextResultSet(sf.Context(e))
+			require.ErrorIs(t, err, io.EOF)
 			_ = res.Close(sf.Context(e))
 			time.Sleep(time.Millisecond) // time for reaction for closing channel
 			require.Empty(t, completed)
@@ -332,7 +336,12 @@ func TestTxOnCompleted(t *testing.T) {
 			})
 
 			res, err := tx.Query(sf.Context(e), "", options.WithCommit())
-			_ = res.Close(sf.Context(e))
+			require.NoError(t, err)
+			_, err = res.NextResultSet(sf.Context(e))
+			require.NoError(t, err)
+			_, err = res.NextResultSet(sf.Context(e))
+			require.ErrorIs(t, err, io.EOF)
+			err = res.Close(sf.Context(e))
 			require.NoError(t, err)
 			xtest.SpinWaitCondition(t, &completedMutex, func() bool {
 				return len(completed) != 0
@@ -384,7 +393,10 @@ func TestTxOnCompleted(t *testing.T) {
 			_, err = res.NextResultSet(sf.Context(e))
 			require.NoError(t, err)
 
-			_ = res.Close(sf.Context(e))
+			_, err = res.NextResultSet(sf.Context(e))
+			require.ErrorIs(t, err, io.EOF)
+
+			err = res.Close(sf.Context(e))
 			require.NoError(t, err)
 			xtest.SpinWaitCondition(t, &completedMutex, func() bool {
 				return len(completed) != 0

internal/xtest/leak.go

@@ -62,7 +62,9 @@ func findGoroutinesLeak() error {
 		unexpectedGoroutines = append(unexpectedGoroutines, g)
 	}
 	if l := len(unexpectedGoroutines); l > 0 {
-		return fmt.Errorf("found %d unexpected goroutines:\n%s", len(goroutines), strings.Join(goroutines, "\n"))
+		return fmt.Errorf("found %d unexpected goroutines:\n\n%s",
+			len(goroutines), strings.Join(goroutines, "\n\n"),
+		)
 	}
 
 	return nil