feat: implement consistent retry logic for topic writer (#223)

* Do not send messages that have a timeout by cancelToken.
* If your value serializer throws an exception, this will be wrapped in a WriterException with unspecified status.
* Added BufferOverflowRetryTimeoutMs to the next try write.
* Rename _disposeTokenSource -> _disposeCts.
* Optimize write worker: if (_toSendBuffer.IsEmpty) continue.
* On RPC errors create DummyWriterSession.
* Message has been skipped because its sequence number is less than or equal to the last processed server's SeqNo.
* Calculate the next sequence number from the calculated previous messages.

Author: KirillKurdyukov

CHANGELOG.md

@@ -1,3 +1,5 @@
+- Topic Writer updated release candidate
+
 ## v0.9.0-rc0
 - Topic Writer release candidate
 - Fixed: grpc requests go via proxy on Grpc.NET.Client >= 2.44 

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

@@ -4,20 +4,15 @@ public class WriterException : Exception
 {
     public WriterException(string message) : base(message)
     {
-        Status = new Status(StatusCode.Unspecified);
     }
 
     public WriterException(string message, Status status) : base(message + ": " + status)
     {
-        Status = status;
     }
 
-    public WriterException(string message, Driver.TransportException e) : base(message, e)
+    public WriterException(string message, Exception inner) : base(message, inner)
     {
-        Status = e.Status;
     }
-
-    public Status Status { get; }
 }
 
 public class ReaderException : Exception

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

@@ -5,7 +5,6 @@ namespace Ydb.Sdk.Services.Topic;
 internal abstract class TopicSession<TFromClient, TFromServer> : IDisposable
 {
     private readonly Func<Task> _initialize;
-    private readonly Action<WriterException> _resetSessionOnTransportError;
 
     protected readonly IBidirectionalStream<TFromClient, TFromServer> Stream;
     protected readonly ILogger Logger;
@@ -17,17 +16,15 @@ protected TopicSession(
         IBidirectionalStream<TFromClient, TFromServer> stream,
         ILogger logger,
         string sessionId,
-        Func<Task> initialize,
-        Action<WriterException> resetSessionOnTransportError)
+        Func<Task> initialize)
     {
         Stream = stream;
         Logger = logger;
         SessionId = sessionId;
         _initialize = initialize;
-        _resetSessionOnTransportError = resetSessionOnTransportError;
     }
 
-    protected async void ReconnectSession(WriterException exception)
+    protected async void ReconnectSession()
     {
         if (Interlocked.CompareExchange(ref _isActive, 0, 1) == 0)
         {
@@ -36,9 +33,7 @@ protected async void ReconnectSession(WriterException exception)
             return;
         }
 
-        _resetSessionOnTransportError(exception);
-
-        Logger.LogInformation("WriterSession[{SessionId}] has been deactivated, starting to reconnect", SessionId);
+        Logger.LogInformation("TopicSession[{SessionId}] has been deactivated, starting to reconnect", SessionId);
 
         await _initialize();
     }

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)