Disassociate the wrapped handle when a throttled association stops

src/core/Akka.Remote.Tests/Transport/ThrottlerManagerLifecycleSpec.cs

@@ -54,13 +54,17 @@ public StubHandle(Address localAddress, Address remoteAddress) : base(localAddre
 
             public bool Disassociated { get; private set; }
 
+            // counts calls so tests can tell a single disassociate from a double one
+            public int DisassociateCallCount { get; private set; }
+
             public override bool Write(ByteString payload) => true;
 
 #pragma warning disable CS0672 // the base member is obsolete, but it is the abstract one we have to implement
             public override void Disassociate()
 #pragma warning restore CS0672
             {
                 Disassociated = true;
+                DisassociateCallCount++;
             }
         }
 
@@ -187,6 +191,43 @@ await EventFilter.Warning(contains: "received an InboundPayload before it was in
                 });
 
             originalHandle.Disassociated.Should().BeTrue("the association has to be torn down so remoting can re-establish it");
+            // this state used to disassociate directly as well as through PostStop; guards against that regressing
+            originalHandle.DisassociateCallCount.Should().Be(1, "PostStop alone must be the one that tears the handle down");
+        }
+
+        [Fact(DisplayName = "ThrottlerHandle.Disassociate should close the wrapped handle promptly (#8637)")]
+        public async Task Should_disassociate_wrapped_handle_on_explicit_disassociate()
+        {
+            var manager = StartManager();
+            var throttlerHandle = await AssociateAsync(manager);
+            var throttler = throttlerHandle.ThrottlerActor;
+            await WatchAsync(throttler);
+
+#pragma warning disable CS0618 // this is the exact API the issue is about
+            throttlerHandle.Disassociate();
+#pragma warning restore CS0618
+            await ExpectTerminatedAsync(throttler);
+
+            var originalHandle = (StubHandle)throttlerHandle.WrappedHandle;
+            originalHandle.Disassociated.Should().BeTrue("disassociating the throttled handle must tear down the wrapped connection too");
+        }
+
+        [Fact(DisplayName = "ThrottledAssociation should disassociate the wrapped handle exactly once, via PostStop, on inbound disassociation")]
+        public async Task Should_disassociate_wrapped_handle_once_on_inbound_disassociation()
+        {
+            var manager = StartManager();
+            var throttlerHandle = await AssociateAsync(manager);
+            var throttler = throttlerHandle.ThrottlerActor;
+            var originalHandle = (StubHandle)throttlerHandle.WrappedHandle;
+            await WatchAsync(throttler);
+
+            // simulates the wrapped handle notifying us that the peer tore the connection down;
+            // nothing on this path calls Disassociate directly, so PostStop must be the sole caller
+            throttler.Tell(new Disassociated(DisassociateInfo.Unknown));
+            await ExpectTerminatedAsync(throttler);
+
+            originalHandle.DisassociateCallCount.Should().Be(1,
+                "PostStop must disassociate the wrapped handle exactly once, even though the peer disassociated first");
         }
     }
 }

src/core/Akka.Remote/Transport/ThrottleTransportAdapter.cs

@@ -1114,6 +1114,24 @@ public ThrottledAssociation(IActorRef manager, IAssociationEventListener associa
             InitializeFSM();
         }
 
+        /// <summary>
+        /// Disassociates the wrapped handle whenever this actor stops, for any reason (explicit
+        /// disassociate, Blackhole, crash, or transport shutdown). Mirrors canonical Akka's
+        /// <c>ThrottledAssociation.postStop</c>. The wrapped handle's Disassociate() is required to be
+        /// idempotent, so this is safe even if the peer already tore the connection down first.
+        /// </summary>
+        protected override void PostStop()
+        {
+            try
+            {
+                OriginalHandle.Disassociate("the owning ThrottledAssociation stopped", _log);
+            }
+            finally
+            {
+                base.PostStop();
+            }
+        }
+
         private void InitializeFSM()
         {
             When(ThrottlerState.WaitExposedHandle, @event =>
@@ -1143,7 +1161,7 @@ private void InitializeFSM()
                     _log.Warning(
                         "Throttler for [{0}] received an InboundPayload before it was initialized - most likely restarted. Disassociating so the association can be re-established.",
                         OriginalHandle.RemoteAddress);
-                    OriginalHandle.Disassociate("throttler received inbound data before it was initialized", _log);
+                    // PostStop disassociates the wrapped handle; no need to do it here too
                     return Stop();
                 }