From 6fb51b8e17990160097a6009a1e8420eef75ee36 Mon Sep 17 00:00:00 2001 From: Aaron Stannard Date: Thu, 27 Aug 2026 22:13:07 +0000 Subject: [PATCH 1/3] Add TestKitBase.ShutdownAsync and use it in StressSpec churn teardown Closes #8497. TestKitBase.Shutdown is a blocking `Terminate().Wait(duration)`. On a loaded agent that pins a thread pool thread for the whole wait, and the work the shutdown is waiting on - coordinated shutdown phases, cluster heartbeats - competes for the threads that are left. In StressSpec's churn phase the wait ran its full 10s and starved the hosting node's heartbeat sender for 12+ seconds, so the survivors marked it unreachable and SBR downed it. The wait made its own timeout more likely. ShutdownAsync races Terminate() against a timer instead of blocking, using the AwaitWithTimeout helper Akka.TestKit already ships. Everything else matches the sync overloads: same default duration, same forced guardian stop, same verifySystemShutdown throw-vs-log, same message. Both paths now share the force-stop tail, so the two cannot drift. The sync Shutdown keeps its own Terminate().Wait rather than blocking on the async one - sync-over-async here would need two continuations off the pool while a thread is pinned, which is worse under the starvation this fixes. StressSpec's post-churn teardown now awaits ShutdownAsync. Same wait budget, same fallback; the wait just no longer holds a thread. --- ...APISpec.ApproveTestKit.DotNet.verified.txt | 2 + ...oreAPISpec.ApproveTestKit.Net.verified.txt | 2 + .../StressSpec.cs | 9 +- .../TestKitBaseTests/ShutdownAsyncTests.cs | 98 +++++++++++++++++++ src/core/Akka.TestKit/TestKitBase.cs | 85 +++++++++++++--- 5 files changed, 181 insertions(+), 15 deletions(-) create mode 100644 src/core/Akka.TestKit.Tests/TestKitBaseTests/ShutdownAsyncTests.cs diff --git a/src/core/Akka.API.Tests/verify/CoreAPISpec.ApproveTestKit.DotNet.verified.txt b/src/core/Akka.API.Tests/verify/CoreAPISpec.ApproveTestKit.DotNet.verified.txt index 2b0f85a28d8..aa74f5da8d9 100644 --- a/src/core/Akka.API.Tests/verify/CoreAPISpec.ApproveTestKit.DotNet.verified.txt +++ b/src/core/Akka.API.Tests/verify/CoreAPISpec.ApproveTestKit.DotNet.verified.txt @@ -484,6 +484,8 @@ namespace Akka.TestKit public void SetAutoPilot(Akka.TestKit.AutoPilot pilot) { } public virtual void Shutdown(System.TimeSpan? duration = default, bool verifySystemShutdown = false) { } protected virtual void Shutdown(Akka.Actor.ActorSystem system, System.TimeSpan? duration = default, bool verifySystemShutdown = false) { } + public virtual System.Threading.Tasks.Task ShutdownAsync(System.TimeSpan? duration = default, bool verifySystemShutdown = false) { } + protected virtual System.Threading.Tasks.Task ShutdownAsync(Akka.Actor.ActorSystem system, System.TimeSpan? duration = default, bool verifySystemShutdown = false) { } public bool TryPeekOne(out Akka.TestKit.MessageEnvelope envelope, System.TimeSpan? max, System.Threading.CancellationToken cancellationToken) { } [return: System.Runtime.CompilerServices.TupleElementNames(new string[] { "success", diff --git a/src/core/Akka.API.Tests/verify/CoreAPISpec.ApproveTestKit.Net.verified.txt b/src/core/Akka.API.Tests/verify/CoreAPISpec.ApproveTestKit.Net.verified.txt index 2e51a3787cb..3b5e31d3ecd 100644 --- a/src/core/Akka.API.Tests/verify/CoreAPISpec.ApproveTestKit.Net.verified.txt +++ b/src/core/Akka.API.Tests/verify/CoreAPISpec.ApproveTestKit.Net.verified.txt @@ -484,6 +484,8 @@ namespace Akka.TestKit public void SetAutoPilot(Akka.TestKit.AutoPilot pilot) { } public virtual void Shutdown(System.TimeSpan? duration = default, bool verifySystemShutdown = false) { } protected virtual void Shutdown(Akka.Actor.ActorSystem system, System.TimeSpan? duration = default, bool verifySystemShutdown = false) { } + public virtual System.Threading.Tasks.Task ShutdownAsync(System.TimeSpan? duration = default, bool verifySystemShutdown = false) { } + protected virtual System.Threading.Tasks.Task ShutdownAsync(Akka.Actor.ActorSystem system, System.TimeSpan? duration = default, bool verifySystemShutdown = false) { } public bool TryPeekOne(out Akka.TestKit.MessageEnvelope envelope, System.TimeSpan? max, System.Threading.CancellationToken cancellationToken) { } [return: System.Runtime.CompilerServices.TupleElementNames(new string[] { "success", diff --git a/src/core/Akka.Cluster.Tests.MultiNode/StressSpec.cs b/src/core/Akka.Cluster.Tests.MultiNode/StressSpec.cs index 26135771beb..23f0b74dc2f 100644 --- a/src/core/Akka.Cluster.Tests.MultiNode/StressSpec.cs +++ b/src/core/Akka.Cluster.Tests.MultiNode/StressSpec.cs @@ -1279,10 +1279,11 @@ await AwaitMembersUpAsync(NbrUsedRoles + activeRoles.Length, return await Loop(counter + 1, nextAs, nextAddresses); } - (await Loop(1, Option.None, ImmutableHashSet
.Empty)).OnSuccess(aSys => - { - Shutdown(aSys); - }); + // await the teardown instead of blocking on it: the sync Shutdown pins a thread pool thread for the + // whole wait, which on a busy agent starves this node's own heartbeat sender and gets it downed. + var lastAs = await Loop(1, Option.None, ImmutableHashSet
.Empty); + if (lastAs.HasValue) + await ShutdownAsync(lastAs.Value); await WithinAsync(loopDuration, async () => { diff --git a/src/core/Akka.TestKit.Tests/TestKitBaseTests/ShutdownAsyncTests.cs b/src/core/Akka.TestKit.Tests/TestKitBaseTests/ShutdownAsyncTests.cs new file mode 100644 index 00000000000..5704a411727 --- /dev/null +++ b/src/core/Akka.TestKit.Tests/TestKitBaseTests/ShutdownAsyncTests.cs @@ -0,0 +1,98 @@ +//----------------------------------------------------------------------- +// +// Copyright (C) 2009-2022 Lightbend Inc. +// Copyright (C) 2013-2025 .NET Foundation +// +//----------------------------------------------------------------------- + +using System; +using System.Threading.Tasks; +using Akka.Actor; +using Akka.Event; +using Akka.TestKit.Extensions; +using FluentAssertions; +using Xunit; + +namespace Akka.TestKit.Tests.TestKitBaseTests +{ + public class ShutdownAsyncTests : AkkaSpec + { + private static readonly TimeSpan ShortTimeout = TimeSpan.FromMilliseconds(200); + private static readonly TimeSpan CleanupTimeout = TimeSpan.FromSeconds(10); + + [Fact(DisplayName = "ShutdownAsync should terminate the system before the returned task completes")] + public async Task Should_TerminateSystem_When_Awaited() + { + var sys = ActorSystem.Create("shutdown-async-completes", Sys.Settings.Config); + + await ShutdownAsync(sys); + + sys.WhenTerminated.IsCompleted.Should().BeTrue( + "ShutdownAsync must not complete before the system has terminated"); + } + + [Fact(DisplayName = "ShutdownAsync should throw TimeoutException when verifySystemShutdown is true and the system will not stop")] + public async Task Should_Throw_When_SystemDoesNotStop_And_VerifyIsSet() + { + var release = NewRelease(); + var sys = CreateStuckSystem("shutdown-async-throws", release); + + try + { + var ex = await Assert.ThrowsAsync( + () => ShutdownAsync(sys, ShortTimeout, verifySystemShutdown: true)); + + ex.Message.Should().Contain("Failed to stop [shutdown-async-throws]"); + } + finally + { + await ReleaseAndCleanUp(sys, release); + } + } + + [Fact(DisplayName = "ShutdownAsync should log instead of throwing when verifySystemShutdown is not set")] + public async Task Should_Log_When_SystemDoesNotStop_And_VerifyIsNotSet() + { + var release = NewRelease(); + var sys = CreateStuckSystem("shutdown-async-logs", release); + var probe = CreateTestProbe(); + sys.EventStream.Subscribe(probe.Ref, typeof(Warning)); + + try + { + await ShutdownAsync(sys, ShortTimeout); + + await probe.FishForMessageAsync( + w => w.Message.ToString().Contains("Failed to stop [shutdown-async-logs]"), + TimeSpan.FromSeconds(3)); + } + finally + { + await ReleaseAndCleanUp(sys, release); + } + } + + private static TaskCompletionSource NewRelease() + => new(TaskCreationOptions.RunContinuationsAsynchronously); + + /// + /// Builds a system whose coordinated shutdown parks on a task we control, so + /// cannot finish until the test releases it. + /// + private ActorSystem CreateStuckSystem(string name, TaskCompletionSource release) + { + var sys = ActorSystem.Create(name, Sys.Settings.Config); + CoordinatedShutdown.Get(sys).AddTask( + CoordinatedShutdown.PhaseBeforeActorSystemTerminate, + "block-until-released", + () => release.Task); + return sys; + } + + private static async Task ReleaseAndCleanUp(ActorSystem sys, TaskCompletionSource release) + { + release.TrySetResult(Done.Instance); + await sys.Terminate().AwaitWithTimeout(CleanupTimeout); + } + } +} diff --git a/src/core/Akka.TestKit/TestKitBase.cs b/src/core/Akka.TestKit/TestKitBase.cs index 01792fe5515..03713972aad 100644 --- a/src/core/Akka.TestKit/TestKitBase.cs +++ b/src/core/Akka.TestKit/TestKitBase.cs @@ -575,19 +575,82 @@ protected virtual void Shutdown( { system ??= _testState.System; - var durationValue = duration.GetValueOrDefault(Dilated(TimeSpan.FromSeconds(5)).Min(TimeSpan.FromSeconds(10))); + var durationValue = ShutdownDurationOrDefault(duration); var wasShutdownDuringWait = system.Terminate().Wait(durationValue); - if(!wasShutdownDuringWait) - { - // Forcefully close the ActorSystem to make sure we exit the test cleanly - ((ExtendedActorSystem) system).Guardian.Stop(); - - const string msg = "Failed to stop [{0}] within [{1}]. ActorSystem is being forcefully shut down.\n{2}"; - if(verifySystemShutdown) - throw new TimeoutException(string.Format(msg, system.Name, durationValue, ((ExtendedActorSystem) system).PrintTree())); - system.Log.Warning(msg, system.Name, durationValue, ((ExtendedActorSystem) system).PrintTree()); - } + ForceShutdownIfNeeded(system, wasShutdownDuringWait, durationValue, verifySystemShutdown); + } + + /// + /// Shuts down this system without blocking the calling thread. + /// On failure debug output will be logged about the remaining actors in the system. + /// If verifySystemShutdown is true, then an exception will be thrown on failure. + /// + /// + /// Prefer this over . The synchronous form pins a thread pool + /// thread for up to the full duration. On a loaded machine that pinned thread can starve the very + /// system work the shutdown is waiting on - cluster heartbeats, coordinated shutdown phases - so the + /// wait itself makes the timeout more likely. + /// + /// Optional. The duration to wait for shutdown. Default is 5 seconds multiplied with the config value "akka.test.timefactor". + /// if set to true an exception will be thrown on failure. + /// Thrown when the system did not stop in time and is true. + public virtual Task ShutdownAsync( + TimeSpan? duration = null, + bool verifySystemShutdown = false) + => ShutdownAsync(_testState.System, duration, verifySystemShutdown); + + /// + /// Shuts down the specified system without blocking the calling thread. + /// On failure debug output will be logged about the remaining actors in the system. + /// If verifySystemShutdown is true, then an exception will be thrown on failure. + /// + /// + /// Prefer this over . The synchronous form pins a + /// thread pool thread for up to the full duration. On a loaded machine that pinned thread can starve + /// the very system work the shutdown is waiting on - cluster heartbeats, coordinated shutdown phases - + /// so the wait itself makes the timeout more likely. + /// + /// The system to shutdown. + /// The duration to wait for shutdown. Default is 5 seconds multiplied with the config value "akka.test.timefactor" + /// if set to true an exception will be thrown on failure. + /// Thrown when the system did not stop in time and is true. + protected virtual async Task ShutdownAsync( + ActorSystem system, + TimeSpan? duration = null, + bool verifySystemShutdown = false) + { + system ??= _testState.System; + + var durationValue = ShutdownDurationOrDefault(duration); + + // race Terminate() against a timer rather than blocking a thread on it + var wasShutdownDuringWait = await system.Terminate().AwaitWithTimeout(durationValue); + ForceShutdownIfNeeded(system, wasShutdownDuringWait, durationValue, verifySystemShutdown); + } + + private TimeSpan ShutdownDurationOrDefault(TimeSpan? duration) + => duration.GetValueOrDefault(Dilated(TimeSpan.FromSeconds(5)).Min(TimeSpan.FromSeconds(10))); + + /// + /// Force-stops a system that failed to terminate on its own inside the timeout. + /// + private static void ForceShutdownIfNeeded( + ActorSystem system, + bool wasShutdownDuringWait, + TimeSpan durationValue, + bool verifySystemShutdown) + { + if(wasShutdownDuringWait) + return; + + // Forcefully close the ActorSystem to make sure we exit the test cleanly + ((ExtendedActorSystem) system).Guardian.Stop(); + + const string msg = "Failed to stop [{0}] within [{1}]. ActorSystem is being forcefully shut down.\n{2}"; + if(verifySystemShutdown) + throw new TimeoutException(string.Format(msg, system.Name, durationValue, ((ExtendedActorSystem) system).PrintTree())); + system.Log.Warning(msg, system.Name, durationValue, ((ExtendedActorSystem) system).PrintTree()); } /// From 0206ad44cf9523e0c3195b6db9ab7724d99c14e8 Mon Sep 17 00:00:00 2001 From: Aaron Stannard Date: Thu, 27 Aug 2026 22:15:47 +0000 Subject: [PATCH 2/3] Await the in-loop churn teardown in StressSpec too The post-loop teardown already awaits ShutdownAsync. This is the other blocking Shutdown in ExerciseJoinRemoveAsync - it tears down the previous round's system at the top of each churn round, so it is the one that blocks while the cluster is under load. Leaving it on Terminate().Wait would keep pinning a thread pool thread for the whole wait, which is what starves the hosting node's heartbeat sender. Same wait budget, same forced-guardian fallback, same ordering: the teardown still finishes before the next ActorSystem is created. --- src/core/Akka.Cluster.Tests.MultiNode/StressSpec.cs | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/src/core/Akka.Cluster.Tests.MultiNode/StressSpec.cs b/src/core/Akka.Cluster.Tests.MultiNode/StressSpec.cs index 23f0b74dc2f..638c2b77820 100644 --- a/src/core/Akka.Cluster.Tests.MultiNode/StressSpec.cs +++ b/src/core/Akka.Cluster.Tests.MultiNode/StressSpec.cs @@ -1236,10 +1236,10 @@ await WithinAsync(loopDuration, async () => if (activeRoles.Contains(Myself)) { - previousAs.OnSuccess(s => - { - Shutdown(s); - }); + // await the teardown instead of blocking on it: the sync Shutdown pins a thread + // pool thread for the whole wait, starving this node's own heartbeat sender. + if (previousAs.HasValue) + await ShutdownAsync(previousAs.Value); var sys = ActorSystem.Create(Sys.Name, Sys.Settings.Config); MuteLog(sys); From 2b083ff990e37df4e3bbed59fb360dfda6d75f12 Mon Sep 17 00:00:00 2001 From: Aaron Stannard Date: Fri, 28 Aug 2026 00:07:10 +0000 Subject: [PATCH 3/3] Drop the redundant happy-path ShutdownAsync test The verify-throws test already proves ShutdownAsync waits: a fire-and-forget implementation cannot observe that the system failed to stop, so it cannot throw the TimeoutException the test requires. StressSpec exercises the happy path on every MNTR run. The two remaining tests stay separate because each call runs the forced-guardian-stop tail, which kills the stuck system - so the throw branch and the log branch each need their own system. --- .../TestKitBaseTests/ShutdownAsyncTests.cs | 11 ----------- 1 file changed, 11 deletions(-) diff --git a/src/core/Akka.TestKit.Tests/TestKitBaseTests/ShutdownAsyncTests.cs b/src/core/Akka.TestKit.Tests/TestKitBaseTests/ShutdownAsyncTests.cs index 5704a411727..c787a309463 100644 --- a/src/core/Akka.TestKit.Tests/TestKitBaseTests/ShutdownAsyncTests.cs +++ b/src/core/Akka.TestKit.Tests/TestKitBaseTests/ShutdownAsyncTests.cs @@ -20,17 +20,6 @@ public class ShutdownAsyncTests : AkkaSpec private static readonly TimeSpan ShortTimeout = TimeSpan.FromMilliseconds(200); private static readonly TimeSpan CleanupTimeout = TimeSpan.FromSeconds(10); - [Fact(DisplayName = "ShutdownAsync should terminate the system before the returned task completes")] - public async Task Should_TerminateSystem_When_Awaited() - { - var sys = ActorSystem.Create("shutdown-async-completes", Sys.Settings.Config); - - await ShutdownAsync(sys); - - sys.WhenTerminated.IsCompleted.Should().BeTrue( - "ShutdownAsync must not complete before the system has terminated"); - } - [Fact(DisplayName = "ShutdownAsync should throw TimeoutException when verifySystemShutdown is true and the system will not stop")] public async Task Should_Throw_When_SystemDoesNotStop_And_VerifyIsSet() {