* Fixed data race on session stream queries

Author: asmyasnikov

CHANGELOG.md

@@ -1,3 +1,4 @@
+* Fixed data race on session stream queries
 * Renamed `internal/router` package to `internal/balancer` for unambiguous understanding of package mission
 * Implemented detection of local data-center with measuring tcp dial RTT
 * Added `trace.Driver.OnBalancer{Init,Close,ChooseEndpoint,Update}` events

internal/table/session.go

@@ -917,25 +917,20 @@ func (s *session) StreamReadTable(
 
 	ctx, cancel := context.WithCancel(ctx)
 
+	t := s.trailer()
+	defer t.processHints()
+
 	stream, err = s.tableService.StreamReadTable(
 		balancer.WithEndpoint(ctx, s),
 		&request,
+		t.Trailer(),
 	)
 
 	if err != nil {
 		cancel()
 		return nil, xerrors.WithStackTrace(err)
 	}
 
-	select {
-	case <-stream.Context().Done():
-		// nop
-	default:
-		if checkHintSessionClose(stream.Trailer()) {
-			s.SetStatus(options.SessionClosing)
-		}
-	}
-
 	return scanner.NewStream(
 		func(ctx context.Context) (
 			set *Ydb.ResultSet,
@@ -1006,25 +1001,20 @@ func (s *session) StreamExecuteScanQuery(
 
 	ctx, cancel := context.WithCancel(ctx)
 
+	t := s.trailer()
+	defer t.processHints()
+
 	stream, err = s.tableService.StreamExecuteScanQuery(
 		balancer.WithEndpoint(ctx, s),
 		&request,
+		t.Trailer(),
 	)
 
 	if err != nil {
 		cancel()
 		return nil, xerrors.WithStackTrace(err)
 	}
 
-	select {
-	case <-stream.Context().Done():
-		// nop
-	default:
-		if checkHintSessionClose(stream.Trailer()) {
-			s.SetStatus(options.SessionClosing)
-		}
-	}
-
 	return scanner.NewStream(
 		func(ctx context.Context) (
 			set *Ydb.ResultSet,