dev: reader first unit test on start fail

Author: KirillKurdyukov

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

@@ -204,6 +204,8 @@ public void Dispose()
         try
         {
             _disposeCts.Cancel();
+
+            _receivedMessagesChannel.Writer.Complete();
         }
         finally
         {
@@ -246,7 +248,6 @@ internal class ReaderSession<TValue> : TopicSession<MessageFromClient, MessageFr
         Channel.CreateUnbounded<MessageFromClient>(
             new UnboundedChannelOptions
             {
-                SingleWriter = true,
                 SingleReader = true,
                 AllowSynchronousContinuations = false
             }

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

@@ -1,6 +1,8 @@
 using Google.Protobuf;
+using Google.Protobuf.WellKnownTypes;
 using Grpc.Core;
 using Moq;
+using Xunit;
 using Ydb.Sdk.Services.Topic.Reader;
 using Ydb.Topic;
 
@@ -25,23 +27,42 @@ public ReaderUnitTests()
         _mockIDriver.Setup(driver => driver.LoggerFactory).Returns(Utils.GetLoggerFactory);
     }
 
-    // [Fact]
+    [Fact]
     public async Task Initialize_WhenFailWriteMessage_ShouldRetryInitializeAndReadThenCommitMessage()
     {
-        var tcs = new TaskCompletionSource<bool>();
+        var tcsMoveNext = new TaskCompletionSource<bool>();
+        var tcsCommitMessage = new TaskCompletionSource<bool>();
 
         _mockStream.SetupSequence(stream => stream.Write(It.IsAny<FromClient>()))
+            // Write Throws Exception
             .ThrowsAsync(new Driver.TransportException(new RpcException(Grpc.Core.Status.DefaultCancelled)))
+            // Write init
             .Returns(Task.CompletedTask)
+            // Write ReadRequest { 200 bytes }
             .Returns(Task.CompletedTask)
-            .Returns(Task.CompletedTask);
+            // Write StartSessionPartitionRequest
+            .Returns(() =>
+            {
+                tcsMoveNext.SetResult(true);
+
+                return Task.CompletedTask;
+            })
+            // Write ReadRequest { 50 bytes }
+            .Returns(Task.CompletedTask)
+            // Write CommitRequest
+            .Returns(() =>
+            {
+                tcsCommitMessage.SetResult(true);
+
+                return Task.CompletedTask;
+            });
 
         _mockStream.SetupSequence(stream => stream.MoveNextAsync())
-            .Returns(new ValueTask<bool>(true))
-            .Returns(new ValueTask<bool>(true))
-            .Returns(new ValueTask<bool>(true))
-            .Returns(new ValueTask<bool>(true))
-            .Returns(new ValueTask<bool>(tcs.Task));
+            .ReturnsAsync(true)
+            .ReturnsAsync(true)
+            .Returns(new ValueTask<bool>(tcsMoveNext.Task))
+            .Returns(new ValueTask<bool>(tcsCommitMessage.Task))
+            .Returns(new ValueTask<bool>(new TaskCompletionSource<bool>().Task));
 
         _mockStream.SetupSequence(stream => stream.Current)
             .Returns(new FromServer
@@ -64,20 +85,27 @@ public async Task Initialize_WhenFailWriteMessage_ShouldRetryInitializeAndReadTh
             .Returns(
                 new FromServer
                 {
+                    Status = StatusIds.Types.StatusCode.Success,
                     ReadResponse = new StreamReadMessage.Types.ReadResponse
                     {
                         BytesSize = 50, PartitionData =
                         {
                             new StreamReadMessage.Types.ReadResponse.Types.PartitionData
                             {
+                                PartitionSessionId = 1,
                                 Batches =
                                 {
                                     new StreamReadMessage.Types.ReadResponse.Types.Batch
                                     {
+                                        ProducerId = "ProducerId",
                                         MessageData =
                                         {
                                             new StreamReadMessage.Types.ReadResponse.Types.MessageData
-                                                { Data = ByteString.CopyFrom(BitConverter.GetBytes(100)) }
+                                            {
+                                                Data = ByteString.CopyFrom(BitConverter.GetBytes(100)),
+                                                Offset = 1,
+                                                CreatedAt = new Timestamp()
+                                            }
                                         }
                                     }
                                 }
@@ -89,6 +117,7 @@ public async Task Initialize_WhenFailWriteMessage_ShouldRetryInitializeAndReadTh
             .Returns(
                 new FromServer
                 {
+                    Status = StatusIds.Types.StatusCode.Success,
                     CommitOffsetResponse =
                         new StreamReadMessage.Types.CommitOffsetResponse
                         {
@@ -97,7 +126,7 @@ public async Task Initialize_WhenFailWriteMessage_ShouldRetryInitializeAndReadTh
                                 new StreamReadMessage.Types.CommitOffsetResponse.Types.PartitionCommittedOffset
                                 {
                                     PartitionSessionId = 1,
-                                    CommittedOffset = 50
+                                    CommittedOffset = 2
                                 }
                             }
                         }
@@ -112,8 +141,8 @@ public async Task Initialize_WhenFailWriteMessage_ShouldRetryInitializeAndReadTh
             SubscribeSettings = { new SubscribeSettings("/topic") }
         }.Build();
 
-        await reader.ReadAsync();
-        // await message.CommitAsync();
-        // Assert.Equal(100, message.Data);
+        var message = await reader.ReadAsync();
+        await message.CommitAsync();
+        Assert.Equal(100, message.Data);
     }
 }

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

@@ -6,6 +6,7 @@
 using Ydb.Sdk.Services.Topic.Writer;
 using Ydb.Topic;
 using Codec = Ydb.Sdk.Services.Topic.Codec;
+using Range = Moq.Range;
 
 namespace Ydb.Sdk.Tests.Topic;
 
@@ -132,7 +133,7 @@ public async Task Initialize_WhenFailWriteMessage_ShouldRetryInitializeAndReturn
         // check attempt repeated!!!
         _mockStream.Verify(stream => stream.Write(It.IsAny<FromClient>()), Times.Exactly(3));
         _mockStream.Verify(stream => stream.MoveNextAsync(),
-            Times.AtLeast(2)); // run processing ack may not be able to start on time 
+            Times.Between(2, 3, Range.Inclusive)); // run processing ack may not be able to start on time 
         _mockStream.Verify(stream => stream.Current, Times.Exactly(2));
     }
 
@@ -180,7 +181,7 @@ public async Task Initialize_WhenFailMoveNextAsync_ShouldRetryInitializeAndRetur
         // check attempt repeated!!!
         _mockStream.Verify(stream => stream.Write(It.IsAny<FromClient>()), Times.Exactly(3));
         _mockStream.Verify(stream => stream.MoveNextAsync(),
-            Times.AtLeast(3)); // run processing ack may not be able to start on time 
+            Times.Between(3, 4, Range.Inclusive)); // run processing ack may not be able to start on time 
         _mockStream.Verify(stream => stream.Current, Times.Exactly(2));
     }
 
@@ -257,7 +258,7 @@ public async Task Initialize_WhenInitResponseStatusIsRetryable_ShouldRetryInitia
         // check attempt repeated!!!
         _mockStream.Verify(stream => stream.Write(It.IsAny<FromClient>()), Times.Exactly(3));
         _mockStream.Verify(stream => stream.MoveNextAsync(),
-            Times.AtLeast(3)); // run processing ack may not be able to start on time 
+            Times.Between(3, 4, Range.Inclusive)); // run processing ack may not be able to start on time 
         _mockStream.Verify(stream => stream.Current, Times.Exactly(3));
     }
 
@@ -391,7 +392,7 @@ public async Task WriteAsync_WhenTransportExceptionOnWriteInWriterSession_Should
 
         Assert.Equal(PersistenceStatus.Written, (await writer.WriteAsync(100L)).Status);
         _mockStream.Verify(stream => stream.Write(It.IsAny<FromClient>()), Times.Exactly(4));
-        _mockStream.Verify(stream => stream.MoveNextAsync(), Times.AtLeast(4));
+        _mockStream.Verify(stream => stream.MoveNextAsync(), Times.Between(4, 5, Range.Inclusive));
         _mockStream.Verify(stream => stream.Current, Times.Exactly(3));
     }
 
@@ -466,7 +467,7 @@ public async Task WriteAsync_WhenTransportExceptionOnProcessingWriteAck_ShouldRe
         Assert.Equal(PersistenceStatus.Written, (await writer.WriteAsync(100L)).Status);
 
         _mockStream.Verify(stream => stream.Write(It.IsAny<FromClient>()), Times.Exactly(3));
-        _mockStream.Verify(stream => stream.MoveNextAsync(), Times.AtLeast(4));
+        _mockStream.Verify(stream => stream.MoveNextAsync(), Times.Between(4, 5, Range.Inclusive));
         _mockStream.Verify(stream => stream.Current, Times.Exactly(3));
     }
 
@@ -543,7 +544,7 @@ public async Task WriteAsync_WhenStreamIsClosingOnProcessingWriteAck_ShouldRecon
         Assert.Equal(PersistenceStatus.Written, (await writer.WriteAsync(100L)).Status);
 
         _mockStream.Verify(stream => stream.Write(It.IsAny<FromClient>()), Times.Exactly(3));
-        _mockStream.Verify(stream => stream.MoveNextAsync(), Times.AtLeast(4));
+        _mockStream.Verify(stream => stream.MoveNextAsync(), Times.Between(4, 5, Range.Inclusive));
         _mockStream.Verify(stream => stream.Current, Times.Exactly(3));
     }