fix-listener-close

CHANGELOG.md

@@ -1,3 +1,4 @@
+* Fixed error on experimental TopicListener.Close
 * Disabled reporting of `ydb_go_sdk_query_session_count` when metrics are disabled
 * Disabled reporting of `ydb_go_sdk_ydb_query_session_create_latency` histogram metrics when metrics are disabled
 * Allowed skip column for `ScanStruct` by tag `-` 

internal/topic/topiclistenerinternal/stream_listener.go

@@ -36,6 +36,8 @@ type streamListener struct {
 	hasNewMessagesToSend empty.Chan
 	syncCommitter        *topicreadercommon.Committer
 
+	closing atomic.Bool
+
 	m              xsync.Mutex
 	messagesToSend []rawtopicreader.ClientMessage
 }
@@ -56,7 +58,7 @@ func newStreamListener(
 
 	res.initVars(sessionIDCounter)
 	if err := res.initStream(connectionCtx, client); err != nil {
-		res.closeWithTimeout(connectionCtx, err)
+		res.goClose(connectionCtx, err)
 
 		return nil, err
 	}
@@ -75,17 +77,22 @@ func newStreamListener(
 }
 
 func (l *streamListener) Close(ctx context.Context, reason error) error {
+	if !l.closing.CompareAndSwap(false, true) {
+		return errTopicListenerClosed
+	}
+
 	var resErrors []error
 
+	// should be first because background wait stop of steams
 	if l.stream != nil {
 		l.streamClose(reason)
 	}
 
-	if err := l.syncCommitter.Close(ctx, reason); err != nil {
+	if err := l.background.Close(ctx, reason); err != nil {
 		resErrors = append(resErrors, err)
 	}
 
-	if err := l.background.Close(ctx, reason); err != nil {
+	if err := l.syncCommitter.Close(ctx, reason); err != nil {
 		resErrors = append(resErrors, err)
 	}
 
@@ -109,10 +116,12 @@ func (l *streamListener) Close(ctx context.Context, reason error) error {
 	return errors.Join(resErrors...)
 }
 
-func (l *streamListener) closeWithTimeout(ctx context.Context, reason error) {
+func (l *streamListener) goClose(ctx context.Context, reason error) {
 	ctx, cancel := context.WithTimeout(xcontext.ValueOnly(ctx), time.Second)
 	l.streamClose(reason)
-	_ = l.background.Close(ctx, reason)
+	go func() {
+		_ = l.background.Close(ctx, reason)
+	}()
 
 	cancel()
 }
@@ -145,7 +154,7 @@ func (l *streamListener) initStream(ctx context.Context, client TopicClient) err
 			err := xerrors.WithStackTrace(xerrors.Wrap(fmt.Errorf(
 				"ydb: topic listener stream init timeout: %w", ctx.Err(),
 			)))
-			l.closeWithTimeout(ctx, err)
+			l.goClose(ctx, err)
 			l.streamClose(err)
 		case <-initDone:
 			// pass
@@ -216,7 +225,7 @@ func (l *streamListener) sendMessagesLoop(ctx context.Context) {
 
 			for _, m := range messages {
 				if err := l.stream.Send(m); err != nil {
-					l.closeWithTimeout(ctx, xerrors.WithStackTrace(xerrors.Wrap(fmt.Errorf(
+					l.goClose(ctx, xerrors.WithStackTrace(xerrors.Wrap(fmt.Errorf(
 						"ydb: failed send message by grpc to topic reader stream from listener: %w",
 						err,
 					))))
@@ -236,7 +245,7 @@ func (l *streamListener) receiveMessagesLoop(ctx context.Context) {
 
 		mess, err := l.stream.Recv()
 		if err != nil {
-			l.closeWithTimeout(ctx, xerrors.WithStackTrace(
+			l.goClose(ctx, xerrors.WithStackTrace(
 				fmt.Errorf("ydb: failed read message from the stream in the topic reader listener: %w", err),
 			))
 
@@ -263,7 +272,7 @@ func (l *streamListener) onReceiveServerMessage(ctx context.Context, mess rawtop
 		// todo log
 	}
 	if err != nil {
-		l.closeWithTimeout(ctx, err)
+		l.goClose(ctx, err)
 	}
 }
 

internal/topic/topiclistenerinternal/topic_listener_reconnector.go

@@ -10,7 +10,10 @@ import (
 	"github.com/ydb-platform/ydb-go-sdk/v3/internal/empty"
 )
 
-var ErrUserCloseTopic = errors.New("ydb: user closed topic listener")
+var (
+	ErrUserCloseTopic      = errors.New("ydb: user closed topic listener")
+	errTopicListenerClosed = errors.New("ydb: the topic listener already closed")
+)
 
 type TopicListenerReconnector struct {
 	streamConfig *StreamListenerConfig
@@ -22,6 +25,7 @@ type TopicListenerReconnector struct {
 	connectionResult    error
 	connectionCompleted empty.Chan
 	connectionIDCounter atomic.Int64
+	closing             atomic.Bool
 
 	m              sync.Mutex
 	streamListener *streamListener
@@ -45,6 +49,9 @@ func NewTopicListenerReconnector(
 }
 
 func (lr *TopicListenerReconnector) Close(ctx context.Context, reason error) error {
+	if !lr.closing.CompareAndSwap(false, true) {
+		return errTopicListenerClosed
+	}
 	var closeErrors []error
 	err := lr.background.Close(ctx, reason)
 	closeErrors = append(closeErrors, err)