This is an automated email from the ASF dual-hosted git repository. lukaszzborek pushed a commit to branch fix/dotnet-problem in repository https://gitbox.apache.org/repos/asf/iggy.git
commit 16aab11ace5dce71958a921265591e64cce36574 Author: Łukasz Zborek <[email protected]> AuthorDate: Wed Oct 14 16:20:20 2026 +0200 fix(csharp): report refusal when routed budget expires before resend When a partition primary refuses a routed request with TRANSIENT_NOT_ACCEPTED or NOT_PRIMARY, SendRoutedAsync backs off and looks up the route again before it resends. If the request budget ran out in that backoff or in the route lookup, the caller got VsrRequestOutcomeUnknownException, even though nothing was written after the refusal. Java's PollRouter reports the refusal in this case. The loop now keeps the last refusal and clears it right before each send starts, and the budget catch rethrows the kept refusal. A budget expiry inside a send still reports an unknown outcome. --- .../TcpMessageStream.PollRouting.cs | 12 +++++++++ .../Iggy_SDK_Tests/VsrTests/PrimaryPollingTests.cs | 31 ++++++++++++++++++++++ 2 files changed, 43 insertions(+) diff --git a/foreign/csharp/Iggy_SDK/IggyClient/Implementations/TcpMessageStream.PollRouting.cs b/foreign/csharp/Iggy_SDK/IggyClient/Implementations/TcpMessageStream.PollRouting.cs index a4c3a9f59..d22766529 100644 --- a/foreign/csharp/Iggy_SDK/IggyClient/Implementations/TcpMessageStream.PollRouting.cs +++ b/foreign/csharp/Iggy_SDK/IggyClient/Implementations/TcpMessageStream.PollRouting.cs @@ -176,6 +176,7 @@ public sealed partial class TcpMessageStream var refreshHistory = code == CommandCodes.SEND_MESSAGES_CODE && context is null; var refusedPrimary = false; var lifecycleInterval = LifecycleRetryIntervalMs; + IggyInvalidStatusCodeException? refusal = null; try { while (true) @@ -192,6 +193,7 @@ public sealed partial class TcpMessageStream frame ??= _consensusSession.Resolve(VsrOperations.ForCode(code)); if (Volatile.Read(ref _clusterNodeCount) <= 1) { + refusal = null; var attempt = await SendVsrAttemptAsync(code, payload, Environment.TickCount64, deadline, false, cancellation.Token, captured.Value, frame, retryTransient: VsrOperations.ForCode(code) != VsrOperation.NonReplicated); @@ -206,6 +208,7 @@ public sealed partial class TcpMessageStream } return attempt.Response!; } + refusal = null; return await SendPrimaryOnceAsync(target.Key, code, payload, route, captured.Value, frame.Value, deadline, cancellation.Token); } @@ -213,6 +216,7 @@ public sealed partial class TcpMessageStream VsrError.TRANSIENT_NOT_ACCEPTED or VsrError.NOT_PRIMARY) { RemovePartitionRoute(target.Key); + refusal = error; if ((generation is not null && generation != _consensusSession.Generation) || Environment.TickCount64 + PollRoutingRetryIntervalMs >= deadline) { @@ -249,6 +253,14 @@ public sealed partial class TcpMessageStream catch (OperationCanceledException error) when (!token.IsCancellationRequested) { RemovePartitionRoute(target.Key); + // The budget can expire in the backoff or in a read-only control request that follows it, before the + // send call for the request starts again, so the last refusal still decides the outcome. An expiry + // inside that send call reports an unknown outcome even before the write, which is conservative. + if (refusal is not null) + { + ExceptionDispatchInfo.Throw(refusal); + } + throw new VsrRequestOutcomeUnknownException(error); } } diff --git a/foreign/csharp/Iggy_SDK_Tests/VsrTests/PrimaryPollingTests.cs b/foreign/csharp/Iggy_SDK_Tests/VsrTests/PrimaryPollingTests.cs index e933c4a18..832e8a29d 100644 --- a/foreign/csharp/Iggy_SDK_Tests/VsrTests/PrimaryPollingTests.cs +++ b/foreign/csharp/Iggy_SDK_Tests/VsrTests/PrimaryPollingTests.cs @@ -228,6 +228,37 @@ public sealed class PrimaryPollingTests Assert.All(cluster.Primaries, primary => Assert.Equal(0, primary.Connections)); } + [Fact] + public async Task given_primary_refusal_when_the_budget_expires_in_the_next_route_lookup_should_report_not_accepted() + { + using var cluster = new PollCluster(); + var release = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var lookups = 0; + cluster.OnPrimaryPoll = request => Reply(request.Operation, [], TRANSIENT_NOT_ACCEPTED); + cluster.OnCoordinator = request => + { + if (request.Code == CommandCodes.GET_PARTITION_ROUTE_CODE && Interlocked.Increment(ref lookups) == 2) + { + release.Task.GetAwaiter().GetResult(); + } + + return null; + }; + using var client = await cluster.ConnectAsync(); + try + { + var error = await Assert.ThrowsAsync<IggyInvalidStatusCodeException>(() => PollAsync(client)); + + Assert.Equal(VsrError.TRANSIENT_NOT_ACCEPTED, error.StatusCode); + Assert.Equal(2, Volatile.Read(ref lookups)); + Assert.Equal(1, cluster.Primaries[0].Requests(CommandCodes.POLL_MESSAGES_CODE)); + } + finally + { + release.SetResult(); + } + } + [Fact] public async Task given_poll_waiting_for_connection_when_metadata_commits_should_revalidate_its_cached_route() {
