dev: ydb topic example & chore changes writer (#282)

KirillKurdyukov · web-flow · commit c697584abda9 · 2025-03-06T15:33:20.000+03:00
diff --git a/.github/workflows/tests.yml b/.github/workflows/tests.yml
@@ -156,7 +156,7 @@ jobs:
     strategy:
       fail-fast: false
       matrix:
-        ydb-version: [ 'latest', '24.1' ]
+        ydb-version: [ 'latest', 'trunk' ]
         dotnet-version: [ 6.0.x, 7.0.x ]
         include:
           - dotnet-version: 6.0.x
@@ -181,7 +181,7 @@ jobs:
         uses: actions/setup-dotnet@v4
         with:
           dotnet-version: ${{ matrix.dotnet-version }}
-      - name: Run ADO.NET examples
+      - name: Run ADO.NET example
         run: |
           docker cp ydb-local:/ydb_certs/ca.pem ~/
           cd ./examples/src/AdoNet
@@ -190,3 +190,7 @@ jobs:
         run: |
           cd ./examples/src/DapperExample
           dotnet run
+      - name: YDB Topic example
+        run: |
+          cd ./examples/src/Topic
+          dotnet run
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -1,3 +1,7 @@
+- Fixed Writer: possible creation of a session after `DisposeAsync()`, which this could happen when there are canceled tasks in `InFlightMessages`.
+- Dev: `Writer.MoveNext()` changed exception on cancelToken from `WriterException` to `TaskCanceledException`.
+- Dev: changed log level from `Warning` to `Information` in `(Reader / Writer).Initialize()` when it is disposed.
+
 ## v0.15.0
 - Dev: added `ValueTask<string?> GetAuthInfoAsync()` in ICredentialProvider.
 - Feat: `Writer.DisposeAsync()` waits for all in-flight messages to complete.
diff --git a/examples/src/Topic/Program.cs b/examples/src/Topic/Program.cs
@@ -0,0 +1,84 @@
+﻿using Microsoft.Extensions.Logging;
+using Ydb.Sdk;
+using Ydb.Sdk.Services.Topic;
+using Ydb.Sdk.Services.Topic.Reader;
+using Ydb.Sdk.Services.Topic.Writer;
+
+const int countMessages = 100;
+const string topicName = "topic_name";
+
+var loggerFactory = LoggerFactory.Create(builder => builder.AddConsole().SetMinimumLevel(LogLevel.Information));
+
+var logger = loggerFactory.CreateLogger<Program>();
+
+var config = new DriverConfig(
+    endpoint: "grpc://localhost:2136",
+    database: "/local"
+);
+
+await using var driver = await Driver.CreateInitialized(
+    config: config,
+    loggerFactory: loggerFactory
+);
+
+var topicClient = new TopicClient(driver);
+
+await topicClient.CreateTopic(new CreateTopicSettings
+{
+    Path = topicName,
+    Consumers = { new Consumer("Consumer_Example") }
+});
+
+var readerCts = new CancellationTokenSource();
+
+var writerJob = Task.Run(async () =>
+{
+    // ReSharper disable once AccessToDisposedClosure
+    await using var writer = new WriterBuilder<string>(driver, topicName)
+    {
+        ProducerId = "ProducerId_Example"
+    }.Build();
+
+    for (var i = 0; i < countMessages; i++)
+    {
+        await writer.WriteAsync($"Message num {i}: Hello Example YDB Topics!");
+    }
+
+    readerCts.CancelAfter(TimeSpan.FromSeconds(3));
+});
+
+var readerJob = Task.Run(async () =>
+{
+    // ReSharper disable once AccessToDisposedClosure
+    await using var reader = new ReaderBuilder<string>(driver)
+    {
+        ConsumerName = "Consumer_Example",
+        SubscribeSettings = { new SubscribeSettings(topicName) }
+    }.Build();
+
+    try
+    {
+        while (!readerCts.IsCancellationRequested)
+        {
+            var message = await reader.ReadAsync(readerCts.Token);
+
+            logger.LogInformation("Received message: [{MessageData}]", message.Data);
+
+            try
+            {
+                await message.CommitAsync();
+            }
+            catch (Exception e)
+            {
+                logger.LogError(e, "Failed commit message");
+            }
+        }
+    }
+    catch (OperationCanceledException)
+    {
+    }
+});
+
+await writerJob;
+await readerJob;
+await topicClient.DropTopic(new DropTopicSettings { Path = topicName });
diff --git a/examples/src/Topic/Topic.csproj b/examples/src/Topic/Topic.csproj
@@ -0,0 +1,17 @@
+﻿<Project Sdk="Microsoft.NET.Sdk">
+
+    <PropertyGroup>
+        <OutputType>Exe</OutputType>
+        <TargetFramework>net8.0</TargetFramework>
+        <ImplicitUsings>enable</ImplicitUsings>
+        <Nullable>enable</Nullable>
+    </PropertyGroup>
+
+    <ItemGroup>
+        <PackageReference Include="Microsoft.Extensions.Logging.Console" Version="5.0.0" />
+        <PackageReference Include="Microsoft.Extensions.Logging" Version="5.0.0" />
+    </ItemGroup>
+    <ItemGroup>
+        <ProjectReference Include="..\..\..\src\Ydb.Sdk\src\Ydb.Sdk.csproj"/>
+    </ItemGroup>
+</Project>
diff --git a/examples/src/YdbExamples.sln b/examples/src/YdbExamples.sln
@@ -15,6 +15,8 @@ Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "DapperExample", "DapperExam
 EndProject
 Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "YC", "YC\YC.csproj", "{753E4F33-CB08-47B9-864F-4CC037B278C4}"
 EndProject
+Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Topic", "Topic\Topic.csproj", "{0FB9A1C2-4B0C-4AE4-9FA2-E0ED37802A6E}"
+EndProject
 Global
 	GlobalSection(SolutionConfigurationPlatforms) = preSolution
 		Debug|Any CPU = Debug|Any CPU
@@ -45,6 +47,10 @@ Global
 		{753E4F33-CB08-47B9-864F-4CC037B278C4}.Debug|Any CPU.Build.0 = Debug|Any CPU
 		{753E4F33-CB08-47B9-864F-4CC037B278C4}.Release|Any CPU.ActiveCfg = Release|Any CPU
 		{753E4F33-CB08-47B9-864F-4CC037B278C4}.Release|Any CPU.Build.0 = Release|Any CPU
+		{0FB9A1C2-4B0C-4AE4-9FA2-E0ED37802A6E}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+		{0FB9A1C2-4B0C-4AE4-9FA2-E0ED37802A6E}.Debug|Any CPU.Build.0 = Debug|Any CPU
+		{0FB9A1C2-4B0C-4AE4-9FA2-E0ED37802A6E}.Release|Any CPU.ActiveCfg = Release|Any CPU
+		{0FB9A1C2-4B0C-4AE4-9FA2-E0ED37802A6E}.Release|Any CPU.Build.0 = Release|Any CPU
 	EndGlobalSection
 	GlobalSection(SolutionProperties) = preSolution
 		HideSolutionNode = FALSE
diff --git a/src/Ydb.Sdk/src/Services/Topic/Reader/Reader.cs b/src/Ydb.Sdk/src/Services/Topic/Reader/Reader.cs
@@ -96,7 +96,7 @@ private async Task Initialize()
         {
             if (_disposeCts.IsCancellationRequested)
             {
-                _logger.LogWarning("Reader writer is canceled because it has been disposed");
+                _logger.LogDebug("Initialize Reader[{ReaderConfig}] is stopped because it has been disposed", _config);
 
                 return;
             }
@@ -335,6 +335,8 @@ private async Task RunProcessingStreamResponse()
                         throw new ArgumentOutOfRangeException();
                 }
             }
+
+            Logger.LogInformation("ReaderSession[{SessionId}]: ResponseStream is closed", SessionId);
         }
         catch (Driver.TransportException e)
         {
@@ -579,6 +581,8 @@ public override async ValueTask DisposeAsync()
         {
             await _runProcessingStreamRequest;
             await Stream.RequestStreamComplete();
+            Logger.LogInformation("ReaderSession[{SessionId}]: RequestStream is closed", SessionId);
+
             await _runProcessingStreamResponse; // waiting all ack's commits
 
             _lifecycleReaderSessionCts.Cancel();
diff --git a/src/Ydb.Sdk/src/Services/Topic/TopicSession.cs b/src/Ydb.Sdk/src/Services/Topic/TopicSession.cs
@@ -38,7 +38,7 @@ protected async void ReconnectSession()
             return;
         }
 
-        Logger.LogInformation("TopicSession[{SessionId}] has been deactivated, starting to reconnect", SessionId);
+        Logger.LogDebug("TopicSession[{SessionId}] has been deactivated, starting to reconnect", SessionId);
 
         await _initialize();
     }
diff --git a/src/Ydb.Sdk/src/Services/Topic/Writer/Writer.cs b/src/Ydb.Sdk/src/Services/Topic/Writer/Writer.cs
@@ -31,6 +31,7 @@ internal class Writer<TValue> : IWriter<TValue>
     private volatile TaskCompletionSource _tcsBufferAvailableEvent = new();
     private volatile IWriteSession _session = null!;
     private volatile int _limitBufferMaxSize;
+    private volatile bool _isStopped;
 
     internal Writer(IDriver driver, WriterConfig config, ISerializer<TValue> serializer)
     {
@@ -52,9 +53,7 @@ public async Task<WriteResult> WriteAsync(Message<TValue> message, CancellationT
     {
         TaskCompletionSource<WriteResult> tcs = new();
         await using var registrationUserCancellationTokenRegistration = cancellationToken.Register(
-            () => tcs.TrySetException(
-                new WriterException("The write operation was canceled before it could be completed")
-            ), useSynchronizationContext: false
+            () => tcs.TrySetCanceled(), useSynchronizationContext: false
         );
         await using var writerDisposedCancellationTokenRegistration = _disposeCts.Token.Register(
             () => tcs.TrySetException(new WriterException($"Writer[{_config}] is disposed")),
@@ -194,9 +193,9 @@ private async Task Initialize()
 
         try
         {
-            if (_disposeCts.IsCancellationRequested && _inFlightMessages.IsEmpty)
+            if (_isStopped)
             {
-                _logger.LogWarning("Initialize writer is canceled because it has been disposed");
+                _logger.LogDebug("Initialize Writer[{WriterConfig}] is stopped because it has been disposed", _config);
 
                 return;
             }
@@ -313,7 +312,6 @@ private async Task Initialize()
                 }
 
                 _session = newSession;
-                newSession.RunProcessingWriteAck();
                 WakeUpWorker(); // attempt send buffer     
             }
             finally
@@ -367,6 +365,8 @@ public async ValueTask DisposeAsync()
             }
         }
 
+        _isStopped = true;
+
         await _session.DisposeAsync();
 
         _logger.LogInformation("Writer[{WriterConfig}] is disposed", _config);
@@ -443,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;
 
@@ -466,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)
@@ -513,7 +516,7 @@ public async Task Write(ConcurrentQueue<MessageSending> toSendBuffer)
         }
     }
 
-    internal async void RunProcessingWriteAck()
+    private async Task RunProcessingWriteAck()
     {
         try
         {
@@ -573,7 +576,7 @@ Completing task on exception...
                 }
             }
 
-            Logger.LogWarning("WriterSession[{SessionId}]: stream is closed", SessionId);
+            Logger.LogInformation("WriterSession[{SessionId}]: ResponseStream is closed", SessionId);
         }
         catch (Driver.TransportException e)
         {
@@ -601,6 +604,7 @@ public override async ValueTask DisposeAsync()
         Logger.LogDebug("WriterSession[{SessionId}]: start dispose process", SessionId);
 
         await Stream.RequestStreamComplete();
+        await _processingResponseStream;
 
         Stream.Dispose();
     }
diff --git a/src/Ydb.Sdk/tests/Topic/WriterUnitTests.cs b/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
@@ -569,7 +575,6 @@ public async Task WriteAsync_WhenStreamIsClosingOnProcessingWriteAck_ShouldRecon
     public async Task WriteAsync_WhenCancellationTokenIsClosed_ThrowCancellationException()
     {
         var cancellationTokenSource = new CancellationTokenSource();
-        var nextCompleted = new TaskCompletionSource<bool>();
         _mockStream.Setup(stream => stream.Write(It.IsAny<FromClient>()))
             .Returns(Task.CompletedTask);
         _mockStream.SetupSequence(stream => stream.MoveNextAsync())
@@ -582,10 +587,8 @@ public async Task WriteAsync_WhenCancellationTokenIsClosed_ThrowCancellationExce
 
         var task = writer.WriteAsync(123L, cancellationTokenSource.Token);
         cancellationTokenSource.Cancel();
-        nextCompleted.SetResult(true);
 
-        Assert.Equal("The write operation was canceled before it could be completed",
-            (await Assert.ThrowsAsync<WriterException>(() => task)).Message);
+        await Assert.ThrowsAsync<TaskCanceledException>(() => task);
     }
 
     [Fact]
@@ -638,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)
@@ -657,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
@@ -713,8 +721,7 @@ public async Task WriteAsync_WhenInFlightBufferSendInInitialize_ReturnCompletedT
 
         moveTcs.SetResult(false); // Fail write ack stream => start reconnect
 
-        Assert.Equal("The write operation was canceled before it could be completed",
-            (await Assert.ThrowsAsync<WriterException>(() => runTaskWithCancel)).Message);
+        await Assert.ThrowsAsync<TaskCanceledException>(() => runTaskWithCancel);
         Assert.Equal(PersistenceStatus.AlreadyWritten, (await runTask1).Status);
         Assert.Equal(PersistenceStatus.Written, (await runTask2).Status);
 
@@ -863,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)
@@ -872,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)

Original file line number	Diff line number	Diff line change
`@@ -96,7 +96,7 @@ private async Task Initialize()`
`96`	`96`	`{`
`97`	`97`	`if (_disposeCts.IsCancellationRequested)`
`98`	`98`	`{`
`99`		`- _logger.LogWarning("Reader writer is canceled because it has been disposed");`
	`99`	`+ _logger.LogDebug("Initialize Reader[{ReaderConfig}] is stopped because it has been disposed", _config);`
`100`	`100`
`101`	`101`	`return;`
`102`	`102`	`}`
`@@ -335,6 +335,8 @@ private async Task RunProcessingStreamResponse()`
`335`	`335`	`throw new ArgumentOutOfRangeException();`
`336`	`336`	`}`
`337`	`337`	`}`
	`338`	`+`
	`339`	`+ Logger.LogInformation("ReaderSession[{SessionId}]: ResponseStream is closed", SessionId);`
`338`	`340`	`}`
`339`	`341`	`catch (Driver.TransportException e)`
`340`	`342`	`{`
`@@ -579,6 +581,8 @@ public override async ValueTask DisposeAsync()`
`579`	`581`	`{`
`580`	`582`	`await _runProcessingStreamRequest;`
`581`	`583`	`await Stream.RequestStreamComplete();`
	`584`	`+ Logger.LogInformation("ReaderSession[{SessionId}]: RequestStream is closed", SessionId);`
	`585`	`+`
`582`	`586`	`await _runProcessingStreamResponse; // waiting all ack's commits`
`583`	`587`
`584`	`588`	`_lifecycleReaderSessionCts.Cancel();`
Original file line number	Diff line number	Diff line change
`@@ -38,7 +38,7 @@ protected async void ReconnectSession()`
`38`	`38`	`return;`
`39`	`39`	`}`
`40`	`40`
`41`		`- Logger.LogInformation("TopicSession[{SessionId}] has been deactivated, starting to reconnect", SessionId);`
	`41`	`+ Logger.LogDebug("TopicSession[{SessionId}] has been deactivated, starting to reconnect", SessionId);`
`42`	`42`
`43`	`43`	`await _initialize();`
`44`	`44`	`}`
Original file line number	Diff line number	Diff line change
`@@ -31,6 +31,7 @@ internal class Writer<TValue> : IWriter<TValue>`
`31`	`31`	`private volatile TaskCompletionSource _tcsBufferAvailableEvent = new();`
`32`	`32`	`private volatile IWriteSession _session = null!;`
`33`	`33`	`private volatile int _limitBufferMaxSize;`
	`34`	`+ private volatile bool _isStopped;`
`34`	`35`
`35`	`36`	`internal Writer(IDriver driver, WriterConfig config, ISerializer<TValue> serializer)`
`36`	`37`	`{`
`@@ -52,9 +53,7 @@ public async Task<WriteResult> WriteAsync(Message<TValue> message, CancellationT`
`52`	`53`	`{`
`53`	`54`	`TaskCompletionSource<WriteResult> tcs = new();`
`54`	`55`	`await using var registrationUserCancellationTokenRegistration = cancellationToken.Register(`
`55`		`- () => tcs.TrySetException(`
`56`		`- new WriterException("The write operation was canceled before it could be completed")`
`57`		`- ), useSynchronizationContext: false`
	`56`	`+ () => tcs.TrySetCanceled(), useSynchronizationContext: false`
`58`	`57`	`);`
`59`	`58`	`await using var writerDisposedCancellationTokenRegistration = _disposeCts.Token.Register(`
`60`	`59`	`() => tcs.TrySetException(new WriterException($"Writer[{_config}] is disposed")),`
`@@ -194,9 +193,9 @@ private async Task Initialize()`
`194`	`193`
`195`	`194`	`try`
`196`	`195`	`{`
`197`		`- if (_disposeCts.IsCancellationRequested && _inFlightMessages.IsEmpty)`
	`196`	`+ if (_isStopped)`
`198`	`197`	`{`
`199`		`- _logger.LogWarning("Initialize writer is canceled because it has been disposed");`
	`198`	`+ _logger.LogDebug("Initialize Writer[{WriterConfig}] is stopped because it has been disposed", _config);`
`200`	`199`
`201`	`200`	`return;`
`202`	`201`	`}`
`@@ -313,7 +312,6 @@ private async Task Initialize()`
`313`	`312`	`}`
`314`	`313`
`315`	`314`	`_session = newSession;`
`316`		`- newSession.RunProcessingWriteAck();`
`317`	`315`	`WakeUpWorker(); // attempt send buffer`
`318`	`316`	`}`
`319`	`317`	`finally`
`@@ -367,6 +365,8 @@ public async ValueTask DisposeAsync()`
`367`	`365`	`}`
`368`	`366`	`}`
`369`	`367`
	`368`	`+ _isStopped = true;`
	`369`	`+`
`370`	`370`	`await _session.DisposeAsync();`
`371`	`371`
`372`	`372`	`_logger.LogInformation("Writer[{WriterConfig}] is disposed", _config);`
`@@ -443,6 +443,7 @@ internal class WriterSession : TopicSession<MessageFromClient, MessageFromServer`
`443`	`443`	`{`
`444`	`444`	`private readonly WriterConfig _config;`
`445`	`445`	`private readonly ConcurrentQueue<MessageSending> _inFlightMessages;`
	`446`	`+ private readonly Task _processingResponseStream;`
`446`	`447`
`447`	`448`	`private long _seqNum;`
`448`	`449`
`@@ -466,6 +467,8 @@ ConcurrentQueue<MessageSending> inFlightMessages`
`466`	`467`	`_config = config;`
`467`	`468`	`_inFlightMessages = inFlightMessages;`
`468`	`469`	`Volatile.Write(ref _seqNum, lastSeqNo); // happens-before for Volatile.Read`
	`470`	`+`
	`471`	`+ _processingResponseStream = RunProcessingWriteAck();`
`469`	`472`	`}`
`470`	`473`
`471`	`474`	`public async Task Write(ConcurrentQueue<MessageSending> toSendBuffer)`
`@@ -513,7 +516,7 @@ public async Task Write(ConcurrentQueue<MessageSending> toSendBuffer)`
`513`	`516`	`}`
`514`	`517`	`}`
`515`	`518`
`516`		`- internal async void RunProcessingWriteAck()`
	`519`	`+ private async Task RunProcessingWriteAck()`
`517`	`520`	`{`
`518`	`521`	`try`
`519`	`522`	`{`
`@@ -573,7 +576,7 @@ Completing task on exception...`
`573`	`576`	`}`
`574`	`577`	`}`
`575`	`578`
`576`		`- Logger.LogWarning("WriterSession[{SessionId}]: stream is closed", SessionId);`
	`579`	`+ Logger.LogInformation("WriterSession[{SessionId}]: ResponseStream is closed", SessionId);`
`577`	`580`	`}`
`578`	`581`	`catch (Driver.TransportException e)`
`579`	`582`	`{`
`@@ -601,6 +604,7 @@ public override async ValueTask DisposeAsync()`
`601`	`604`	`Logger.LogDebug("WriterSession[{SessionId}]: start dispose process", SessionId);`
`602`	`605`
`603`	`606`	`await Stream.RequestStreamComplete();`
	`607`	`+ await _processingResponseStream;`
`604`	`608`
`605`	`609`	`Stream.Dispose();`
`606`	`610`	`}`