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 3903224fc611ebacdb927189d634752c7c9f7d32 Author: Łukasz Zborek <[email protected]> AuthorDate: Wed Oct 14 21:08:11 2026 +0200 fix(csharp): flaky tests --- .../Fixtures/VsrCluster.cs | 9 + .../Helpers/ConsumerGroupAssignment.cs | 4 +- .../Vsr/VsrMetadataTests.cs | 3 +- .../Exceptions/IggyInvalidStatusCodeException.cs | 5 + .../TcpMessageStream.PollRouting.cs | 16 +- .../Implementations/TcpMessageStream.Vsr.cs | 22 +- foreign/csharp/Iggy_SDK/Vsr/VsrConnection.cs | 52 ++++- foreign/csharp/Iggy_SDK/Vsr/VsrError.cs | 11 +- foreign/csharp/Iggy_SDK/Vsr/VsrReplyDecoder.cs | 3 +- .../VsrTests/EndpointFailoverTests.cs | 234 +++++++++++++++++++-- .../VsrTests/VsrReplyDecoderTests.cs | 10 +- .../Iggy_SDK_Tests/VsrTests/VsrTestPayloads.cs | 5 + 12 files changed, 332 insertions(+), 42 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 d22766529..8b4c94cf9 100644 --- a/foreign/csharp/Iggy_SDK/IggyClient/Implementations/TcpMessageStream.PollRouting.cs +++ b/foreign/csharp/Iggy_SDK/IggyClient/Implementations/TcpMessageStream.PollRouting.cs @@ -191,9 +191,9 @@ public sealed partial class TcpMessageStream } generation ??= route.Generation; frame ??= _consensusSession.Resolve(VsrOperations.ForCode(code)); + refusal = null; 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); @@ -208,7 +208,6 @@ 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); } @@ -238,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; @@ -263,6 +266,15 @@ public sealed partial class TcpMessageStream 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. A kept refusal means no send started since, so it still decides the outcome. + catch (IOException) when (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..4abe4feb5 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,41 @@ 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. + /// </summary> + private static bool IsFinalRefusal(IggyInvalidStatusCodeException error) + { + return error is + { + FromResultSection: 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..03e6223d9 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,17 @@ 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. Only server verdicts, + /// these and <see cref="FromResultSection" /> ones, may drive retry and failover. /// </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/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];
