Skip to content

De-flake MNTR specs: stop inner waits from consuming their enclosing Within budget #8429

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
Aaronontheweb merged 8 commits into from
Jul 29, 2026
Merged

De-flake MNTR specs: stop inner waits from consuming their enclosing Within budget #8429

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
8 commits
Select commit Hold shift + click to select a range
718c7e4
Fix double-dilated timeouts in MNTR specs
Aaronontheweb Jul 24, 2026
32877f0
StressSpec: correct the timefactor/Within note
Aaronontheweb Jul 29, 2026
557cae0
De-flake ClusterShardingFailureSpec: keep barriers and assertions out…
Aaronontheweb Jul 29, 2026
4e5c282
StressSpec: fresh probe per attempt for the watchee lookup, and bound…
Aaronontheweb Jul 29, 2026
ab407c3
De-flake the async CircuitBreakerSpec tests: await the failing call
Aaronontheweb Jul 29, 2026
1f92ff0
Merge branch 'dev' into fix/ci-flake-batch-2
Aaronontheweb Jul 29, 2026
0cd21a1
StressSpec: bracket the aggregator resolution with a barrier
Aaronontheweb Jul 29, 2026
252053b
Merge remote-tracking branch 'akkadotnet/dev' into fix/ci-flake-batch-2
Aaronontheweb Jul 29, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
232 changes: 123 additions & 109 deletions src/contrib/cluster/Akka.Cluster.Sharding.Tests.MultiNode/ClusterShardingFailureSpec.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -170,136 +170,150 @@ public async Task ClusterSharding_with_flaky_journal_network_specs()

private async Task ClusterSharding_with_flaky_journal_network_must_join_cluster()
{
await WithinAsync(TimeSpan.FromSeconds(20), async () =>
{
await StartPersistenceIfNeededAsync(Config.Controller, CancellationToken.None, Config.First, Config.Second);
// No outer Within: EnterBarrierAsync derives its timeout from RemainingOr(barrier-timeout)
// (MultiNodeSpec.cs:603), which clamps to zero once an enclosing Within's deadline has
// passed. Each expectation below carries its own explicit bound instead, so the barrier
// always gets the full akka.testconductor.barrier-timeout (30s default) rather than
// whatever scraps a shared clock left over.
await StartPersistenceIfNeededAsync(Config.Controller, CancellationToken.None, Config.First, Config.Second);

await JoinAsync(Config.First, Config.First);
await JoinAsync(Config.Second, Config.First);
await JoinAsync(Config.First, Config.First);
await JoinAsync(Config.Second, Config.First);

await RunOnAsync(async () =>
{
var region = _region.Value;
region.Tell(new Add("10", 1));
region.Tell(new Add("20", 2));
region.Tell(new Add("21", 3));
region.Tell(new Get("10"));
await ExpectMsgAsync<Value>(v => v.Id == "10" && v.N == 1);
region.Tell(new Get("20"));
await ExpectMsgAsync<Value>(v => v.Id == "20" && v.N == 2);
region.Tell(new Get("21"));
await ExpectMsgAsync<Value>(v => v.Id == "21" && v.N == 3);
}, Config.First);
await EnterBarrierAsync("after-2");
});
await RunOnAsync(async () =>
{
var region = _region.Value;
region.Tell(new Add("10", 1));
region.Tell(new Add("20", 2));
region.Tell(new Add("21", 3));
region.Tell(new Get("10"));
await ExpectMsgAsync<Value>(v => v.Id == "10" && v.N == 1, TimeSpan.FromSeconds(5));
region.Tell(new Get("20"));
await ExpectMsgAsync<Value>(v => v.Id == "20" && v.N == 2, TimeSpan.FromSeconds(5));
region.Tell(new Get("21"));
await ExpectMsgAsync<Value>(v => v.Id == "21" && v.N == 3, TimeSpan.FromSeconds(5));
}, Config.First);
await EnterBarrierAsync("after-2");
}

private async Task ClusterSharding_with_flaky_journal_network_must_recover_after_journal_network_failure()
{
await WithinAsync(TimeSpan.FromSeconds(20), async () =>
// No outer Within here (see the join-cluster phase above for why barriers must stay
// outside one). The Get("21") retry below is the one operation in this phase that is
// *expected* to take multiple seconds: the coordinator/shard must ride out
// coordinator-failure-backoff / shard-failure-backoff (3s each,
// ClusterShardingFailureSpecConfig) against the shared journal's 5s ask-timeout
// (MemoryJournalShared, MultiNodeClusterShardingConfig.PersistenceConfig) before they can
// recover from the blackhole. It gets its own bound, sized off those constants, instead
// of borrowing from a shared clock that would starve the barriers and assertions after it.
await RunOnAsync(async () =>
{
await RunOnAsync(async () =>
if (PersistenceIsNeeded)
{
await TestConductor.BlackholeAsync(Config.Controller, Config.First, ThrottleTransportAdapter.Direction.Both);
await TestConductor.BlackholeAsync(Config.Controller, Config.Second, ThrottleTransportAdapter.Direction.Both);
}
else
{
if (PersistenceIsNeeded)
{
await TestConductor.BlackholeAsync(Config.Controller, Config.First, ThrottleTransportAdapter.Direction.Both);
await TestConductor.BlackholeAsync(Config.Controller, Config.Second, ThrottleTransportAdapter.Direction.Both);
}
else
{
await TestConductor.BlackholeAsync(Config.First, Config.Second, ThrottleTransportAdapter.Direction.Both);
}
}, Config.Controller);
await EnterBarrierAsync("journal-backholded");

await RunOnAsync(async () =>
await TestConductor.BlackholeAsync(Config.First, Config.Second, ThrottleTransportAdapter.Direction.Both);
}
}, Config.Controller);
await EnterBarrierAsync("journal-backholded");

await RunOnAsync(async () =>
{
// try with a new shard, will not reply until journal/network is available again
var region = _region.Value;
region.Tell(new Add("40", 4));
var probe = CreateTestProbe();
region.Tell(new Get("40"), probe.Ref);
await probe.ExpectNoMsgAsync(TimeSpan.FromSeconds(1));
}, Config.First);
await EnterBarrierAsync("first-delayed");

await RunOnAsync(async () =>
{
if (PersistenceIsNeeded)
{
// try with a new shard, will not reply until journal/network is available again
var region = _region.Value;
region.Tell(new Add("40", 4));
var probe = CreateTestProbe();
region.Tell(new Get("40"), probe.Ref);
await probe.ExpectNoMsgAsync(TimeSpan.FromSeconds(1));
}, Config.First);
await EnterBarrierAsync("first-delayed");

await RunOnAsync(async () =>
await TestConductor.PassThroughAsync(Config.Controller, Config.First, ThrottleTransportAdapter.Direction.Both);
await TestConductor.PassThroughAsync(Config.Controller, Config.Second, ThrottleTransportAdapter.Direction.Both);
}
else
{
if (PersistenceIsNeeded)
{
await TestConductor.PassThroughAsync(Config.Controller, Config.First, ThrottleTransportAdapter.Direction.Both);
await TestConductor.PassThroughAsync(Config.Controller, Config.Second, ThrottleTransportAdapter.Direction.Both);
}
else
{
await TestConductor.PassThroughAsync(Config.First, Config.Second, ThrottleTransportAdapter.Direction.Both);
}
}, Config.Controller);
await EnterBarrierAsync("journal-ok");

await RunOnAsync(async () =>
await TestConductor.PassThroughAsync(Config.First, Config.Second, ThrottleTransportAdapter.Direction.Both);
}
}, Config.Controller);
await EnterBarrierAsync("journal-ok");

await RunOnAsync(async () =>
{
var region = _region.Value;

// Confirm the ShardCoordinator/Shard have recovered and are routing again.
// A single-shot expectation can legitimately lose a race against the coordinator's
// and shard's own failure-backoff retry loop, so retry the request itself - bounded
// by roughly two worst-case retry cycles (5s journal timeout + 3s backoff each) -
// rather than gambling on one wait being long enough.
IActorRef entity21 = null;
await AwaitAssertAsync(async () =>
{
var region = _region.Value;
region.Tell(new Get("21"));
await ExpectMsgAsync<Value>(v => v.Id == "21" && v.N == 3);
var entity21 = LastSender;
var shard2 = Sys.ActorSelection(entity21.Path.Parent);

await ExpectMsgAsync<Value>(v => v.Id == "21" && v.N == 3, TimeSpan.FromSeconds(3));
entity21 = LastSender;
}, TimeSpan.FromSeconds(30));
var shard2 = Sys.ActorSelection(entity21.Path.Parent);

//Test the ShardCoordinator allocating shards after a journal/network failure
region.Tell(new Add("30", 3));
//Test the ShardCoordinator allocating shards after a journal/network failure
region.Tell(new Add("30", 3));

//Test the Shard starting entities and persisting after a journal/network failure
region.Tell(new Add("11", 1));
//Test the Shard starting entities and persisting after a journal/network failure
region.Tell(new Add("11", 1));

//Test the Shard passivate works after a journal failure
shard2.Tell(new Passivate(PoisonPill.Instance), entity21);
//Test the Shard passivate works after a journal failure
shard2.Tell(new Passivate(PoisonPill.Instance), entity21);

await AwaitAssertAsync(async () =>
{
// Note that the order between this Get message to 21 and the above Passivate to 21 is undefined.
// If this Get arrives first the reply will be Value("21", 3) and then it is retried by the
// awaitAssert.
// Also note that there is no timeout parameter on below expectMsg because messages should not
// be lost here. They should be buffered and delivered also after Passivate completed.
region.Tell(new Get("21"));
// counter reset to 0 when started again
await ExpectMsgAsync<Value>(v => v.Id == "21" && v.N == 0, hint: "Passivating did not reset Value down to 0");
});
await AwaitAssertAsync(async () =>
{
// Note that the order between this Get message to 21 and the above Passivate to 21 is undefined.
// If this Get arrives first the reply will be Value("21", 3) and then it is retried by the
// awaitAssert.
region.Tell(new Get("21"));
// counter reset to 0 when started again
await ExpectMsgAsync<Value>(v => v.Id == "21" && v.N == 0, TimeSpan.FromSeconds(3), hint: "Passivating did not reset Value down to 0");
}, TimeSpan.FromSeconds(10));

region.Tell(new Add("21", 1));
region.Tell(new Add("21", 1));

region.Tell(new Get("21"));
await ExpectMsgAsync<Value>(v => v.Id == "21" && v.N == 1);
region.Tell(new Get("21"));
await ExpectMsgAsync<Value>(v => v.Id == "21" && v.N == 1, TimeSpan.FromSeconds(5));

region.Tell(new Get("30"));
await ExpectMsgAsync<Value>(v => v.Id == "30" && v.N == 3);
region.Tell(new Get("30"));
await ExpectMsgAsync<Value>(v => v.Id == "30" && v.N == 3, TimeSpan.FromSeconds(5));

region.Tell(new Get("11"));
await ExpectMsgAsync<Value>(v => v.Id == "11" && v.N == 1);
region.Tell(new Get("11"));
await ExpectMsgAsync<Value>(v => v.Id == "11" && v.N == 1, TimeSpan.FromSeconds(5));

region.Tell(new Get("40"));
await ExpectMsgAsync<Value>(v => v.Id == "40" && v.N == 4);
}, Config.First);
await EnterBarrierAsync("verified-first");
region.Tell(new Get("40"));
await ExpectMsgAsync<Value>(v => v.Id == "40" && v.N == 4, TimeSpan.FromSeconds(5));
}, Config.First);
await EnterBarrierAsync("verified-first");

await RunOnAsync(async () =>
{
var region = _region.Value;
region.Tell(new Add("10", 1));
region.Tell(new Add("20", 2));
region.Tell(new Add("30", 3));
region.Tell(new Add("11", 4));
region.Tell(new Get("10"));
await ExpectMsgAsync<Value>(v => v.Id == "10" && v.N == 2);
region.Tell(new Get("11"));
await ExpectMsgAsync<Value>(v => v.Id == "11" && v.N == 5);
region.Tell(new Get("20"));
await ExpectMsgAsync<Value>(v => v.Id == "20" && v.N == 4);
region.Tell(new Get("30"));
await ExpectMsgAsync<Value>(v => v.Id == "30" && v.N == 6);
}, Config.Second);
await EnterBarrierAsync("after-3");
});
await RunOnAsync(async () =>
{
var region = _region.Value;
region.Tell(new Add("10", 1));
region.Tell(new Add("20", 2));
region.Tell(new Add("30", 3));
region.Tell(new Add("11", 4));
region.Tell(new Get("10"));
await ExpectMsgAsync<Value>(v => v.Id == "10" && v.N == 2, TimeSpan.FromSeconds(5));
region.Tell(new Get("11"));
await ExpectMsgAsync<Value>(v => v.Id == "11" && v.N == 5, TimeSpan.FromSeconds(5));
region.Tell(new Get("20"));
await ExpectMsgAsync<Value>(v => v.Id == "20" && v.N == 4, TimeSpan.FromSeconds(5));
region.Tell(new Get("30"));
await ExpectMsgAsync<Value>(v => v.Id == "30" && v.N == 6, TimeSpan.FromSeconds(5));
}, Config.Second);
await EnterBarrierAsync("after-3");
}
}
6 changes: 3 additions & 3 deletions src/contrib/cluster/Akka.Cluster.Sharding.Tests.MultiNode/ClusterShardingQueriesSpec.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -171,7 +171,7 @@ await AwaitAssertAsync(async () =>
var probe = CreateTestProbe();
var region = ClusterSharding.Get(Sys).ShardRegion(ShardTypeName);
region.Tell(new GetClusterShardingStats(Dilated(TimeSpan.FromSeconds(10))), probe.Ref);
var regions = (await probe.ExpectMsgAsync<ClusterShardingStats>(Dilated(TimeSpan.FromSeconds(10)))).Regions;
var regions = (await probe.ExpectMsgAsync<ClusterShardingStats>(TimeSpan.FromSeconds(10))).Regions;
regions.Count.Should().Be(3);
var timeouts = NumberOfShards / regions.Count;

Expand All @@ -197,7 +197,7 @@ await AwaitAssertAsync(async () =>
var probe = CreateTestProbe();
var region = ClusterSharding.Get(Sys).ShardRegion(ShardTypeName);
region.Tell(GetShardRegionState.Instance, probe.Ref);
var state = await probe.ExpectMsgAsync<CurrentShardRegionState>(Dilated(TimeSpan.FromSeconds(10)));
var state = await probe.ExpectMsgAsync<CurrentShardRegionState>(TimeSpan.FromSeconds(10));
state.Shards.Should().BeEmpty();
state.Failed.Should().HaveCount(2);
}, Dilated(TimeSpan.FromSeconds(30)), TimeSpan.FromSeconds(1));
Expand All @@ -215,7 +215,7 @@ await AwaitAssertAsync(async () =>
var probe = CreateTestProbe();
var region = ClusterSharding.Get(Sys).ShardRegion(ShardTypeName);
region.Tell(GetShardRegionState.Instance, probe.Ref);
var state = await probe.ExpectMsgAsync<CurrentShardRegionState>(Dilated(TimeSpan.FromSeconds(10)));
var state = await probe.ExpectMsgAsync<CurrentShardRegionState>(TimeSpan.FromSeconds(10));
state.Shards.Should().HaveCount(2);
state.Failed.Should().BeEmpty();
}, Dilated(TimeSpan.FromSeconds(30)), TimeSpan.FromSeconds(1));
Expand Down
Loading
Loading