From 718c7e453a9301a235ac50aeaf0f09e67a266ec0 Mon Sep 17 00:00:00 2001 From: Aaron Stannard Date: Fri, 24 Jul 2026 21:30:41 +0000 Subject: [PATCH 1/6] Fix double-dilated timeouts in MNTR specs MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit TestKit dilates any explicitly-passed duration: RemainingOrDilated(duration) returns Dilated(duration.Value) (TestKitBase.cs:516-521). Four call sites pre-dilated the value themselves, so it was dilated twice. StressSpec.cs:842 is the clearest case — 'ExpectMsgAsync(Dilated(3s))' with akka.test.timefactor = 3 became a 27s wait (3 -> 9 -> 27), matching the 'Timeout 00:00:27 while waiting for a message of type ActorIdentity' failures seen in CI (Azure build 129447). That is self-defeating for a retry helper: ClusterResultAggregatorAsync retries inside a 90s budget, so 27s attempts allow ~3 tries where 9s attempts allow ~10. Pass the raw TimeSpan and let TestKit dilate once. No timeout is widened — the effective waits get shorter and the retry density under load goes up. Sites: StressSpec (ActorIdentity), ClusterShardingQueriesSpec (ClusterShardingStats, CurrentShardRegionState x2). --- .../ClusterShardingQueriesSpec.cs | 6 +++--- src/core/Akka.Cluster.Tests.MultiNode/StressSpec.cs | 2 +- 2 files changed, 4 insertions(+), 4 deletions(-) diff --git a/src/contrib/cluster/Akka.Cluster.Sharding.Tests.MultiNode/ClusterShardingQueriesSpec.cs b/src/contrib/cluster/Akka.Cluster.Sharding.Tests.MultiNode/ClusterShardingQueriesSpec.cs index 584c5d7222e..b1fbccff68f 100644 --- a/src/contrib/cluster/Akka.Cluster.Sharding.Tests.MultiNode/ClusterShardingQueriesSpec.cs +++ b/src/contrib/cluster/Akka.Cluster.Sharding.Tests.MultiNode/ClusterShardingQueriesSpec.cs @@ -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(Dilated(TimeSpan.FromSeconds(10)))).Regions; + var regions = (await probe.ExpectMsgAsync(TimeSpan.FromSeconds(10))).Regions; regions.Count.Should().Be(3); var timeouts = NumberOfShards / regions.Count; @@ -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(Dilated(TimeSpan.FromSeconds(10))); + var state = await probe.ExpectMsgAsync(TimeSpan.FromSeconds(10)); state.Shards.Should().BeEmpty(); state.Failed.Should().HaveCount(2); }, Dilated(TimeSpan.FromSeconds(30)), TimeSpan.FromSeconds(1)); @@ -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(Dilated(TimeSpan.FromSeconds(10))); + var state = await probe.ExpectMsgAsync(TimeSpan.FromSeconds(10)); state.Shards.Should().HaveCount(2); state.Failed.Should().BeEmpty(); }, Dilated(TimeSpan.FromSeconds(30)), TimeSpan.FromSeconds(1)); diff --git a/src/core/Akka.Cluster.Tests.MultiNode/StressSpec.cs b/src/core/Akka.Cluster.Tests.MultiNode/StressSpec.cs index f271024ac15..c0a795bef61 100644 --- a/src/core/Akka.Cluster.Tests.MultiNode/StressSpec.cs +++ b/src/core/Akka.Cluster.Tests.MultiNode/StressSpec.cs @@ -839,7 +839,7 @@ await AwaitAssertAsync(async () => var probe = CreateTestProbe(); Sys.ActorSelection(new RootActorPath(GetAddress(Roles.First())) / "user" / ("result" + Step)) .Tell(new Identify(Step), probe.Ref); - aggregatorRef = (await probe.ExpectMsgAsync(Dilated(TimeSpan.FromSeconds(3)))).Subject; + aggregatorRef = (await probe.ExpectMsgAsync(TimeSpan.FromSeconds(3))).Subject; }, TimeSpan.FromSeconds(30), TimeSpan.FromSeconds(1)); return Option.Create(aggregatorRef); } From 32877f0b4db3d941b89da5febde708c60e84701b Mon Sep 17 00:00:00 2001 From: Aaron Stannard Date: Wed, 29 Jul 2026 11:22:18 +0000 Subject: [PATCH 2/6] StressSpec: correct the timefactor/Within note The config comment claimed Within(max) is not scaled by akka.test.timefactor, and said the raw Within bounds were therefore 'addressed per-site'. That is wrong: TestKitBase.WithinAsync applies Dilated(max) to the bound (TestKitBase_Within.cs:261) and every Within/WithinAsync overload routes through that core, as the XML docs on each overload state. Believing otherwise invites two mistakes this suite has already made: pre-multiplying a Within bound to 'compensate', and pre-Dilating a duration handed to a TestKit expect (which squares the factor - see the ActorIdentity fix in this branch). Replace the note with what the framework actually does. --- src/core/Akka.Cluster.Tests.MultiNode/StressSpec.cs | 9 ++++++--- 1 file changed, 6 insertions(+), 3 deletions(-) diff --git a/src/core/Akka.Cluster.Tests.MultiNode/StressSpec.cs b/src/core/Akka.Cluster.Tests.MultiNode/StressSpec.cs index c0a795bef61..c3ad0159894 100644 --- a/src/core/Akka.Cluster.Tests.MultiNode/StressSpec.cs +++ b/src/core/Akka.Cluster.Tests.MultiNode/StressSpec.cs @@ -91,9 +91,12 @@ public StressSpecConfig() akka.loglevel = INFO akka.remote.log-remote-lifecycle-events = off -# NOTE: Akka.NET's Within(max) does NOT dilate by timefactor (unlike JVM/Pekko), so these -# only reach the Dilated/single-expect paths -- notably the per-phase aggregator Identify -- -# while the raw Within bounds are addressed per-site. +# NOTE: timefactor scales Within(max) too -- TestKitBase.WithinAsync applies Dilated(max) to the +# bound, and every Within/WithinAsync overload routes through it. So a phase written as Within(30s) +# actually gets 30s * timefactor, and inner waits that inherit RemainingOrDefault are bounded by +# whatever that dilated phase budget has left. Do NOT pre-multiply a Within bound (or pre-Dilate a +# duration handed to a TestKit expect) to compensate -- it is already scaled once, and doing it +# twice squares the factor. akka.test.single-expect-default = 10s akka.test.timefactor = 3 akka.actor.default-dispatcher = { From 557cae02ea9a561d4befe53db113d4b527be70b8 Mon Sep 17 00:00:00 2001 From: Aaron Stannard Date: Wed, 29 Jul 2026 11:49:46 +0000 Subject: [PATCH 3/6] De-flake ClusterShardingFailureSpec: keep barriers and assertions out of one shared Within Both phases wrapped everything in a single WithinAsync(20s): 5 barriers, a deliberate 1s ExpectNoMsg, a journal blackhole/restore cycle, 8 ExpectMsg calls and 2 AwaitAssert loops across three roles. Every wait that omits an explicit duration resolves through RemainingOrDefault, which returns the enclosing Within's remainder clamped at zero - and EnterBarrierAsync does the same (MultiNodeSpec.cs:603, RemainingOr(barrier-timeout)), so barriers silently traded their 30s default for whatever scraps were left. Once the shared budget was consumed - by the deliberate wait, the blackhole round trips, and the coordinator/shard riding out coordinator-failure-backoff/shard-failure-backoff (3s) against the journal's 5s ask-timeout - every later expectation and barrier in the method inherited near-zero time. That is the CI signature exactly: 'Timeout 00:00:00' on second, a small insufficient remainder on first, and a barrier TimeoutException on controller (Azure build 129447). Shrinking the original Within to 4s reproduces that same three-node signature deterministically, confirming the mechanism. Remove both outer Within wrappers so barriers get their real timeout back, give each ordinary expectation its own explicit bound, and make the one recovery-critical check (the first Get after journal-ok) a retrying AwaitAssert bounded by the spec's own backoff constants rather than a single-shot expectation racing an ambient clock. Same fix shape as f13a8fa01 for this bug class. Verified: both variants pass, including under sustained CPU saturation. --- .../ClusterShardingFailureSpec.cs | 232 ++++++++++-------- 1 file changed, 123 insertions(+), 109 deletions(-) diff --git a/src/contrib/cluster/Akka.Cluster.Sharding.Tests.MultiNode/ClusterShardingFailureSpec.cs b/src/contrib/cluster/Akka.Cluster.Sharding.Tests.MultiNode/ClusterShardingFailureSpec.cs index 86e4a1b38f7..ef9b51362f1 100644 --- a/src/contrib/cluster/Akka.Cluster.Sharding.Tests.MultiNode/ClusterShardingFailureSpec.cs +++ b/src/contrib/cluster/Akka.Cluster.Sharding.Tests.MultiNode/ClusterShardingFailureSpec.cs @@ -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(v => v.Id == "10" && v.N == 1); - region.Tell(new Get("20")); - await ExpectMsgAsync(v => v.Id == "20" && v.N == 2); - region.Tell(new Get("21")); - await ExpectMsgAsync(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(v => v.Id == "10" && v.N == 1, TimeSpan.FromSeconds(5)); + region.Tell(new Get("20")); + await ExpectMsgAsync(v => v.Id == "20" && v.N == 2, TimeSpan.FromSeconds(5)); + region.Tell(new Get("21")); + await ExpectMsgAsync(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(v => v.Id == "21" && v.N == 3); - var entity21 = LastSender; - var shard2 = Sys.ActorSelection(entity21.Path.Parent); - + await ExpectMsgAsync(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(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(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(v => v.Id == "21" && v.N == 1); + region.Tell(new Get("21")); + await ExpectMsgAsync(v => v.Id == "21" && v.N == 1, TimeSpan.FromSeconds(5)); - region.Tell(new Get("30")); - await ExpectMsgAsync(v => v.Id == "30" && v.N == 3); + region.Tell(new Get("30")); + await ExpectMsgAsync(v => v.Id == "30" && v.N == 3, TimeSpan.FromSeconds(5)); - region.Tell(new Get("11")); - await ExpectMsgAsync(v => v.Id == "11" && v.N == 1); + region.Tell(new Get("11")); + await ExpectMsgAsync(v => v.Id == "11" && v.N == 1, TimeSpan.FromSeconds(5)); - region.Tell(new Get("40")); - await ExpectMsgAsync(v => v.Id == "40" && v.N == 4); - }, Config.First); - await EnterBarrierAsync("verified-first"); + region.Tell(new Get("40")); + await ExpectMsgAsync(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(v => v.Id == "10" && v.N == 2); - region.Tell(new Get("11")); - await ExpectMsgAsync(v => v.Id == "11" && v.N == 5); - region.Tell(new Get("20")); - await ExpectMsgAsync(v => v.Id == "20" && v.N == 4); - region.Tell(new Get("30")); - await ExpectMsgAsync(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(v => v.Id == "10" && v.N == 2, TimeSpan.FromSeconds(5)); + region.Tell(new Get("11")); + await ExpectMsgAsync(v => v.Id == "11" && v.N == 5, TimeSpan.FromSeconds(5)); + region.Tell(new Get("20")); + await ExpectMsgAsync(v => v.Id == "20" && v.N == 4, TimeSpan.FromSeconds(5)); + region.Tell(new Get("30")); + await ExpectMsgAsync(v => v.Id == "30" && v.N == 6, TimeSpan.FromSeconds(5)); + }, Config.Second); + await EnterBarrierAsync("after-3"); } } \ No newline at end of file From 4e5c2826757675c52d25d41e4566fb91f4568973 Mon Sep 17 00:00:00 2001 From: Aaron Stannard Date: Wed, 29 Jul 2026 12:44:41 +0000 Subject: [PATCH 4/6] StressSpec: fresh probe per attempt for the watchee lookup, and bound its retry loop RemoveOneAsync resolved the watchee inside AwaitAssertAsync but sent every Identify to the shared, long-lived IdentifyProbe. When an attempt's 1s expect timed out, that attempt's ActorIdentity stayed queued on the probe and the next attempt consumed it instead of its own reply. A reply resolved before the watchee existed carries a null Subject, so once the probe held one the retry loop could never recover: 'Expected object not to be ' at the guard (Azure build 129814, node-1). Use a fresh probe per attempt, matching ClusterResultAggregatorAsync, so a stale reply can never be read by a later attempt. Also give that AwaitAssertAsync an explicit duration. With none it resolved RemainingOrDefault - the enclosing phase's leftover budget - so a slow watch could consume the phase and starve the removal work that follows. Establishing a watch on a just-created actor is one round trip; 10s of retries is ample and leaves the phase intact. --- .../Akka.Cluster.Tests.MultiNode/StressSpec.cs | 15 ++++++++++++--- 1 file changed, 12 insertions(+), 3 deletions(-) diff --git a/src/core/Akka.Cluster.Tests.MultiNode/StressSpec.cs b/src/core/Akka.Cluster.Tests.MultiNode/StressSpec.cs index c3ad0159894..9232f44fe4e 100644 --- a/src/core/Akka.Cluster.Tests.MultiNode/StressSpec.cs +++ b/src/core/Akka.Cluster.Tests.MultiNode/StressSpec.cs @@ -1006,8 +1006,13 @@ await RunOnAsync(async () => { await AwaitAssertAsync(async () => { - Sys.ActorSelection(new RootActorPath(removeAddress) / "user" / "watchee").Tell(new Identify("watchee"), IdentifyProbe.Ref); - var identity = await IdentifyProbe.ExpectMsgAsync(TimeSpan.FromSeconds(1)); + // Fresh probe per attempt, matching ClusterResultAggregatorAsync. The shared IdentifyProbe + // kept a timed-out attempt's late ActorIdentity queued, so the next attempt consumed that + // stale reply instead of its own -- and a reply resolved before the watchee existed carries + // a null Subject, so the retry could never recover no matter how often it ran. + var probe = CreateTestProbe(); + Sys.ActorSelection(new RootActorPath(removeAddress) / "user" / "watchee").Tell(new Identify("watchee"), probe.Ref); + var identity = await probe.ExpectMsgAsync(TimeSpan.FromSeconds(1)); // Under load the selection can resolve to an ActorIdentity with a null Subject; guard // here so this attempt fails fast and the retry loop retries, instead of passing null // into WatchAsync (ArgumentNullException inside the TestActor -> AskTimeout). @@ -1022,7 +1027,11 @@ await AwaitAssertAsync(async () => // degeneracy), regardless of whether we call the sync or async TestKit method. Bound it // explicitly instead so a slow attempt here still leaves room for AwaitAssertAsync to retry. await WatchAsync(watchee).WaitAsync(Dilated(TimeSpan.FromSeconds(3))); - }, interval:TimeSpan.FromSeconds(1.25d)); + // Explicit bound: with no duration this retry loop inherited RemainingOrDefault, i.e. the + // enclosing phase's leftover budget, so it could consume the phase and starve the removal + // work that follows. Establishing a watch on a just-created actor is a single round trip; + // 10s of retries is ample and leaves the phase intact. + }, TimeSpan.FromSeconds(10), TimeSpan.FromSeconds(1.25d)); }, Roles.First()); await EnterBarrierAsync("watchee-established-" + Step); From ab407c384c9c3be4bda0649b7be07b1d8d4c53e6 Mon Sep 17 00:00:00 2001 From: Aaron Stannard Date: Wed, 29 Jul 2026 17:10:04 +0000 Subject: [PATCH 5/6] De-flake the async CircuitBreakerSpec tests: await the failing call Ten tests fired the call that is supposed to trip the breaker as detached work: _ = breaker.Instance.WithCircuitBreaker(ct => Task.Run(ThrowException, ct)); Assert.True(CheckLatch(breaker.OpenLatch)); The breaker cannot record a failure - and therefore cannot open, half-open or close - until that call actually runs, so every latch assertion after it was racing thread-pool scheduling rather than testing breaker behaviour. Under pool starvation on a loaded agent the latch simply never trips within its window. Observed in CI as 'An asynchronous circuit breaker that is half open must pass through next call and close on success' failing with Assert.True() Failure. Await the call instead, intercepting the TestException it is expected to throw. This is already the idiom elsewhere in this file (the re-open tests use InterceptException the same way), and the file also carries a WaitForTaskToBeScheduled helper showing the hazard was known and only partly addressed. Awaiting is strictly stronger: it guarantees the failure is recorded before the assertion rather than merely that the task started. Methods that had no other await are converted to async Task. No timeout was changed. Verified: full CircuitBreaker suite 29/29, four consecutive runs at 2x core saturation. --- .../Akka.Tests/Pattern/CircuitBreakerSpec.cs | 41 ++++++++++++------- 1 file changed, 27 insertions(+), 14 deletions(-) diff --git a/src/core/Akka.Tests/Pattern/CircuitBreakerSpec.cs b/src/core/Akka.Tests/Pattern/CircuitBreakerSpec.cs index 70a4d58447e..75219f6307b 100644 --- a/src/core/Akka.Tests/Pattern/CircuitBreakerSpec.cs +++ b/src/core/Akka.Tests/Pattern/CircuitBreakerSpec.cs @@ -232,10 +232,14 @@ await InterceptException(() => } [Fact(DisplayName = "An asynchronous circuit breaker that is closed must increment failure count on async failure")] - public void Must_increment_failure_count_on_async_failure() + public async Task Must_increment_failure_count_on_async_failure() { var breaker = LongCallTimeoutCb(); - _ = breaker.Instance.WithCircuitBreaker(ct => Task.Run(ThrowException, ct)); + // Await the failing call rather than firing it and hoping the thread pool schedules it. The + // breaker cannot record the failure - and so cannot trip any latch - until this call actually + // runs, so a detached call left every assertion below racing pool scheduling. + await InterceptException(() => + breaker.Instance.WithCircuitBreaker(ct => Task.Run(ThrowException, ct))); Assert.True(CheckLatch(breaker.OpenLatch)); breaker.Instance.CurrentFailureCount.ShouldBe(1); } @@ -271,10 +275,11 @@ public async Task Must_increment_failure_count_on_callTimeout() } [Fact(DisplayName = "An asynchronous circuit breaker that is closed must invoke onOpen if call fails and breaker transits to open state")] - public void Must_invoke_onOpen_if_call_fails_and_breaker_transits_to_open_state() + public async Task Must_invoke_onOpen_if_call_fails_and_breaker_transits_to_open_state() { var breaker = ShortCallTimeoutCb(); - _ = breaker.Instance.WithCircuitBreaker(ct => Task.Run(ThrowException, ct)); + await InterceptException(() => + breaker.Instance.WithCircuitBreaker(ct => Task.Run(ThrowException, ct))); Assert.True(CheckLatch(breaker.OpenLatch)); } } @@ -285,7 +290,8 @@ public class AnAsynchronousCircuitBreakerThatIsHalfOpen : CircuitBreakerSpecBase public async Task Must_pass_through_next_call_and_close_on_success() { var breaker = ShortResetTimeoutCb(); - _ = breaker.Instance.WithCircuitBreaker(ct => Task.Run(ThrowException, ct)); + await InterceptException(() => + breaker.Instance.WithCircuitBreaker(ct => Task.Run(ThrowException, ct))); Assert.True(CheckLatch(breaker.HalfOpenLatch)); var result = await breaker.Instance.WithCircuitBreaker(ct => Task.Run(SayHi, ct)).WaitAsync(AwaitTimeout); Assert.Equal("hi", result); @@ -296,7 +302,8 @@ public async Task Must_pass_through_next_call_and_close_on_success() public async Task Must_reopen_on_exception_in_call() { var breaker = ShortResetTimeoutCb(); - _ = breaker.Instance.WithCircuitBreaker(ct => Task.Run(ThrowException, ct)); + await InterceptException(() => + breaker.Instance.WithCircuitBreaker(ct => Task.Run(ThrowException, ct))); Assert.True(CheckLatch(breaker.HalfOpenLatch)); breaker.OpenLatch.Reset(); await InterceptException(() => @@ -305,14 +312,16 @@ await InterceptException(() => } [Fact(DisplayName = "An asynchronous circuit breaker that is half open must re-open on async failure")] - public void Must_reopen_on_async_failure() + public async Task Must_reopen_on_async_failure() { var breaker = ShortResetTimeoutCb(); - _ = breaker.Instance.WithCircuitBreaker(ct => Task.Run(ThrowException, ct)); + await InterceptException(() => + breaker.Instance.WithCircuitBreaker(ct => Task.Run(ThrowException, ct))); Assert.True(CheckLatch(breaker.HalfOpenLatch)); breaker.OpenLatch.Reset(); - _ = breaker.Instance.WithCircuitBreaker(ct => Task.Run(ThrowException, ct)); + await InterceptException(() => + breaker.Instance.WithCircuitBreaker(ct => Task.Run(ThrowException, ct))); Assert.True(CheckLatch(breaker.OpenLatch)); } } @@ -323,7 +332,8 @@ public class AnAsynchronousCircuitBreakerThatIsOpen : CircuitBreakerSpecBase public async Task Must_throw_exceptions_when_called_before_reset_timeout() { var breaker = LongResetTimeoutCb(); - _ = breaker.Instance.WithCircuitBreaker(ct => Task.Run(ThrowException, ct)); + await InterceptException(() => + breaker.Instance.WithCircuitBreaker(ct => Task.Run(ThrowException, ct))); Assert.True(CheckLatch(breaker.OpenLatch)); @@ -332,10 +342,11 @@ await InterceptException( } [Fact(DisplayName = "An asynchronous circuit breaker that is open must transition to half-open on reset timeout")] - public void Must_transition_to_half_open_on_reset_timeout() + public async Task Must_transition_to_half_open_on_reset_timeout() { var breaker = ShortResetTimeoutCb(); - _ = breaker.Instance.WithCircuitBreaker(ct => Task.Run(ThrowException, ct)); + await InterceptException(() => + breaker.Instance.WithCircuitBreaker(ct => Task.Run(ThrowException, ct))); Assert.True(CheckLatch(breaker.HalfOpenLatch)); } @@ -343,7 +354,8 @@ public void Must_transition_to_half_open_on_reset_timeout() public async Task Must_increase_reset_timeout_after_it_transits_to_open_again() { var breaker = NonOneFactorCb(); - _ = breaker.Instance.WithCircuitBreaker(ct => Task.Run(ThrowException, ct)); + await InterceptException(() => + breaker.Instance.WithCircuitBreaker(ct => Task.Run(ThrowException, ct))); Assert.True(CheckLatch(breaker.OpenLatch)); var e1 = await InterceptException( @@ -355,7 +367,8 @@ public async Task Must_increase_reset_timeout_after_it_transits_to_open_again() // transit to open again breaker.OpenLatch.Reset(); - _ = breaker.Instance.WithCircuitBreaker(ct => Task.Run(ThrowException, ct)); + await InterceptException(() => + breaker.Instance.WithCircuitBreaker(ct => Task.Run(ThrowException, ct))); Assert.True(CheckLatch(breaker.OpenLatch)); var e2 = await InterceptException(() => From 0cd21a1fcc84523210e04235ddf59b0c896574bd Mon Sep 17 00:00:00 2001 From: Aaron Stannard Date: Wed, 29 Jul 2026 21:34:01 +0000 Subject: [PATCH 6/6] StressSpec: bracket the aggregator resolution with a barrier CreateResultAggregatorAsync enters 'result-aggregator-created-' so the aggregator provably exists, then performs a REMOTE Identify inside RunOnAsync - and then returns. RunOnAsync provides no synchronization at all; it is just 'if (IsNode(nodes)) await thunkAsync()'. So 'the aggregator has been resolved' was only ever a per-node fact, never a cluster-wide one. PartitionSeveral is where that gap becomes a failure. The node that owns the aggregator returns from this method and immediately starts blackholing the very nodes that are still mid-Identify, in Direction.Both and never lifted. Their lookup can then never complete, so they burn the entire retry budget and fail with 'Timeout ... while waiting for a message of type Akka.Actor.ActorIdentity' - observed repeatedly on node-9 in CI (builds 129447, 129814, 129905), with the stack running through ClusterResultAggregatorAsync <- CreateResultAggregatorAsync <- PartitionSeveral. Add a trailing barrier so every node has finished resolving before the phase proceeds. This closes the hole for every phase, not only the partition one. Verified: StressSpec passes locally, 10/10 nodes, 2m46s, zero barrier failures. --- src/core/Akka.Cluster.Tests.MultiNode/StressSpec.cs | 7 +++++++ 1 file changed, 7 insertions(+) diff --git a/src/core/Akka.Cluster.Tests.MultiNode/StressSpec.cs b/src/core/Akka.Cluster.Tests.MultiNode/StressSpec.cs index 9232f44fe4e..26135771beb 100644 --- a/src/core/Akka.Cluster.Tests.MultiNode/StressSpec.cs +++ b/src/core/Akka.Cluster.Tests.MultiNode/StressSpec.cs @@ -875,6 +875,13 @@ await RunOnAsync(async () => StatsObserver.Value.Tell(new ReportTo(resultAggregator)); }, Roles.Take(NbrUsedRoles).ToArray()); + // The RunOnAsync above performs a REMOTE Identify, and RunOnAsync provides no synchronization + // whatsoever - it is just `if (IsNode(nodes)) await thunkAsync()`. Without a barrier here, nodes + // still resolving the aggregator race whatever the caller does next. PartitionSeveral is where it + // bites: the node that owns the aggregator returns from this method and starts blackholing the very + // nodes that are still mid-Identify, so their lookup can never complete and they burn the whole + // retry budget. Bracket the resolution so every node has finished it before the phase proceeds. + await EnterBarrierAsync("result-aggregator-identified-" + Step); } public async Task AwaitClusterResultAsync()