dev: unit reader tests on fails (#269)

* PartitionSession.Stop uses committedOffset to complete commit tasks when stopPartitionSessionRequest.Graceful is true.
* Changed batch type: IReadOnlyCollection<Message<TValue>> -> IReadOnlyList<Message<TValue>>.
* Invoking TryReadRequestBytes before deserializing message.
* Updated Ydb.Protos 1.0.6 -> 1.1.0: Updated version of the Grpc.Net.Client library to 2.67.0.

Author: KirillKurdyukov

CHANGELOG.md

@@ -1,3 +1,7 @@
+* PartitionSession.Stop uses committedOffset to complete commit tasks when stopPartitionSessionRequest.Graceful is true.
+* Changed batch type: IReadOnlyCollection<Message<TValue>> -> IReadOnlyList<Message<TValue>>.
+* Invoking TryReadRequestBytes before deserializing message.
+* Updated Ydb.Protos 1.0.6 -> 1.1.0: Updated version of the Grpc.Net.Client library to 2.67.0.
 * Fixed: YdbDataReader.GetDataTypeName for optional values.
 * Added support for "Columns" collectionName in YdbConnection.GetSchema(Async).
 

src/Ydb.Sdk/src/Driver.cs

@@ -130,7 +130,7 @@ private async Task<Status> DiscoverEndpoints()
         };
 
         var options = GetCallOptions(requestSettings);
-        options.Headers.Add(Metadata.RpcSdkInfoHeader, _sdkInfo);
+        options.Headers?.Add(Metadata.RpcSdkInfoHeader, _sdkInfo);
 
         var response = await client.ListEndpointsAsync(
             request: request,

src/Ydb.Sdk/src/Services/Table/Transaction.cs

@@ -28,7 +28,7 @@ internal Transaction(string txId)
         if (!string.IsNullOrEmpty(proto.Id))
         {
             tx.TxNum = IncTxCounter();
-            logger.LogTrace($"Received tx #{tx.TxNum}");
+            logger?.LogTrace($"Received tx #{tx.TxNum}");
         }
 
         return tx;

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

@@ -40,9 +40,9 @@ internal bool TryDequeueMessage([MaybeNullWhen(false)] out Message<TValue> messa
         }
 
         var index = _startMessageDataIndex++;
-        var approximatelyMessageBytesSize = Utils
-            .CalculateApproximatelyBytesSize(_approximatelyBatchSize, OriginalMessageCount, index);
         var messageData = _batch.MessageData[index];
+        _readerSession.TryReadRequestBytes(Utils
+            .CalculateApproximatelyBytesSize(_approximatelyBatchSize, OriginalMessageCount, index));
 
         TValue value;
         try
@@ -54,7 +54,6 @@ internal bool TryDequeueMessage([MaybeNullWhen(false)] out Message<TValue> messa
             throw new ReaderException("Error when deserializing message data", e);
         }
 
-        _readerSession.TryReadRequestBytes(approximatelyMessageBytesSize);
         var nextCommitedOffset = messageData.Offset + 1;
 
         message = new Message<TValue>(

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

@@ -59,10 +59,10 @@ public class BatchMessages<TValue>
     private readonly OffsetsRange _offsetsRange;
     private readonly long _partitionSessionId;
 
-    public IReadOnlyCollection<Message<TValue>> Batch { get; }
+    public IReadOnlyList<Message<TValue>> Batch { get; }
 
     internal BatchMessages(
-        IReadOnlyCollection<Message<TValue>> batch,
+        IReadOnlyList<Message<TValue>> batch,
         ReaderSession<TValue> readerSession,
         OffsetsRange offsetsRange,
         long partitionSessionId)

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

@@ -37,7 +37,7 @@ public PartitionSession(
     internal long PrevEndOffsetMessage { get; set; }
 
     // Each offset up to and including (committed_offset - 1) was fully processed.
-    internal long CommitedOffset { get; private set; }
+    private long CommitedOffset { get; set; }
 
     internal void RegisterCommitRequest(CommitSending commitSending)
     {
@@ -78,12 +78,19 @@ internal void HandleCommitedOffset(long commitedOffset)
         }
     }
 
-    internal void Stop()
+    internal void Stop(long commitedOffset)
     {
         _isStopped = true;
         while (_waitCommitMessages.TryDequeue(out var commitSending))
         {
-            Utils.SetPartitionClosedException(commitSending, PartitionSessionId);
+            if (commitSending.OffsetsRange.End <= commitedOffset)
+            {
+                commitSending.TcsCommit.SetResult();
+            }
+            else
+            {
+                Utils.SetPartitionClosedException(commitSending, PartitionSessionId);
+            }
         }
     }
 }

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

@@ -2,7 +2,6 @@
 using System.Threading.Channels;
 using Google.Protobuf.WellKnownTypes;
 using Microsoft.Extensions.Logging;
-using Ydb.Sdk.Ado;
 using Ydb.Topic;
 using Ydb.Topic.V1;
 using static Ydb.Topic.StreamReadMessage.Types.FromServer;
@@ -203,6 +202,8 @@ public void Dispose()
     {
         try
         {
+            _receivedMessagesChannel.Writer.TryComplete();
+
             _disposeCts.Cancel();
         }
         finally
@@ -246,7 +247,6 @@ internal class ReaderSession<TValue> : TopicSession<MessageFromClient, MessageFr
         Channel.CreateUnbounded<MessageFromClient>(
             new UnboundedChannelOptions
             {
-                SingleWriter = true,
                 SingleReader = true,
                 AllowSynchronousContinuations = false
             }
@@ -420,14 +420,19 @@ private async Task StopPartitionSessionRequest(
         {
             if (stopPartitionSessionRequest.Graceful)
             {
+                partitionSession.Stop(stopPartitionSessionRequest.CommittedOffset);
+
                 await _channelFromClientMessageSending.Writer.WriteAsync(new MessageFromClient
                 {
                     StopPartitionSessionResponse = new StreamReadMessage.Types.StopPartitionSessionResponse
                         { PartitionSessionId = partitionSession.PartitionSessionId }
                 });
             }
-
-            partitionSession.Stop();
+            else
+            {
+                // Maybe a race condition with the server dropping all waiters before they can commit.
+                partitionSession.Stop(-1);
+            }
         }
         else
         {
@@ -441,7 +446,7 @@ public async Task CommitOffsetRange(OffsetsRange offsetsRange, long partitionSes
         var tcsCommit = new TaskCompletionSource();
 
         await using var register = _lifecycleReaderSessionCts.Token.Register(
-            () => tcsCommit.TrySetException(new YdbException($"ReaderSession[{SessionId}] was deactivated"))
+            () => tcsCommit.TrySetException(new ReaderException($"ReaderSession[{SessionId}] was deactivated"))
         );
 
         var commitSending = new CommitSending(offsetsRange, tcsCommit);

src/Ydb.Sdk/src/Ydb.Sdk.csproj

@@ -26,7 +26,7 @@
   </PropertyGroup>
 
   <ItemGroup>
-    <PackageReference Include="Ydb.Protos" Version="1.0.6" />
+    <PackageReference Include="Ydb.Protos" Version="1.1.0" />
     <PackageReference Include="Portable.BouncyCastle" Version="1.9.0" />
     <PackageReference Include="System.IdentityModel.Tokens.Jwt" Version="7.0.0" />
   </ItemGroup>