feat: writer retry always

Author: KirillKurdyukov

src/Ydb.Sdk/src/Services/Topic/Writer/WriteResult.cs

@@ -4,6 +4,8 @@ namespace Ydb.Sdk.Services.Topic.Writer;
 
 public class WriteResult
 {
+    internal static readonly WriteResult Skipped = new();
+
     private readonly long _offset;
 
     internal WriteResult(StreamWriteMessage.Types.WriteResponse.Types.WriteAck ack)
@@ -23,6 +25,11 @@ internal WriteResult(StreamWriteMessage.Types.WriteResponse.Types.WriteAck ack)
         }
     }
 
+    private WriteResult()
+    {
+        Status = PersistenceStatus.AlreadyWritten;
+    }
+
     public PersistenceStatus Status { get; }
 
     public bool TryGetOffset(out long offset)

src/Ydb.Sdk/src/Services/Topic/Writer/Writer.cs

@@ -7,7 +7,6 @@
 
 namespace Ydb.Sdk.Services.Topic.Writer;
 
-using InitResponse = StreamWriteMessage.Types.InitResponse;
 using MessageData = StreamWriteMessage.Types.WriteRequest.Types.MessageData;
 using MessageFromClient = StreamWriteMessage.Types.FromClient;
 using MessageFromServer = StreamWriteMessage.Types.FromServer;
@@ -134,6 +133,11 @@ private async void StartWriteWorker()
             await _tcsWakeUp.Task;
             _tcsWakeUp = new TaskCompletionSource();
 
+            if (_toSendBuffer.IsEmpty)
+            {
+                continue;
+            }
+
             await _session.Write(_toSendBuffer);
         }
     }
@@ -179,8 +183,8 @@ private async Task Initialize()
             await stream.Write(new MessageFromClient { InitRequest = initRequest });
             if (!await stream.MoveNextAsync())
             {
-                _session = new NotStartedWriterSession(
-                    $"Stream unexpectedly closed by YDB server. Current InitRequest: {initRequest}");
+                _logger.LogError("Stream unexpectedly closed by YDB server. Current InitRequest: {initRequest}",
+                    initRequest);
 
                 _ = Task.Run(Initialize, _disposeCts.Token);
 
@@ -193,14 +197,18 @@ private async Task Initialize()
 
             if (status.IsNotSuccess)
             {
-                _session = new NotStartedWriterSession("Initialization failed", status);
-
-                if (status.StatusCode != StatusCode.SchemeError)
+                if (RetrySettings.DefaultInstance.GetRetryRule(status.StatusCode).Policy != RetryPolicy.None)
                 {
+                    _logger.LogError("Writer initialization failed to start. Reason: {Status}", status);
+
                     _ = Task.Run(Initialize, _disposeCts.Token);
                 }
+                else
+                {
+                    _logger.LogCritical("Writer initialization failed to start. Reason: {Status}", status);
 
-                _logger.LogCritical("Writer initialization failed to start. Reason: {Status}", status);
+                    _session = new NotStartedWriterSession("Initialization failed", status);
+                }
 
                 return;
             }
@@ -222,45 +230,59 @@ private async Task Initialize()
                 return;
             }
 
-            var newSession = new WriterSession(
-                _config,
-                stream,
-                initResponse,
-                Initialize,
-                _logger,
-                _toSendBuffer
-            );
-
-            if (!_inFlightMessages.IsEmpty)
+            var copyInFlightMessages = new ConcurrentQueue<MessageSending>();
+            var lastSeqNo = initResponse.LastSeqNo;
+            while (_inFlightMessages.TryDequeue(out var sendData))
             {
-                var copyInFlightMessages = new ConcurrentQueue<MessageSending>();
-                while (_inFlightMessages.TryDequeue(out var sendData))
+                if (sendData.Tcs.Task.IsFaulted)
                 {
-                    if (sendData.Tcs.Task.IsFaulted)
-                    {
-                        _logger.LogWarning("Message[SeqNo={SeqNo}] is cancelled", sendData.MessageData.SeqNo);
+                    _logger.LogWarning("Message[SeqNo={SeqNo}] is cancelled", sendData.MessageData.SeqNo);
 
-                        continue;
-                    }
+                    continue;
+                }
 
-                    copyInFlightMessages.Enqueue(sendData);
+                if (lastSeqNo >= sendData.MessageData.SeqNo)
+                {
+                    _logger.LogWarning(
+                        "Message[SeqNo={SeqNo}] has been skipped because its sequence number " +
+                        "is less than or equal to the last processed server's SeqNo[{LastSeqNo}]",
+                        sendData.MessageData.SeqNo, lastSeqNo);
+
+                    sendData.Tcs.TrySetResult(WriteResult.Skipped);
+
+                    continue;
                 }
 
-                await newSession.Write(copyInFlightMessages); // retry prev in flight messages
+
+                // Calculate the next sequence number from the calculated previous messages.
+                lastSeqNo = Math.Max(lastSeqNo, sendData.MessageData.SeqNo);
+
+                copyInFlightMessages.Enqueue(sendData);
             }
 
-            _session = newSession;
-            newSession.RunProcessingWriteAck();
-            if (!_toSendBuffer.IsEmpty)
+            var newSession = new WriterSession(
+                config: _config,
+                stream: stream,
+                lastSeqNo: lastSeqNo,
+                sessionId: initResponse.SessionId,
+                initialize: Initialize,
+                logger: _logger,
+                inFlightMessages: _inFlightMessages
+            );
+
+            if (!copyInFlightMessages.IsEmpty)
             {
-                WakeUpWorker(); // send buffer    
+                await newSession.Write(copyInFlightMessages); // retry prev in flight messages    
             }
+
+            _session = newSession;
+            newSession.RunProcessingWriteAck();
+            WakeUpWorker(); // attempt send buffer        
         }
         catch (Driver.TransportException e)
         {
             _logger.LogError(e, "Transport error on creating WriterSession");
 
-            _session = DummyWriteSession.Instance;
             _ = Task.Run(Initialize, _disposeCts.Token);
         }
     }
@@ -321,6 +343,10 @@ internal class DummyWriteSession : IWriteSession
 {
     internal static readonly DummyWriteSession Instance = new();
 
+    private DummyWriteSession()
+    {
+    }
+
     public void Dispose()
     {
     }
@@ -341,20 +367,21 @@ internal class WriterSession : TopicSession<MessageFromClient, MessageFromServer
     public WriterSession(
         WriterConfig config,
         WriterStream stream,
-        InitResponse initResponse,
+        long lastSeqNo,
+        string sessionId,
         Func<Task> initialize,
         ILogger logger,
         ConcurrentQueue<MessageSending> inFlightMessages
     ) : base(
         stream,
         logger,
-        initResponse.SessionId,
+        sessionId,
         initialize
     )
     {
         _config = config;
         _inFlightMessages = inFlightMessages;
-        Volatile.Write(ref _seqNum, initResponse.LastSeqNo); // happens-before for Volatile.Read
+        Volatile.Write(ref _seqNum, lastSeqNo); // happens-before for Volatile.Read
     }
 
     public async Task Write(ConcurrentQueue<MessageSending> toSendBuffer)
@@ -406,7 +433,7 @@ internal async void RunProcessingWriteAck()
 
                 if (status.IsNotSuccess)
                 {
-                    Logger.LogWarning(
+                    Logger.LogError(
                         "WriterSession[{SessionId}] received unsuccessful status while processing writeAck: {Status}",
                         SessionId, status);
                     return;

src/Ydb.Sdk/tests/Topic/WriterIntegrationTests.cs

@@ -45,8 +45,10 @@ public async Task WriteAsync_WhenTopicNotFound_ReturnNotFoundException()
         using var writer = new WriterBuilder<string>(_driver, _topicName + "_not_found")
             { ProducerId = "producerId" }.Build();
 
-        Assert.Equal("StatusCode.SchemeError", (await Assert.ThrowsAsync<WriterException>(
-            () => writer.WriteAsync("hello world"))).Message);
+        Assert.Contains(
+            $"Initialization failed: Status: SchemeError, Issues:\n[500017] Error: no path 'local/{_topicName + "_not_found"}'",
+            (await Assert.ThrowsAsync<WriterException>(() => writer.WriteAsync("hello world"))).Message
+        );
     }
 
     [Fact]