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 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..26135771beb 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 = { @@ -839,7 +842,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); } @@ -872,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() @@ -1003,8 +1013,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). @@ -1019,7 +1034,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); 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(() =>