Merge branch 'main' into feat-update-slo

Author: polRk

.github/scripts/publish.sh

@@ -8,51 +8,50 @@ then
   exit 1;
 fi;
 
-MAJOR=$(cat $VERSION_FILE | grep Major | grep -Eo '[0-9]*');
-MINOR=$(cat $VERSION_FILE | grep Minor | grep -Eo '[0-9]*');
-PATCH=$(cat $VERSION_FILE | grep Patch | grep -Eo '[0-9]*');
+LAST_TAG=$(git tag | tail -n 1);
+MAJOR=$(echo $LAST_TAG | sed -E 's/v([0-9]+)\..*/\1/');
+MINOR=$(echo $LAST_TAG | sed -E 's/v[0-9]+\.([0-9]+)\..*/\1/');
+PATCH=$(echo $LAST_TAG | sed -E 's/v[0-9]+\.[0-9]+\.([0-9]+)-rc[0-9]+/\1/');
+RC=0;
 
-VERSION="$MAJOR.$MINOR.$PATCH"
-
-LAST_TAG="v$MAJOR.$MINOR.$PATCH";
-if [ "$VERSION_CHANGE" = "MINOR" ]
+if [ "$RELEASE_CANDIDATE" = true ]
+then
+  RC=$(git tag | grep "v$MAJOR.$MINOR.$PATCH-rc" | wc -l | xargs || true); 
+fi  
+if [ "$VERSION_CHANGE" = "MINOR" ] && [ $RC = 0 ]
 then
   MINOR=$((MINOR+1));
   PATCH=0;
 fi;
-if [ "$VERSION_CHANGE" = "PATCH" ]
+if [ "$VERSION_CHANGE" = "PATCH" ] && [ $RC = 0 ]
 then
   PATCH=$((PATCH+1));
 fi;
 if [ "$RELEASE_CANDIDATE" = true ]
 then
-  RC=$(git tag | grep "v$MAJOR.$MINOR.$PATCH-rc" | wc -l);
-  TAG="v$MAJOR.$MINOR.$PATCH-rc$RC";
+  VERSION="$MAJOR.$MINOR.$PATCH-rc$RC";
 else
   sed -e "s/Minor = [0-9]*/Minor = $MINOR/g" -i $VERSION_FILE
   sed -e "s/Patch = [0-9]*/Patch = $PATCH/g" -i $VERSION_FILE
   git add $VERSION_FILE;
-  echo "## v$MAJOR.$MINOR.$PATCH" >> $CHANGELOG_FILE.tmp
-  cat $CHANGELOG_FILE >> $CHANGELOG_FILE.tmp
-  mv $CHANGELOG_FILE.tmp $CHANGELOG_FILE;
-  git add $CHANGELOG_FILE;
-  git config --global user.email "robot@umbrella";
-  git config --global user.name "robot";
-  git commit -m "Release v$MAJOR.$MINOR.$PATCH";
-  TAG="v$MAJOR.$MINOR.$PATCH";
+  VERSION="$MAJOR.$MINOR.$PATCH";
 fi;
+TAG="v$VERSION"
+echo "## v$VERSION" >> $CHANGELOG_FILE.tmp
+cat $CHANGELOG_FILE >> $CHANGELOG_FILE.tmp
+mv $CHANGELOG_FILE.tmp $CHANGELOG_FILE;
+git add $CHANGELOG_FILE;
+git config --global user.email "robot@umbrella";
+git config --global user.name "robot";
+git commit -m "Release v$VERSION";
 git tag $TAG
 git push --tags && git push
 CHANGELOG="$CHANGELOG
 
 Full Changelog: [$LAST_TAG...$TAG](https://github.com/ydb-platform/ydb-dotnet-sdk/compare/$LAST_TAG...$TAG)"
-if [ "$RELEASE_CANDIDATE" = true ]
-then
-  gh release create -d $TAG -t "$TAG" --notes "$CHANGELOG"
-else
-  VERSION=$MAJOR.$MINOR.$PATCH
-  cd src
-  dotnet pack -c Release -o out /p:Version=$VERSION
-  gh release create $TAG -t "$TAG" --notes "$CHANGELOG"
-  dotnet nuget push out/Ydb.Sdk.$VERSION.nupkg  --skip-duplicate --api-key $NUGET_TOKEN --source https://api.nuget.org/v3/index.json
-fi;
+
+cd src
+dotnet pack -c Release -o out /p:Version=$VERSION
+gh release create $TAG -t "$TAG" --notes "$CHANGELOG"
+dotnet nuget push out/Ydb.Sdk.$VERSION.nupkg  --skip-duplicate --api-key $NUGET_TOKEN --source https://api.nuget.org/v3/index.json
+

.github/workflows/tests.yml

@@ -14,7 +14,7 @@ jobs:
     strategy:
       fail-fast: false
       matrix:
-        os: [ ubuntu-22.04, windows-2022, macos-12 ]
+        os: [ ubuntu-22.04, windows-2022, macos-15 ]
         dotnet-version: [ 6.0.x, 7.0.x ]
         include:
           - dotnet-version: 6.0.x
@@ -33,7 +33,6 @@ jobs:
         run: |
           cd src
           dotnet test --filter "Category=Unit" -f ${{ matrix.dotnet-target-framework }}
-
   ado-net-tests:
     runs-on: ubuntu-22.04
     strategy:
@@ -74,7 +73,46 @@ jobs:
           docker cp ydb-local:/ydb_certs/ca.pem ~/
           cd src
           dotnet test --filter "(FullyQualifiedName~Ado) | (FullyQualifiedName~Dapper)" -l "console;verbosity=normal"
-  
+  topic-tests:
+    runs-on: ubuntu-22.04
+    strategy:
+      fail-fast: false
+      matrix:
+        ydb-version: [ 'trunk' ]
+        dotnet-version: [ 6.0.x, 7.0.x ]
+        include:
+          - dotnet-version: 6.0.x
+            dotnet-target-framework: net6.0
+          - dotnet-version: 7.0.x
+            dotnet-target-framework: net7.0
+    services:
+      ydb:
+        image: cr.yandex/yc/yandex-docker-local-ydb:${{ matrix.ydb-version }}
+        ports:
+          - 2135:2135
+          - 2136:2136
+          - 8765:8765
+        env:
+          YDB_LOCAL_SURVIVE_RESTART: true
+          YDB_USE_IN_MEMORY_PDISKS: true
+        options: '--name ydb-local -h localhost'
+    env:
+      OS: ubuntu-22.04
+      YDB_VERSION: ${{ matrix.ydb-version }}
+      YDB_CONNECTION_STRING: grpc://localhost:2136/local
+      YDB_CONNECTION_STRING_SECURE: grpcs://localhost:2135/local
+    steps:
+      - name: Checkout code
+        uses: actions/checkout@v4
+      - name: Install Dotnet
+        uses: actions/setup-dotnet@v4
+        with:
+          dotnet-version: ${{ matrix.dotnet-version }}
+      - name: Run Topic tests
+        run: |
+          docker cp ydb-local:/ydb_certs/ca.pem ~/
+          cd src
+          dotnet test --filter "FullyQualifiedName~Topic" -l "console;verbosity=normal"  
   integration-tests:
     runs-on: ubuntu-22.04
     strategy:
@@ -154,4 +192,4 @@ jobs:
       - name: Run Dapper example
         run: |
           cd ./examples/src/DapperExample
-          dotnet run
+          dotnet run

CHANGELOG.md

@@ -1,3 +1,25 @@
+## v0.9.0-rc1
+- Topic Writer updated release candidate:
+  * Do not send messages that have a timeout by cancelToken.
+  * If your value serializer throws an exception, this will be wrapped in a WriterException with unspecified status.
+  * Added BufferOverflowRetryTimeoutMs to the next try write.
+  * Rename _disposeTokenSource -> _disposeCts.
+  * Optimize write worker: if (_toSendBuffer.IsEmpty) continue.
+  * On RPC errors create DummyWriterSession.
+  * Message has been skipped because its sequence number is less than or equal to the last processed server's SeqNo.
+  * Calculate the next sequence number from the calculated previous messages.
+
+## v0.9.0-rc0
+- Topic Writer release candidate:
+  * Updated CAS semantics for enqueuing in the buffer.
+  * Processed buffer overflow on WriteAsync.
+  * Setting NotStartedWriterSession with a fail reason on RPC and more errors.
+  * New initialization strategy for WriterSession (background task).
+  * Supported cancellation token for sending tasks.
+  * Fixed setting the SeqNo field in the message (in-flight buffer already has a seqNo) and added a check on canceled TCS.
+  * Using BitConverter for Serializer / Deserializer.
+- Fixed: grpc requests go via proxy on Grpc.NET.Client >= 2.44 
+
 ## v0.8.0
 - Fixed bug on commit with fail, no set failed flag for rollback invocation 
 - Supported UUID (Guid)

src/Ydb.Sdk/src/IDriver.cs

@@ -6,21 +6,21 @@ namespace Ydb.Sdk;
 
 public interface IDriver : IAsyncDisposable, IDisposable
 {
-    internal Task<TResponse> UnaryCall<TRequest, TResponse>(
+    public Task<TResponse> UnaryCall<TRequest, TResponse>(
         Method<TRequest, TResponse> method,
         TRequest request,
         GrpcRequestSettings settings)
         where TRequest : class
         where TResponse : class;
 
-    internal ServerStream<TResponse> ServerStreamCall<TRequest, TResponse>(
+    public ServerStream<TResponse> ServerStreamCall<TRequest, TResponse>(
         Method<TRequest, TResponse> method,
         TRequest request,
         GrpcRequestSettings settings)
         where TRequest : class
         where TResponse : class;
 
-    internal 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,15 @@ internal BidirectionalStream<TRequest, TResponse> BidirectionalStreamCall<TReque
     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 +104,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,7 +222,7 @@ public async ValueTask<bool> MoveNextAsync()
     }
 }
 
-public sealed class BidirectionalStream<TRequest, TResponse> : IDisposable
+public class BidirectionalStream<TRequest, TResponse> : IBidirectionalStream<TRequest, TResponse>
 {
     private readonly AsyncDuplexStreamingCall<TRequest, TResponse> _stream;
     private readonly Action<RpcException> _rpcErrorAction;

src/Ydb.Sdk/src/Pool/ChannelPool.cs

@@ -134,6 +134,8 @@ public GrpcChannel CreateChannel(string endpoint)
 
                 return true;
             };
+        // https://github.com/grpc/grpc-dotnet/issues/2312#issuecomment-1790661801
+        httpHandler.Properties["__GrpcLoadBalancingDisabled"] = true;
 
         channelOptions.HttpHandler = httpHandler;
         channelOptions.DisposeHttpClient = true;

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

@@ -55,28 +55,15 @@ private class Int64Deserializer : IDeserializer<long>
     {
         public long Deserialize(byte[] data)
         {
-            if (data.Length != 8)
-            {
-                throw new ArgumentException(
-                    $"Deserializer<Long> encountered data of length ${data.Length}. Expecting data length to be 8");
-            }
-
-            return ((long)data[0] << 56) | ((long)data[1] << 48) | ((long)data[2] << 40) | ((long)data[3] << 32) |
-                   ((long)data[4] << 24) | ((long)data[5] << 16) | ((long)data[6] << 8) | data[7];
+            return BitConverter.ToInt64(data);
         }
     }
 
     private class Int32Deserializer : IDeserializer<int>
     {
         public int Deserialize(byte[] data)
         {
-            if (data.Length != 4)
-            {
-                throw new ArgumentException(
-                    $"Deserializer<Int32> encountered data of length ${data.Length}. Expecting data length to be 4");
-            }
-
-            return (data[0] << 24) | (data[1] << 16) | (data[2] << 8) | data[3];
+            return BitConverter.ToInt32(data);
         }
     }
 

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

@@ -1,22 +1,23 @@
 namespace Ydb.Sdk.Services.Topic;
 
-public class YdbTopicException : Exception
+public class WriterException : Exception
 {
-    protected YdbTopicException(string message) : base(message)
+    public WriterException(string message) : base(message)
     {
     }
-}
 
-public class YdbWriterException : YdbTopicException
-{
-    public YdbWriterException(string message) : base(message)
+    public WriterException(string message, Status status) : base(message + ": " + status)
+    {
+    }
+
+    public WriterException(string message, Exception inner) : base(message, inner)
     {
     }
 }
 
-public class YdbReaderException : YdbTopicException
+public class ReaderException : Exception
 {
-    protected YdbReaderException(string message) : base(message)
+    protected ReaderException(string message) : base(message)
     {
     }
 }

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

@@ -2,9 +2,9 @@
 
 namespace Ydb.Sdk.Services.Topic;
 
-public interface IWriter<TValue>
+public interface IWriter<TValue> : IDisposable
 {
-    public Task<WriteResult> WriteAsync(TValue data);
+    public Task<WriteResult> WriteAsync(TValue data, CancellationToken cancellationToken = default);
 
-    public Task<WriteResult> WriteAsync(Message<TValue> message);
+    public Task<WriteResult> WriteAsync(Message<TValue> message, CancellationToken cancellationToken = default);
 }

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/Serializer.cs

@@ -49,31 +49,15 @@ private class Int64Serializer : ISerializer<long>
     {
         public byte[] Serialize(long data)
         {
-            return new[]
-            {
-                (byte)(data >> 56),
-                (byte)(data >> 48),
-                (byte)(data >> 40),
-                (byte)(data >> 32),
-                (byte)(data >> 24),
-                (byte)(data >> 16),
-                (byte)(data >> 8),
-                (byte)data
-            };
+            return BitConverter.GetBytes(data);
         }
     }
 
     private class Int32Serializer : ISerializer<int>
     {
         public byte[] Serialize(int data)
         {
-            return new[]
-            {
-                (byte)(data >> 24),
-                (byte)(data >> 16),
-                (byte)(data >> 8),
-                (byte)data
-            };
+            return BitConverter.GetBytes(data);
         }
     }