feat: update CAS on buffer

Author: KirillKurdyukov

.github/workflows/tests.yml

@@ -108,7 +108,7 @@ jobs:
         uses: actions/setup-dotnet@v4
         with:
           dotnet-version: ${{ matrix.dotnet-version }}
-      - name: Run ADO.NET tests
+      - name: Run Topic tests
         run: |
           docker cp ydb-local:/ydb_certs/ca.pem ~/
           cd src

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

@@ -1,28 +1,28 @@
 namespace Ydb.Sdk.Services.Topic;
 
-public class YdbWriterException : Exception
+public class WriterException : Exception
 {
-    public YdbWriterException(string message) : base(message)
+    public WriterException(string message) : base(message)
     {
         Status = new Status(StatusCode.Unspecified);
     }
 
-    public YdbWriterException(string message, Status status) : base(message + ": " + status)
+    public WriterException(string message, Status status) : base(message + ": " + status)
     {
         Status = status;
     }
 
-    public YdbWriterException(string message, Driver.TransportException e) : base(message, e)
+    public WriterException(string message, Driver.TransportException e) : base(message, e)
     {
         Status = e.Status;
     }
 
     public Status Status { get; }
 }
 
-public class YdbReaderException : Exception
+public class ReaderException : Exception
 {
-    protected YdbReaderException(string message) : base(message)
+    protected ReaderException(string message) : base(message)
     {
     }
 }

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

@@ -22,7 +22,7 @@ public Task<IReader<TValue>> Build()
         //     Deserializer ?? (IDeserializer<TValue>)(
         //         Deserializers.DefaultDeserializers.TryGetValue(typeof(TValue), out var deserializer)
         //             ? deserializer
-        //             : throw new YdbWriterException("The serializer is not set")
+        //             : throw new WriterException("The serializer is not set")
         //     )
         // );
         //

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

@@ -19,7 +19,7 @@ internal WriteResult(StreamWriteMessage.Types.WriteResponse.Types.WriteAck ack)
                 break;
             case StreamWriteMessage.Types.WriteResponse.Types.WriteAck.MessageWriteStatusOneofCase.None:
             default:
-                throw new YdbWriterException($"Unexpected WriteAck status: {ack.MessageWriteStatusCase}");
+                throw new WriterException($"Unexpected WriteAck status: {ack.MessageWriteStatusCase}");
         }
     }
 

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

@@ -74,8 +74,8 @@ public async Task<WriteResult> WriteAsync(Message<TValue> message)
                 curLimitBufferSize == _config.BufferMaxSize && data.Length > curLimitBufferSize
                 || curLimitBufferSize >= data.Length)
             {
-                if (Interlocked.CompareExchange(ref _limitBufferMaxSize, curLimitBufferSize,
-                        curLimitBufferSize - data.Length) == curLimitBufferSize)
+                if (Interlocked.CompareExchange(ref _limitBufferMaxSize,
+                        curLimitBufferSize - data.Length, curLimitBufferSize) == curLimitBufferSize)
                 {
                     _toSendBuffer.Enqueue(new MessageSending(messageData, completeTask));
 
@@ -92,7 +92,7 @@ public async Task<WriteResult> WriteAsync(Message<TValue> message)
                 "Buffer overflow: the data size [{DataLength}] exceeds the current buffer limit ({CurLimitBufferSize}) [BufferMaxSize = {BufferMaxSize}]",
                 data.Length, curLimitBufferSize, _config.BufferMaxSize);
 
-            throw new YdbWriterException("Buffer overflow");
+            throw new WriterException("Buffer overflow");
         }
 
         try
@@ -199,7 +199,7 @@ private async Task Initialize()
             _logger.LogError(e, "Unable to connect the session");
 
             _session = new NotStartedWriterSession(
-                new YdbWriterException("Transport error on creating write session", e));
+                new WriterException("Transport error on creating write session", e));
         }
     }
 
@@ -227,19 +227,19 @@ internal interface IWriteSession : IDisposable
 
 internal class NotStartedWriterSession : IWriteSession
 {
-    private readonly YdbWriterException _reasonException;
+    private readonly WriterException _reasonException;
 
     public NotStartedWriterSession(string reasonExceptionMessage)
     {
-        _reasonException = new YdbWriterException(reasonExceptionMessage);
+        _reasonException = new WriterException(reasonExceptionMessage);
     }
 
     public NotStartedWriterSession(string reasonExceptionMessage, Status status)
     {
-        _reasonException = new YdbWriterException(reasonExceptionMessage, status);
+        _reasonException = new WriterException(reasonExceptionMessage, status);
     }
 
-    public NotStartedWriterSession(YdbWriterException reasonException)
+    public NotStartedWriterSession(WriterException reasonException)
     {
         _reasonException = reasonException;
     }
@@ -365,7 +365,7 @@ Completing task on exception...
 Client SeqNo: {SeqNo}, WriteAck: {WriteAck}",
                             messageFromClient.MessageData.SeqNo, ack);
 
-                        messageFromClient.TaskCompletionSource.SetException(new YdbWriterException(
+                        messageFromClient.TaskCompletionSource.SetException(new WriterException(
                             $"Client SeqNo[{messageFromClient.MessageData.SeqNo}] is less then server's WriteAck[{ack}]"));
                     }
                     else

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

@@ -21,7 +21,7 @@ public IWriter<TValue> Build()
             Serializer ?? (ISerializer<TValue>)(
                 Serializers.DefaultSerializers.TryGetValue(typeof(TValue), out var serializer)
                     ? serializer
-                    : throw new YdbWriterException("The serializer is not set")
+                    : throw new WriterException("The serializer is not set")
             )
         );
     }

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

@@ -44,7 +44,7 @@ public async Task WriteAsync_WhenTopicNotFound_ReturnNotFoundException()
                 new WriterConfig(_topicName + "_not_found") { ProducerId = "producerId" })
             .Build();
 
-        Assert.Equal(StatusCode.SchemeError, (await Assert.ThrowsAsync<YdbWriterException>(
-                () => writer.WriteAsync("hello world"))).Status.StatusCode);
+        Assert.Equal(StatusCode.SchemeError, (await Assert.ThrowsAsync<WriterException>(
+            () => writer.WriteAsync("hello world"))).Status.StatusCode);
     }
 }