Skip to content

Commit 7b2a131

Browse files
KSemenenkoKSemenenko
committed
test: await converged membership throughout silo recovery
Restart scenarios treated RoutingReady as membership convergence, allowing MCP initialization to route to a lost silo. Reuse ClusterMembershipReadiness for surviving members and complete rejoins, including joined recovery. Keep missing-copy refusals, original replay/data assertions and deadlines. Endpoint ports remain automatically allocated by Aspire. Validation: strict Debug ClusterTests build (0 warnings/errors), 10/10 real Aspire cluster cases, scoped format verification, repository governance, shard partition verification and git diff --check.
1 parent cfe8137 commit 7b2a131

6 files changed

Lines changed: 23 additions & 80 deletions

File tree

‎tests/KeyLoad.ClusterTests/EventStreams/Cases/AggregateReplay/AggregateReplayClusterSiloRestartTests.cs‎

Lines changed: 6 additions & 21 deletions
Original file line numberDiff line numberDiff line change
@@ -48,7 +48,8 @@ private async Task RunSiloRestartAsync(AggregateReplayClusterScenario scenario,
4848
stoppedNode = nodes[selectedIndex];
4949
await fixture.KillServerAsync(stoppedNode, AggregateReplayClusterTokens.FailureScenario, cancellationToken);
5050
var survivorIndex = Enumerable.Range(0, nodes.Length).First(index => index != selectedIndex);
51-
await WaitUntilReadyAsync(administrators[survivorIndex], cancellationToken);
51+
await ClusterMembershipReadiness.WaitForActiveSilosAsync(
52+
administrators.Where((_, index) => index != selectedIndex).ToArray(), cancellationToken);
5253
var frozen = scenario.FailoverCommand(Guid.NewGuid());
5354
await ClusterReplicationTestSupport.MissingCopiesAsync(await clients[survivorIndex].CommitAsync(frozen, cancellationToken));
5455

@@ -66,7 +67,7 @@ private async Task RunSiloRestartAsync(AggregateReplayClusterScenario scenario,
6667
}
6768
await fixture.App.ResourceNotifications.WaitForResourceHealthyAsync(stoppedNode,
6869
WaitBehavior.WaitOnResourceUnavailable, cancellationToken);
69-
await WaitForClusterReadyAsync(nodes, administrators, cancellationToken);
70+
await ClusterMembershipReadiness.WaitForActiveSilosAsync(administrators, cancellationToken);
7071
var receipt = ClusterReplicationTestSupport.Success(await ClusterReplicationTestSupport.RetryDuringSiloRecoveryAsync(
7172
() => clients[survivorIndex].CommitAsync(frozen, cancellationToken), cancellationToken));
7273
await Assert.That(receipt.Durability).IsEqualTo(DurabilityProfile.ProcessDurable);
@@ -92,7 +93,7 @@ await SubmissionColdAssertions.ReceiptAsync(ClusterReplicationTestSupport.Succes
9293
if (stoppedNode is not null && !restarted)
9394
{
9495
using var recovery = new CancellationTokenSource(TimeSpan.FromSeconds(45), TimeProvider.System);
95-
await RestoreSiloAsync(stoppedNode, nodes, administrators, originalFailure, recovery.Token);
96+
await RestoreSiloAsync(stoppedNode, administrators, originalFailure, recovery.Token);
9697
}
9798
}
9899
}
@@ -144,7 +145,7 @@ await Assert.That(sdkPage.Snapshot!.SnapshotVersion)
144145
}
145146
}
146147

147-
private async Task RestoreSiloAsync(string stoppedNode, string[] nodes, KeyLoadClient[] administrators,
148+
private async Task RestoreSiloAsync(string stoppedNode, KeyLoadClient[] administrators,
148149
Exception? originalFailure, CancellationToken cancellationToken)
149150
{
150151
try
@@ -153,7 +154,7 @@ private async Task RestoreSiloAsync(string stoppedNode, string[] nodes, KeyLoadC
153154
await fixture.App.ResourceNotifications.WaitForResourceHealthyAsync(stoppedNode,
154155
WaitBehavior.WaitOnResourceUnavailable, cancellationToken);
155156
await Assert.That(Directory.Exists(Path.Combine(fixture.Root, stoppedNode))).IsTrue();
156-
await WaitForClusterReadyAsync(nodes, administrators, cancellationToken);
157+
await ClusterMembershipReadiness.WaitForActiveSilosAsync(administrators, cancellationToken);
157158
}
158159
catch (Exception recoveryFailure) when (ClusterReplicationTestSupport.IsNonFatalCleanupFailure(recoveryFailure))
159160
{
@@ -166,20 +167,4 @@ await fixture.App.ResourceNotifications.WaitForResourceHealthyAsync(stoppedNode,
166167
}
167168
}
168169

169-
private static async Task WaitUntilReadyAsync(KeyLoadClient client, CancellationToken cancellationToken)
170-
=> await ClusterReplicationTestSupport.EventuallyAsync(async () =>
171-
{
172-
var status = await client.StatusAsync(cancellationToken);
173-
return status.IsSuccess && status.Value.RoutingReady;
174-
}, cancellationToken);
175-
176-
private static async Task WaitForClusterReadyAsync(string[] nodes, KeyLoadClient[] clients,
177-
CancellationToken cancellationToken)
178-
{
179-
foreach (var index in Enumerable.Range(0, nodes.Length))
180-
{
181-
await WaitUntilReadyAsync(clients[index], cancellationToken);
182-
}
183-
}
184-
185170
}

‎tests/KeyLoad.ClusterTests/QueryExecution/Scenarios/Distribution/PartitionQueryClusterSiloRestart.cs‎

Lines changed: 3 additions & 20 deletions
Original file line numberDiff line numberDiff line change
@@ -59,11 +59,6 @@ private static async Task VerifySurvivorReadWriteAsync(ClusterFixture fixture,
5959
.First(node => !string.Equals(node, selectedSilo, StringComparison.Ordinal));
6060
using var http = McpCallerHttp.Create(fixture, survivor);
6161
var sdk = new KeyLoadClient(http, fixture.AdminKey, IntegrationClientOptions.Execution());
62-
await ClusterReplicationTestSupport.EventuallyAsync(async () =>
63-
{
64-
var status = await sdk.StatusAsync(cancellationToken).ConfigureAwait(false);
65-
return status.IsSuccess && status.Value.RoutingReady;
66-
}, cancellationToken).ConfigureAwait(false);
6762
var target = scenario.Partitions[1];
6863
var row = new PartitionQueryClusterInputRow(target, AddedId, 0, AddedValue);
6964
var payload = System.Text.Json.JsonSerializer.Serialize(new Dictionary<string, object?>(StringComparer.Ordinal)
@@ -100,7 +95,9 @@ await ServerFailureObserver.ObserveAsync(async () =>
10095
{
10196
await fixture.App.ResourceNotifications.WaitForResourceHealthyAsync(selectedSilo,
10297
WaitBehavior.WaitOnResourceUnavailable, recovery.Token).ConfigureAwait(false);
103-
await VerifyAllNodesReadyAsync(fixture, recovery.Token).ConfigureAwait(false);
98+
await ClusterMembershipReadiness.WaitForActiveSilosAsync(fixture,
99+
[McpCallerProtocol.Node1, McpCallerProtocol.Node2, McpCallerProtocol.Node3], recovery.Token)
100+
.ConfigureAwait(false);
104101
}, failures).ConfigureAwait(false);
105102
}
106103
if (failures.Count == 0)
@@ -115,18 +112,4 @@ await fixture.App.ResourceNotifications.WaitForResourceHealthyAsync(selectedSilo
115112
throw new AggregateException("Partition query silo-restart cleanup failed.", failures);
116113
}
117114

118-
private static async Task VerifyAllNodesReadyAsync(ClusterFixture fixture,
119-
CancellationToken cancellationToken)
120-
{
121-
foreach (var node in new[] { McpCallerProtocol.Node1, McpCallerProtocol.Node2, McpCallerProtocol.Node3 })
122-
{
123-
using var http = McpCallerHttp.Create(fixture, node);
124-
var sdk = new KeyLoadClient(http, fixture.AdminKey, IntegrationClientOptions.Execution());
125-
await ClusterReplicationTestSupport.EventuallyAsync(async () =>
126-
{
127-
var status = await sdk.StatusAsync(cancellationToken).ConfigureAwait(false);
128-
return status.IsSuccess && status.Value.RoutingReady;
129-
}, cancellationToken).ConfigureAwait(false);
130-
}
131-
}
132115
}

‎tests/KeyLoad.ClusterTests/Search/Scenarios/Projections/EventProjectionClusterSiloRestart.cs‎

Lines changed: 3 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -22,7 +22,8 @@ internal async Task RunAsync(EventProjectionClusterScenario scenario, EventProje
2222
await fixture.KillServerAsync(stoppedNode, EventProjectionClusterScenario.FailureScenario,
2323
cancellationToken);
2424
var survivor = Enumerable.Range(0, nodes.Length).First(index => index != selectedIndex);
25-
await WaitReadyAsync(clients.Administrators[survivor], cancellationToken);
25+
await ClusterMembershipReadiness.WaitForActiveSilosAsync(
26+
clients.Administrators.Where((_, index) => index != selectedIndex).ToArray(), cancellationToken);
2627
await ReplayAndCheckSurvivorAsync(scenario, clients, survivor, nodes[survivor],
2728
cancellationToken);
2829
await RestartAndCheckClusterAsync(stoppedNode, nodes, clients.Administrators, cancellationToken);
@@ -102,21 +103,14 @@ private static async Task<int> SelectSiloIndexAsync(string[] nodes, KeyLoadClien
102103
private async Task WaitForClusterAsync(string[] nodes, KeyLoadClient[] administrators,
103104
CancellationToken cancellationToken)
104105
{
106+
await ClusterMembershipReadiness.WaitForActiveSilosAsync(administrators, cancellationToken);
105107
foreach (var index in Enumerable.Range(0, nodes.Length))
106108
{
107-
await WaitReadyAsync(administrators[index], cancellationToken);
108109
await fixture.App.ResourceNotifications.WaitForResourceHealthyAsync(nodes[index],
109110
WaitBehavior.WaitOnResourceUnavailable, cancellationToken);
110111
}
111112
}
112113

113-
private static async Task WaitReadyAsync(KeyLoadClient administrator, CancellationToken cancellationToken)
114-
=> await ClusterReplicationTestSupport.EventuallyAsync(async () =>
115-
{
116-
var status = await administrator.StatusAsync(cancellationToken);
117-
return status.IsSuccess && status.Value.RoutingReady;
118-
}, cancellationToken);
119-
120114
private async Task RestoreStoppedNodeAsync(string stoppedNode, string[] nodes,
121115
KeyLoadClient[] administrators, Exception? primaryFailure)
122116
{

‎tests/KeyLoad.ClusterTests/StorageReplication/Fixtures/SiloRestartRecoveryScenario.cs‎

Lines changed: 1 addition & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -33,8 +33,7 @@ internal static async Task VerifyUnavailableCopiesAndRecoveryAsync(ClusterFixtur
3333
await Task.WhenAll(nodes.Select(node => fixture.RestartServerAsync(node, cancellationToken)));
3434
stoppedProcesses.Clear();
3535

36-
await ClusterReplicationTestSupport.EventuallyAsync(
37-
async () => (await surviving.StatusAsync(cancellationToken)).IsSuccess, cancellationToken);
36+
await ClusterMembershipReadiness.WaitForActiveSilosAsync(state.Clients, cancellationToken);
3837
var restored = ClusterReplicationTestSupport.Success(await ClusterReplicationTestSupport.RetryDuringSiloRecoveryAsync(
3938
() => surviving.CommitAsync(pending, cancellationToken), cancellationToken));
4039
await Assert.That(restored.Durability).IsEqualTo(DurabilityProfile.ProcessDurable);
@@ -68,11 +67,6 @@ private static async Task VerifyAllNodesReadyAsync(ClusterFixture fixture, SiloR
6867
{
6968
foreach (var index in Enumerable.Range(0, NodeCount))
7069
{
71-
await ClusterReplicationTestSupport.EventuallyAsync(async () =>
72-
{
73-
var current = await state.Clients[index].StatusAsync(cancellationToken);
74-
return current.IsSuccess && current.Value.RoutingReady;
75-
}, cancellationToken);
7670
await fixture.App.ResourceNotifications.WaitForResourceHealthyAsync(
7771
ClusterReplicationTestSupport.NodeName(index + 1), WaitBehavior.WaitOnResourceUnavailable, cancellationToken);
7872
await Assert.That(ClusterReplicationTestSupport.Success(

‎tests/KeyLoad.ClusterTests/StorageReplication/Scenarios/Faults/SiloRestartScenario.cs‎

Lines changed: 9 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -100,12 +100,14 @@ await fixture.KillServerAsync(selectedNode,
100100
SiloFailure, cancellationToken);
101101
var survivors = Enumerable.Range(0, NodeCount).Where(index => index != selectedIndex).ToArray();
102102
var surviving = state.Clients[survivors[0]];
103-
await EventuallyAsync(async () => (await surviving.StatusAsync(cancellationToken)).IsSuccess, cancellationToken);
103+
await ClusterMembershipReadiness.WaitForActiveSilosAsync(
104+
survivors.Select(index => state.Clients[index]).ToArray(), cancellationToken);
104105
await VerifyInitialReplicasAsync(survivors.Select(index => state.Clients[index]).ToArray(),
105106
state.Partition, state.Subscription, cancellationToken);
106107
await MissingCopiesAsync(await surviving.CommitAsync(state.Command, cancellationToken));
107108
await fixture.RestartServerAsync(selectedNode, cancellationToken);
108109
await fixture.App.ResourceNotifications.WaitForResourceHealthyAsync(selectedNode, cancellationToken);
110+
await ClusterMembershipReadiness.WaitForActiveSilosAsync(state.Clients, cancellationToken);
109111
stoppedProcesses.Remove(selectedNode);
110112
return state with
111113
{
@@ -133,20 +135,13 @@ private static async Task RestoreFailedClusterAsync(ClusterFixture fixture, KeyL
133135
{
134136
return;
135137
}
136-
foreach (var client in clients)
138+
try
137139
{
138-
try
139-
{
140-
await EventuallyAsync(async () =>
141-
{
142-
var status = await client.StatusAsync(recoveryTimeout.Token);
143-
return status.IsSuccess && status.Value.RoutingReady;
144-
}, recoveryTimeout.Token);
145-
}
146-
catch (Exception readinessFailure) when (IsNonFatalCleanupFailure(readinessFailure))
147-
{
148-
failure.Data[ClusterReadinessFailureKey] = readinessFailure;
149-
}
140+
await ClusterMembershipReadiness.WaitForActiveSilosAsync(clients, recoveryTimeout.Token);
141+
}
142+
catch (Exception readinessFailure) when (IsNonFatalCleanupFailure(readinessFailure))
143+
{
144+
failure.Data[ClusterReadinessFailureKey] = readinessFailure;
150145
}
151146
}
152147

‎tests/KeyLoad.ClusterTests/TimeSeries/Cases/TimeSeriesClusterReplicaRestartTests.cs‎

Lines changed: 1 addition & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -154,21 +154,13 @@ private async Task RestoreSiloAsync(string node, KeyLoadClient[] clients, Except
154154
try
155155
{
156156
await fixture.RestartServerAsync(node, recovery.Token);
157-
var index = NodeIndex(node);
158-
await ClusterReplicationTestSupport.EventuallyAsync(async () =>
159-
{
160-
var status = await clients[index].StatusAsync(recovery.Token);
161-
return status.IsSuccess && status.Value.RoutingReady;
162-
}, recovery.Token);
157+
await ClusterMembershipReadiness.WaitForActiveSilosAsync(clients, recovery.Token);
163158
}
164159
catch (Exception recoveryFailure) when (ClusterReplicationTestSupport.IsNonFatalCleanupFailure(recoveryFailure))
165160
{
166161
failure.Data[RestartFailureKey] = recoveryFailure;
167162
}
168163
}
169164

170-
private static int NodeIndex(string node) => int.Parse(node.AsSpan("node".Length),
171-
System.Globalization.CultureInfo.InvariantCulture) - FirstNode;
172-
173165
private static string NodeName(int number) => "node" + number.ToString(System.Globalization.CultureInfo.InvariantCulture);
174166
}

0 commit comments

Comments
 (0)