fix

Author: KirillKurdyukov

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

@@ -7,9 +7,9 @@ protected YdbTopicException(string message) : base(message)
     }
 }
 
-public class YdbProducerException : YdbTopicException
+public class YdbWriterException : YdbTopicException
 {
-    public YdbProducerException(string message) : base(message)
+    public YdbWriterException(string message) : base(message)
     {
     }
 }

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

@@ -0,0 +1,8 @@
+namespace Ydb.Sdk.Services.Topic;
+
+public interface IWriter<TValue>
+{
+    public Task<WriteResult> WriteAsync(TValue data);
+
+    public Task<WriteResult> WriteAsync(Message<TValue> message);
+}

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

@@ -10,6 +10,7 @@ internal abstract class TopicSession : IDisposable
     protected readonly string SessionId;
 
     private int _isActive = 1;
+    private bool _disposed;
 
     protected TopicSession(ILogger logger, string sessionId, Func<Task> initialize)
     {
@@ -22,13 +23,14 @@ protected async void ReconnectSession()
     {
         if (Interlocked.CompareExchange(ref _isActive, 0, 1) == 0)
         {
-            Logger.LogWarning("The reconnect has already been launched");
+            Logger.LogWarning("Skipping reconnect. A reconnect session has already been initiated");
 
             return;
         }
-        
 
-        while (true)
+        Logger.LogInformation("WriterSession[{SessionId}] has been deactivated, starting to reconnect", SessionId);
+
+        while (!_disposed)
         {
             try
             {
@@ -37,12 +39,16 @@ protected async void ReconnectSession()
             }
             catch (Exception e)
             {
-                Logger.LogError(e, "Error reconnect session!");
+                Logger.LogError(e, "Unable to reconnect the session due to the following error");
             }
         }
     }
 
     public void Dispose()
     {
+        lock (this)
+        {
+            _disposed = true;
+        }
     }
 }

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

@@ -2,11 +2,11 @@
 
 namespace Ydb.Sdk.Services.Topic;
 
-public class SendResult
+public class WriteResult
 {
     private readonly long _offset;
 
-    internal SendResult(StreamWriteMessage.Types.WriteResponse.Types.WriteAck ack)
+    internal WriteResult(StreamWriteMessage.Types.WriteResponse.Types.WriteAck ack)
     {
         switch (ack.MessageWriteStatusCase)
         {
@@ -19,7 +19,7 @@ internal SendResult(StreamWriteMessage.Types.WriteResponse.Types.WriteAck ack)
                 break;
             case StreamWriteMessage.Types.WriteResponse.Types.WriteAck.MessageWriteStatusOneofCase.None:
             default:
-                throw new YdbProducerException($"Unexpected WriteAck status: {ack.MessageWriteStatusCase}");
+                throw new YdbWriterException($"Unexpected WriteAck status: {ack.MessageWriteStatusCase}");
         }
     }
 

src/Ydb.Sdk/src/Services/Topic/SendResult.cs renamed to src/Ydb.Sdk/src/Services/Topic/WriteResult.cs

@@ -11,33 +11,33 @@ namespace Ydb.Sdk.Services.Topic;
 using InitResponse = StreamWriteMessage.Types.InitResponse;
 using MessageData = StreamWriteMessage.Types.WriteRequest.Types.MessageData;
 using MessageFromClient = StreamWriteMessage.Types.FromClient;
-using ProducerStream = Driver.BidirectionalStream<
+using WriterStream = Driver.BidirectionalStream<
     StreamWriteMessage.Types.FromClient,
     StreamWriteMessage.Types.FromServer
 >;
 
-internal class Producer<TValue> : IProducer<TValue>
+internal class Writer<TValue> : IWriter<TValue>
 {
-    private readonly ProducerConfig _config;
-    private readonly ILogger<Producer<TValue>> _logger;
+    private readonly WriterConfig _config;
+    private readonly ILogger<Writer<TValue>> _logger;
     private readonly ISerializer<TValue> _serializer;
 
     private readonly ConcurrentQueue<MessageSending> _inFlightMessages = new();
     private readonly ConcurrentQueue<MessageSending> _toSendBuffer = new();
     private readonly SemaphoreSlim _writeSemaphoreSlim = new(1);
 
-    private volatile ProducerSession _session = null!;
+    private volatile WriterSession _session = null!;
 
-    internal Producer(ProducerConfig producerConfig, ISerializer<TValue> serializer)
+    internal Writer(WriterConfig config, ISerializer<TValue> serializer)
     {
-        _config = producerConfig;
+        _config = config;
         _serializer = serializer;
-        _logger = producerConfig.Driver.LoggerFactory.CreateLogger<Producer<TValue>>();
+        _logger = config.Driver.LoggerFactory.CreateLogger<Writer<TValue>>();
     }
 
     internal async Task Initialize()
     {
-        _logger.LogInformation("Producer session initialization started. ProducerConfig: {ProducerConfig}", _config);
+        _logger.LogInformation("Writer session initialization started. WriterConfig: {WriterConfig}", _config);
 
         var stream = _config.Driver.BidirectionalStreamCall(
             TopicService.StreamWriteMethod,
@@ -55,10 +55,12 @@ internal async Task Initialize()
             initRequest.MessageGroupId = _config.MessageGroupId;
         }
 
+        _logger.LogDebug("Sending initialization request for the write stream: {InitRequest}", initRequest);
+
         await stream.Write(new MessageFromClient { InitRequest = initRequest });
         if (!await stream.MoveNextAsync())
         {
-            throw new YdbProducerException(
+            throw new YdbWriterException(
                 $"Stream unexpectedly closed by YDB server. Current InitRequest: {initRequest}");
         }
 
@@ -68,23 +70,39 @@ internal async Task Initialize()
 
         var initResponse = receivedInitMessage.InitResponse;
 
+        _logger.LogDebug("Received a response for the initialization request on the write stream: {InitResponse}",
+            initResponse);
+
         if (!initResponse.SupportedCodecs.Codecs.Contains((int)_config.Codec))
         {
-            throw new YdbProducerException($"Topic is not supported codec: {_config.Codec}");
+            throw new YdbWriterException($"Topic[{_config.TopicPath}] is not supported codec: {_config.Codec}");
+        }
+
+        _session = new WriterSession(_config, stream, initResponse, Initialize, _logger);
+
+        await _writeSemaphoreSlim.WaitAsync();
+        try
+        {
+            _logger.LogDebug("Retrying to send pending in-flight messages after stream restart");
+
+            await _session.Write(_inFlightMessages, _inFlightMessages);
+        }
+        finally
+        {
+            _writeSemaphoreSlim.Release();
         }
 
-        _session = new ProducerSession(_config, stream, initResponse, Initialize, _logger);
         _ = _session.RunProcessingWriteAck(_inFlightMessages);
     }
 
-    public Task<SendResult> SendAsync(TValue data)
+    public Task<WriteResult> WriteAsync(TValue data)
     {
-        return SendAsync(new Message<TValue>(data));
+        return WriteAsync(new Message<TValue>(data));
     }
 
-    public async Task<SendResult> SendAsync(Message<TValue> message)
+    public async Task<WriteResult> WriteAsync(Message<TValue> message)
     {
-        TaskCompletionSource<SendResult> completeTask = new();
+        TaskCompletionSource<WriteResult> completeTask = new();
 
         var data = _serializer.Serialize(message.Data);
         var messageData = new MessageData
@@ -123,16 +141,16 @@ public async Task<SendResult> SendAsync(Message<TValue> message)
 }
 
 // No thread safe
-internal class ProducerSession : TopicSession
+internal class WriterSession : TopicSession
 {
-    private readonly ProducerConfig _config;
-    private readonly ProducerStream _stream;
+    private readonly WriterConfig _config;
+    private readonly WriterStream _stream;
 
     private long _seqNum;
 
-    public ProducerSession(
-        ProducerConfig config,
-        ProducerStream stream,
+    public WriterSession(
+        WriterConfig config,
+        WriterStream stream,
         InitResponse initResponse,
         Func<Task> initialize,
         ILogger logger) : base(logger, initResponse.SessionId, initialize)
@@ -146,7 +164,7 @@ internal async Task RunProcessingWriteAck(ConcurrentQueue<MessageSending> inFlig
     {
         try
         {
-            Logger.LogInformation("ProducerSession[{SessionId}] is running processing writeAck", SessionId);
+            Logger.LogInformation("WriterSession[{SessionId}] is running processing writeAck", SessionId);
 
             await foreach (var messageFromServer in _stream)
             {
@@ -155,25 +173,55 @@ internal async Task RunProcessingWriteAck(ConcurrentQueue<MessageSending> inFlig
                 if (status.IsNotSuccess)
                 {
                     Logger.LogWarning(
-                        "ProducerSession[{SessionId}] received unsuccessful status while processing writeAck: {Status}",
+                        "WriterSession[{SessionId}] received unsuccessful status while processing writeAck: {Status}",
                         SessionId, status);
                     return;
                 }
 
                 foreach (var ack in messageFromServer.WriteResponse.Acks)
                 {
-                    if (!inFlightMessages.TryDequeue(out var messageFromClient))
+                    if (!inFlightMessages.TryPeek(out var messageFromClient))
                     {
+                        Logger.LogCritical("No client message was found upon receipt of an acknowledgement: {WriteAck}",
+                            ack);
+
                         break;
                     }
 
-                    messageFromClient.TaskCompletionSource.SetResult(new SendResult(ack));
+                    if (messageFromClient.MessageData.SeqNo > ack.SeqNo)
+                    {
+                        Logger.LogCritical(
+                            @"The sequence number of the client's message in the queue is greater than the server's write acknowledgment number. 
+Skipping the WriteAck... 
+Client SeqNo: {SeqNo}, WriteAck: {WriteAck}",
+                            messageFromClient.MessageData.SeqNo, ack);
+
+                        continue;
+                    }
+
+                    if (messageFromClient.MessageData.SeqNo < ack.SeqNo)
+                    {
+                        Logger.LogCritical(
+                            @"The sequence number of the client's message in the queue is less than the server's write acknowledgment number. 
+Completing task on exception...
+Client SeqNo: {SeqNo}, WriteAck: {WriteAck}",
+                            messageFromClient.MessageData.SeqNo, ack);
+
+                        messageFromClient.TaskCompletionSource.SetException(new YdbWriterException(
+                            $"Client SeqNo[{messageFromClient.MessageData.SeqNo}] is less then server's WriteAck[{ack}]"));
+                    }
+                    else
+                    {
+                        messageFromClient.TaskCompletionSource.SetResult(new WriteResult(ack));
+                    }
+
+                    inFlightMessages.TryDequeue(out _); // Dequeue 
                 }
             }
         }
         catch (Exception e)
         {
-            Logger.LogError(e, "ProducerSession[{SessionId}] have error on processing writeAck", SessionId);
+            Logger.LogError(e, "WriterSession[{SessionId}] have error on processing writeAck", SessionId);
         }
         finally
         {
@@ -207,10 +255,10 @@ internal async Task Write(ConcurrentQueue<MessageSending> toSendBuffer,
         }
         catch (TransactionException e)
         {
-            ReconnectSession();
+            Logger.LogError(e, "WriterSession[{SessionId}] have error on Write, last SeqNo={SeqNo}",
+                SessionId, Volatile.Read(ref _seqNum));
 
-            Console.WriteLine(e);
-            throw;
+            ReconnectSession();
         }
     }
 
@@ -220,4 +268,4 @@ public ValueTask DisposeAsync()
     }
 }
 
-internal record MessageSending(MessageData MessageData, TaskCompletionSource<SendResult> TaskCompletionSource);
+internal record MessageSending(MessageData MessageData, TaskCompletionSource<WriteResult> TaskCompletionSource);

src/Ydb.Sdk/src/Services/Topic/Producer.cs renamed to src/Ydb.Sdk/src/Services/Topic/Writer.cs

@@ -1,29 +1,29 @@
 namespace Ydb.Sdk.Services.Topic;
 
-public class ProducerBuilder<TValue>
+public class WriterBuilder<TValue>
 {
-    private readonly ProducerConfig _config;
+    private readonly WriterConfig _config;
 
-    public ProducerBuilder(ProducerConfig config)
+    public WriterBuilder(WriterConfig config)
     {
         _config = config;
     }
 
     public ISerializer<TValue>? Serializer { get; set; }
 
-    public async Task<IProducer<TValue>> Build()
+    public async Task<IWriter<TValue>> Build()
     {
-        var producer = new Producer<TValue>(
+        var writer = new Writer<TValue>(
             _config,
             Serializer ?? (ISerializer<TValue>)(
                 Serializers.DefaultSerializers.TryGetValue(typeof(TValue), out var serializer)
                     ? serializer
-                    : throw new YdbProducerException("The serializer is not set")
+                    : throw new YdbWriterException("The serializer is not set")
             )
         );
 
-        await producer.Initialize();
+        await writer.Initialize();
 
-        return producer;
+        return writer;
     }
 }

src/Ydb.Sdk/src/Services/Topic/ProducerBuilder.cs renamed to src/Ydb.Sdk/src/Services/Topic/WriterBuilder.cs

@@ -2,9 +2,9 @@
 
 namespace Ydb.Sdk.Services.Topic;
 
-public class ProducerConfig
+public class WriterConfig
 {
-    public ProducerConfig(Driver driver, string topicPath)
+    public WriterConfig(Driver driver, string topicPath)
     {
         Driver = driver;
         TopicPath = topicPath;