This is an automated email from the ASF dual-hosted git repository.
spetz pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iggy.git
The following commit(s) were added to refs/heads/master by this push:
new e06fb351d fix(csharp): report refusal when routed budget expires
before resend (#4547)
e06fb351d is described below
commit e06fb351d66baec3cbb2a7b1372eb7118c02a467
Author: Łukasz Zborek <[email protected]>
AuthorDate: Wed Oct 14 23:41:32 2026 +0200
fix(csharp): report refusal when routed budget expires before resend (#4547)
Under a full parallel run the C# integration suite failed at random.
Backups read the 5s heartbeat timeout as a dead metadata primary, HTTP
clients were refused after a partition primary moved because nodes had
no shared JWT secret, consumer group convergence outran its 10s wait,
and the user delete test checked a numeric id another test could reuse.
Raise the test cluster heartbeat timeout to 15s, share one JWT secret
across nodes, wait up to 30s for convergence and check deleted users by
name. In the SDK, a committed rejection of a replay now settles an
uncertain write, REQUEST_TOO_OLD and INVALID_COMMAND end the replay at
once, no attempt starts past the deadline and a finished roster walk
restarts once. Failover unit tests send 3 refusals instead of 50 where
masking is not under test.
---
.../Fixtures/VsrCluster.cs | 9 +
.../Helpers/ConsumerGroupAssignment.cs | 4 +-
.../Vsr/VsrMetadataTests.cs | 3 +-
.../Exceptions/IggyInvalidStatusCodeException.cs | 5 +
.../TcpMessageStream.PollRouting.cs | 26 +++
.../Implementations/TcpMessageStream.Vsr.cs | 22 +-
foreign/csharp/Iggy_SDK/Vsr/VsrConnection.cs | 53 ++++-
foreign/csharp/Iggy_SDK/Vsr/VsrError.cs | 12 +-
foreign/csharp/Iggy_SDK/Vsr/VsrReplyDecoder.cs | 3 +-
.../VsrTests/EndpointFailoverTests.cs | 234 +++++++++++++++++++--
.../Iggy_SDK_Tests/VsrTests/PrimaryPollingTests.cs | 31 +++
.../VsrTests/VsrReplyDecoderTests.cs | 10 +-
.../Iggy_SDK_Tests/VsrTests/VsrTestPayloads.cs | 5 +
13 files changed, 377 insertions(+), 40 deletions(-)
diff --git a/foreign/csharp/Iggy_SDK.Tests.Integration/Fixtures/VsrCluster.cs
b/foreign/csharp/Iggy_SDK.Tests.Integration/Fixtures/VsrCluster.cs
index 924f9047d..e86b27a36 100644
--- a/foreign/csharp/Iggy_SDK.Tests.Integration/Fixtures/VsrCluster.cs
+++ b/foreign/csharp/Iggy_SDK.Tests.Integration/Fixtures/VsrCluster.cs
@@ -34,6 +34,7 @@ namespace Apache.Iggy.Tests.Integrations.Fixtures;
internal sealed class VsrCluster : IAsyncDisposable
{
private const string ClusterName = "test-vsr-cluster";
+ private const string HttpJwtSecret = "test-vsr-cluster-http-jwt-secret!";
// TCP and HTTP get real host ports on every node: a view change can make
any replica the
// primary clients are redirected to, so every advertised 127.0.0.1:port
must be dialable from
@@ -297,6 +298,14 @@ internal sealed class VsrCluster : IAsyncDisposable
environment["IGGY_CLUSTER_NAME"] = ClusterName;
environment["IGGY_MESSAGE_BUS_RECONNECT_PERIOD"] = "100ms";
+ // The metadata primary seeds a new partition group seconds after its
backups when the whole suite runs
+ // at once, and at the default 5s the backups read that as a dead
primary and elect a new one.
+ environment["IGGY_CLUSTER_HEARTBEAT_TIMEOUT"] = "15s";
+ // A partition primary can move off the node HTTP clients talk to.
Without a JWT secret every node
+ // verifies, that node cannot forward to the new primary and refuses
the partition until the budget
+ // runs out.
+ environment["IGGY_HTTP_JWT_ENCODING_SECRET"] = HttpJwtSecret;
+ environment["IGGY_HTTP_JWT_DECODING_SECRET"] = HttpJwtSecret;
for (var node = 0; node < _nodeCount; node++)
{
diff --git
a/foreign/csharp/Iggy_SDK.Tests.Integration/Helpers/ConsumerGroupAssignment.cs
b/foreign/csharp/Iggy_SDK.Tests.Integration/Helpers/ConsumerGroupAssignment.cs
index 068f7337a..93b884bf8 100644
---
a/foreign/csharp/Iggy_SDK.Tests.Integration/Helpers/ConsumerGroupAssignment.cs
+++
b/foreign/csharp/Iggy_SDK.Tests.Integration/Helpers/ConsumerGroupAssignment.cs
@@ -28,7 +28,9 @@ namespace Apache.Iggy.Tests.Integrations.Helpers;
/// </summary>
public static class ConsumerGroupAssignment
{
- private static readonly TimeSpan ConvergenceTimeout =
TimeSpan.FromSeconds(10);
+ // The fence install runs off the reconciler of the shard that owns the
partition, and on the metadata
+ // primary's shard 0 that lags by seconds under a full suite.
+ private static readonly TimeSpan ConvergenceTimeout =
TimeSpan.FromSeconds(30);
private static readonly TimeSpan PollInterval =
TimeSpan.FromMilliseconds(100);
/// <summary>
diff --git a/foreign/csharp/Iggy_SDK.Tests.Integration/Vsr/VsrMetadataTests.cs
b/foreign/csharp/Iggy_SDK.Tests.Integration/Vsr/VsrMetadataTests.cs
index 8f898d8e2..6c092ed52 100644
--- a/foreign/csharp/Iggy_SDK.Tests.Integration/Vsr/VsrMetadataTests.cs
+++ b/foreign/csharp/Iggy_SDK.Tests.Integration/Vsr/VsrMetadataTests.cs
@@ -100,7 +100,8 @@ public class VsrMetadataTests
fetched.Username.ShouldBe(name);
await client.DeleteUserAsync(Identifier.Numeric(user.Id));
- (await
client.GetUserAsync(Identifier.Numeric(user.Id))).ShouldBeNull();
+ // The freed id goes to the next user any parallel test creates, so
only the name proves the delete.
+ (await client.GetUserAsync(Identifier.String(name))).ShouldBeNull();
}
/// <summary>
diff --git
a/foreign/csharp/Iggy_SDK/Exceptions/IggyInvalidStatusCodeException.cs
b/foreign/csharp/Iggy_SDK/Exceptions/IggyInvalidStatusCodeException.cs
index ba5a72915..26238f08e 100644
--- a/foreign/csharp/Iggy_SDK/Exceptions/IggyInvalidStatusCodeException.cs
+++ b/foreign/csharp/Iggy_SDK/Exceptions/IggyInvalidStatusCodeException.cs
@@ -36,6 +36,11 @@ public sealed class IggyInvalidStatusCodeException :
Exception
internal int? CommandCode { get; set; }
+ /// <summary>
+ /// Whether the code came from the result section of a reply rather
than its status word or an eviction.
+ /// </summary>
+ internal bool FromResultSection { get; init; }
+
internal IggyInvalidStatusCodeException(int statusCode, string message,
bool fromServer = false) : base(message)
{
StatusCode = statusCode;
diff --git
a/foreign/csharp/Iggy_SDK/IggyClient/Implementations/TcpMessageStream.PollRouting.cs
b/foreign/csharp/Iggy_SDK/IggyClient/Implementations/TcpMessageStream.PollRouting.cs
index a4c3a9f59..41e17322e 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)
@@ -190,6 +191,7 @@ public sealed partial class TcpMessageStream
}
generation ??= route.Generation;
frame ??=
_consensusSession.Resolve(VsrOperations.ForCode(code));
+ refusal = null;
if (Volatile.Read(ref _clusterNodeCount) <= 1)
{
var attempt = await SendVsrAttemptAsync(code, payload,
Environment.TickCount64, deadline,
@@ -213,6 +215,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)
{
@@ -234,12 +237,16 @@ public sealed partial class TcpMessageStream
refreshHistory = false;
captured = null;
frame = null;
+ refusal = error;
}
+ // The partition plane does not commit LIFECYCLE_BUSY today. A
routed request still handles it, so a
+ // server that starts committing it needs no client change.
catch (IggyInvalidStatusCodeException error) when
(error.FromServer
&& error.StatusCode == VsrError.LIFECYCLE_BUSY
&& error.CommandCode == code
&& Environment.TickCount64 + lifecycleInterval < deadline)
{
+ refusal = error;
await Task.Delay(lifecycleInterval, cancellation.Token);
lifecycleInterval = Math.Min(lifecycleInterval * 2,
LifecycleRetryMaxIntervalMs);
frame = null;
@@ -249,8 +256,27 @@ 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);
}
+ // The read timer of a route lookup fires at the same deadline as the
budget and turns into a timeout
+ // IOException when it wins, and a redial can fail the same way. A
kept refusal means no send started
+ // since, so it still decides the outcome.
+ catch (Exception error) when (error is IOException or SocketException
+ && refusal is not null
+ && Environment.TickCount64 + PollRoutingRetryIntervalMs >=
deadline)
+ {
+ RemovePartitionRoute(target.Key);
+ ExceptionDispatchInfo.Throw(refusal);
+ throw;
+ }
}
private async Task<IMemoryOwner<byte>>
SendPrimaryOnceAsync(PartitionRouteKey key, int code,
diff --git
a/foreign/csharp/Iggy_SDK/IggyClient/Implementations/TcpMessageStream.Vsr.cs
b/foreign/csharp/Iggy_SDK/IggyClient/Implementations/TcpMessageStream.Vsr.cs
index 9f1592a00..26fb64ee3 100644
--- a/foreign/csharp/Iggy_SDK/IggyClient/Implementations/TcpMessageStream.Vsr.cs
+++ b/foreign/csharp/Iggy_SDK/IggyClient/Implementations/TcpMessageStream.Vsr.cs
@@ -641,6 +641,7 @@ public sealed partial class TcpMessageStream :
ISessionGenerationProvider
var leaderRedirects = 0;
var redirectBudgetLogged = false;
var walkingRoster = false;
+ var rosterWalkRestarted = false;
HashSet<string> walkedRosterEndpoints =
new(StringComparer.OrdinalIgnoreCase);
VsrConnection? lastConnection = null;
@@ -668,7 +669,7 @@ public sealed partial class TcpMessageStream :
ISessionGenerationProvider
}
&& !isLoginRegister
&& allowRedirect
- && Environment.TickCount64 < overallDeadline)
+ && Environment.TickCount64 + PollRoutingRetryIntervalMs <
overallDeadline)
{
var moved = false;
if (leaderRedirects < VsrMaxLeaderRedirects &&
!walkingRoster)
@@ -683,6 +684,16 @@ public sealed partial class TcpMessageStream :
ISessionGenerationProvider
if (!moved && attempt.Error is not
IggyInvalidStatusCodeException { StatusCode: VsrError.NOT_PRIMARY })
{
moved = await
RedirectToNextRosterNodeAsync(walkedRosterEndpoints, token);
+ if (!moved && walkingRoster && !rosterWalkRestarted)
+ {
+ // A node that refused earlier can serve the
request once it catches up, like a
+ // partition primary that had not materialised the
partition yet. Every hop also drops
+ // the poll connections, so the walk restarts once
and then stays on the current node.
+ rosterWalkRestarted = true;
+ walkedRosterEndpoints.Clear();
+ moved = await
RedirectToNextRosterNodeAsync(walkedRosterEndpoints, token);
+ }
+
walkingRoster |= moved;
}
@@ -701,7 +712,12 @@ public sealed partial class TcpMessageStream :
ISessionGenerationProvider
_currentAddress);
}
- continue;
+ // The redirect and reconnect run without a deadline, so a
fresh attempt past the budget
+ // would read no reply and turn this refusal into an
unknown outcome.
+ if (Environment.TickCount64 + PollRoutingRetryIntervalMs <
overallDeadline)
+ {
+ continue;
+ }
}
if (attempt.Error is VsrSessionEvictedException evicted)
@@ -741,7 +757,7 @@ public sealed partial class TcpMessageStream :
ISessionGenerationProvider
/// <summary>
/// Moves to the next unvisited roster endpoint after a never-admitted
refusal. Metadata and partition
/// groups elect independently, so the metadata leader is not
necessarily the primary for the request's
- /// partition. The caller's request deadline and redirect budget bound
the walk.
+ /// partition. The caller's request deadline bounds the walk, and the
caller restarts a finished walk once.
/// </summary>
private async Task<bool> RedirectToNextRosterNodeAsync(HashSet<string>
visited, CancellationToken token)
{
diff --git a/foreign/csharp/Iggy_SDK/Vsr/VsrConnection.cs
b/foreign/csharp/Iggy_SDK/Vsr/VsrConnection.cs
index ba13d8759..eddac2817 100644
--- a/foreign/csharp/Iggy_SDK/Vsr/VsrConnection.cs
+++ b/foreign/csharp/Iggy_SDK/Vsr/VsrConnection.cs
@@ -144,16 +144,19 @@ internal sealed class VsrConnection : IDisposable
catch (IggyInvalidStatusCodeException e)
{
uncertain |= e is { FromServer: true, StatusCode:
VsrError.TRANSIENT_NOT_COMMITTED };
- var verdict = uncertain && e is
+ var masked = uncertain && e is
{
FromServer: true,
StatusCode: not VsrError.TRANSIENT_NOT_COMMITTED
- }
+ } && !SettlesOutcome(e);
+ var verdict = masked
? VsrError.FromServer(VsrError.TRANSIENT_NOT_COMMITTED,
- "A later refusal does not resolve the original
request outcome.")
+ $"A later refusal with status {e.StatusCode} does
not resolve the original request "
+ + "outcome.")
: e;
verdict.CommandCode = code;
- if (!retryTransient || !IsReplayableTransient(verdict,
transientDeadline, readDeadline))
+ if (!retryTransient || (masked && IsFinalRefusal(e))
+ || !IsReplayableTransient(verdict, transientDeadline,
readDeadline))
{
return VsrAttempt.Failed(encoded, verdict,
requestStarted, this);
}
@@ -162,6 +165,12 @@ internal sealed class VsrConnection : IDisposable
: transientDeadline;
var remaining = governingDeadline -
Environment.TickCount64;
await Task.Delay((int)Math.Clamp(remaining, 0,
TransientRetryIntervalMs), token);
+ if (Environment.TickCount64 + TransientRetryIntervalMs >=
readDeadline)
+ {
+ // A replay written now has too little time left for
its reply, and the read timeout would
+ // turn a refusal the caller can act on into an
unknown outcome.
+ return VsrAttempt.Failed(encoded, verdict,
requestStarted, this);
+ }
}
catch (Exception e) when (IsConnectionException(e))
{
@@ -341,6 +350,42 @@ internal sealed class VsrConnection : IDisposable
_ => false
};
}
+
+ /// <summary>
+ /// Whether a refusal of a replay settles the outcome of the original
request. The primary answers a
+ /// same-id replay from its session table, so a result-section
rejection is the cached reply of the
+ /// original, the commit of a request that never entered the pipeline,
or a refusal of a request neither
+ /// the session table nor the pipeline holds. The codes below settle
nothing: most come from the checks
+ /// that run before that lookup, and REQUEST_TOO_OLD means the
original applied and its reply is gone.
+ /// PERSONAL_ACCESS_TOKEN_ALREADY_EXISTS settles even as the refusal
to replay a token the original
+ /// minted: either way the name is taken and the caller holds no
token. Apply can commit INVALID_COMMAND
+ /// too, but nothing tells it apart from the preflight one. The server
never puts HISTORY_UNAVAILABLE in a
+ /// result section, and it stays listed because settling on it would
be wrong if it ever did. The caller
+ /// filters TRANSIENT_NOT_COMMITTED out before asking.
+ /// </summary>
+ private static bool SettlesOutcome(IggyInvalidStatusCodeException error)
+ {
+ return error is
+ {
+ FromResultSection: true,
+ StatusCode: not (VsrError.TRANSIENT_NOT_ACCEPTED or
VsrError.NOT_PRIMARY or VsrError.REQUEST_TOO_OLD
+ or VsrError.INVALID_COMMAND or VsrError.UNAUTHENTICATED or
VsrError.HISTORY_UNAVAILABLE)
+ };
+ }
+
+ /// <summary>
+ /// Whether an unsettled refusal of a replay is final: the primary's
preflight answers every replay of the
+ /// same request id with it again, so replaying until the deadline
only ends in the same unknown outcome.
+ /// The metadata plane carries it in the result section and the
partition plane on the status word.
+ /// </summary>
+ private static bool IsFinalRefusal(IggyInvalidStatusCodeException error)
+ {
+ return error is
+ {
+ FromServer: true,
+ StatusCode: VsrError.REQUEST_TOO_OLD or VsrError.INVALID_COMMAND
+ };
+ }
}
/// <summary>Outcome of one <see cref="VsrConnection.SendAttemptAsync" />
call.</summary>
diff --git a/foreign/csharp/Iggy_SDK/Vsr/VsrError.cs
b/foreign/csharp/Iggy_SDK/Vsr/VsrError.cs
index bf6dd65bf..8ab14048d 100644
--- a/foreign/csharp/Iggy_SDK/Vsr/VsrError.cs
+++ b/foreign/csharp/Iggy_SDK/Vsr/VsrError.cs
@@ -36,6 +36,7 @@ internal static class VsrError
internal const int INVALID_CREDENTIALS = 42;
internal const int INVALID_USERNAME = 43;
internal const int INVALID_PASSWORD = 44;
+ internal const int PERSONAL_ACCESS_TOKEN_ALREADY_EXISTS = 51;
internal const int INVALID_PERSONAL_ACCESS_TOKEN = 53;
internal const int TRANSIENT_NOT_COMMITTED = 57;
internal const int TRANSIENT_NOT_ACCEPTED = 58;
@@ -59,11 +60,18 @@ internal static class VsrError
}
/// <summary>
- /// A verdict the server reported, on the reply status word, in the
committed result section or in an
- /// eviction frame. Only these may drive retry and failover.
+ /// A verdict the server reported, on the reply status word or in an
eviction frame. Retry and failover act
+ /// on server verdicts, these and <see cref="FromResultSection" />
ones, plus the TRANSIENT_NOT_ACCEPTED the
+ /// routed poll path raises itself when a primary does not admit it.
/// </summary>
internal static IggyInvalidStatusCodeException FromServer(int code, string
message)
{
return new IggyInvalidStatusCodeException(code, message, true);
}
+
+ /// <summary>A verdict the server reported in the result section of a
reply.</summary>
+ internal static IggyInvalidStatusCodeException FromResultSection(int code,
string message)
+ {
+ return new IggyInvalidStatusCodeException(code, message, true) {
FromResultSection = true };
+ }
}
diff --git a/foreign/csharp/Iggy_SDK/Vsr/VsrReplyDecoder.cs
b/foreign/csharp/Iggy_SDK/Vsr/VsrReplyDecoder.cs
index 72a5e109c..fa4ca6616 100644
--- a/foreign/csharp/Iggy_SDK/Vsr/VsrReplyDecoder.cs
+++ b/foreign/csharp/Iggy_SDK/Vsr/VsrReplyDecoder.cs
@@ -180,7 +180,8 @@ internal static class VsrReplyDecoder
if (code != 0)
{
- throw VsrError.FromServer((int)code.Value, $"Server rejected the
request with status {code.Value}.");
+ throw VsrError.FromResultSection((int)code.Value,
+ $"Server rejected the request with status {code.Value}.");
}
var payloadStart = ReadResultSectionLength(body.Span)
diff --git a/foreign/csharp/Iggy_SDK_Tests/VsrTests/EndpointFailoverTests.cs
b/foreign/csharp/Iggy_SDK_Tests/VsrTests/EndpointFailoverTests.cs
index 1d4aa5a14..be65af3ce 100644
--- a/foreign/csharp/Iggy_SDK_Tests/VsrTests/EndpointFailoverTests.cs
+++ b/foreign/csharp/Iggy_SDK_Tests/VsrTests/EndpointFailoverTests.cs
@@ -42,17 +42,23 @@ namespace Apache.Iggy.Tests.VsrTests;
/// </summary>
public sealed class EndpointFailoverTests
{
+ // An unmasked TRANSIENT_NOT_ACCEPTED replays the same id for the 2s
failover check anyway, so only more
+ // refusals than fit in it (50 x 50ms) tell masking apart. The other codes
are not replayed unmasked.
[Theory]
- [InlineData(VsrError.TRANSIENT_NOT_ACCEPTED)]
- [InlineData(VsrError.UNAUTHORIZED)]
- [InlineData(VsrError.UNAUTHENTICATED)]
- [InlineData(VsrError.STALE_CLIENT)]
- [InlineData(VsrError.NOT_PRIMARY)]
- [InlineData(VsrError.LIFECYCLE_BUSY)]
- [InlineData(VsrError.FEATURE_UNAVAILABLE)]
- public async Task
LaterRefusalsKeepTheIdentityOfAnUncertainMetadataWrite(int refusal)
+ [InlineData(VsrError.TRANSIENT_NOT_ACCEPTED, false, 50)]
+ [InlineData(VsrError.UNAUTHORIZED, false, 3)]
+ [InlineData(VsrError.UNAUTHENTICATED, false, 3)]
+ [InlineData(VsrError.STALE_CLIENT, false, 3)]
+ [InlineData(VsrError.NOT_PRIMARY, false, 3)]
+ [InlineData(VsrError.LIFECYCLE_BUSY, false, 3)]
+ [InlineData(VsrError.FEATURE_UNAVAILABLE, false, 3)]
+ [InlineData(VsrError.TRANSIENT_NOT_ACCEPTED, true, 50)]
+ [InlineData(VsrError.NOT_PRIMARY, true, 3)]
+ [InlineData(VsrError.UNAUTHENTICATED, true, 3)]
+ [InlineData(VsrError.HISTORY_UNAVAILABLE, true, 3)]
+ public async Task
LaterRefusalsKeepTheIdentityOfAnUncertainMetadataWrite(int refusal, bool
resultSection,
+ int refusalCount)
{
- const int RefusalCount = 50;
using var node = new MockNode();
var attempts = new ConcurrentQueue<MockRequest>();
node.Serve(request =>
@@ -63,27 +69,23 @@ public sealed class EndpointFailoverTests
var status = attempts.Count switch
{
1 => VsrError.TRANSIENT_NOT_COMMITTED,
- <= RefusalCount => refusal,
+ _ when attempts.Count <= refusalCount => refusal,
_ => 0
};
- return Reply(request.Operation, status == 0 ? new byte[4] :
[], (uint)status);
+ return status == 0 ? Reply(request.Operation, new byte[4])
+ : resultSection ? Reply(request.Operation,
VsrTestPayloads.Rejection(status))
+ : Reply(request.Operation, [], (uint)status);
}
return request.Code == GET_CLUSTER_METADATA_CODE
? Reply(OPERATION_NON_REPLICATED, ClusterMetadata(node.Port,
node.Port, node.Port))
: Answer(request);
});
- using var client = new TcpMessageStream(new IggyClientConfigurator
- {
- BaseAddress = $"127.0.0.1:{node.Port}",
- Protocol = Protocol.Tcp,
- HeartbeatInterval = TimeSpan.FromHours(1)
- }, NullLoggerFactory.Instance);
- await client.ConnectAsync(TestContext.Current.CancellationToken);
- await client.LoginUserAsync("iggy", "iggy",
TestContext.Current.CancellationToken);
+ using var client = await SignedInClientAsync(node);
+
await client.CreatePartitionsAsync(Identifier.Numeric(1),
Identifier.Numeric(1), 1,
TestContext.Current.CancellationToken);
- Assert.Equal(RefusalCount + 1, attempts.Count);
+ Assert.Equal(refusalCount + 1, attempts.Count);
var original = attempts.First();
Assert.All(attempts, replay =>
{
@@ -94,6 +96,124 @@ public sealed class EndpointFailoverTests
});
}
+ [Theory]
+ [InlineData(VsrError.REQUEST_TOO_OLD)]
+ [InlineData(VsrError.INVALID_COMMAND)]
+ public async Task
FinalRejectionOfAReplayReportsAnUncertainMetadataWriteAsUnknownAtOnce(int
rejection)
+ {
+ using var node = new MockNode();
+ var attempts = new ConcurrentQueue<MockRequest>();
+ node.Serve(request =>
+ {
+ if (request.Operation == (byte)VsrOperation.CreatePartitions)
+ {
+ attempts.Enqueue(request);
+ return Reply(request.Operation,
VsrTestPayloads.Rejection(attempts.Count == 1
+ ? VsrError.TRANSIENT_NOT_COMMITTED
+ : rejection));
+ }
+ return request.Code == GET_CLUSTER_METADATA_CODE
+ ? Reply(OPERATION_NON_REPLICATED, ClusterMetadata(node.Port,
node.Port, node.Port))
+ : Answer(request);
+ });
+ using var client = await SignedInClientAsync(node);
+
+ await Assert.ThrowsAsync<VsrRequestOutcomeUnknownException>(() =>
+ client.CreatePartitionsAsync(Identifier.Numeric(1),
Identifier.Numeric(1), 1,
+ TestContext.Current.CancellationToken));
+
+ var sent = attempts.ToArray();
+ Assert.Equal(2, sent.Length);
+ Assert.Equal(sent[0].RequestId, sent[1].RequestId);
+ }
+
+ [Fact]
+ public async Task
CommittedRejectionOfAReplaySettlesAnUncertainMetadataWrite()
+ {
+ const int rejectionCode = VsrError.TOPIC_ID_NOT_FOUND;
+ using var node = new MockNode();
+ var attempts = 0;
+ node.Serve(request =>
+ {
+ if (request.Operation == (byte)VsrOperation.CreatePartitions)
+ {
+ return Interlocked.Increment(ref attempts) == 1
+ ? Reply(request.Operation,
VsrTestPayloads.Rejection(VsrError.TRANSIENT_NOT_COMMITTED))
+ : Reply(request.Operation,
VsrTestPayloads.Rejection(rejectionCode));
+ }
+ return request.Code == GET_CLUSTER_METADATA_CODE
+ ? Reply(OPERATION_NON_REPLICATED, ClusterMetadata(node.Port,
node.Port, node.Port))
+ : Answer(request);
+ });
+ using var client = await SignedInClientAsync(node);
+
+ var rejection = await
Assert.ThrowsAsync<IggyInvalidStatusCodeException>(() =>
+ client.CreatePartitionsAsync(Identifier.Numeric(1),
Identifier.Numeric(1), 1,
+ TestContext.Current.CancellationToken));
+
+ Assert.Equal(rejectionCode, rejection.StatusCode);
+ Assert.Equal(2, Volatile.Read(ref attempts));
+ }
+
+ [Fact]
+ public async Task RefusedTokenReplaySettlesAnUncertainTokenCreation()
+ {
+ using var node = new MockNode();
+ var attempts = 0;
+ node.Serve(request =>
+ {
+ if (request.Operation ==
(byte)VsrOperation.CreatePersonalAccessToken)
+ {
+ return Interlocked.Increment(ref attempts) == 1
+ ? Reply(request.Operation,
VsrTestPayloads.Rejection(VsrError.TRANSIENT_NOT_COMMITTED))
+ : Reply(request.Operation,
+
VsrTestPayloads.Rejection(VsrError.PERSONAL_ACCESS_TOKEN_ALREADY_EXISTS));
+ }
+ return request.Code == GET_CLUSTER_METADATA_CODE
+ ? Reply(OPERATION_NON_REPLICATED, ClusterMetadata(node.Port,
node.Port, node.Port))
+ : Answer(request);
+ });
+ using var client = await SignedInClientAsync(node);
+
+ var rejection = await
Assert.ThrowsAsync<IggyInvalidStatusCodeException>(() =>
+ client.CreatePersonalAccessTokenAsync("token", token:
TestContext.Current.CancellationToken));
+
+ Assert.Equal(VsrError.PERSONAL_ACCESS_TOKEN_ALREADY_EXISTS,
rejection.StatusCode);
+ Assert.Equal(2, Volatile.Read(ref attempts));
+ }
+
+ [Fact]
+ public async Task
CommittedLifecycleBusyOfAReplayResendsAnUncertainMetadataWriteUnderANewId()
+ {
+ using var node = new MockNode();
+ var attempts = new ConcurrentQueue<MockRequest>();
+ node.Serve(request =>
+ {
+ if (request.Operation == (byte)VsrOperation.CreatePartitions)
+ {
+ attempts.Enqueue(request);
+ return attempts.Count switch
+ {
+ 1 => Reply(request.Operation,
VsrTestPayloads.Rejection(VsrError.TRANSIENT_NOT_COMMITTED)),
+ 2 => Reply(request.Operation,
VsrTestPayloads.Rejection(VsrError.LIFECYCLE_BUSY)),
+ _ => Reply(request.Operation, new byte[4])
+ };
+ }
+ return request.Code == GET_CLUSTER_METADATA_CODE
+ ? Reply(OPERATION_NON_REPLICATED, ClusterMetadata(node.Port,
node.Port, node.Port))
+ : Answer(request);
+ });
+ using var client = await SignedInClientAsync(node);
+
+ await client.CreatePartitionsAsync(Identifier.Numeric(1),
Identifier.Numeric(1), 1,
+ TestContext.Current.CancellationToken);
+
+ var sent = attempts.ToArray();
+ Assert.Equal(3, sent.Length);
+ Assert.Equal(sent[0].RequestId, sent[1].RequestId);
+ Assert.NotEqual(sent[0].RequestId, sent[2].RequestId);
+ }
+
[Theory]
[InlineData(false)]
[InlineData(true)]
@@ -353,6 +473,67 @@ public sealed class EndpointFailoverTests
"the arbitrary metadata redirect cap stopped the bounded roster
walk");
}
+ [Fact]
+ public async Task
ReturnsToANodeThatRefusedEarlierOnceTheWalkFindsNoOtherPrimary()
+ {
+ const uint commandCode = 60_042;
+ using var leader = new MockNode();
+ using var follower = new MockNode();
+ var roster = new[] { leader.Port, follower.Port };
+ var followerRefusals = 0;
+ var leaderAnswers = 0;
+
+ // The leader refuses until the walk has been to the follower, as a
primary does until it has
+ // materialised the partition.
+ leader.Serve(request =>
+ {
+ if (request.Code == GET_CLUSTER_METADATA_CODE)
+ {
+ return Reply(OPERATION_NON_REPLICATED,
RosterMetadata(leader.Port, roster));
+ }
+
+ if (request.Code != (int)commandCode)
+ {
+ return Answer(request);
+ }
+
+ if (Volatile.Read(ref followerRefusals) == 0)
+ {
+ return Reply(request.Operation, [], TRANSIENT_NOT_ACCEPTED);
+ }
+
+ Interlocked.Increment(ref leaderAnswers);
+ return Answer(request);
+ });
+ follower.Serve(request =>
+ {
+ if (request.Code != (int)commandCode)
+ {
+ return Answer(request);
+ }
+
+ Interlocked.Increment(ref followerRefusals);
+ return Reply(request.Operation, [], TRANSIENT_NOT_ACCEPTED);
+ });
+
+ using var client = new TcpMessageStream(new IggyClientConfigurator
+ {
+ BaseAddress = $"127.0.0.1:{leader.Port}",
+ Protocol = Protocol.Tcp,
+ ReconnectionSettings = new ReconnectionSettings
+ {
+ Enabled = true,
+ MaxRetries = 1,
+ InitialDelay = TimeSpan.FromMilliseconds(20)
+ }
+ }, NullLoggerFactory.Instance);
+ await client.ConnectAsync(TestContext.Current.CancellationToken);
+ await client.LoginUserAsync("iggy", "iggy",
TestContext.Current.CancellationToken);
+
+ Assert.Empty(await client.SendBinaryRequestAsync(commandCode, [],
TestContext.Current.CancellationToken));
+ Assert.Equal(1, Volatile.Read(ref leaderAnswers));
+ }
+
/// <summary>
/// Mirrors the integration contract (HeartbeatTests
/// EvictedClient_WithoutAutoLogin_Should_ReestablishItsSession): an
eviction comes off the server's
@@ -671,6 +852,19 @@ public sealed class EndpointFailoverTests
return frame;
}
+ private static async Task<TcpMessageStream> SignedInClientAsync(MockNode
node)
+ {
+ var client = new TcpMessageStream(new IggyClientConfigurator
+ {
+ BaseAddress = $"127.0.0.1:{node.Port}",
+ Protocol = Protocol.Tcp,
+ HeartbeatInterval = TimeSpan.FromHours(1)
+ }, NullLoggerFactory.Instance);
+ await client.ConnectAsync(TestContext.Current.CancellationToken);
+ await client.LoginUserAsync("iggy", "iggy",
TestContext.Current.CancellationToken);
+ return client;
+ }
+
[Fact]
public async Task FailsFastWhenNothingEverSignedIn()
{
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()
{
diff --git a/foreign/csharp/Iggy_SDK_Tests/VsrTests/VsrReplyDecoderTests.cs
b/foreign/csharp/Iggy_SDK_Tests/VsrTests/VsrReplyDecoderTests.cs
index c7788172c..e9a8aa5fb 100644
--- a/foreign/csharp/Iggy_SDK_Tests/VsrTests/VsrReplyDecoderTests.cs
+++ b/foreign/csharp/Iggy_SDK_Tests/VsrTests/VsrReplyDecoderTests.cs
@@ -52,12 +52,6 @@ public sealed class VsrReplyDecoderTests
return VsrTestPayloads.Concat(VsrTestPayloads.UInt32(0), payload);
}
- private static byte[] RejectionBody(uint code)
- {
- return VsrTestPayloads.Concat(VsrTestPayloads.UInt32(1),
VsrTestPayloads.UInt32(0),
- VsrTestPayloads.UInt32((int)code));
- }
-
[Fact]
public void Decode_StripsTheResultSectionFromACommittedMetadataReply()
{
@@ -72,7 +66,7 @@ public sealed class VsrReplyDecoderTests
[Fact]
public void Decode_CommittedRejectionThrowsTheTypedError()
{
- var body = RejectionBody(1009);
+ var body = VsrTestPayloads.Rejection(1009);
var exception = Assert.Throws<IggyInvalidStatusCodeException>(() =>
VsrReplyDecoder.Decode(ReplyHeader(VsrOperation.CreateStream,
body.Length), body));
@@ -284,7 +278,7 @@ public sealed class VsrReplyDecoderTests
public void ReadResultSectionLength_CoversTheCountAndItsEntries()
{
Assert.Equal(4, VsrReplyDecoder.ReadResultSectionLength(SuccessBody(1,
2)));
- Assert.Equal(12,
VsrReplyDecoder.ReadResultSectionLength(RejectionBody(1009)));
+ Assert.Equal(12,
VsrReplyDecoder.ReadResultSectionLength(VsrTestPayloads.Rejection(1009)));
Assert.Null(VsrReplyDecoder.ReadResultSectionLength(VsrTestPayloads.UInt32(1)));
// Shorter than the count itself, so there is no section to measure.
diff --git a/foreign/csharp/Iggy_SDK_Tests/VsrTests/VsrTestPayloads.cs
b/foreign/csharp/Iggy_SDK_Tests/VsrTests/VsrTestPayloads.cs
index c7c18d979..11d908402 100644
--- a/foreign/csharp/Iggy_SDK_Tests/VsrTests/VsrTestPayloads.cs
+++ b/foreign/csharp/Iggy_SDK_Tests/VsrTests/VsrTestPayloads.cs
@@ -81,6 +81,11 @@ internal static class VsrTestPayloads
return Concat(streamId, topicId, UInt32((int)partitionId),
UInt32((int)segmentsCount));
}
+ internal static byte[] Rejection(int code)
+ {
+ return Concat(UInt32(1), UInt32(0), UInt32(code));
+ }
+
internal static byte[] UInt32(int value)
{
var bytes = new byte[4];
