Artery: honour the reconnect backoff, return the held control element on stage stop, publish the inbound context before bind, host the materializer under /system

BREAKING_CHANGES_V1.6.md

@@ -44,7 +44,7 @@ of `Behavior`, `Wire`, `API` (combine with `+`).
 | Planned | [#8324](https://github.com/akkadotnet/akka.net/pull/8324) | `Akka.Routing` | Behavior | `ConsistentHash<T>` no longer retains the `SortedDictionary` passed to its public constructor — the ring is snapshotted into internal sorted arrays. Mutating that dictionary after construction no longer affects the instance (previously the aliasing was inconsistent: `IsEmpty` and `operator +`/`-` read it live, while `NodeFor` froze it after the first lookup). A `null` dictionary now throws `ArgumentNullException` from the constructor instead of surfacing later as a `NullReferenceException`. No public API removed; the ring built by `ConsistentHash.Create` is byte-identical. (#8293) | None for normal use — `ConsistentHash.Create` already builds the dictionary fully before constructing, so routers/receptionists are unaffected. If you call the `ConsistentHash(SortedDictionary, int)` constructor directly, populate the dictionary before passing it and don't rely on post-construction mutation being visible. |
 | Planned | `fix/gossip-removal-tombstones` | `Akka.Cluster` | Wire + Behavior | Cluster gossip now carries a removal tombstone for every member the leader removes (new `repeated Tombstone tombstones = 7` on the `Gossip` proto message, previously the one vacant field number). A member the leader removed can no longer be put back into the gossip by a peer that has not caught up, which previously blocked convergence permanently and needed a full cluster restart to clear. Tombstones expire after the new `akka.cluster.prune-gossip-tombstones-after` setting (default `24h`), pruned by the leader on a converged tick and reclaimed for good: gossip reception keeps the tombstones of whichever gossip wins the causal comparison, so a peer that has not pruned yet cannot hand a pruned tombstone back. Tombstones are only unioned across the two gossips when neither descends from the other -- equal clocks, or concurrent clocks, where each side may hold a removal the other has not heard about. The field is additive and proto3 ignores unknown fields in both directions, so rolling upgrades are safe; an older node parses the gossip fine but drops the tombstones when it re-emits it, and if that stripped gossip is causally newer than an upgraded node's, the upgraded node adopts it and loses those tombstones too -- so the protection only holds between upgraded nodes and the fix is not fully in effect until every node is upgraded. | No action required. Raise `akka.cluster.prune-gossip-tombstones-after` if the cluster must survive partitions longer than 24 hours, or lower it if a cluster with heavy join/leave churn needs the gossip message kept small -- each tombstone adds a full address plus a timestamp to every gossip message until it expires. |
 | Planned | `feature/default-bounded-shard-rebalancing` | `Akka.Cluster.Sharding` | Behavior | The default `rebalance-absolute-limit` is now `20`, selecting the bounded shard allocation strategy instead of the legacy threshold-based strategy. | To retain the legacy strategy temporarily, explicitly set `akka.cluster.sharding.least-shard-allocation-strategy.rebalance-absolute-limit = 0`. Review `rebalance-threshold` and `max-simultaneous-rebalance`, which do not apply while the bounded strategy is active. |
-| Planned | `fix/artery-shutdown-flush` | `Akka.Remote` (Artery) | Behavior | Graceful transport shutdown now flushes. `ArteryRemoting.Shutdown()` completes each association's outbound channels, waits up to the new `akka.remote.artery.advanced.flush-wait-on-shutdown` (default `2 s`, mirroring classic remoting's `akka.remote.flush-wait-on-shutdown`) for the outbound streams to finish writing what they had already accepted, and only then drains the remainder to `Dropped`. Previously it completed and drained in the same breath, so a message accepted microseconds before shutdown -- an ack, a graceful notice, a handshake reply -- was published as `Dropped` instead of reaching the socket. Two consequences: graceful shutdown may now take up to `flush-wait-on-shutdown` longer per transport (only when a backlog cannot reach its peer; an association with nothing left to write completes at once), and messages that previously surfaced as `Dropped` may now be delivered. Abrupt termination (materializer death without a graceful `Shutdown()`) is unchanged. | No action required. Set `akka.remote.artery.advanced.flush-wait-on-shutdown = 0` to restore the previous drain-immediately behavior, or lower it to cap how long shutdown may wait. Monitoring that counts shutdown-time `Dropped` events will see fewer of them. |
+| Planned | `fix/artery-shutdown-flush` | `Akka.Remote` (Artery) | Behavior | Graceful transport shutdown now flushes. `ArteryRemoting.Shutdown()` completes each association's outbound channels, waits up to the new `akka.remote.artery.advanced.flush-wait-on-shutdown` (default `2 s`, mirroring classic remoting's `akka.remote.flush-wait-on-shutdown`) for the outbound streams to finish writing what they had already accepted, and only then drains the remainder to `Dropped`. Previously it completed and drained in the same breath, so a message accepted microseconds before shutdown -- an ack, a graceful notice, a handshake reply -- was published as `Dropped` instead of reaching the socket. Two consequences: graceful shutdown may now take up to `flush-wait-on-shutdown` longer per transport (only when a backlog cannot reach its peer; an association with nothing left to write completes at once), and messages that previously surfaced as `Dropped` may now be delivered. Abrupt termination (materializer death without a graceful `Shutdown()`) is unchanged. **Takes effect starting with `fix/artery-control-backoff-and-held-element` (#8554)**, which hosts the transport's materializer under `/system` (`ArteryRemoting.CreateSystemMaterializer`) so its `StreamSupervisor` survives `/user` guardian teardown instead of being killed by it; before #8554 the outbound streams were already dead by the time this flush ran (killed out from under a live graph interpreter during ordinary `/user` teardown, well before `Shutdown()` itself executed), so this row's two consequences were never actually observable on `dev`. That same `/system` move adds two further consequences not previously listed here: the `AbruptStageTerminationException` warning that used to fire on every graceful `ActorSystem.Terminate()` (because the stream was killed live) no longer fires, since the stream now completes gracefully instead; and a terminating node's INBOUND connections also survive `/user` teardown, so traffic arriving late during shutdown is now drained and dead-lettered rather than met with a connection reset. | No action required. Set `akka.remote.artery.advanced.flush-wait-on-shutdown = 0` to restore the previous drain-immediately behavior, or lower it to cap how long shutdown may wait. Monitoring that counts shutdown-time `Dropped` events will see fewer of them. Log-scraping or alerting keyed on the `AbruptStageTerminationException`/"outbound stream has ended" warning firing during ordinary graceful shutdown will go quiet -- that is expected, not a regression. Peers that inferred an in-flight terminate from a reset connection will instead observe a clean, drained FIN. |
 | Planned | `fix/artery-inbound-quarantine-check` | `Akka.Remote` (Artery) | Behavior | Quarantine is now enforced on the INBOUND path too. Previously Artery only gated outbound sends -- an envelope arriving FROM a uid this system has quarantined was still delivered, and the quarantined peer was only notified once, proactively, at the moment `Quarantine()` was called. A new `InboundQuarantineCheckStage`, woven into the inbound pipeline right after handshake, now drops every inbound envelope (ordinary or control, including system messages) whose origin uid is quarantined, and reactively re-sends a `Quarantined` control notice to the origin for each drop (except for a heartbeat or the peer's own `Quarantined` notice, to avoid a reply storm). The existing one-shot proactive notice in `Quarantine()` is unchanged. Additionally, an ordinary/large outbound stream that terminates while its association is quarantined no longer wedges permanently: the materialize-once gate is released (timer-driven auto-reconnect stays suppressed), so a quarantine-piercing `ActorSelection` send -- or any send after a new incarnation's handshake lifts the quarantine -- re-materializes the stream on demand and can reach a restarted peer at the same address. | No action required -- this is a bug fix restoring the documented "no further communication" guarantee of quarantine and the documented new-incarnation piercing behavior; code that (incorrectly) depended on a quarantined peer's replies still arriving is unsupported. |
 | Planned | `fix/artery-daemonmsgcreate-control-stream` | `Akka.Remote` (Artery) | Behavior | Remote deployment's `DaemonMsgCreate` now travels over Artery's CONTROL stream (as a plain envelope, no delivery/ack sequencing) instead of the ordinary stream, ordering it ahead of the `Watch` that remote deployment sends immediately afterwards. Previously the two rode independent, unordered TCP connections and `Watch` systematically arrived first, so the receiver replied `DeathWatchNotification(existenceConfirmed: false)` for a not-yet-created actor and the deployer reaped the freshly-deployed routee before its `Supervise` registration landed, emptying cluster router pools. Additionally (Pekko parity), inbound ordinary messages addressed to a remote-deployed recipient that has not been created yet are no longer dead-lettered immediately: the resolve is retried on a bounded schedule (20 attempts x 50ms, buffered per recipient in FIFO order) so first messages that arrive ahead of the in-flight `DaemonMsgCreate` are delivered once the actor exists; paths that never resolve are banned and dead-letter as before. | No action required -- this is a bug fix restoring correct create-before-watch ordering; code that (incorrectly) depended on the old race is unsupported. |
 | Planned | `feature/artery-test-mode` | `Akka.Remote` | Behavior | System UID generation (`AddressUidExtension` / `AddressUid`) now uses a cryptographic RNG instead of the `Environment.TickCount`-seeded `ThreadLocalRandom`. Previously, multiple processes started within the same millisecond tick (e.g. every node of a multi-node test spawning at once) could draw IDENTICAL system UIDs, silently corrupting Artery's uid-keyed identity (handshakes, quarantine, association reverse index). UIDs remain nonzero and in the legacy `[1, int.MaxValue]` range at the default (`use-64bit-system-uids = off`). | Nothing required -- UIDs were always documented as random; only code depending on the (buggy) time-seeded determinism could observe a difference. |

src/core/Akka.Remote.Tests/Artery/ArteryGracefulTerminateFlushSpec.cs

@@ -0,0 +1,150 @@
+//-----------------------------------------------------------------------
+// <copyright file="ArteryGracefulTerminateFlushSpec.cs" company="Akka.NET Project">
+//     Copyright (C) 2009-2022 Lightbend Inc. <http://www.lightbend.com>
+//     Copyright (C) 2013-2025 .NET Foundation <https://github.com/akkadotnet/akka.net>
+// </copyright>
+//-----------------------------------------------------------------------
+
+#nullable enable
+
+using System;
+using System.Threading.Tasks;
+using Akka.Actor;
+using Akka.Configuration;
+using Akka.TestKit;
+using Akka.TestKit.Extensions;
+using FluentAssertions;
+using Xunit;
+
+namespace Akka.Remote.Tests.Artery
+{
+    /// <summary>
+    /// Covers the actual, end-user-visible shutdown path: <see cref="ActorSystem.Terminate"/>, NOT
+    /// a direct call to the transport's own <c>Shutdown()</c> (that path is
+    /// <see cref="ArteryShutdownFlushSpec"/>, and it already passes today -- it never exercises
+    /// <c>/user</c> guardian teardown at all).
+    ///
+    /// <para>
+    /// <b>Why the assertion is "no AbruptStageTerminationException", not message delivery.</b> A
+    /// warm, already-handshake-complete association's outbound stream never completes on its own --
+    /// the association-owned channel it reads from stays open indefinitely until
+    /// <c>ArteryRemoting.Shutdown()</c> explicitly calls <c>CompleteOutbound()</c> -- so on a fast,
+    /// local, healthy loopback connection a small burst of messages typically reaches the peer
+    /// before ANY teardown-related race even matters, fix or no fix. That makes raw message
+    /// delivery an unreliable discriminator here (verified empirically while writing this spec).
+    /// What the placement bug actually determines is HOW the still-open stream ends: abruptly
+    /// (an active <c>ActorGraphInterpreter</c> killed out from under a live stream, which
+    /// unconditionally faults its <c>WatchTermination</c> monitor with
+    /// <see cref="Akka.Streams.AbruptStageTerminationException"/>) or gracefully (the channel is
+    /// completed first, and the stream drains and finishes on its own). That is exactly what the
+    /// "Artery {stream} outbound connection to [...] failed; ... outbound stream has ended" WARNING
+    /// reports, so its presence/absence is the deterministic signal.
+    /// </para>
+    ///
+    /// <para>
+    /// Before this fix, <c>ArteryRemoting.Start()</c> materialized every Artery stream on
+    /// <c>ActorMaterializer.Create(System)</c>, whose <c>StreamSupervisor</c> is a <c>/user</c>
+    /// actor. A graceful <see cref="ActorSystem.Terminate"/> stops the <c>/user</c> guardian FIRST --
+    /// and a parent's own stop does not complete until every child (recursively) has stopped, so the
+    /// <c>StreamSupervisor</c> and every graph interpreter under it are ALREADY DEAD by the time
+    /// <c>RemotingTerminator</c> (a <c>/system</c> actor) observes that and finally calls
+    /// <c>Transport.Shutdown()</c> -- the method that would otherwise complete the channel
+    /// gracefully first. That is a strict ordering, not a race, which is why the WARNING below fires
+    /// on every run pre-fix and never post-fix. Moving the materializer's <c>StreamSupervisor</c> to
+    /// a <c>/system</c> actor (<c>ArteryRemoting.CreateSystemMaterializer</c>) is what changes the
+    /// ordering: the outbound stream now survives the <c>/user</c> teardown, so
+    /// <c>Shutdown()</c>'s own graceful <c>CompleteOutbound()</c> + kill-switch sequence gets to run
+    /// while it is still alive.
+    /// </para>
+    /// </summary>
+    public class ArteryGracefulTerminateFlushSpec : AkkaSpec
+    {
+        public ArteryGracefulTerminateFlushSpec(ITestOutputHelper output) : base(ArteryConfig, output)
+        {
+        }
+
+        private static readonly Config ArteryConfig = ConfigurationFactory.ParseString("""
+            akka.actor.provider = "Akka.Remote.RemoteActorRefProvider, Akka.Remote"
+            akka.loggers = ["Akka.TestKit.TestEventListener, Akka.TestKit"]
+            akka.remote.artery.enabled = on
+            akka.remote.artery.canonical.hostname = "127.0.0.1"
+            akka.remote.artery.canonical.port = 0
+            akka.remote.artery.advanced.flush-wait-on-shutdown = 5s
+            akka.loglevel = INFO
+            """);
+
+        private static Address AddressOf(ActorSystem system) => RARP.For(system).Provider.DefaultAddress;
+
+        /// <summary>Forwards everything it receives to <paramref name="target"/>.</summary>
+        private sealed class Forwarder : ReceiveActor
+        {
+            public Forwarder(IActorRef target)
+            {
+                ReceiveAny(msg => target.Forward(msg));
+            }
+        }
+
+        /// <summary>
+        /// Sends one-way markers until one lands at this spec's test actor, establishing the
+        /// association (connection plus handshake) before the assertion under test -- the same idea
+        /// <c>ArteryShutdownFlushSpec</c> uses for the same purpose. A WARM association is
+        /// deliberate: this spec is about how an already-open stream ends, not about connection
+        /// establishment, and a cold first-contact handshake racing a peer's own reply connection
+        /// introduces timing behavior this spec has nothing to do with.
+        /// </summary>
+        private async Task AwaitAssociationAsync(ActorSelection selection)
+        {
+            var attempt = 0;
+            await AwaitAssertAsync(async () =>
+            {
+                var marker = $"warmup-{++attempt}";
+                selection.Tell(marker, ActorRefs.NoSender);
+                await FishForMessageAsync(msg => Equals(msg, marker), TimeSpan.FromSeconds(1));
+            }, TimeSpan.FromSeconds(30), TimeSpan.FromMilliseconds(200));
+        }
+
+        [Fact(DisplayName = "Should_NotAbortTheOutboundStream_When_TheSenderGracefullyTerminates")]
+        public async Task Should_NotAbortTheOutboundStream_When_TheSenderGracefullyTerminates()
+        {
+            // Sys is the RECEIVER here and outlives the sender's system.
+            var senderSys = ActorSystem.Create("artery-graceful-terminate-sender", ArteryConfig);
+            try
+            {
+                Sys.ActorOf(Props.Create(() => new Forwarder(TestActor)), "graceful-terminate-receiver");
+                var selection = senderSys.ActorSelection(
+                    $"akka://{Sys.Name}@127.0.0.1:{AddressOf(Sys).Port}/user/graceful-terminate-receiver");
+
+                await AwaitAssociationAsync(selection);
+
+                // The deterministic proof: pre-fix, /user teardown always kills the still-open
+                // outbound stream out from under a live ActorGraphInterpreter, which ALWAYS faults
+                // its WatchTermination monitor and logs this WARNING. Post-fix, Shutdown()'s own
+                // graceful CompleteOutbound() + kill-switch sequence gets to run first, so the
+                // stream finishes on its own and this WARNING never fires. The warning is also
+                // cadence-gated on the association's reconnect attempt counter being 1 (only the
+                // FIRST outage after a successful connect logs at Warning; a later one logs at
+                // Debug) -- true here because AwaitAssociationAsync leaves the association warm and
+                // never disconnected, but a future edit that adds a reconnect to the warm-up would
+                // silently defang this assertion.
+                //
+                // 3 s, explicit rather than the implicit akka.test.single-expect-default: an
+                // expected count of 0 polls the FULL window before concluding nothing was logged, so
+                // this is the actual width of the "no warning" observation -- pin it so a future
+                // config or `single-expect-default` change cannot silently shrink it.
+                await CreateEventFilter(senderSys)
+                    .Warning(contains: "outbound stream has ended")
+                    .ExpectAsync(0, TimeSpan.FromSeconds(3), async () =>
+                    {
+                        (await senderSys.Terminate().AwaitWithTimeout(TimeSpan.FromSeconds(30)))
+                            .Should().BeTrue("graceful termination must actually complete");
+                    });
+            }
+            finally
+            {
+                // senderSys is already terminated by the assertion above in the success case; this
+                // is a no-op then and a safety net if an earlier assertion throws first.
+                await senderSys.Terminate().AwaitWithTimeout(TimeSpan.FromSeconds(30));
+            }
+        }
+    }
+}