feat(csharp): add VSR framing to the csharp SDK

.github/actions/csharp-dotnet/pre-merge/action.yml

@@ -21,7 +21,7 @@ description: .NET pre-merge testing github iggy actions
 
 inputs:
   task:
-    description: "Task to run (lint, test, build, e2e)"
+    description: "Task to run (lint, test, build, e2e, e2e-vsr)"
     required: true
 
 runs:
@@ -33,7 +33,7 @@ runs:
         dotnet-version: "10.0.x"
 
     - name: Setup Rust with cache
-      if: inputs.task == 'test' || inputs.task == 'e2e'
+      if: inputs.task == 'test' || inputs.task == 'e2e' || inputs.task == 'e2e-vsr'
       uses: ./.github/actions/utils/setup-rust-with-cache
 
     - name: Restore dependencies
@@ -90,6 +90,51 @@ runs:
       env:
         IGGY_SERVER_DOCKER_IMAGE: ${{ steps.docker_build.outputs.docker_image }}
         IGGY_TEST_LOGS_DIR: ./reports/container-logs
+        # The suite runs once per server; this job only built the classic image.
+        IGGY_TEST_SERVER: classic
+      run: |
+        dotnet test --project Iggy_SDK.Tests.Integration \
+          --no-build \
+          --verbosity normal \
+          --coverage \
+          --coverage-output-format cobertura \
+          --coverage-output coverage.cobertura.xml \
+          --results-directory ./reports \
+          -- --report-trx --retry-failed-tests 3
+      shell: bash
+
+    - name: Build iggy-server-ng Docker image
+      if: inputs.task == 'e2e-vsr'
+      shell: bash
+      run: |
+        set -euo pipefail
+        IMAGE_TAG="iggy-server-ng:test"
+        SERVER_BINARY="target/debug/iggy-server-ng"
+        CLI_BINARY="target/debug/iggy"
+
+        # server-ng only speaks the VSR wire protocol when built with the vsr feature.
+        cargo build --locked --features vsr --bin iggy-server-ng --bin iggy
+        ls -lh "$SERVER_BINARY" "$CLI_BINARY"
+
+        docker build \
+          -f core/server-ng/Dockerfile \
+          --target runtime-prebuilt \
+          -t "$IMAGE_TAG" \
+          --build-arg PREBUILT_IGGY_SERVER_NG="$SERVER_BINARY" \
+          --build-arg PREBUILT_IGGY_CLI="$CLI_BINARY" \
+          .
+
+        echo "IGGY_SERVER_NG_DOCKER_IMAGE=$IMAGE_TAG" >> "$GITHUB_ENV"
+
+    - name: Run VSR integration tests
+      if: inputs.task == 'e2e-vsr'
+      working-directory: foreign/csharp
+      env:
+        IGGY_TEST_LOGS_DIR: ./reports/container-logs
+        # Runs the whole suite plus the server-ng-only tests against a two-node iggy-server-ng cluster the
+        # fixture starts, so every case commits through consensus. TCP only on this leg, framed with the VSR
+        # wire protocol.
+        IGGY_TEST_SERVER: ng
       run: |
         dotnet test --project Iggy_SDK.Tests.Integration \
           --no-build \
@@ -102,17 +147,17 @@ runs:
       shell: bash
 
     - name: Collect container logs
-      if: inputs.task == 'e2e' && always()
+      if: (inputs.task == 'e2e' || inputs.task == 'e2e-vsr') && always()
       shell: bash
       run: |
         mkdir -p foreign/csharp/reports/container-logs
         find foreign/csharp/Iggy_SDK.Tests.Integration/bin -name "*.log" -path "*/container-logs/*" -exec cp {} foreign/csharp/reports/container-logs/ \; || true
 
     - name: Upload Test Results
       uses: actions/upload-artifact@v7
-      if: inputs.task == 'e2e' && always()
+      if: (inputs.task == 'e2e' || inputs.task == 'e2e-vsr') && always()
       with:
-        name: dotnet-test-results
+        name: dotnet-test-results-${{ inputs.task }}
         path: foreign/csharp/reports
         retention-days: 7
 

.github/config/components.yml

@@ -281,7 +281,7 @@ components:
     paths:
       - "foreign/csharp/**"
       - "examples/csharp/**"
-    tasks: ["lint", "test", "build", "e2e"]
+    tasks: ["lint", "test", "build", "e2e", "e2e-vsr"]
 
   sdk-cpp:
     depends_on:

.github/workflows/_test.yml

@@ -197,7 +197,7 @@ jobs:
           override_pr: ${{ github.event.pull_request.number }}
 
       - name: Upload C# coverage to Codecov
-        if: inputs.component == 'sdk-csharp' && (inputs.task == 'test' || inputs.task == 'e2e')
+        if: inputs.component == 'sdk-csharp' && (inputs.task == 'test' || inputs.task == 'e2e' || inputs.task == 'e2e-vsr')
         uses: codecov/codecov-action@v7.0.0
         with:
           token: ${{ secrets.CODECOV_TOKEN }}

.github/workflows/coverage-baseline.yml

@@ -195,6 +195,7 @@ jobs:
         working-directory: foreign/csharp
         env:
           IGGY_SERVER_DOCKER_IMAGE: ${{ steps.docker_build.outputs.docker_image }}
+          IGGY_TEST_SERVER: classic
         run: |
           dotnet test \
             --project Iggy_SDK.Tests.Integration \
@@ -205,6 +206,43 @@ jobs:
             --coverage-output coverage.cobertura.xml \
             --results-directory ./reports/integration
 
+      - name: Build iggy-server-ng Docker image
+        shell: bash
+        run: |
+          set -euo pipefail
+          IMAGE_TAG="iggy-server-ng:test"
+          SERVER_BINARY="target/debug/iggy-server-ng"
+          CLI_BINARY="target/debug/iggy"
+
+          # server-ng only speaks the VSR wire protocol when built with the vsr feature.
+          cargo build --locked --features vsr --bin iggy-server-ng --bin iggy
+
+          docker build \
+            -f core/server-ng/Dockerfile \
+            --target runtime-prebuilt \
+            -t "$IMAGE_TAG" \
+            --build-arg PREBUILT_IGGY_SERVER_NG="$SERVER_BINARY" \
+            --build-arg PREBUILT_IGGY_CLI="$CLI_BINARY" \
+            .
+
+          echo "IGGY_SERVER_NG_DOCKER_IMAGE=$IMAGE_TAG" >> "$GITHUB_ENV"
+
+      # The PR side uploads its e2e-vsr run under the same csharp flag, so a baseline without this leg
+      # compares every VSR line against nothing and reports the whole transport as uncovered.
+      - name: Run VSR integration tests with coverage
+        working-directory: foreign/csharp
+        env:
+          IGGY_TEST_SERVER: ng
+        run: |
+          dotnet test \
+            --project Iggy_SDK.Tests.Integration \
+            --no-build \
+            --verbosity normal \
+            --coverage \
+            --coverage-output-format cobertura \
+            --coverage-output coverage.cobertura.xml \
+            --results-directory ./reports/vsr
+
       - name: Merge coverage reports
         working-directory: foreign/csharp
         run: |

foreign/csharp/Directory.Packages.props

@@ -35,6 +35,7 @@
         <PackageVersion Include="Moq" Version="4.20.72" />
         <PackageVersion Include="Reqnroll.xunit.v3" Version="3.3.4" />
         <PackageVersion Include="Shouldly" Version="4.3.0" />
+        <PackageVersion Include="System.IO.Hashing" Version="10.0.0" />
         <PackageVersion Include="Testcontainers" Version="4.13.0" />
         <PackageVersion Include="TUnit" Version="1.61.15" />
         <PackageVersion Include="xunit.runner.visualstudio" Version="3.1.5">

foreign/csharp/Iggy_SDK.Tests.Integration/Attributes/ServerAttributes.cs

@@ -0,0 +1,50 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+using Apache.Iggy.Tests.Integrations.Fixtures;
+
+namespace Apache.Iggy.Tests.Integrations.Attributes;
+
+/// <summary>
+///     The operation is not served by iggy-server-ng yet, so the test only runs against the classic server.
+///     The reason names the gap.
+/// </summary>
+internal class SkipServerNgAttribute(string reason) : SkipAttribute($"Skipped for server-ng: {reason}")
+{
+    public override Task<bool> ShouldSkip(TestRegisteredContext context)
+    {
+        return Task.FromResult(IggyServerFixture.IsServerNg);
+    }
+}
+
+/// <summary>Pins a test to iggy-server-ng: a classic run has no cluster to exercise.</summary>
+internal class RequiresServerNgAttribute() : SkipAttribute("Requires IGGY_TEST_SERVER=ng")
+{
+    public override Task<bool> ShouldSkip(TestRegisteredContext context)
+    {
+        return Task.FromResult(!IggyServerFixture.IsServerNg);
+    }
+}
+
+/// <summary>Pins a test to the classic server, which owns the only image it can start.</summary>
+internal class RequiresClassicServerAttribute() : SkipAttribute("Requires IGGY_TEST_SERVER=classic")
+{
+    public override Task<bool> ShouldSkip(TestRegisteredContext context)
+    {
+        return Task.FromResult(IggyServerFixture.IsServerNg);
+    }
+}

foreign/csharp/Iggy_SDK.Tests.Integration/ClusterRedirectionTests.cs

@@ -19,11 +19,13 @@
 using Apache.Iggy.Contracts;
 using Apache.Iggy.Enums;
 using Apache.Iggy.Factory;
+using Apache.Iggy.Tests.Integrations.Attributes;
 using Apache.Iggy.Tests.Integrations.Fixtures;
 using Shouldly;
 
 namespace Apache.Iggy.Tests.Integrations;
 
+[RequiresClassicServer]
 public class ClusterRedirectionTests
 {
     [ClassDataSource<IggyClusterFixture>(Shared = SharedType.PerAssembly)]

foreign/csharp/Iggy_SDK.Tests.Integration/ConsumerGroupTests.cs

@@ -81,8 +81,8 @@ public async Task GetConsumerGroupById_Should_Return_ValidResponse(Protocol prot
         var cg = await client.CreateConsumerGroupAsync(Identifier.String(streamName),
             Identifier.String(TopicName), GroupName);
 
-        var response = await client.GetConsumerGroupByIdAsync(
-            Identifier.String(streamName), Identifier.String(TopicName),
+        var response = await client.GetConsumerGroupByIdAsync(Identifier.String(streamName),
+            Identifier.String(TopicName),
             Identifier.Numeric(cg!.Id));
 
         response.ShouldNotBeNull();
@@ -194,15 +194,14 @@ public async Task GetConsumerGroupById_WithMembers_Should_Return_ValidResponse(P
         var clients = new List<IIggyClient>();
         for (var i = 0; i < 2; i++)
         {
-            var memberClient = await Fixture.CreateClient(Protocol.Tcp);
+            var memberClient = await Fixture.CreateAuthenticatedClient(Protocol.Tcp);
             clients.Add(memberClient);
-            await memberClient.LoginUserAsync("iggy", "iggy");
             await memberClient.JoinConsumerGroupAsync(Identifier.String(streamName),
                 Identifier.String(TopicName), Identifier.Numeric(cg!.Id));
         }
 
-        var response = await client.GetConsumerGroupByIdAsync(
-            Identifier.String(streamName), Identifier.String(TopicName),
+        var response = await client.GetConsumerGroupByIdAsync(Identifier.String(streamName),
+            Identifier.String(TopicName),
             Identifier.Numeric(cg!.Id));
 
         response.ShouldNotBeNull();

foreign/csharp/Iggy_SDK.Tests.Integration/FetchMessagesTests.cs

@@ -23,6 +23,7 @@
 using Apache.Iggy.IggyClient;
 using Apache.Iggy.Kinds;
 using Apache.Iggy.Messages;
+using Apache.Iggy.Tests.Integrations.Attributes;
 using Apache.Iggy.Tests.Integrations.Fixtures;
 using Shouldly;
 using Partitioning = Apache.Iggy.Kinds.Partitioning;
@@ -90,6 +91,8 @@ public async Task PollMessages_WithNoHeaders_Should_PollMessages_Successfully(Pr
 
     [Test]
     [MethodDataSource<IggyServerFixture>(nameof(IggyServerFixture.ProtocolData))]
+    [SkipServerNg(
+        "server-ng replies the empty-poll shape for an unresolved topic; a zero-byte error body would break the poll decoder")]
     public async Task PollMessages_InvalidTopic_Should_Throw_InvalidResponse(Protocol protocol)
     {
         var (client, streamName) = await CreateStreamWithMessages(protocol);

foreign/csharp/Iggy_SDK.Tests.Integration/Fixtures/IggyClusterFixture.cs

@@ -39,12 +39,6 @@ public class IggyClusterFixture : IAsyncInitializer, IAsyncDisposable
     private const ushort PortRangeSize = 100;
     private static readonly ushort BasePort = (ushort)(30000 + Environment.Version.Major * 100);
     private static readonly ushort EndPort = (ushort)(BasePort + PortRangeSize);
-
-    // Listeners only need to outlive the eight ReservePort() calls in the
-    // constructor so we don't pick the same port twice within one fixture.
-    // Partitioned ranges already guarantee sibling processes can't race us, so
-    // we can release them as soon as picking is done.
-    private readonly List<TcpListener> _portReservations = [];
     private readonly IContainer _followerContainer;
     private readonly ushort _followerHttpPort;
     private readonly ushort _followerQuicPort;
@@ -60,6 +54,12 @@ public class IggyClusterFixture : IAsyncInitializer, IAsyncDisposable
 
     private readonly INetwork _network;
 
+    // Listeners only need to outlive the eight ReservePort() calls in the
+    // constructor so we don't pick the same port twice within one fixture.
+    // Partitioned ranges already guarantee sibling processes can't race us, so
+    // we can release them as soon as picking is done.
+    private readonly List<TcpListener> _portReservations = [];
+
     private string DockerImage =>
         Environment.GetEnvironmentVariable("IGGY_SERVER_DOCKER_IMAGE") ?? "apache/iggy:edge";
 
@@ -108,7 +108,7 @@ public IggyClusterFixture()
             ["IGGY_CLUSTER_NODES_1_PORTS_TCP"] = _followerTcpPort.ToString(),
             ["IGGY_CLUSTER_NODES_1_PORTS_QUIC"] = _followerQuicPort.ToString(),
             ["IGGY_CLUSTER_NODES_1_PORTS_HTTP"] = _followerHttpPort.ToString(),
-            ["IGGY_CLUSTER_NODES_1_PORTS_WEBSOCKET"] = _followerWsPort.ToString(),
+            ["IGGY_CLUSTER_NODES_1_PORTS_WEBSOCKET"] = _followerWsPort.ToString()
         };
 
         _leaderContainer = new ContainerBuilder(DockerImage)
@@ -183,7 +183,7 @@ public string GetFollowerAddress()
 
     private ushort ReservePort()
     {
-        for (ushort candidate = BasePort; candidate < EndPort; candidate++)
+        for (var candidate = BasePort; candidate < EndPort; candidate++)
         {
             try
             {