From dfaf3cd3daa71ece0a447917a3d0715b21085abb Mon Sep 17 00:00:00 2001 From: Aaron Stannard Date: Wed, 9 Sep 2026 15:26:40 +0000 Subject: [PATCH 1/4] De-flake DistributedPubSubRestartSpec: the restarted node makes first contact, and the survivor waits on it first's association to third's old incarnation carries no signal that a same-address restart even started; the survivor's association only learns the new incarnation exists from a frame the restarted node itself sends. The old test started its 45s clock at Shutdown()'s return, before any of third's restart cost had begun, and none of the ten prior fixes changed who makes first contact after the restart - they all kept first polling a stale association instead. Now third's Shutdown actor pings a forwarder on first the moment it exists (PreStart), so first waits on that ping (dilated 60s) before running a short 20s closed-loop kill over ActorSelection.Tell (only an ActorSelectionMessage pierces a quarantined association). The ping's own inbound handshake is what heals first's association as a side effect. barrier-timeout goes to 180s to fit the new worst case with headroom, and a bound-address assertion on third settles, on the next Linux failure, whether the fresh listener came up on the pinned port. --- .../DistributedPubSubRestartSpec.cs | 136 ++++++++++++------ 1 file changed, 90 insertions(+), 46 deletions(-) diff --git a/src/contrib/cluster/Akka.Cluster.Tools.Tests.MultiNode/PublishSubscribe/DistributedPubSubRestartSpec.cs b/src/contrib/cluster/Akka.Cluster.Tools.Tests.MultiNode/PublishSubscribe/DistributedPubSubRestartSpec.cs index 57423baf34a..67b8bec37d1 100644 --- a/src/contrib/cluster/Akka.Cluster.Tools.Tests.MultiNode/PublishSubscribe/DistributedPubSubRestartSpec.cs +++ b/src/contrib/cluster/Akka.Cluster.Tools.Tests.MultiNode/PublishSubscribe/DistributedPubSubRestartSpec.cs @@ -26,6 +26,13 @@ public class DistributedPubSubRestartSpecConfig : MultiNodeConfig public RoleName Second { get; } public RoleName Third { get; } + // Well-known actor on `first` that the restarted `third`'s Shutdown actor pings the moment + // it exists (see Shutdown.PreStart below). Must live under /user: a TestKit probe backs onto + // SystemActorOf and cannot be addressed at a fixed /user path itself, but a plain + // ForwardActor standing under /user in front of one can be. + internal const string ReadyActorName = "restart-ready"; + internal const string ReadySignal = "third-restarted"; + public DistributedPubSubRestartSpecConfig() { First = Role("first"); @@ -61,9 +68,13 @@ public DistributedPubSubRestartSpecConfig() # via the dot-netty handshake-timeout/connection-timeout conflation starved the # re-association handshake; the default gives it the full 15s budget). - # second waits at the ""end"" barrier while first runs its closed-loop identify/kill - # window (dilated) plus the conductor-shutdown ack - kept well past the 30s default. - akka.testconductor.barrier-timeout = 120s + # The barrier-timeout clock arms the instant the FIRST node reaches a barrier, not + # when every node does - second parks at ""end"" almost immediately, while first is + # still working through, in order: the 30s Shutdown(third) cap, then up to a + # dilated 60s wait for third's restarted incarnation to make first contact (see + # the ready-ping comment below), then a 20s closed-loop kill. + # 30s + 60s + 20s = 110s; 180s leaves 70s of headroom on top of that. + akka.testconductor.barrier-timeout = 180s ").WithFallback(DistributedPubSub.DefaultConfig()); TestTransport = true; @@ -71,8 +82,11 @@ public DistributedPubSubRestartSpecConfig() internal class Shutdown : ReceiveActor { - public Shutdown() + private readonly Address _firstAddress; + + public Shutdown(Address firstAddress) { + _firstAddress = firstAddress; Context.GetLogger().Info("Shutdown actor started on {0}", Context.System.Name); Receive(str => str.Equals("shutdown"), _ => { @@ -84,6 +98,22 @@ public Shutdown() Context.System.Terminate(); }); } + + protected override void PreStart() + { + base.PreStart(); + + // This actor existing IS the fact first is waiting on below, so make the ping and + // that fact the same event instead of signalling readiness some other way that could + // race this actor's own creation. This fresh system has no stale association to + // first to fight through - unlike first's outbound path to US, which still carries + // the dead old incarnation's handshake state until something re-associates it - so + // this send's own HandshakeReq is normally the first thing that reaches first after + // the restart, and completing THAT handshake on first's end is what un-gates the + // ordinary lane the kill loop below needs. + Context.System.ActorSelection(new RootActorPath(_firstAddress) / "user" / ReadyActorName) + .Tell(ReadySignal); + } } } @@ -143,6 +173,12 @@ await RunOnAsync(() => // be read AFTER pub-sub gossip goes quiet, not merely after CountAsync - see the comment on // ReadStableDeltaCountAsync for why those are not the same moment. var oldDeltaCount = await ReadStableDeltaCountAsync(); + + // Captured on EVERY node while all three systems are still alive. Third's original Sys + // terminates partway through its own restart below, and NodeAsync needs the + // TestConductor client that dies along with it - so first's address has to be in hand + // before that happens, not looked up afterward. + var firstAddress = (await NodeAsync(_config.First)).Address; await EnterBarrierAsync("old-delta-count"); await RunOnAsync(async () => @@ -159,49 +195,46 @@ await RunOnAsync(async () => await RunOnAsync(async () => { var thirdAddress = (await NodeAsync(_config.Third)).Address; + + // Stand this up BEFORE Shutdown(third) so it exists no matter how fast the restarted + // incarnation comes back. It is the receiving end of the ready-ping the fresh + // Shutdown actor sends from its own PreStart (see that class, above) - reversing the + // direction of first contact from "first polls a stale association" to "third + // announces itself over an association with no history to carry". + var readyProbe = CreateTestProbe(); + Sys.ActorOf(Akka.TestKit.TestActors.ForwardActor.Props(readyProbe.Ref), + DistributedPubSubRestartSpecConfig.ReadyActorName); + await TestConductor.Shutdown(_config.Third).WaitAsync(30.Seconds()); - // Await the association, then assert on it. After the same-address restart, first's - // outbound endpoint to third is still gated by the dead old incarnation, so a blind - // "shutdown" Tell drops to dead letters for as long as that gate lasts. Blind resends - // over an at-most-once gated path have no time bound. - // - // ResolveOne drives an Identify round trip, and that round trip IS the delivery - // confirmation: it only completes once traffic flows to the restarted incarnation and - // its /user/shutdown exists. Every retry re-pokes the association, and the final - // failure raises ActorNotFoundException naming the path that never resolved instead - // of a bare "expected ActorIdentity" timeout on a null Subject. - // - // Resolve and kill must stay in ONE retry. Splitting them - resolve to a ref, then - // send the kill outside the loop - fails on artery: the ordinary outbound stream - // restarts with backoff and does not resend, so the association can drop in the gap - // between a successful resolve and the next Tell. A local artery soak reproduced - // exactly that (resolve succeeded, then ~19s of "Still unable to reconnect Artery - // Control outbound connection ... Tcp command Connect ... failed" swallowed the kill - // and the ack never came). Keeping both inside the retry re-sends the kill against a - // freshly resolved ref on the next attempt. - // - // The 45s budget comes from arithmetic, not taste. Worst case first absorbs: - // third comes back: WhenTerminated + fresh system + join + its 5s ExpectNoMsg ~15s - // first's endpoint: 15s associate timeout on the dead incarnation + 5s gate 20s - // fresh handshake + Identify round trip + ack ~5s - // = ~40s. It nests inside the 120s "end" barrier second is already parked on: - // 30s Shutdown cap + 45s here = 75s, leaving 45s of headroom. - var shutdownPath = new RootActorPath(thirdAddress) / "user" / "shutdown"; + // Wait for third's restarted incarnation to make first contact, instead of starting + // this window's clock at Shutdown()'s return - before graceful CoordinatedShutdown, + // the fresh ActorSystem, the rebind, the self-join and third's own 5s ExpectNoMsg + // have even begun. This is not merely a signal: the inbound HandshakeReq that carries + // it is what completes first's outbound handshake to the new incarnation + // (InboundHandshakeStage.HandleReq -> CompleteHandshake), so by the time this + // returns, the ordinary lane the kill loop below uses is no longer gated on the dead + // old uid. + await readyProbe.ExpectMsgAsync( + msg => msg == DistributedPubSubRestartSpecConfig.ReadySignal, 60.Seconds()); + + // ActorSelection.Tell, not ResolveOne + a resolved ref: only an ActorSelectionMessage + // pierces a quarantined association (ArteryRemoting.Send drops a plain Tell to a + // quarantined peer; Pekko's Association.scala carries the identical carve-out, and + // upstream's own restart spec relies on exactly that). The ready-ping above should + // already have healed the association, so this loop is a short closed-loop + // confirmation - not the thing racing third's restart cost the way the old, + // Shutdown()-anchored window did. + var shutdownSelection = Sys.ActorSelection(new RootActorPath(thirdAddress) / "user" / "shutdown"); await AwaitAssertAsync(async () => { - // ResolveOne is not a TestKit call, so dilate its per-attempt bound by hand. - // Its temp actor is fresh per attempt, so no probe can carry a stale - // ActorIdentity(null) from the burst that flushes when the association comes up. - var target = await Sys.ActorSelection(shutdownPath).ResolveOne(Dilated(2.Seconds())); - - // The resolve proved the association carries traffic. Kill the actor and require - // the ack: the Shutdown actor replies before it terminates, so the ack is proof - // that THIS incarnation received the message. + // Fresh probe per attempt: a late ack from a previous attempt must never satisfy + // the next one. The Shutdown actor replies before it terminates, so the ack is + // proof that THIS incarnation received the message. var killProbe = CreateTestProbe(); - target.Tell("shutdown", killProbe.Ref); - await killProbe.ExpectMsgAsync(msg => msg == "shutdown-ack", 3.Seconds()); - }, 45.Seconds(), 1.Seconds()); + shutdownSelection.Tell("shutdown", killProbe.Ref); + await killProbe.ExpectMsgAsync(msg => msg == "shutdown-ack", 2.Seconds()); + }, 20.Seconds(), 500.Milliseconds()); await EnterBarrierAsync("end"); @@ -242,6 +275,16 @@ await RunOnAsync(async () => $"akka.remote.artery.canonical.port={node3Address.Port}") .WithFallback(Sys.Settings.Config)); + // Settle, on the next Linux failure, whether the listener came up on the pinned port + // at all. First's whole restart-detection path - the ready ping above and the kill + // loop's ActorSelection - depends on this system actually listening on + // node3Address's port, and nothing upstream of this line would surface a silent + // mismatch; it would just look like first's ready-ping wait timing out. + var actualAddress = Cluster.Get(newSystem).SelfAddress; + newSystem.Log.Info("Restarted system bound to [{0}] (pinned [{1}])", actualAddress, node3Address); + actualAddress.Port.Should().Be(node3Address.Port, + "the fresh system must rebind the address the other nodes still address it by"); + try { // don't join the old cluster @@ -262,14 +305,15 @@ await RunOnAsync(async () => // We must complete the DeltaCount check above before this, otherwise there's // a race where First triggers shutdown while we're still verifying. newSystem.Log.Info("Creating shutdown actor on {0}", node3Address); - newSystem.ActorOf("shutdown"); + newSystem.ActorOf( + Props.Create(() => new DistributedPubSubRestartSpecConfig.Shutdown(firstAddress)), + "shutdown"); // First's closed-loop kill (above) normally drives this WhenTerminated: it keeps // re-poking the association until third's /user/shutdown acks the kill, at which // point newSystem terminates and this wait completes. Give it a generous 120s - // upper bound so first's whole worst-case pipeline (Shutdown-ack anchor sliding up - // to ~30s past third's restart clock, plus the 45s dilated identify/kill window) - // fits comfortably. + // upper bound so first's whole worst-case pipeline (the 30s Shutdown cap, plus + // the dilated 60s ready-ping wait, plus the 20s closed-loop kill) fits comfortably. // // This wait is BEST-EFFORT: the spec's REAL assertions - the SubscribeAck / // ExpectNoMsg / DeltaCount == 0 gossip-isolation checks above - have already run From 8c7a2708fd5af82ae3fca161d66c6fd0f8d6b8e2 Mon Sep 17 00:00:00 2001 From: Aaron Stannard Date: Wed, 9 Sep 2026 17:17:05 +0000 Subject: [PATCH 2/4] DistributedPubSubRestartSpec: repeat the ready ping until acknowledged, use the conductor's async shutdown, keep the port assertion inside the cleanup Addresses review findings on PR #8552: - Shutdown.PreStart no longer sends the ready ping once; it resends every 500ms on an undilated self-scheduled timer until first's ReadyPingForwarder acks it back to the sender, cancelling the timer on ack (and defensively in PostStop). first still waits once for the first ping it sees and ignores later resends. This removes the dependency on any Artery handshake-stage fix - a lost ping is simply retried instead of failing the 60s wait. - Replace TestConductor.Shutdown(...) with the ShutdownAsync(RoleName, ...) overload, keeping the same 30s bound. - Move the restarted-address port assertion inside the try/finally that owns newSystem, so a failed assertion still terminates it. - State the measured worst case behind the 60s ready-ping wait and re-derive newSystem.WhenTerminated's bound (120s -> 155s) against the current 110s worst-case pipeline, restoring its original 45s of slack. --- .../DistributedPubSubRestartSpec.cs | 138 +++++++++++++----- 1 file changed, 100 insertions(+), 38 deletions(-) diff --git a/src/contrib/cluster/Akka.Cluster.Tools.Tests.MultiNode/PublishSubscribe/DistributedPubSubRestartSpec.cs b/src/contrib/cluster/Akka.Cluster.Tools.Tests.MultiNode/PublishSubscribe/DistributedPubSubRestartSpec.cs index 67b8bec37d1..a702669605e 100644 --- a/src/contrib/cluster/Akka.Cluster.Tools.Tests.MultiNode/PublishSubscribe/DistributedPubSubRestartSpec.cs +++ b/src/contrib/cluster/Akka.Cluster.Tools.Tests.MultiNode/PublishSubscribe/DistributedPubSubRestartSpec.cs @@ -26,12 +26,14 @@ public class DistributedPubSubRestartSpecConfig : MultiNodeConfig public RoleName Second { get; } public RoleName Third { get; } - // Well-known actor on `first` that the restarted `third`'s Shutdown actor pings the moment - // it exists (see Shutdown.PreStart below). Must live under /user: a TestKit probe backs onto - // SystemActorOf and cannot be addressed at a fixed /user path itself, but a plain - // ForwardActor standing under /user in front of one can be. + // Well-known actor on `first` that the restarted `third`'s Shutdown actor pings, on a + // repeating timer, the moment it exists (see Shutdown.PreStart and ReadyPingForwarder + // below). Must live under /user: a TestKit probe backs onto SystemActorOf and cannot be + // addressed at a fixed /user path itself, but a plain actor standing under /user in front + // of one can be. internal const string ReadyActorName = "restart-ready"; internal const string ReadySignal = "third-restarted"; + internal const string ReadyAck = "third-restarted-ack"; public DistributedPubSubRestartSpecConfig() { @@ -83,11 +85,21 @@ public DistributedPubSubRestartSpecConfig() internal class Shutdown : ReceiveActor { private readonly Address _firstAddress; + private ICancelable? _readyPingTimer; public Shutdown(Address firstAddress) { _firstAddress = firstAddress; Context.GetLogger().Info("Shutdown actor started on {0}", Context.System.Name); + + // The ping's ack (see PreStart). Stop resending the instant first proves it saw us - + // anything after that is a wasted send into an association that is already healed. + Receive(str => str.Equals(ReadyAck), _ => + { + _readyPingTimer?.Cancel(); + _readyPingTimer = null; + }); + Receive(str => str.Equals("shutdown"), _ => { // Reply BEFORE terminating so the sender (first) gets an observable ack that @@ -105,14 +117,51 @@ protected override void PreStart() // This actor existing IS the fact first is waiting on below, so make the ping and // that fact the same event instead of signalling readiness some other way that could - // race this actor's own creation. This fresh system has no stale association to - // first to fight through - unlike first's outbound path to US, which still carries - // the dead old incarnation's handshake state until something re-associates it - so - // this send's own HandshakeReq is normally the first thing that reaches first after - // the restart, and completing THAT handshake on first's end is what un-gates the - // ordinary lane the kill loop below needs. - Context.System.ActorSelection(new RootActorPath(_firstAddress) / "user" / ReadyActorName) - .Tell(ReadySignal); + // race this actor's own creation. + // + // A single send is not enough to rely on: it is an ActorSelectionMessage, which rides + // the ORDINARY lane, and Artery's OutboundHandshakeStage holds one element on that + // lane while the handshake to this fresh incarnation is still gating. If that stage's + // materialization stops before the handshake completes, the held element is discarded + // and the ping is gone for good - first would then wait out its whole 60s window for + // nothing. So resend on a self-scheduled timer instead of sending once: every 500ms, + // undilated (this is a resend cadence, not a test assertion bound - see Scheduler + // .Advanced), until first proves it got one. The forwarder on first + // (ReadyPingForwarder, below) acks the sender directly, and that ack is what cancels + // this timer (also cancelled defensively in PostStop, in case the ack is somehow + // missed). First itself still only ever waits ONCE, on the first ping it sees, and + // ignores every later resend - it never calls ExpectMsg on the probe a second time. + var self = Self; + var readyActor = Context.System.ActorSelection(new RootActorPath(_firstAddress) / "user" / ReadyActorName); + var timer = new Cancelable(Context.System.Scheduler); + Context.System.Scheduler.Advanced.ScheduleRepeatedly( + TimeSpan.Zero, TimeSpan.FromMilliseconds(500), () => readyActor.Tell(ReadySignal, self), timer); + _readyPingTimer = timer; + } + + protected override void PostStop() + { + _readyPingTimer?.Cancel(); + base.PostStop(); + } + } + + /// + /// Stands in for `first`'s side of the ready-ping handshake (see + /// above). Acks every ping straight back to its sender - which is what lets the restarted + /// node's resend timer stop - and separately forwards the signal to + /// so the spec can observe that first contact was made. The spec only ever waits on `target` + /// once, so a resend that lands after that wait is simply left unread. + /// + internal sealed class ReadyPingForwarder : ReceiveActor + { + public ReadyPingForwarder(IActorRef target) + { + Receive(str => str.Equals(ReadySignal), _ => + { + Sender.Tell(ReadyAck); + target.Tell(ReadySignal); + }); } } } @@ -140,7 +189,7 @@ public async Task DistributedPubSubRestartSpecs() public async Task A_Cluster_with_DistributedPubSub_must_startup_3_node_cluster() { // No Within wrapper: JoinAsync and "after-1" are barrier waits, and barriers must - // not run inside a Within (see the comment on the restart test below). + // not run inside a Within block (see the comment on the restart test below). await JoinAsync(_config.First, _config.First); await JoinAsync(_config.Second, _config.First); await JoinAsync(_config.Third, _config.First); @@ -198,14 +247,16 @@ await RunOnAsync(async () => // Stand this up BEFORE Shutdown(third) so it exists no matter how fast the restarted // incarnation comes back. It is the receiving end of the ready-ping the fresh - // Shutdown actor sends from its own PreStart (see that class, above) - reversing the - // direction of first contact from "first polls a stale association" to "third - // announces itself over an association with no history to carry". + // Shutdown actor repeats from its own PreStart (see that class and + // ReadyPingForwarder, above) - reversing the direction of first contact from "first + // polls a stale association" to "third announces itself over an association with no + // history to carry", and acking every ping so third's resend timer knows to stop. var readyProbe = CreateTestProbe(); - Sys.ActorOf(Akka.TestKit.TestActors.ForwardActor.Props(readyProbe.Ref), + Sys.ActorOf( + Props.Create(() => new DistributedPubSubRestartSpecConfig.ReadyPingForwarder(readyProbe.Ref)), DistributedPubSubRestartSpecConfig.ReadyActorName); - await TestConductor.Shutdown(_config.Third).WaitAsync(30.Seconds()); + await TestConductor.ShutdownAsync(_config.Third).WaitAsync(30.Seconds()); // Wait for third's restarted incarnation to make first contact, instead of starting // this window's clock at Shutdown()'s return - before graceful CoordinatedShutdown, @@ -214,17 +265,24 @@ await RunOnAsync(async () => // it is what completes first's outbound handshake to the new incarnation // (InboundHandshakeStage.HandleReq -> CompleteHandshake), so by the time this // returns, the ordinary lane the kill loop below uses is no longer gated on the dead - // old uid. + // old uid. Measured worst case for that whole sequence: 12.9s of graceful shutdown + + // 6s before the restarted actors even exist + 21s of re-association = 39.9s: the 60s + // bound leaves about 20s of margin above the measured worst case. And unlike a single + // send, this wait cannot be defeated by the ping itself getting lost in flight - see + // the resend timer in Shutdown.PreStart above - so it does not depend on any fix to + // how Artery's handshake stage handles a held element on stop. await readyProbe.ExpectMsgAsync( msg => msg == DistributedPubSubRestartSpecConfig.ReadySignal, 60.Seconds()); // ActorSelection.Tell, not ResolveOne + a resolved ref: only an ActorSelectionMessage // pierces a quarantined association (ArteryRemoting.Send drops a plain Tell to a // quarantined peer; Pekko's Association.scala carries the identical carve-out, and - // upstream's own restart spec relies on exactly that). The ready-ping above should - // already have healed the association, so this loop is a short closed-loop - // confirmation - not the thing racing third's restart cost the way the old, - // Shutdown()-anchored window did. + // upstream's own restart spec relies on exactly that). The ready-ping above is + // guaranteed to have healed the association by this point: readyProbe could only have + // received the ping above once first's inbound handshake to third's new incarnation + // had already completed (see the comment on that wait). So this loop is a short + // closed-loop confirmation, not the thing racing third's restart cost the way the + // old, Shutdown()-anchored window did. var shutdownSelection = Sys.ActorSelection(new RootActorPath(thirdAddress) / "user" / "shutdown"); await AwaitAssertAsync(async () => { @@ -275,18 +333,20 @@ await RunOnAsync(async () => $"akka.remote.artery.canonical.port={node3Address.Port}") .WithFallback(Sys.Settings.Config)); - // Settle, on the next Linux failure, whether the listener came up on the pinned port - // at all. First's whole restart-detection path - the ready ping above and the kill - // loop's ActorSelection - depends on this system actually listening on - // node3Address's port, and nothing upstream of this line would surface a silent - // mismatch; it would just look like first's ready-ping wait timing out. - var actualAddress = Cluster.Get(newSystem).SelfAddress; - newSystem.Log.Info("Restarted system bound to [{0}] (pinned [{1}])", actualAddress, node3Address); - actualAddress.Port.Should().Be(node3Address.Port, - "the fresh system must rebind the address the other nodes still address it by"); - try { + // Settle, on the next Linux failure, whether the listener came up on the pinned + // port at all. First's whole restart-detection path - the ready ping above and + // the kill loop's ActorSelection - depends on this system actually listening on + // node3Address's port, and nothing upstream of this line would surface a silent + // mismatch; it would just look like first's ready-ping wait timing out. Kept + // inside this try so a failed assertion still runs the finally below and shuts + // newSystem down instead of leaking it (and its pinned port). + var actualAddress = Cluster.Get(newSystem).SelfAddress; + newSystem.Log.Info("Restarted system bound to [{0}] (pinned [{1}])", actualAddress, node3Address); + actualAddress.Port.Should().Be(node3Address.Port, + "the fresh system must rebind the address the other nodes still address it by"); + // don't join the old cluster await Cluster.Get(newSystem).JoinAsync(Cluster.Get(newSystem).SelfAddress); var newMediator = DistributedPubSub.Get(newSystem).Mediator; @@ -311,9 +371,11 @@ await RunOnAsync(async () => // First's closed-loop kill (above) normally drives this WhenTerminated: it keeps // re-poking the association until third's /user/shutdown acks the kill, at which - // point newSystem terminates and this wait completes. Give it a generous 120s - // upper bound so first's whole worst-case pipeline (the 30s Shutdown cap, plus - // the dilated 60s ready-ping wait, plus the 20s closed-loop kill) fits comfortably. + // point newSystem terminates and this wait completes. First's own worst-case + // pipeline before it can even send that kill is 30s (Shutdown cap) + 60s + // (ready-ping wait) + 20s (closed-loop kill) = 110s, so this bound has to clear + // that first. 155s = 110s + 45s of slack (the same 45s of slack this wait + // originally carried, back when the pipeline above it was 75s instead of 110s). // // This wait is BEST-EFFORT: the spec's REAL assertions - the SubscribeAck / // ExpectNoMsg / DeltaCount == 0 gossip-isolation checks above - have already run @@ -323,12 +385,12 @@ await RunOnAsync(async () => // of the isolation assertions that already succeeded. try { - await newSystem.WhenTerminated.WaitAsync(120.Seconds()); + await newSystem.WhenTerminated.WaitAsync(155.Seconds()); } catch (TimeoutException) { newSystem.Log.Warning( - "newSystem did not observe first's shutdown within 120s; terminating self (best-effort). " + + "newSystem did not observe first's shutdown within 155s; terminating self (best-effort). " + "The gossip-isolation assertions (SubscribeAck / ExpectNoMsg / DeltaCount == 0) already passed, " + "so the spec's subject-under-test is verified regardless."); } From e9425c26cf7934175f2f64645598a32688a07c7f Mon Sep 17 00:00:00 2001 From: Aaron Stannard Date: Thu, 10 Sep 2026 03:01:49 +0000 Subject: [PATCH 3/4] DistributedPubSubRestartSpec: let the shutdown ack leave before the restarted system terminates Build 131332 (this PR's own CI run, Linux Artery) showed third's Shutdown actor replying "shutdown-ack" and calling Context.System.Terminate() inline in the same handler; Artery aborted third's outbound streams before the ack could flush, so it never reached first. First's closed-loop kill then retried "shutdown" every 500ms against an already-exited process and its 20s bound timed out. Local runs only passed because the ack happened to win that race on a fast machine. Slide the terminate behind a 2s timer instead of calling it inline - same shape as the Subject actor in RemoteNodeRestartDeathWatchSpec (PR #8557) - so a live system, and the warmed-up lane under it, always outlasts whichever "shutdown" attempt first's loop last sends. Cancel the timer in PostStop. --- .../DistributedPubSubRestartSpec.cs | 39 ++++++++++++++++++- 1 file changed, 37 insertions(+), 2 deletions(-) diff --git a/src/contrib/cluster/Akka.Cluster.Tools.Tests.MultiNode/PublishSubscribe/DistributedPubSubRestartSpec.cs b/src/contrib/cluster/Akka.Cluster.Tools.Tests.MultiNode/PublishSubscribe/DistributedPubSubRestartSpec.cs index a702669605e..2a445c7576c 100644 --- a/src/contrib/cluster/Akka.Cluster.Tools.Tests.MultiNode/PublishSubscribe/DistributedPubSubRestartSpec.cs +++ b/src/contrib/cluster/Akka.Cluster.Tools.Tests.MultiNode/PublishSubscribe/DistributedPubSubRestartSpec.cs @@ -86,6 +86,7 @@ internal class Shutdown : ReceiveActor { private readonly Address _firstAddress; private ICancelable? _readyPingTimer; + private ICancelable? _terminateTimer; public Shutdown(Address firstAddress) { @@ -105,9 +106,34 @@ public Shutdown(Address firstAddress) // Reply BEFORE terminating so the sender (first) gets an observable ack that // proves this incarnation received the kill. This ack is what lets first run a // CLOSED-LOOP, self-verifying retry instead of an open-loop blind resend - // (mirrors the Subject actor in RemoteNodeRestartDeathWatchSpec, PR #8404). + // (mirrors the Subject actor in RemoteNodeRestartDeathWatchSpec, PR #8404) - but + // the ack still has to actually leave before the transport underneath it dies. + // Build 131332 (this PR's own CI run, Linux Artery) showed why an inline + // Terminate() cannot be trusted to let that happen: third's Shutdown actor replied + // "shutdown-ack" and called Context.System.Terminate() in the same handler at + // 02:26:57.990, and Artery tore down third's outbound streams before any of them + // could flush - "Artery Ordinary outbound connection to [first] failed ... with + // AbruptStageTerminationException" at 57.992 - so the ack never reached the socket. + // First then kept retrying "shutdown" every 500ms against a process that had + // already exited, and its 20s closed-loop kill timed out ("AwaitAssert failed, + // timeout [00:00:20] is over after [9] attempts"). Local runs only passed because + // the ack happened to win that race on a fast machine. + // + // So slide the terminate behind a timer instead of calling it inline - the same + // shape as the Subject actor in RemoteNodeRestartDeathWatchSpec (PR #8557): cancel + // any previous timer and re-arm a fresh one on every "shutdown", so a retried kill + // can never have its ack racing a system that is already tearing itself down. + // Capture the system reference, not Context, since the scheduled callback runs off + // the actor after this handler returns. 2s is far above a loopback round trip + // (this spec's own retry interval is 500ms) and far below first's 20s closed-loop + // kill budget, so whichever "shutdown" attempt first's loop last sends always gets + // a live system - and the warmed-up lane under it - for its ack to leave on before + // this timer fires. Sender.Tell("shutdown-ack"); - Context.System.Terminate(); + _terminateTimer?.Cancel(); + var system = Context.System; + _terminateTimer = system.Scheduler.Advanced.ScheduleOnceCancelable( + TimeSpan.FromSeconds(2), () => system.Terminate()); }); } @@ -142,6 +168,11 @@ protected override void PreStart() protected override void PostStop() { _readyPingTimer?.Cancel(); + + // The sliding terminate timer above re-arms on every "shutdown" and is otherwise left + // running. Cancel it here so a stopped Shutdown actor can't have it fire later and + // terminate whatever ActorSystem happens to own it at that point. + _terminateTimer?.Cancel(); base.PostStop(); } } @@ -376,6 +407,10 @@ await RunOnAsync(async () => // (ready-ping wait) + 20s (closed-loop kill) = 110s, so this bound has to clear // that first. 155s = 110s + 45s of slack (the same 45s of slack this wait // originally carried, back when the pipeline above it was 75s instead of 110s). + // Shutdown's own sliding terminate timer (see that class) adds at most 2s on top + // of whichever "shutdown" attempt first's loop last sends before newSystem actually + // goes down - negligible against the 45s of slack already in this bound, so 155s + // does not need to move. // // This wait is BEST-EFFORT: the spec's REAL assertions - the SubscribeAck / // ExpectNoMsg / DeltaCount == 0 gossip-isolation checks above - have already run From 78c25696e38372214445d648d9c4e377fdedab4a Mon Sep 17 00:00:00 2001 From: Aaron Stannard Date: Thu, 10 Sep 2026 04:48:55 +0000 Subject: [PATCH 4/4] DistributedPubSubRestartSpec: terminate timer outlives the retry cycle; restore the 120 s termination wait; last sync call migrated --- .../DistributedPubSubRestartSpec.cs | 42 ++++++++++--------- 1 file changed, 22 insertions(+), 20 deletions(-) diff --git a/src/contrib/cluster/Akka.Cluster.Tools.Tests.MultiNode/PublishSubscribe/DistributedPubSubRestartSpec.cs b/src/contrib/cluster/Akka.Cluster.Tools.Tests.MultiNode/PublishSubscribe/DistributedPubSubRestartSpec.cs index 2a445c7576c..9e611574cbc 100644 --- a/src/contrib/cluster/Akka.Cluster.Tools.Tests.MultiNode/PublishSubscribe/DistributedPubSubRestartSpec.cs +++ b/src/contrib/cluster/Akka.Cluster.Tools.Tests.MultiNode/PublishSubscribe/DistributedPubSubRestartSpec.cs @@ -124,16 +124,19 @@ public Shutdown(Address firstAddress) // any previous timer and re-arm a fresh one on every "shutdown", so a retried kill // can never have its ack racing a system that is already tearing itself down. // Capture the system reference, not Context, since the scheduled callback runs off - // the actor after this handler returns. 2s is far above a loopback round trip - // (this spec's own retry interval is 500ms) and far below first's 20s closed-loop - // kill budget, so whichever "shutdown" attempt first's loop last sends always gets - // a live system - and the warmed-up lane under it - for its ack to leave on before - // this timer fires. + // the actor after this handler returns. 5s, not 2s: the timer has to outlive + // first's retry CYCLE, not just its retry interval. First's cycle is a 2s + // ExpectMsgAsync on the kill probe plus the 500ms AwaitAssertAsync interval between + // attempts = 2.5s, so a 2s timer can fire half a second before the next attempt + // lands - if THIS ack is lost, the system is already gone before first retries. 5s + // covers one full 2.5s cycle plus the same again, so the acked-and-then-lost case + // still gets a second attempt while this incarnation is alive, and it stays far + // below first's 20s closed-loop kill budget. Sender.Tell("shutdown-ack"); _terminateTimer?.Cancel(); var system = Context.System; _terminateTimer = system.Scheduler.Advanced.ScheduleOnceCancelable( - TimeSpan.FromSeconds(2), () => system.Terminate()); + TimeSpan.FromSeconds(5), () => system.Terminate()); }); } @@ -402,15 +405,15 @@ await RunOnAsync(async () => // First's closed-loop kill (above) normally drives this WhenTerminated: it keeps // re-poking the association until third's /user/shutdown acks the kill, at which - // point newSystem terminates and this wait completes. First's own worst-case - // pipeline before it can even send that kill is 30s (Shutdown cap) + 60s - // (ready-ping wait) + 20s (closed-loop kill) = 110s, so this bound has to clear - // that first. 155s = 110s + 45s of slack (the same 45s of slack this wait - // originally carried, back when the pipeline above it was 75s instead of 110s). - // Shutdown's own sliding terminate timer (see that class) adds at most 2s on top - // of whichever "shutdown" attempt first's loop last sends before newSystem actually - // goes down - negligible against the 45s of slack already in this bound, so 155s - // does not need to move. + // point newSystem terminates and this wait completes. This wait's clock starts + // HERE, after the Shutdown actor above is already created - and that actor's own + // PreStart is what sends the ready ping - so by this instant first's 30s + // Shutdown(third) cap and its 60s ready-ping wait are already behind it; first is + // already inside (or past) its closed-loop kill. What is actually left of first's + // pipeline from this point is the 20s closed-loop kill, its last 2s ExpectMsgAsync + // attempt, and Shutdown's own 5s sliding terminate timer (see that class): about + // 27s worst case. 120s leaves roughly 90s of margin on top of that, which is why + // it is restored here rather than widened - there was no shortfall to fix. // // This wait is BEST-EFFORT: the spec's REAL assertions - the SubscribeAck / // ExpectNoMsg / DeltaCount == 0 gossip-isolation checks above - have already run @@ -420,12 +423,12 @@ await RunOnAsync(async () => // of the isolation assertions that already succeeded. try { - await newSystem.WhenTerminated.WaitAsync(155.Seconds()); + await newSystem.WhenTerminated.WaitAsync(120.Seconds()); } catch (TimeoutException) { newSystem.Log.Warning( - "newSystem did not observe first's shutdown within 155s; terminating self (best-effort). " + + "newSystem did not observe first's shutdown within 120s; terminating self (best-effort). " + "The gossip-isolation assertions (SubscribeAck / ExpectNoMsg / DeltaCount == 0) already passed, " + "so the spec's subject-under-test is verified regardless."); } @@ -463,11 +466,10 @@ private IActorRef Mediator private async Task JoinAsync(RoleName from, RoleName to) { - await RunOnAsync(() => + await RunOnAsync(async () => { - Cluster.Get(Sys).Join(Node(to).Address); + Cluster.Get(Sys).Join((await NodeAsync(to)).Address); CreateMediator(); - return Task.CompletedTask; }, from); await EnterBarrierAsync(from.Name + "-joined"); }