feat: mock tests on init WriterSession

KirillKurdyukov · KirillKurdyukov · commit cb32d5e9c5ae · 2024-11-20T15:49:30.000+03:00
diff --git a/src/Ydb.Sdk/src/IDriver.cs b/src/Ydb.Sdk/src/IDriver.cs
@@ -20,7 +20,7 @@ public ServerStream<TResponse> ServerStreamCall<TRequest, TResponse>(
         where TRequest : class
         where TResponse : class;
 
-    public BidirectionalStream<TRequest, TResponse> BidirectionalStreamCall<TRequest, TResponse>(
+    public IBidirectionalStream<TRequest, TResponse> BidirectionalStreamCall<TRequest, TResponse>(
         Method<TRequest, TResponse> method,
         GrpcRequestSettings settings)
         where TRequest : class
@@ -29,6 +29,13 @@ public BidirectionalStream<TRequest, TResponse> BidirectionalStreamCall<TRequest
     ILoggerFactory LoggerFactory { get; }
 }
 
+public interface IBidirectionalStream<in TRequest, out TResponse> : IDisposable
+{
+    public Task Write(TRequest request);
+    public ValueTask<bool> MoveNextAsync();
+    public TResponse Current { get; }
+}
+
 public abstract class BaseDriver : IDriver
 {
     protected readonly DriverConfig Config;
@@ -95,7 +102,7 @@ public ServerStream<TResponse> ServerStreamCall<TRequest, TResponse>(
         return new ServerStream<TResponse>(call, e => { OnRpcError(endpoint, e); });
     }
 
-    public BidirectionalStream<TRequest, TResponse> BidirectionalStreamCall<TRequest, TResponse>(
+    public IBidirectionalStream<TRequest, TResponse> BidirectionalStreamCall<TRequest, TResponse>(
         Method<TRequest, TResponse> method,
         GrpcRequestSettings settings)
         where TRequest : class
@@ -213,16 +220,11 @@ public async ValueTask<bool> MoveNextAsync()
     }
 }
 
-public class BidirectionalStream<TRequest, TResponse> : IDisposable
+public class BidirectionalStream<TRequest, TResponse> : IBidirectionalStream<TRequest, TResponse>
 {
     private readonly AsyncDuplexStreamingCall<TRequest, TResponse> _stream;
     private readonly Action<RpcException> _rpcErrorAction;
 
-    public BidirectionalStream()
-    {
-        
-    }
-    
     internal BidirectionalStream(
         AsyncDuplexStreamingCall<TRequest, TResponse> stream,
         Action<RpcException> rpcErrorAction)
diff --git a/src/Ydb.Sdk/src/Services/Topic/TopicSession.cs b/src/Ydb.Sdk/src/Services/Topic/TopicSession.cs
@@ -6,13 +6,13 @@ internal abstract class TopicSession<TFromClient, TFromServer> : IDisposable
 {
     private readonly Func<Task> _initialize;
 
-    protected readonly BidirectionalStream<TFromClient, TFromServer> Stream;
+    protected readonly IBidirectionalStream<TFromClient, TFromServer> Stream;
     protected readonly ILogger Logger;
     protected readonly string SessionId;
 
     private int _isActive = 1;
 
-    protected TopicSession(BidirectionalStream<TFromClient, TFromServer> stream, ILogger logger,
+    protected TopicSession(IBidirectionalStream<TFromClient, TFromServer> stream, ILogger logger,
         string sessionId, Func<Task> initialize)
     {
         Stream = stream;
diff --git a/src/Ydb.Sdk/src/Services/Topic/Writer/Writer.cs b/src/Ydb.Sdk/src/Services/Topic/Writer/Writer.cs
@@ -11,7 +11,7 @@ namespace Ydb.Sdk.Services.Topic.Writer;
 using MessageData = StreamWriteMessage.Types.WriteRequest.Types.MessageData;
 using MessageFromClient = StreamWriteMessage.Types.FromClient;
 using MessageFromServer = StreamWriteMessage.Types.FromServer;
-using WriterStream = BidirectionalStream<
+using WriterStream = IBidirectionalStream<
     StreamWriteMessage.Types.FromClient,
     StreamWriteMessage.Types.FromServer
 >;
@@ -172,6 +172,8 @@ private async Task Initialize()
                     _ = Task.Run(Initialize, _disposeTokenSource.Token);
                 }
 
+                _logger.LogCritical("Writer initialization failed to start. Reason: {Status}", status);
+
                 return;
             }
 
@@ -183,11 +185,12 @@ private async Task Initialize()
             if (initResponse.SupportedCodecs != null &&
                 !initResponse.SupportedCodecs.Codecs.Contains((int)_config.Codec))
             {
-                _logger.LogCritical("Topic[{TopicPath}] is not supported codec: {Codec}", _config.TopicPath,
-                    _config.Codec);
+                _logger.LogCritical(
+                    "Writer initialization failed to start. Reason: topic[Path=\"{TopicPath}\"] is not supported codec {Codec}",
+                    _config.TopicPath, _config.Codec);
 
                 _session = new NotStartedWriterSession(
-                    $"Topic[{_config.TopicPath}] is not supported codec: {_config.Codec}");
+                    $"Topic[Path=\"{_config.TopicPath}\"] is not supported codec: {_config.Codec}");
                 return;
             }
 
@@ -199,6 +202,8 @@ private async Task Initialize()
 
             _session = new NotStartedWriterSession(
                 new WriterException("Transport error on creating write session", e));
+
+            _ = Task.Run(Initialize, _disposeTokenSource.Token);
         }
     }
 
@@ -245,7 +250,7 @@ public NotStartedWriterSession(WriterException reasonException)
 
     public Task Write(ConcurrentQueue<MessageSending> toSendBuffer)
     {
-        foreach (var messageSending in toSendBuffer)
+        while (toSendBuffer.TryDequeue(out var messageSending))
         {
             messageSending.TaskCompletionSource.SetException(_reasonException);
         }
diff --git a/src/Ydb.Sdk/src/Status.cs b/src/Ydb.Sdk/src/Status.cs
@@ -248,6 +248,7 @@ internal static Status ConvertStatus(this Grpc.Core.Status rpcStatus)
                 Grpc.Core.StatusCode.DeadlineExceeded => StatusCode.ClientTransportTimeout,
                 Grpc.Core.StatusCode.ResourceExhausted => StatusCode.ClientTransportResourceExhausted,
                 Grpc.Core.StatusCode.Unimplemented => StatusCode.ClientTransportUnimplemented,
+                Grpc.Core.StatusCode.Cancelled => StatusCode.Cancelled,
                 _ => StatusCode.ClientTransportUnknown
             },
             new List<Issue> { new(rpcStatus.Detail) }
diff --git a/src/Ydb.Sdk/tests/Topic/WriterMockTests.cs b/src/Ydb.Sdk/tests/Topic/WriterMockTests.cs
@@ -1,13 +1,15 @@
 using Grpc.Core;
 using Moq;
 using Xunit;
+using Ydb.Issue;
 using Ydb.Sdk.Services.Topic;
 using Ydb.Sdk.Services.Topic.Writer;
 using Ydb.Topic;
+using Codec = Ydb.Sdk.Services.Topic.Codec;
 
 namespace Ydb.Sdk.Tests.Topic;
 
-using WriterStream = BidirectionalStream<StreamWriteMessage.Types.FromClient, StreamWriteMessage.Types.FromServer>;
+using WriterStream = IBidirectionalStream<StreamWriteMessage.Types.FromClient, StreamWriteMessage.Types.FromServer>;
 
 public class WriterMockTests
 {
@@ -20,24 +22,219 @@ public WriterMockTests()
             It.IsAny<Method<StreamWriteMessage.Types.FromClient, StreamWriteMessage.Types.FromServer>>(),
             It.IsAny<GrpcRequestSettings>())
         ).Returns(_mockStream.Object);
+
+        _mockIDriver.Setup(driver => driver.LoggerFactory).Returns(Utils.GetLoggerFactory);
     }
 
-    // [Fact]
-    public async Task NotStarted_Failed_Test()
+    [Fact]
+    public async Task Initialize_WhenStreamIsClosedByServer_ThrowWriterExceptionOnWriteAsyncAndTryNextInitialize()
     {
         var moveNextTry = new TaskCompletionSource<bool>();
+        var taskNextComplete = new TaskCompletionSource();
 
-        _mockStream
-            .Setup(stream => stream.Write(It.IsAny<StreamWriteMessage.Types.FromClient>()))
+        _mockStream.Setup(stream => stream.Write(It.IsAny<StreamWriteMessage.Types.FromClient>()))
             .Returns(Task.CompletedTask);
-
         _mockStream.SetupSequence(stream => stream.MoveNextAsync())
             .ReturnsAsync(false)
-            .Returns(new ValueTask<bool>(moveNextTry.Task)); // For retry
+            .Returns(() =>
+            {
+                taskNextComplete.SetResult();
+                return new ValueTask<bool>(moveNextTry.Task);
+            });
 
         using var writer = new WriterBuilder<int>(_mockIDriver.Object, new WriterConfig("/topic")
             { ProducerId = "producerId" }).Build();
 
-        await Assert.ThrowsAsync<WriterException>(() => writer.WriteAsync(100));
+        Assert.Equal("Stream unexpectedly closed by YDB server. " +
+                     "Current InitRequest: { \"path\": \"/topic\", \"producerId\": \"producerId\" }",
+            (await Assert.ThrowsAsync<WriterException>(() => writer.WriteAsync(100))).Message);
+
+        await taskNextComplete.Task;
+        // check attempt repeated!!!
+        _mockStream.Verify(stream => stream.Write(It.IsAny<StreamWriteMessage.Types.FromClient>()), Times.Exactly(2));
+        _mockStream.Verify(stream => stream.MoveNextAsync(), Times.Exactly(2));
+    }
+
+    [Fact]
+    public async Task Initialize_WhenFailWriteMessage_ThrowWriterExceptionOnWriteAsyncAndTryNextInitialize()
+    {
+        var taskSource = new TaskCompletionSource();
+        var taskNextComplete = new TaskCompletionSource();
+        _mockStream.SetupSequence(stream => stream.Write(It.IsAny<StreamWriteMessage.Types.FromClient>()))
+            .ThrowsAsync(new Driver.TransportException(new RpcException(Grpc.Core.Status.DefaultCancelled)))
+            .Returns(() =>
+            {
+                taskNextComplete.SetResult();
+                return taskSource.Task;
+            });
+
+        using var writer = new WriterBuilder<string>(_mockIDriver.Object, new WriterConfig("/topic")
+            { ProducerId = "producerId" }).Build();
+
+        var writerException = await Assert.ThrowsAsync<WriterException>(() => writer.WriteAsync("abacaba"));
+        Assert.Equal("Transport error on creating write session", writerException.Message);
+        Assert.Equal(StatusCode.Cancelled, writerException.Status.StatusCode);
+
+        await taskNextComplete.Task;
+        // check attempt repeated!!!
+        _mockStream.Verify(stream => stream.Write(It.IsAny<StreamWriteMessage.Types.FromClient>()), Times.Exactly(2));
+    }
+
+    [Fact]
+    public async Task Initialize_WhenFailMoveNextAsync_ThrowWriterExceptionOnWriteAsyncAndTryNextInitialize()
+    {
+        var taskSource = new TaskCompletionSource<bool>();
+        var taskNextComplete = new TaskCompletionSource();
+        _mockStream.Setup(stream => stream.Write(It.IsAny<StreamWriteMessage.Types.FromClient>()))
+            .Returns(Task.CompletedTask);
+        _mockStream.SetupSequence(stream => stream.MoveNextAsync())
+            .ThrowsAsync(new Driver.TransportException(
+                new RpcException(new Grpc.Core.Status(Grpc.Core.StatusCode.DeadlineExceeded, "Some message"))))
+            .Returns(() =>
+            {
+                taskNextComplete.SetResult();
+                return new ValueTask<bool>(taskSource.Task);
+            });
+
+        using var writer = new WriterBuilder<string>(_mockIDriver.Object, new WriterConfig("/topic")
+            { ProducerId = "producerId" }).Build();
+
+        var writerException = await Assert.ThrowsAsync<WriterException>(() => writer.WriteAsync("abacaba"));
+        Assert.Equal("Transport error on creating write session", writerException.Message);
+        Assert.Equal(StatusCode.ClientTransportTimeout, writerException.Status.StatusCode);
+
+        await taskNextComplete.Task;
+        // check attempt repeated!!!
+        _mockStream.Verify(stream => stream.Write(It.IsAny<StreamWriteMessage.Types.FromClient>()), Times.Exactly(2));
+        _mockStream.Verify(stream => stream.MoveNextAsync(), Times.Exactly(2));
     }
+
+    [Fact]
+    public async Task Initialize_WhenInitResponseNotSuccess_ThrowWriterExceptionOnWriteAsyncAndTryNextInitialize()
+    {
+        var taskSource = new TaskCompletionSource<bool>();
+        var taskNextComplete = new TaskCompletionSource();
+        _mockStream.Setup(stream => stream.Write(It.IsAny<StreamWriteMessage.Types.FromClient>()))
+            .Returns(Task.CompletedTask);
+        _mockStream.SetupSequence(stream => stream.MoveNextAsync())
+            .Returns(new ValueTask<bool>(true))
+            .Returns(() =>
+            {
+                taskNextComplete.SetResult();
+                return new ValueTask<bool>(taskSource.Task);
+            });
+        _mockStream.Setup(stream => stream.Current)
+            .Returns(new StreamWriteMessage.Types.FromServer
+            {
+                Status = StatusIds.Types.StatusCode.BadSession,
+                Issues = { new IssueMessage { Message = "Some message" } }
+            });
+
+        using var writer = new WriterBuilder<long>(_mockIDriver.Object, new WriterConfig("/topic")
+            { ProducerId = "producerId" }).Build();
+
+        Assert.Equal("Initialization failed: Status: BadSession, Issues:\n[0] Fatal: Some message\n",
+            (await Assert.ThrowsAsync<WriterException>(() => writer.WriteAsync(123L))).Message);
+
+        await taskNextComplete.Task;
+        // check attempt repeated!!!
+        _mockStream.Verify(stream => stream.Write(It.IsAny<StreamWriteMessage.Types.FromClient>()), Times.Exactly(2));
+        _mockStream.Verify(stream => stream.MoveNextAsync(), Times.Exactly(2));
+    }
+
+    [Fact]
+    public async Task Initialize_WhenInitResponseIsSchemaError_ThrowWriterExceptionOnWriteAsyncAndStopInitializing()
+    {
+        _mockStream.Setup(stream => stream.Write(It.IsAny<StreamWriteMessage.Types.FromClient>()))
+            .Returns(Task.CompletedTask);
+        _mockStream.Setup(stream => stream.MoveNextAsync())
+            .Returns(new ValueTask<bool>(true));
+        _mockStream.Setup(stream => stream.Current)
+            .Returns(new StreamWriteMessage.Types.FromServer
+            {
+                Status = StatusIds.Types.StatusCode.SchemeError,
+                Issues = { new IssueMessage { Message = "Topic not found" } }
+            });
+
+        using var writer = new WriterBuilder<long>(_mockIDriver.Object, new WriterConfig("/topic")
+            { ProducerId = "producerId" }).Build();
+
+        Assert.Equal("Initialization failed: Status: SchemeError, Issues:\n[0] Fatal: Topic not found\n",
+            (await Assert.ThrowsAsync<WriterException>(() => writer.WriteAsync(123L))).Message);
+
+        // check not attempt repeated!!!
+        _mockStream.Verify(stream => stream.Write(It.IsAny<StreamWriteMessage.Types.FromClient>()), Times.Once);
+        _mockStream.Verify(stream => stream.MoveNextAsync(), Times.Once);
+    }
+
+    [Fact]
+    public async Task Initialize_WhenNotSupportedCodec_ThrowWriterExceptionOnWriteAsyncAndStopInitializing()
+    {
+        _mockStream.Setup(stream => stream.Write(It.IsAny<StreamWriteMessage.Types.FromClient>()))
+            .Returns(Task.CompletedTask);
+        _mockStream.Setup(stream => stream.MoveNextAsync())
+            .Returns(new ValueTask<bool>(true));
+        _mockStream.Setup(stream => stream.Current)
+            .Returns(new StreamWriteMessage.Types.FromServer
+            {
+                InitResponse = new StreamWriteMessage.Types.InitResponse
+                {
+                    LastSeqNo = 1, PartitionId = 1, SessionId = "SessionId",
+                    SupportedCodecs = new SupportedCodecs { Codecs = { 2 /* Gzip */, 3 /* Lzop */ } }
+                },
+                Status = StatusIds.Types.StatusCode.Success,
+            });
+
+        using var writer = new WriterBuilder<long>(_mockIDriver.Object, new WriterConfig("/topic")
+            { ProducerId = "producerId", Codec = Codec.Raw }).Build();
+
+        Assert.Equal("Topic[Path=\"/topic\"] is not supported codec: Raw",
+            (await Assert.ThrowsAsync<WriterException>(() => writer.WriteAsync(123L))).Message);
+
+        // check not attempt repeated!!!
+        _mockStream.Verify(stream => stream.Write(It.IsAny<StreamWriteMessage.Types.FromClient>()), Times.Once);
+        _mockStream.Verify(stream => stream.MoveNextAsync(), Times.Once);
+    }
+    
+    
+
+    /*
+     *         _mockStream.Setup(stream => stream.Current)
+           .Returns(new StreamWriteMessage.Types.FromServer
+           {
+               InitResponse = new StreamWriteMessage.Types.InitResponse
+                   { LastSeqNo = 1, PartitionId = 1, SessionId = "SessionId" },
+               Status = StatusIds.Types.StatusCode.Success,
+           });
+       moveNextTry.SetResult(true);
+       await Task.Yield();
+
+       var writeTask = writer.WriteAsync(100);
+       moveNextTryWriteAck.SetResult(true);
+
+       _mockStream.Setup(stream => stream.Current).Returns(
+           new StreamWriteMessage.Types.FromServer
+           {
+               WriteResponse = new StreamWriteMessage.Types.WriteResponse
+               {
+                   Acks =
+                   {
+                       new StreamWriteMessage.Types.WriteResponse.Types.WriteAck
+                       {
+                           SeqNo = 1, Written =
+                               new StreamWriteMessage.Types.WriteResponse.Types.WriteAck.Types.Written
+                                   { Offset = 2 }
+                       }
+                   }
+               },
+               Status = StatusIds.Types.StatusCode.Success
+           });
+       _mockStream.Setup(stream => stream.MoveNextAsync()).ReturnsAsync(true);
+
+       var writeResult = await writeTask;
+       Assert.Equal(PersistenceStatus.Written, writeResult.Status);
+       Assert.True(writeResult.TryGetOffset(out var offset));
+       Assert.Equal(2, offset);
+       _mockStream.Setup(stream => stream.MoveNextAsync()).ReturnsAsync(false);
+     */
 }

Original file line number	Diff line number	Diff line change
`@@ -6,13 +6,13 @@ internal abstract class TopicSession<TFromClient, TFromServer> : IDisposable`
`6`	`6`	`{`
`7`	`7`	`private readonly Func<Task> _initialize;`
`8`	`8`
`9`		`- protected readonly BidirectionalStream<TFromClient, TFromServer> Stream;`
	`9`	`+ protected readonly IBidirectionalStream<TFromClient, TFromServer> Stream;`
`10`	`10`	`protected readonly ILogger Logger;`
`11`	`11`	`protected readonly string SessionId;`
`12`	`12`
`13`	`13`	`private int _isActive = 1;`
`14`	`14`
`15`		`- protected TopicSession(BidirectionalStream<TFromClient, TFromServer> stream, ILogger logger,`
	`15`	`+ protected TopicSession(IBidirectionalStream<TFromClient, TFromServer> stream, ILogger logger,`
`16`	`16`	`string sessionId, Func<Task> initialize)`
`17`	`17`	`{`
`18`	`18`	`Stream = stream;`
Original file line number	Diff line number	Diff line change
`@@ -11,7 +11,7 @@ namespace Ydb.Sdk.Services.Topic.Writer;`
`11`	`11`	`using MessageData = StreamWriteMessage.Types.WriteRequest.Types.MessageData;`
`12`	`12`	`using MessageFromClient = StreamWriteMessage.Types.FromClient;`
`13`	`13`	`using MessageFromServer = StreamWriteMessage.Types.FromServer;`
`14`		`-using WriterStream = BidirectionalStream<`
	`14`	`+using WriterStream = IBidirectionalStream<`
`15`	`15`	`StreamWriteMessage.Types.FromClient,`
`16`	`16`	`StreamWriteMessage.Types.FromServer`
`17`	`17`	`>;`
`@@ -172,6 +172,8 @@ private async Task Initialize()`
`172`	`172`	`_ = Task.Run(Initialize, _disposeTokenSource.Token);`
`173`	`173`	`}`
`174`	`174`
	`175`	`+ _logger.LogCritical("Writer initialization failed to start. Reason: {Status}", status);`
	`176`	`+`
`175`	`177`	`return;`
`176`	`178`	`}`
`177`	`179`
`@@ -183,11 +185,12 @@ private async Task Initialize()`
`183`	`185`	`if (initResponse.SupportedCodecs != null &&`
`184`	`186`	`!initResponse.SupportedCodecs.Codecs.Contains((int)_config.Codec))`
`185`	`187`	`{`
`186`		`- _logger.LogCritical("Topic[{TopicPath}] is not supported codec: {Codec}", _config.TopicPath,`
`187`		`- _config.Codec);`
	`188`	`+ _logger.LogCritical(`
	`189`	`+ "Writer initialization failed to start. Reason: topic[Path=\"{TopicPath}\"] is not supported codec {Codec}",`
	`190`	`+ _config.TopicPath, _config.Codec);`
`188`	`191`
`189`	`192`	`_session = new NotStartedWriterSession(`
`190`		`- $"Topic[{_config.TopicPath}] is not supported codec: {_config.Codec}");`
	`193`	`+ $"Topic[Path=\"{_config.TopicPath}\"] is not supported codec: {_config.Codec}");`
`191`	`194`	`return;`
`192`	`195`	`}`
`193`	`196`
`@@ -199,6 +202,8 @@ private async Task Initialize()`
`199`	`202`
`200`	`203`	`_session = new NotStartedWriterSession(`
`201`	`204`	`new WriterException("Transport error on creating write session", e));`
	`205`	`+`
	`206`	`+ _ = Task.Run(Initialize, _disposeTokenSource.Token);`
`202`	`207`	`}`
`203`	`208`	`}`
`204`	`209`
`@@ -245,7 +250,7 @@ public NotStartedWriterSession(WriterException reasonException)`
`245`	`250`
`246`	`251`	`public Task Write(ConcurrentQueue<MessageSending> toSendBuffer)`
`247`	`252`	`{`
`248`		`- foreach (var messageSending in toSendBuffer)`
	`253`	`+ while (toSendBuffer.TryDequeue(out var messageSending))`
`249`	`254`	`{`
`250`	`255`	`messageSending.TaskCompletionSource.SetException(_reasonException);`
`251`	`256`	`}`