Artery: port InboundQuarantineCheck

BREAKING_CHANGES_V1.6.md

@@ -41,6 +41,7 @@ of `Behavior`, `Wire`, `API` (combine with `+`).
 | Status | PR / Branch | Component | Type | Change | Migration |
 |--------|-------------|-----------|------|--------|-----------|
 | 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-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. |
 | Planned | `feature/artery-test-mode` | `Akka.Remote` (Artery) / `Akka.Remote.TestKit` | Behavior | New `akka.remote.artery.advanced.test-mode` setting (default `off`, Pekko key parity) enables failure-injection test stages in Artery's stream pipelines, and `ArteryRemoting.ManagementCommand` now honors `SetThrottle` with `Blackhole`/`Unthrottled` (returning `true`; previously always `false`) so TestConductor `Blackhole`/`PassThrough` work over Artery. Rate throttles (`TokenBucket`) and `ForceDisassociate` remain unsupported on Artery (still `false`, matching Pekko). `MultiNodeConfig.TestTransport = true` now also sets Artery test-mode on, and the TestKit `Player` now treats a `false` `ManagementCommand` result as a failure (previously reported success for a silently ignored command). | Nothing at the default (`test-mode = off` composes byte-identical pipelines). Test code that relied on `ManagementCommand` always returning `false` on Artery, or on the Player treating `false` as success, must account for the new semantics. |

openspec/changes/artery-tcp-remoting/design.md

@@ -164,7 +164,7 @@ Verified against Pekko `Handshake.scala` / `Association.scala` / `ArteryTranspor
 
 **Association state machine:** `AssociationRegistry` keyed by remote **Address** (one Association per address, CAS-materialized) + an `association(uid)` reverse lookup (None until handshake completes). Per-association `AssociationState` (volatile, CAS-swapped) with `uniqueRemoteAddress`: **Associating** (UID unknown — gates OutboundHandshake) → **Associated** (`completeHandshake` sets it) → **Quarantined**. A **different** incoming UID (remote restart) → `newIncarnation` + atomic swap + clear outbound compression (UID-change → reset); the old UID is not auto-quarantined.
 
-**Quarantine (UID-scoped):** acts only if the uid matches the current `uniqueRemoteAddress().uid` (stale-UID request ignored); swaps `newQuarantined`, emits `QuarantinedEvent`, clears compression, sends `ClearSystemMessageDelivery(incarnation)`. Only `ActorSelectionMessage` + `ClearSystemMessageDelivery` pierce. A **new incarnation re-associates** (keyed by Address; a new UID installs a fresh non-quarantined incarnation while the old UID stays quarantined). Prune after `remove-quarantined-association-after` = 1h.
+**Quarantine (UID-scoped):** acts only if the uid matches the current `uniqueRemoteAddress().uid` (stale-UID request ignored); swaps `newQuarantined`, emits `QuarantinedEvent`, clears compression, sends `ClearSystemMessageDelivery(incarnation)`. Only `ActorSelectionMessage` + `ClearSystemMessageDelivery` pierce. The pierce is not uid-checked and may reach the quarantined incarnation itself, which processes it; the inbound quarantine check drops the reply, so the exchange is one-directional. This is deliberate: the sender cannot learn that the peer restarted without sending to it, and the not-restarted case is the price of that discovery. Classic remoting differs -- it refuses the quarantined uid at handshake (`HopelessAssociation` via `refuseUid`), so a pierce there can only ever reach a new incarnation. A **new incarnation re-associates** (keyed by Address; a new UID installs a fresh non-quarantined incarnation while the old UID stays quarantined). Prune after `remove-quarantined-association-after` = 1h. **Implementation gap:** the config key is not parsed and no pruning exists yet -- `QuarantinedUids` accumulates per association for the life of the process (classic bounds the same state with `prune-quarantine-marker-after = 5 d`, documented there as leak GC, not re-admission).
 
 **Provider integration:** the `RemoteTransport` seam already exists — `RemoteActorRef.Tell → Remote.Send`; the provider creates refs via `new RemoteActorRef(Transport,…)`; `DefaultAddress` / `LocalAddressForRemote` / `Quarantine` all delegate to the transport. So `ArteryRemoting : RemoteTransport` implements **9 abstract members** (two `ManagementCommand` overloads — the design draft said 8) and needs **no change** to `RemoteActorRef` or the ref-creation path. **The one wiring change:** `RemoteActorRefProvider.CreateInternals()` hard-codes `new Remoting(…)` — add a config switch (`akka.remote.artery.enabled = on` → `ArteryRemoting`, else classic) by making it read `RemoteSettings` or overriding in a subclass. **Two nodes must run the same transport** (wire + scheme differ: classic `akka.tcp://`, Artery `akka://`) — homogeneous cluster; fail fast on a mixed config.
 

openspec/changes/artery-tcp-remoting/specs/artery-tcp-remoting/spec.md

@@ -96,6 +96,32 @@ System messages SHALL be delivered with explicit ACK/NACK and resend semantics i
 - **WHEN** the bounded system-message buffer overflows
 - **THEN** the association SHALL be quarantined
 
+### Requirement: Quarantine is one-directional and uid-scoped
+
+Quarantine SHALL block traffic for the quarantined uid in both directions on the
+quarantining system, with a deliberate outbound exemption that lets the sender
+discover a restarted peer. The exemption MAY deliver a message to the quarantined
+incarnation itself; the sender cannot know whether the peer restarted without
+sending to it. The reply from a quarantined incarnation SHALL NOT be delivered.
+
+#### Scenario: ActorSelection pierces outbound quarantine
+- **WHEN** an `ActorSelectionMessage` is sent toward a quarantined association
+- **THEN** it SHALL be transmitted to the peer address, even if the peer is still running as the quarantined uid
+
+#### Scenario: Inbound traffic from a quarantined uid is dropped
+- **WHEN** an envelope arrives whose origin uid this system has quarantined
+- **THEN** the envelope SHALL be dropped before dispatch, system messages included
+- **AND** a quarantine control notice SHALL be sent to the origin, unless the dropped message is itself a quarantine notice or a heartbeat
+
+#### Scenario: New incarnation lifts quarantine
+- **WHEN** a completed handshake installs a different uid for a quarantined association
+- **THEN** traffic to and from the new uid SHALL flow normally
+- **AND** the old uid SHALL remain quarantined
+
+#### Scenario: Quarantined peer does not retaliate
+- **WHEN** a system receives notice that a peer has quarantined it
+- **THEN** it SHALL publish `ThisActorSystemQuarantinedEvent` and SHALL NOT quarantine the peer in return, because a reciprocal quarantine can split a cluster
+
 ### Requirement: Outbound queues are bounded
 
 Artery remoting SHALL use bounded outbound queues for user and control traffic.

src/core/Akka.Remote.Tests.MultiNode/RemoteQuarantinePiercingSpec.cs

@@ -122,9 +122,18 @@ await RunOnAsync(async () =>
                 await EnterBarrierAsync("actor-identified");
                 await Sys.WhenTerminated.WaitAsync(TimeSpan.FromSeconds(30));
 
+                // Set the host and port for both transports, so that the new system starts at the
+                // same address as the old one. Each transport ignores the keys of the other, thus
+                // it is safe to set both.
+                //
+                // The artery keys are necessary. MultiNodeSpec adds a config tier that sets
+                // artery.canonical.port to 0. Without an explicit port here, that tier wins, the
+                // new system binds a random port, and `first` cannot reach it.
                 var freshSystem = ActorSystem.Create(Sys.Name, ConfigurationFactory.ParseString($@"
                     akka.remote.dot-netty.tcp.hostname = {addr.Host}
                     akka.remote.dot-netty.tcp.port = {addr.Port}
+                    akka.remote.artery.canonical.hostname = {addr.Host}
+                    akka.remote.artery.canonical.port = {addr.Port}
                 ").WithFallback(Sys.Settings.Config));
 
                 freshSystem.ActorOf(Props.Create<RemoteQuarantinePiercingSpecConfig.Subject>(), "subject");

src/core/Akka.Remote.Tests.MultiNode/RemoteRestartedQuarantinedSpec.cs

@@ -92,6 +92,14 @@ await RunOnAsync(async () =>
                 var secondAddress = Node(_config.Second).Address;
                 var uid = (await _identifyWithUid(_config.Second, "subject")).Item1;
 
+                // This barrier separates the identify exchanges from the Quarantine() call.
+                //
+                // Without it, the identify that `second` sends can still be in flight when the
+                // quarantine starts. Inbound quarantine enforcement then makes `first` discard
+                // that identify, and `second` receives ThisActorSystemQuarantinedEvent in place of
+                // the uid and ref that it expects.
+                await EnterBarrierAsync("before-quarantined");
+
                 RARP.For(Sys).Provider.Transport.Quarantine(Node(_config.Second).Address, uid);
 
                 await EnterBarrierAsync("quarantined");
@@ -120,30 +128,61 @@ await RunOnAsync(async () =>
 
                 var actorRef = (await _identifyWithUid(_config.First, "subject")).Item2;
 
+                // See the matching barrier comment in `first`'s block.
+                await EnterBarrierAsync("before-quarantined");
+
                 await EnterBarrierAsync("quarantined");
 
-                // Check that quarantine is intact
-                await WithinAsync(TimeSpan.FromSeconds(30), async () =>
+                // Test that the quarantine is still in effect. The two transports tell the
+                // quarantined system about the quarantine in different ways, thus this test needs
+                // two procedures.
+                //
+                // Classic: the endpoint learns about the quarantine only when it tries to write.
+                // It then logs the warning "The remote system has quarantined this system". The
+                // test must therefore send a message to cause the write, and then look for that
+                // warning.
+                //
+                // Artery: the quarantining system sends an ArteryQuarantined control message to the
+                // peer. The peer publishes ThisActorSystemQuarantinedEvent when the message
+                // arrives. No send from the test is necessary, and there is no equivalent warning
+                // to look for.
+                if (Sys.Settings.Config.GetBoolean("akka.remote.artery.enabled", false))
                 {
-                    await AwaitAssertAsync(() =>
+                    await ExpectMsgAsync<ThisActorSystemQuarantinedEvent>(TimeSpan.FromSeconds(10));
+                }
+                else
+                {
+                    await WithinAsync(TimeSpan.FromSeconds(30), async () =>
                     {
-                        EventFilter.Warning(null, null, "The remote system has quarantined this system")
-                            .ExpectOne(TimeSpan.FromSeconds(10), () => actorRef.Tell("boo!"));
+                        await AwaitAssertAsync(() =>
+                        {
+                            EventFilter.Warning(null, null, "The remote system has quarantined this system")
+                                .ExpectOne(TimeSpan.FromSeconds(10), () => actorRef.Tell("boo!"));
+                        });
                     });
-                });
 
-                await ExpectMsgAsync<ThisActorSystemQuarantinedEvent>(TimeSpan.FromSeconds(10));
+                    await ExpectMsgAsync<ThisActorSystemQuarantinedEvent>(TimeSpan.FromSeconds(10));
+                }
 
                 await EnterBarrierAsync("still-quarantined");
 
                 await Sys.WhenTerminated.WaitAsync(TimeSpan.FromSeconds(10));
 
+                // Set the host and port for both transports, so that the new system starts at the
+                // same address as the old one. Each transport ignores the keys of the other, thus
+                // it is safe to set both.
+                //
+                // The artery keys are necessary. MultiNodeSpec adds a config tier that sets
+                // artery.canonical.port to 0. Without an explicit port here, that tier wins, the
+                // new system binds a random port, and `first` cannot reach it.
                 var sb = new StringBuilder()
                     .AppendLine("akka.remote.retry-gate-closed-for = 0.5 s")
                     .AppendLine("akka.remote.dot-netty.tcp {")
                     .AppendLine("hostname = " + addr.Host)
                     .AppendLine("port = " + addr.Port)
-                    .AppendLine("}");
+                    .AppendLine("}")
+                    .AppendLine("akka.remote.artery.canonical.hostname = " + addr.Host)
+                    .AppendLine("akka.remote.artery.canonical.port = " + addr.Port);
                 var freshSystem = ActorSystem.Create(Sys.Name,
                     ConfigurationFactory.ParseString(sb.ToString()).WithFallback(Sys.Settings.Config));