dev: ydb topic example & stop init task with canceled tasks in inFlight buffer

Author: KirillKurdyukov

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

@@ -312,7 +312,6 @@ private async Task Initialize()
                 }
 
                 _session = newSession;
-                newSession.RunProcessingWriteAck();
                 WakeUpWorker(); // attempt send buffer     
             }
             finally
@@ -444,6 +443,7 @@ internal class WriterSession : TopicSession<MessageFromClient, MessageFromServer
 {
     private readonly WriterConfig _config;
     private readonly ConcurrentQueue<MessageSending> _inFlightMessages;
+    private readonly Task _processingResponseStream;
 
     private long _seqNum;
 
@@ -467,6 +467,8 @@ ConcurrentQueue<MessageSending> inFlightMessages
         _config = config;
         _inFlightMessages = inFlightMessages;
         Volatile.Write(ref _seqNum, lastSeqNo); // happens-before for Volatile.Read
+
+        _processingResponseStream = RunProcessingWriteAck();
     }
 
     public async Task Write(ConcurrentQueue<MessageSending> toSendBuffer)
@@ -514,7 +516,7 @@ public async Task Write(ConcurrentQueue<MessageSending> toSendBuffer)
         }
     }
 
-    internal async void RunProcessingWriteAck()
+    private async Task RunProcessingWriteAck()
     {
         try
         {
@@ -574,7 +576,7 @@ Completing task on exception...
                 }
             }
 
-            Logger.LogWarning("WriterSession[{SessionId}]: stream is closed", SessionId);
+            Logger.LogInformation("WriterSession[{SessionId}]: stream is closed", SessionId);
         }
         catch (Driver.TransportException e)
         {
@@ -606,6 +608,7 @@ public override async ValueTask DisposeAsync()
         Logger.LogDebug("WriterSession[{SessionId}]: start dispose process", SessionId);
 
         await Stream.RequestStreamComplete();
+        await _processingResponseStream;
 
         Stream.Dispose();
     }

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

@@ -358,6 +358,7 @@ public async Task Initialize_WhenNotSupportedCodec_ThrowWriterExceptionOnWriteAs
     public async Task WriteAsync_WhenTransportExceptionOnWriteInWriterSession_ShouldReconnectAndReturnWriteResult()
     {
         var moveTcs = new TaskCompletionSource<bool>();
+        var moveTcsRetry = new TaskCompletionSource<bool>();
 
         _mockStream.SetupSequence(stream => stream.Write(It.IsAny<FromClient>()))
             .Returns(Task.CompletedTask)
@@ -367,12 +368,17 @@ public async Task WriteAsync_WhenTransportExceptionOnWriteInWriterSession_Should
                 return new Driver.TransportException(new RpcException(Grpc.Core.Status.DefaultCancelled));
             })
             .Returns(Task.CompletedTask)
-            .Returns(Task.CompletedTask);
+            .Returns(() =>
+            {
+                moveTcsRetry.SetResult(true);
+
+                return Task.CompletedTask;
+            });
         _mockStream.SetupSequence(stream => stream.MoveNextAsync())
             .ReturnsAsync(true)
             .Returns(new ValueTask<bool>(moveTcs.Task))
             .ReturnsAsync(true)
-            .ReturnsAsync(true)
+            .Returns(new ValueTask<bool>(moveTcsRetry.Task))
             .Returns(_lastMoveNext);
         _mockStream.SetupSequence(stream => stream.Current)
             .Returns(new StreamWriteMessage.Types.FromServer
@@ -635,6 +641,7 @@ public async Task WriteAsync_WhenInFlightBufferSendInInitialize_ReturnCompletedT
         var writeTcs2 = new TaskCompletionSource();
         var writeTcs3 = new TaskCompletionSource();
         var moveTcs = new TaskCompletionSource<bool>();
+        var moveTcsRetry = new TaskCompletionSource<bool>();
 
         _mockStream.SetupSequence(stream => stream.Write(It.IsAny<FromClient>()))
             .Returns(Task.CompletedTask)
@@ -654,13 +661,17 @@ public async Task WriteAsync_WhenInFlightBufferSendInInitialize_ReturnCompletedT
                 return Task.CompletedTask;
             })
             .Returns(Task.CompletedTask)
-            .Returns(Task.CompletedTask);
+            .Returns(() =>
+            {
+                moveTcsRetry.SetResult(true);
+                return Task.CompletedTask;
+            });
 
         _mockStream.SetupSequence(stream => stream.MoveNextAsync())
             .ReturnsAsync(true)
             .Returns(new ValueTask<bool>(moveTcs.Task))
             .ReturnsAsync(true)
-            .ReturnsAsync(true)
+            .Returns(new ValueTask<bool>(moveTcsRetry.Task))
             .Returns(_lastMoveNext);
         _mockStream.SetupSequence(stream => stream.Current)
             .Returns(new StreamWriteMessage.Types.FromServer
@@ -859,6 +870,7 @@ public async Task DisposeAsync_WhenInFlightMessages_WaitingInFlightMessages()
     {
         var tcsDetectedWrite = new TaskCompletionSource();
         var writeTcs1 = new TaskCompletionSource<bool>();
+        var moveTcsRetry = new TaskCompletionSource<bool>();
 
         _mockStream.SetupSequence(stream => stream.Write(It.IsAny<FromClient>()))
             .Returns(Task.CompletedTask)
@@ -868,12 +880,16 @@ public async Task DisposeAsync_WhenInFlightMessages_WaitingInFlightMessages()
                 return Task.CompletedTask;
             })
             .Returns(Task.CompletedTask)
-            .Returns(Task.CompletedTask);
+            .Returns(() =>
+            {
+                moveTcsRetry.SetResult(true);
+                return Task.CompletedTask;
+            });
         _mockStream.SetupSequence(stream => stream.MoveNextAsync())
             .ReturnsAsync(true)
             .Returns(new ValueTask<bool>(writeTcs1.Task))
             .ReturnsAsync(true)
-            .ReturnsAsync(true)
+            .Returns(new ValueTask<bool>(moveTcsRetry.Task))
             .Returns(_lastMoveNext);
 
         _mockStream.SetupSequence(stream => stream.Current)