Fixed race condition between close writer queue and receive acks in b…

…ackground stream. (#1054)

Co-authored-by: Aleksey Myasnikov <[email protected]>

CHANGELOG.md

@@ -1,3 +1,4 @@
+* Fixed sometime panic on topic writer closing
 * Added experimental query parameters builder `ydb.ParamsBuilder()`
 * Changed types of `table/table.{QueryParameters,ParameterOption}` to aliases on `internal/params.{Parameters,NamedValue}`
 * Fixed bug with optional decimal serialization 

internal/topic/topicwriterinternal/queue.go

@@ -14,6 +14,7 @@ import (
 
 var (
 	errCloseClosedMessageQueue   = xerrors.Wrap(errors.New("ydb: close closed message queue"))
+	errAckOnClosedMessageQueue   = xerrors.Wrap(errors.New("ydb: ack on closed message queue"))
 	errGetMessageFromClosedQueue = xerrors.Wrap(errors.New("ydb: get message from closed message queue"))
 	errAddUnorderedMessages      = xerrors.Wrap(errors.New("ydb: add unordered messages"))
 	errAckUnexpectedMessage      = xerrors.Wrap(errors.New("ydb: ack unexpected message"))
@@ -151,6 +152,9 @@ func (q *messageQueue) AcksReceived(acks []rawtopicwriter.WriteAck) error {
 			q.OnAckReceived(ackReceivedCounter)
 		}
 	}()
+	if q.closed {
+		return xerrors.WithStackTrace(errAckOnClosedMessageQueue)
+	}
 
 	for i := range acks {
 		if err := q.ackReceivedNeedLock(acks[i].SeqNo); err != nil {

internal/topic/topicwriterinternal/queue_test.go

@@ -407,6 +407,26 @@ func TestQueuePanicOnOverflow(t *testing.T) {
 	})
 }
 
+func TestRegressionIssue1038_ReceiveAckAfterCloseQueue(t *testing.T) {
+	counter := 0
+
+	q := newMessageQueue()
+	q.OnAckReceived = func(count int) {
+		counter -= count
+	}
+	require.NoError(t, q.AddMessages(newTestMessagesWithContent(1)))
+	counter++
+
+	require.NoError(t, q.Close(errors.New("test err")))
+	require.ErrorIs(t, q.AcksReceived([]rawtopicwriter.WriteAck{
+		{
+			SeqNo:              1,
+			MessageWriteStatus: rawtopicwriter.MessageWriteStatus{},
+		},
+	}), errAckOnClosedMessageQueue)
+	require.Zero(t, counter)
+}
+
 func TestQueue_Ack(t *testing.T) {
 	t.Run("First", func(t *testing.T) {
 		q := newMessageQueue()

internal/topic/topicwriterinternal/writer_single_stream.go

@@ -189,7 +189,7 @@ func (w *SingleStreamWriter) receiveMessagesLoop(ctx context.Context) {
 
 		switch m := mess.(type) {
 		case *rawtopicwriter.WriteResult:
-			if err = w.cfg.queue.AcksReceived(m.Acks); err != nil {
+			if err = w.cfg.queue.AcksReceived(m.Acks); err != nil && !errors.Is(err, errCloseClosedMessageQueue) {
 				reason := xerrors.WithStackTrace(err)
 				closeCtx, closeCtxCancel := xcontext.WithCancel(ctx)
 				closeCtxCancel()