Merge branch 'dev' into docs/8439-auto-downing-investigation

Author: orange-dot

BREAKING_CHANGES_V1.6.md

@@ -44,6 +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-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/contrib/cluster/Akka.Cluster.Sharding.Tests.MultiNode/ClusterShardingSpec.cs

@@ -1007,15 +1007,29 @@ await RunOnAsync(async () =>
             await ExpectMsgAsync(new ShardStopped("1"), TimeSpan.FromSeconds(10), "ShardStopped not received");
 
             //Get the path to where the shard now resides
+            // HandOffStopper answers ShardStopped as soon as the last entity terminates,
+            // which is strictly before the Shard actor stops and before the ShardRegion
+            // observes Terminated and drops the shard from its routing table. A Get sent
+            // in that window is forwarded to the dying Shard and dead-lettered - sharding
+            // is at-most-once, so the request has to be re-sent, not merely re-awaited.
+            // Each attempt gets a fresh probe (a late reply from a previous attempt must
+            // not be mistaken for this attempt's) and a 1s bound: that covers one region
+            // retry-interval (1s) plus the shard restart, so the 5s budget admits four
+            // sends instead of the single one an unbounded inner expect allowed - the
+            // inner expect inherited akka.test.single-expect-default (5 s), the whole
+            // budget, so the loop never retried.
+            IActorRef entity13 = null;
             await AwaitAssertAsync(async () =>
             {
-                _persistentEntitiesRegion.Value.Tell(new Get(13));
-                await ExpectMsgAsync(0);
+                var probe0 = CreateTestProbe();
+                _persistentEntitiesRegion.Value.Tell(new Get(13), probe0.Ref);
+                await probe0.ExpectMsgAsync(0, TimeSpan.FromSeconds(1));
+                entity13 = probe0.LastSender;
             }, TimeSpan.FromSeconds(5), TimeSpan.FromMilliseconds(500));
 
             //Check that counter 1 is now alive again, even though we have
             // not sent a message to it via the ShardRegion
-            var counter1 = Sys.ActorSelection(LastSender.Path.Parent / "1");
+            var counter1 = Sys.ActorSelection(entity13.Path.Parent / "1");
             await WithinAsync(TimeSpan.FromSeconds(5), async () =>
             {
                 await AwaitAssertAsync(async () =>
@@ -1058,78 +1072,98 @@ await RunOnAsync(async () =>
 
     private async Task PersistentClusterSharding_should_permanently_stop_entities_which_passivate()
     {
-        await WithinAsync(TimeSpan.FromSeconds(15), async () =>
+        // No outer Within, for the reason documented on
+        // PersistentClusterSharding_should_recover_entities_upon_restart: barriers derive
+        // their timeout from RemainingOr(barrier-timeout), so a Within here hands the
+        // rendezvous whatever is left of the phase budget instead of the configured 70s
+        // testconductor barrier-timeout. The 15s the removed Within granted this phase is
+        // re-attached below to the two waits it was actually protecting - the first Get
+        // against each region, which has to allocate the shard, recover the
+        // remember-entities store and persist through the shared journal - so no wait ends
+        // up with a smaller budget than it had, and the barriers get the full 70s.
+        RunOn(() =>
         {
-            RunOn(() =>
-            {
-                _ = _persistentRegion.Value;
-            }, Config.Third, Config.Fourth, Config.Fifth);
-            await EnterBarrierAsync("cluster-started-12");
+            _ = _persistentRegion.Value;
+        }, Config.Third, Config.Fourth, Config.Fifth);
+        await EnterBarrierAsync("cluster-started-12");
 
-            await RunOnAsync(async () =>
-            {
-                //create and increment counter 1
-                _persistentRegion.Value.Tell(new EntityEnvelope(1, Increment.Instance));
-                _persistentRegion.Value.Tell(new Get(1));
-                await ExpectMsgAsync(1);
+        await RunOnAsync(async () =>
+        {
+            //create and increment counter 1
+            _persistentRegion.Value.Tell(new EntityEnvelope(1, Increment.Instance));
+            _persistentRegion.Value.Tell(new Get(1));
+            await ExpectMsgAsync(1, TimeSpan.FromSeconds(15));
 
-                var counter1 = LastSender;
-                var shard = Sys.ActorSelection(counter1.Path.Parent);
-                var region = Sys.ActorSelection(counter1.Path.Parent.Parent);
+            var counter1 = LastSender;
+            var shard = Sys.ActorSelection(counter1.Path.Parent);
+            var region = Sys.ActorSelection(counter1.Path.Parent.Parent);
 
-                //create and increment counter 13
-                _persistentRegion.Value.Tell(new EntityEnvelope(13, Increment.Instance));
-                _persistentRegion.Value.Tell(new Get(13));
-                await ExpectMsgAsync(1);
+            //create and increment counter 13
+            _persistentRegion.Value.Tell(new EntityEnvelope(13, Increment.Instance));
+            _persistentRegion.Value.Tell(new Get(13));
+            await ExpectMsgAsync(1, TimeSpan.FromSeconds(15));
 
-                var counter13 = LastSender;
+            var counter13 = LastSender;
 
-                counter13.Path.Parent.Should().Be(counter1.Path.Parent);
+            counter13.Path.Parent.Should().Be(counter1.Path.Parent);
 
-                //Send the shard the passivate message from the counter
-                await WatchAsync(counter1);
-                shard.Tell(new Passivate(Stop.Instance), counter1);
+            //Send the shard the passivate message from the counter
+            await WatchAsync(counter1);
+            shard.Tell(new Passivate(Stop.Instance), counter1);
 
-                // watch for the Terminated message
-                await ExpectTerminatedAsync(counter1, TimeSpan.FromSeconds(5));
+            // watch for the Terminated message
+            await ExpectTerminatedAsync(counter1, TimeSpan.FromSeconds(5));
 
+            await AwaitAssertAsync(async () =>
+            {
+                // check counter 1 is dead
                 var probe1 = CreateTestProbe();
-                await AwaitAssertAsync(async () =>
-                {
-                    // check counter 1 is dead
-                    counter1.Tell(new Identify(1), probe1.Ref);
-                    await probe1.ExpectMsgAsync(new ActorIdentity(1, null), TimeSpan.FromSeconds(1), "Entity 1 was still around");
-                }, TimeSpan.FromSeconds(5), TimeSpan.FromMilliseconds(500));
+                counter1.Tell(new Identify(1), probe1.Ref);
+                await probe1.ExpectMsgAsync(new ActorIdentity(1, null), TimeSpan.FromSeconds(1), "Entity 1 was still around");
+            }, TimeSpan.FromSeconds(5), TimeSpan.FromMilliseconds(500));
 
-                // stop shard cleanly
-                region.Tell(new HandOff("1"));
-                await ExpectMsgAsync(new ShardStopped("1"), TimeSpan.FromSeconds(10), "ShardStopped not received");
+            // stop shard cleanly
+            region.Tell(new HandOff("1"));
+            await ExpectMsgAsync(new ShardStopped("1"), TimeSpan.FromSeconds(10), "ShardStopped not received");
 
-            }, Config.Third);
-            await EnterBarrierAsync("shard-shutdonw-12");
+        }, Config.Third);
+        await EnterBarrierAsync("shard-shutdonw-12");
 
-            await RunOnAsync(async () =>
+        await RunOnAsync(async () =>
+        {
+            // force shard backup
+            // ShardStopped - which released the barrier above - is answered by the
+            // HandOffStopper the moment the last entity terminates, before the Shard
+            // actor stops and before its ShardRegion drops it from the routing table.
+            // Shard "1" may be hosted by this node, so this Get can be forwarded to the
+            // dying Shard and dead-lettered. Sharding is at-most-once: re-send until the
+            // shard has been re-allocated. 1s per attempt covers one region
+            // retry-interval (1s) plus the restart, so the 5s budget admits four sends.
+            IActorRef entity25 = null;
+            await AwaitAssertAsync(async () =>
             {
-                // force shard backup
-                _persistentRegion.Value.Tell(new Get(25));
-                await ExpectMsgAsync(0);
+                var probe0 = CreateTestProbe();
+                _persistentRegion.Value.Tell(new Get(25), probe0.Ref);
+                await probe0.ExpectMsgAsync(0, TimeSpan.FromSeconds(1));
+                entity25 = probe0.LastSender;
+            }, TimeSpan.FromSeconds(5), TimeSpan.FromMilliseconds(500));
 
-                var shard = LastSender.Path.Parent;
+            var shard = entity25.Path.Parent;
 
-                // check counter 1 is still dead
-                Sys.ActorSelection(shard / "1").Tell(new Identify(3));
-                await ExpectMsgAsync(new ActorIdentity(3, null));
+            // check counter 1 is still dead
+            Sys.ActorSelection(shard / "1").Tell(new Identify(3));
+            await ExpectMsgAsync(new ActorIdentity(3, null), TimeSpan.FromSeconds(5));
 
-                // check counter 13 is alive again
+            // check counter 13 is alive again
+            await AwaitAssertAsync(async () =>
+            {
                 var probe3 = CreateTestProbe();
-                await AwaitAssertAsync(async () =>
-                {
-                    Sys.ActorSelection(shard / "13").Tell(new Identify(4), probe3.Ref);
-                    await probe3.ExpectMsgAsync<ActorIdentity>(i => i.MessageId.Equals(4) && i.Subject != null);
-                }, TimeSpan.FromSeconds(5), TimeSpan.FromMilliseconds(500));
-            }, Config.Fourth);
-            await EnterBarrierAsync("after-13");
-        });
+                Sys.ActorSelection(shard / "13").Tell(new Identify(4), probe3.Ref);
+                await probe3.ExpectMsgAsync<ActorIdentity>(
+                    i => i.MessageId.Equals(4) && i.Subject != null, TimeSpan.FromSeconds(1));
+            }, TimeSpan.FromSeconds(5), TimeSpan.FromMilliseconds(500));
+        }, Config.Fourth);
+        await EnterBarrierAsync("after-13");
     }
 
     private async Task PersistentClusterSharding_should_restart_entities_which_stop_without_passivation()