Artery: stop dropping the first ordinary message when the peer dialed first

Fixes #8496.

`OutboundHandshakeStage.PreStart` treated "this association already has a
`UniqueRemoteAddress`" as "our handshake is done" and skipped straight to
`Completed`. That field is also set by the INBOUND direction - when WE handle
the peer's `HandshakeReq` - so it proves we know the peer's uid, not that the
peer knows ours. When a peer dialed us first, our first outbound ordinary
stream therefore sent no `HandshakeReq` of its own, and our first user message
raced our `HandshakeRsp` (a different TCP connection, no ordering relationship)
into the peer's unknown-origin gate:

  Dropping inbound message [Akka.Actor.ActorSelectionMessage] from unknown
  origin uid [817823530] (no completed handshake for this uid yet).

Ordinary messages have no resend path, so the message was gone. The receiving
side documented an invariant ("the sender's gate cannot complete before we have
processed its Req") that the `PreStart` shortcut quietly broke.

`AssociationState.OutboundHandshakeCompleted` now records, per INCARNATION,
whether the peer has answered a `HandshakeReq` of ours. Only the new
`CompleteOutboundHandshake` sets it, and only `InboundHandshakeStage.HandleRsp`
calls that - a Rsp is the sole event proving the peer registered our uid. The
flag rides the same immutable snapshot and CAS swap as the rest of the
association state: a uid change resets it, `Quarantine` clears it, so it never
outlives the incarnation it describes.

Two gates consult it, both for ordinary/large/lane streams only:

* `PreStart`'s fast path (`CanSkipOwnHandshake`) - so a stream materialized
  against an inbound-only association runs the normal, well-tested handshake
  path instead of shortcutting.
* `RefreshCompletionFromContext` - load-bearing too, since the generation
  counter it already checked is advanced by the peer's own inbound Req: without
  it, a stream that did send its Req would complete on the peer's Req *retry*
  landing while we wait for our Rsp, which is the same bug through the back
  door.

The CONTROL stream deliberately keeps the old, weaker rule. Its traffic is never
subject to the receiver's unknown-origin drop - every control envelope
(handshake, heartbeat, quarantine notice, system messages and their Ack/Nack) is
dispatched regardless of whether the origin uid is registered - and it must be
able to complete on an inbound handshake: the `HandshakeRsp` we owe a peer is
enqueued on that very control queue, so a strict rule there would deadlock two
systems that dial each other at the same instant, each holding the Rsp the other
is waiting for.

`ForceReqOnStart` is unaffected: it still forces the full handshake path, and the
two conditions compose - the fast path is skipped when either says so.

Also, at both inbound gates (`InboundHandshakeStage` and the lane-routed copy in
`ArteryInboundProcessingStage`), the unknown-origin drop is raised from DEBUG to
a rate-limited WARNING: the first drop warns at once, later ones at most once
per 10s carrying the count suppressed in between. This class of loss should be
diagnosable in the field, not invisible.

Tests. `ArteryPeerDialedFirstSpec` covers the issue's stage-level repro (an
association created by a real inbound `HandshakeReq`; the ordinary stage must
send its own Req and hold traffic until a real `HandshakeRsp` arrives), the
peer's-Req-retry variant, fast-path preservation, the new warning, and an
end-to-end test over two real Artery systems where B dials A first and A's
`HandshakeRsp` is dropped - the deterministic form of the race. Four of the five
fail on unfixed code; the fast-path test passes before and after, as it must.
`AssociationStateSpec` gains the transition, incarnation-reset and quarantine
coverage for the new flag. The invariant comment on `InboundHandshakeStage` now
describes the corrected construction.

Author: Aaronontheweb

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

@@ -102,7 +102,7 @@ public async Task OutboundHandshakeStage_should_inject_req_first_hold_then_flow_
 
             // Simulate what InboundHandshakeStage does when the peer's HandshakeRsp arrives on
             // OUR inbound pipeline for the return direction.
-            registry.CompleteHandshake(remoteAddress, new UniqueAddress(remoteAddress, 222L));
+            registry.CompleteOutboundHandshake(remoteAddress, new UniqueAddress(remoteAddress, 222L));
 
             // Nothing re-triggers the stage directly; per the documented notification mechanism,
             // the retry timer (still running) is what notices completion and delivers the held
@@ -152,7 +152,7 @@ public async Task OutboundHandshakeStage_should_emit_control_envelope_ahead_of_h
             // slow to arrive.
             await sub.ExpectNoMsgAsync(TimeSpan.FromMilliseconds(300));
 
-            registry.CompleteHandshake(remoteAddress, new UniqueAddress(remoteAddress, 222L));
+            registry.CompleteOutboundHandshake(remoteAddress, new UniqueAddress(remoteAddress, 222L));
 
             // The retry timer (still running) is what notices completion and releases the held
             // element - bounded by one retry interval; a legal idempotent retry HandshakeReq may
@@ -268,7 +268,7 @@ public async Task OutboundHandshakeStage_non_control_should_route_req_via_send_c
             await pub.SendNextAsync(new OutboundEnvelope("user-message", null, null));
             await sub.ExpectNoMsgAsync(TimeSpan.FromMilliseconds(200));
 
-            registry.CompleteHandshake(remoteAddress, new UniqueAddress(remoteAddress, 555L));
+            registry.CompleteOutboundHandshake(remoteAddress, new UniqueAddress(remoteAddress, 555L));
 
             var delivered = await sub.ExpectNextAsync(TimeSpan.FromSeconds(3));
             delivered.Message.Should().Be("user-message");
@@ -285,7 +285,7 @@ public async Task OutboundHandshakeStage_non_control_should_not_hold_elements_fo
 
             // Already-associated at PreStart (Completed immediately) -- exercises the
             // "ShouldReinjectForLiveness" path directly rather than the initial handshake path.
-            registry.CompleteHandshake(remoteAddress, new UniqueAddress(remoteAddress, 777L));
+            registry.CompleteOutboundHandshake(remoteAddress, new UniqueAddress(remoteAddress, 777L));
 
             var stage = new OutboundHandshakeStage(
                 context,

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

@@ -94,6 +94,65 @@ public void Quarantine_should_ignore_stale_uid()
             newState.IsQuarantined(firstPeer.Uid).Should().BeFalse();
         }
 
+        [Fact(DisplayName = "Only CompleteOutboundHandshake should set OutboundHandshakeCompleted (issue #8496)")]
+        public void Only_CompleteOutboundHandshake_should_set_outbound_completed()
+        {
+            var peer = new UniqueAddress(RemoteAddress, 1L);
+
+            var fromPeersReq = AssociationState.Create().CompleteHandshake(peer);
+            fromPeersReq.UniqueRemoteAddress.Should().Be(peer, "the peer's own Req still registers its uid");
+            fromPeersReq.OutboundHandshakeCompleted.Should().BeFalse("the peer has learned nothing about US from its own Req");
+
+            AssociationState.Create().CompleteOutboundHandshake(peer).OutboundHandshakeCompleted
+                .Should().BeTrue("a Rsp is proof the peer registered our uid");
+        }
+
+        [Fact(DisplayName = "CompleteOutboundHandshake should flip OutboundHandshakeCompleted after an inbound-only handshake, then be an idempotent no-op")]
+        public void CompleteOutboundHandshake_should_flip_the_flag_then_be_idempotent()
+        {
+            var peer = new UniqueAddress(RemoteAddress, 1L);
+            var inboundOnly = AssociationState.Create().CompleteHandshake(peer);
+
+            var answered = inboundOnly.CompleteOutboundHandshake(peer);
+
+            answered.Should().NotBeSameAs(inboundOnly, "the false -> true flip is a real transition, not a no-op");
+            answered.Incarnation.Should().Be(inboundOnly.Incarnation, "the peer did not restart");
+            answered.OutboundHandshakeCompleted.Should().BeTrue();
+
+            answered.CompleteOutboundHandshake(peer).Should().BeSameAs(answered, "re-answering is an idempotent no-op");
+            answered.CompleteHandshake(peer).Should().BeSameAs(answered, "a later inbound Req must not clear what we already proved");
+        }
+
+        [Fact(DisplayName = "A new incarnation (peer restart) should reset OutboundHandshakeCompleted")]
+        public void New_incarnation_should_reset_outbound_completed()
+        {
+            var oldPeer = new UniqueAddress(RemoteAddress, 1L);
+            var newPeer = new UniqueAddress(RemoteAddress, 2L);
+            var answered = AssociationState.Create().CompleteOutboundHandshake(oldPeer);
+
+            var restarted = answered.CompleteHandshake(newPeer);
+
+            restarted.Incarnation.Should().Be(answered.Incarnation + 1);
+            restarted.OutboundHandshakeCompleted.Should().BeFalse(
+                "the RESTARTED peer has never answered a Req of ours -- what the previous incarnation knew about us died with it");
+
+            answered.CompleteOutboundHandshake(newPeer).OutboundHandshakeCompleted
+                .Should().BeTrue("unless the very message that changed the uid was itself a Rsp");
+        }
+
+        [Fact(DisplayName = "Quarantine should clear OutboundHandshakeCompleted")]
+        public void Quarantine_should_clear_outbound_completed()
+        {
+            var peer = new UniqueAddress(RemoteAddress, 1L);
+            var answered = AssociationState.Create().CompleteOutboundHandshake(peer);
+
+            var (quarantined, acted) = answered.Quarantine(peer.Uid);
+
+            acted.Should().BeTrue();
+            quarantined.OutboundHandshakeCompleted.Should().BeFalse();
+            quarantined.UniqueRemoteAddress.Should().Be(peer, "quarantine is uid-scoped -- it does not forget which uid it applies to");
+        }
+
         [Fact(DisplayName = "Quarantine should act on the current uid and mark it quarantined")]
         public void Quarantine_should_act_on_current_uid()
         {

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

@@ -328,8 +328,17 @@ public LaneWorkItem(ReadOnlySequence<byte> payload, int serializerId, string man
 
         private sealed class Logic : GraphStageLogic, IInHandler, IOutHandler
         {
+            /// <summary>
+            /// Rate limit for this connection's unknown-origin drop warnings -- the lane path's
+            /// half of the pair <see cref="InboundHandshakeStage"/> owns the other half of; see
+            /// that stage for the rationale.
+            /// </summary>
+            private static readonly TimeSpan UnknownOriginWarnInterval = TimeSpan.FromSeconds(10);
+
             private readonly ArteryInboundProcessingStage _stage;
             private readonly Queue<IInboundEnvelope> _pending = new();
+            private DateTime _lastUnknownOriginWarning = DateTime.MinValue;
+            private long _suppressedUnknownOriginDrops;
 
             private readonly byte[] _preambleBuffer = new byte[ArteryConnectionHeader.Length];
             private int _preambleFilled;
@@ -762,9 +771,22 @@ private bool ProcessFrameLaneMode(ReadOnlySequence<byte> frameBody)
                 // user-message dispatch per connection" holds for the lane path too.
                 if (!_stage.InboundContext!.IsKnownOrigin(decoded.Header.OriginUid))
                 {
-                    Log.Debug(
-                        "Dropping inbound lane-routed Artery message from unknown origin uid [{0}] (no completed handshake for this uid yet).",
-                        decoded.Header.OriginUid);
+                    var now = DateTime.UtcNow;
+                    if (_lastUnknownOriginWarning == DateTime.MinValue ||
+                        now - _lastUnknownOriginWarning >= UnknownOriginWarnInterval)
+                    {
+                        Log.Warning(
+                            "Dropping inbound lane-routed Artery message from unknown origin uid [{0}]: no completed handshake for this uid yet. " +
+                            "The message is LOST - ordinary messages are not resent. [{1}] further drop(s) suppressed since the last warning.",
+                            decoded.Header.OriginUid, _suppressedUnknownOriginDrops);
+                        _lastUnknownOriginWarning = now;
+                        _suppressedUnknownOriginDrops = 0;
+                    }
+                    else
+                    {
+                        _suppressedUnknownOriginDrops++;
+                    }
+
                     return true;
                 }
 

src/core/Akka.Remote/Artery/ArteryInboundProcessingStage.cs

@@ -996,14 +996,27 @@ public void DrainLargeToDropped(Action<IOutboundEnvelope> onDrained)
         /// snapshot, so <see cref="AssociationRegistry"/> can tell — without a separate,
         /// racy read — whether (and from what uid) an incarnation change just happened.
         /// </summary>
-        public (AssociationState Previous, AssociationState Updated) CompleteHandshake(UniqueAddress peer)
+        public (AssociationState Previous, AssociationState Updated) CompleteHandshake(UniqueAddress peer) =>
+            Transition(peer, static (state, p) => state.CompleteHandshake(p));
+
+        /// <summary>
+        /// As <see cref="CompleteHandshake"/>, but applying
+        /// <see cref="AssociationState.CompleteOutboundHandshake"/> -- the only path that sets
+        /// <see cref="AssociationState.OutboundHandshakeCompleted"/> (issue #8496).
+        /// </summary>
+        public (AssociationState Previous, AssociationState Updated) CompleteOutboundHandshake(UniqueAddress peer) =>
+            Transition(peer, static (state, p) => state.CompleteOutboundHandshake(p));
+
+        private (AssociationState Previous, AssociationState Updated) Transition(
+            UniqueAddress peer,
+            Func<AssociationState, UniqueAddress, AssociationState> transition)
         {
             Interlocked.Increment(ref _handshakeGeneration);
 
             while (true)
             {
                 var current = _state;
-                var updated = current.CompleteHandshake(peer);
+                var updated = transition(current, peer);
 
                 if (ReferenceEquals(updated, current))
                     return (current, current);
@@ -1137,7 +1150,27 @@ public Association AssociationFor(Address remoteAddress) =>
         public AssociationState CompleteHandshake(Address remoteAddress, UniqueAddress peer)
         {
             var association = AssociationFor(remoteAddress);
-            var (previous, updated) = association.CompleteHandshake(peer);
+            return Complete(association, peer, association.CompleteHandshake(peer));
+        }
+
+        /// <summary>
+        /// As <see cref="CompleteHandshake"/>, but records that the peer answered a
+        /// <see cref="HandshakeReq"/> of OURS -- see
+        /// <see cref="AssociationState.OutboundHandshakeCompleted"/>. Called only from
+        /// <see cref="InboundHandshakeStage"/>'s <c>HandleRsp</c>.
+        /// </summary>
+        public AssociationState CompleteOutboundHandshake(Address remoteAddress, UniqueAddress peer)
+        {
+            var association = AssociationFor(remoteAddress);
+            return Complete(association, peer, association.CompleteOutboundHandshake(peer));
+        }
+
+        private AssociationState Complete(
+            Association association,
+            UniqueAddress peer,
+            (AssociationState Previous, AssociationState Updated) transition)
+        {
+            var (previous, updated) = transition;
 
             if (!ReferenceEquals(previous, updated) &&
                 previous.UniqueRemoteAddress is { } previousPeer &&

src/core/Akka.Remote/Artery/AssociationRegistry.cs

@@ -37,18 +37,24 @@ namespace Akka.Remote.Artery
     /// </summary>
     internal sealed class AssociationState
     {
-        private AssociationState(int incarnation, UniqueAddress? uniqueRemoteAddress, ImmutableHashSet<long> quarantinedUids)
+        private AssociationState(
+            int incarnation,
+            UniqueAddress? uniqueRemoteAddress,
+            bool outboundHandshakeCompleted,
+            ImmutableHashSet<long> quarantinedUids)
         {
             Incarnation = incarnation;
             UniqueRemoteAddress = uniqueRemoteAddress;
+            OutboundHandshakeCompleted = outboundHandshakeCompleted;
             QuarantinedUids = quarantinedUids;
         }
 
         /// <summary>
         /// The initial state for a freshly-materialized association: no peer UID known yet
-        /// (<c>Associating</c>), incarnation 1, no quarantined UIDs.
+        /// (<c>Associating</c>), incarnation 1, our own handshake unanswered, no quarantined UIDs.
         /// </summary>
-        public static AssociationState Create() => new(incarnation: 1, uniqueRemoteAddress: null, quarantinedUids: ImmutableHashSet<long>.Empty);
+        public static AssociationState Create() =>
+            new(incarnation: 1, uniqueRemoteAddress: null, outboundHandshakeCompleted: false, quarantinedUids: ImmutableHashSet<long>.Empty);
 
         /// <summary>
         /// Monotonically increasing incarnation counter. Starts at 1; incremented only when
@@ -63,6 +69,16 @@ private AssociationState(int incarnation, UniqueAddress? uniqueRemoteAddress, Im
         /// </summary>
         public UniqueAddress? UniqueRemoteAddress { get; }
 
+        /// <summary>
+        /// Whether THIS side's own <see cref="HandshakeReq"/> has been answered for the CURRENT
+        /// incarnation. Set ONLY by <see cref="CompleteOutboundHandshake"/>; a uid change resets it
+        /// and <see cref="Quarantine"/> clears it, so it never outlives the incarnation it
+        /// describes. <see cref="UniqueRemoteAddress"/> cannot stand in for it: the inbound
+        /// direction (the peer's own Req) sets that field too, and knowing the peer's uid says
+        /// nothing about whether the peer knows OURS (issue #8496).
+        /// </summary>
+        public bool OutboundHandshakeCompleted { get; }
+
         /// <summary>
         /// The set of peer UIDs (for this association's remote address) that have been
         /// explicitly quarantined. A uid change alone does not add the superseded uid here —
@@ -80,21 +96,37 @@ private AssociationState(int incarnation, UniqueAddress? uniqueRemoteAddress, Im
         /// with <paramref name="peer"/>:
         /// <list type="bullet">
         /// <item><description><c>Associating</c> → <c>Associated</c>: adopts <paramref name="peer"/>, incarnation unchanged.</description></item>
-        /// <item><description>Same uid as the current <see cref="UniqueRemoteAddress"/>: no-op — returns <c>this</c> (reference-equal, so the CAS loop in <see cref="Association"/> can skip the compare-exchange).</description></item>
+        /// <item><description>Same uid as the current <see cref="UniqueRemoteAddress"/>: no-op — returns <c>this</c> (reference-equal, so the CAS loop in <see cref="Association"/> can skip the compare-exchange), except for the one-way <see cref="OutboundHandshakeCompleted"/> flip in <see cref="CompleteOutboundHandshake"/>.</description></item>
         /// <item><description>Different uid (remote restart): a new incarnation — <see cref="Incarnation"/> + 1, <see cref="UniqueRemoteAddress"/> replaced, <see cref="QuarantinedUids"/> carried over UNCHANGED (the old uid is deliberately not auto-quarantined).</description></item>
         /// </list>
         /// </summary>
-        public AssociationState CompleteHandshake(UniqueAddress peer)
+        public AssociationState CompleteHandshake(UniqueAddress peer) => Apply(peer, answeredOurReq: false);
+
+        /// <summary>
+        /// As <see cref="CompleteHandshake"/>, but for the ONE event that proves the peer has
+        /// registered our uid: a <see cref="HandshakeRsp"/>, which a peer only sends after handling
+        /// a <see cref="HandshakeReq"/> of ours. Sets <see cref="OutboundHandshakeCompleted"/> --
+        /// nothing else does. A same-uid call that only flips that flag still returns a NEW
+        /// snapshot (it is a real transition, not the documented no-op).
+        /// </summary>
+        public AssociationState CompleteOutboundHandshake(UniqueAddress peer) => Apply(peer, answeredOurReq: true);
+
+        private AssociationState Apply(UniqueAddress peer, bool answeredOurReq)
         {
             if (UniqueRemoteAddress is { } current)
             {
                 if (current.Uid == peer.Uid)
-                    return this;
+                {
+                    if (!answeredOurReq || OutboundHandshakeCompleted)
+                        return this;
+
+                    return new AssociationState(Incarnation, peer, outboundHandshakeCompleted: true, QuarantinedUids);
+                }
 
-                return new AssociationState(Incarnation + 1, peer, QuarantinedUids);
+                return new AssociationState(Incarnation + 1, peer, answeredOurReq, QuarantinedUids);
             }
 
-            return new AssociationState(Incarnation, peer, QuarantinedUids);
+            return new AssociationState(Incarnation, peer, answeredOurReq, QuarantinedUids);
         }
 
         /// <summary>
@@ -113,7 +145,9 @@ public AssociationState CompleteHandshake(UniqueAddress peer)
             if (QuarantinedUids.Contains(uid))
                 return (this, true);
 
-            return (new AssociationState(Incarnation, UniqueRemoteAddress, QuarantinedUids.Add(uid)), true);
+            // OutboundHandshakeCompleted is cleared with the incarnation it describes: this uid is
+            // cut off, so no later stream may trust "the peer knows us" on its behalf.
+            return (new AssociationState(Incarnation, UniqueRemoteAddress, outboundHandshakeCompleted: false, QuarantinedUids.Add(uid)), true);
         }
     }
 }

src/core/Akka.Remote/Artery/AssociationState.cs

@@ -47,6 +47,15 @@ internal interface IInboundContext
         /// </summary>
         AssociationState CompleteHandshake(UniqueAddress peer);
 
+        /// <summary>
+        /// As <see cref="CompleteHandshake"/>, but for a <see cref="HandshakeRsp"/> -- which the
+        /// peer only sends after registering the uid in a <see cref="HandshakeReq"/> of OURS. This
+        /// is the only call that records <see cref="AssociationState.OutboundHandshakeCompleted"/>,
+        /// the signal an ordinary/large outbound stream needs before releasing user traffic
+        /// (issue #8496).
+        /// </summary>
+        AssociationState CompleteOutboundHandshake(UniqueAddress peer);
+
         /// <summary>
         /// Sends <paramref name="message"/> over the control channel to <paramref name="to"/>.
         /// Used by <see cref="InboundHandshakeStage"/> to reply with a <see cref="HandshakeRsp"/>.
@@ -129,6 +138,9 @@ public AssociationRegistryInboundContext(
         /// <inheritdoc/>
         public AssociationState CompleteHandshake(UniqueAddress peer) => _registry.CompleteHandshake(peer.Address, peer);
 
+        /// <inheritdoc/>
+        public AssociationState CompleteOutboundHandshake(UniqueAddress peer) => _registry.CompleteOutboundHandshake(peer.Address, peer);
+
         /// <inheritdoc/>
         public void SendControl(Address to, object message) => _sendControl(to, message);