diff --git a/src/EventStore.Core.Tests/Integration/grpc_request_forwarding_survives_leader_change.cs b/src/EventStore.Core.Tests/Integration/grpc_request_forwarding_survives_leader_change.cs index ee8429106..83c70d3b5 100644 --- a/src/EventStore.Core.Tests/Integration/grpc_request_forwarding_survives_leader_change.cs +++ b/src/EventStore.Core.Tests/Integration/grpc_request_forwarding_survives_leader_change.cs @@ -1,4 +1,5 @@ using System; +using System.Collections.Generic; using System.Diagnostics; using System.Linq; using System.Net; @@ -25,14 +26,14 @@ public class grpc_request_forwarding_survives_leader_change node is not null)) + { + await AssertRevisions(node.HttpEndPoint, [0, 1, 2], scenario); + } + + var restartedLeaderIndex = initialLeader.DebugIndex; + var restartedLeader = CreateNode( + restartedLeaderIndex, + _nodeEndpoints[restartedLeaderIndex], + _nodeEndpoints.Where((_, index) => index != restartedLeaderIndex) + .Select(endpoints => (EndPoint)endpoints.ClusterEndPoint) + .ToArray()); + _nodes[restartedLeaderIndex] = restartedLeader; + restartedLeader.Start(); + + AssertEx.IsOrBecomesTrue( + () => + _nodes.Count(node => node.NodeState == VNodeState.Leader) == 1 && + _nodes.Count(node => node.NodeState == VNodeState.Follower) == 2, + RemainingScenarioTime(scenario), + "The reconnected node did not rejoin the cluster", + MiniNodeLogging.WriteLogs); + + foreach (var node in _nodes) + { + await AssertRevisions(node.HttpEndPoint, [0, 1, 2], scenario); + } + } + + private static async Task AssertRevisions(IPEndPoint endpoint, ulong[] expected, Stopwatch scenario) + { + while (true) + { + try + { + var actual = await ReadRevisions(endpoint, RemainingScenarioTime(scenario)); + if (actual.Count >= expected.Length) + { + Assert.That(actual, Is.EqualTo(expected), $"Replication differed at {endpoint}"); + return; + } + } + catch (RpcException ex) when ( + ex.StatusCode is StatusCode.Unavailable or StatusCode.DeadlineExceeded && + scenario.Elapsed < ScenarioTimeout) + { + } + + await Task.Delay(AuthenticationRetryDelay); + } + } + + private static async Task> ReadRevisions(IPEndPoint endpoint, TimeSpan remainingScenarioTime) + { + using var handler = new SocketsHttpHandler + { + SslOptions = + { + RemoteCertificateValidationCallback = delegate { return true; } + } + }; + using var httpClient = new HttpClient(handler); + using var channel = GrpcChannel.ForAddress( + new Uri($"https://{endpoint}"), + new GrpcChannelOptions { HttpClient = httpClient }); + var client = new Streams.StreamsClient(channel); + using var call = client.Read(new ReadReq + { + Options = new ReadReq.Types.Options + { + Stream = new ReadReq.Types.Options.Types.StreamOptions + { + StreamIdentifier = new StreamIdentifier + { + StreamName = ByteString.CopyFromUtf8(Stream) + }, + Start = new Empty() + }, + ReadDirection = ReadReq.Types.Options.Types.ReadDirection.Forwards, + Count = 3, + NoFilter = new Empty(), + UuidOption = new ReadReq.Types.Options.Types.UUIDOption { Structured = new Empty() } + } + }, new CallOptions( + credentials: CallCredentials.FromInterceptor((_, metadata) => + { + metadata.Add("authorization", AuthorizationHeaderValue); + return Task.CompletedTask; + }), + deadline: DateTime.UtcNow.Add(remainingScenarioTime < RequestTimeout + ? remainingScenarioTime + : RequestTimeout))); + + var revisions = new List(); + await foreach (var response in call.ResponseStream.ReadAllAsync()) + { + if (response.Event is { } readEvent) + { + revisions.Add(readEvent.Event.StreamRevision); + } + } + return revisions; } private static async Task Append(