[PR] Stop failed ThrottledAssociation children and prune them from the handle table [pekko]

["He-Pin (via GitHub)" <gi...@apache.org>]

He-Pin opened a new pull request, #3706:
URL: https://github.com/apache/pekko/pull/3706

   ### Motivation
   `ThrottlerManager` (the `trttl`/`gremlin` test adapters) declares no `supervisorStrategy` and never watches its `ThrottledAssociation` children. Under the default strategy a failed child restarts into `WaitExposedHandle`/`WaitModeAndUpstreamListener`, but `Handle` is only sent once — the restarted instance silently drops payloads forever. Terminated children also remain in `handleTable`, so later `SetThrottle` calls iterate dead refs and `askModeWithDeathCompletion` replies `SetThrottleAck` while having no effect. The failure mode is that a multi-node test's fault injection silently stops working instead of failing loudly.
   
   ### Modification
   - `ThrottlerManager`: declare `OneForOneStrategy { case NonFatal(_) => Stop }`, matching `PekkoProtocolTransport`. Stopping is safe and visible: `ThrottledAssociation.postStop` disassociates the original handle.
   - Watch each `ThrottledAssociation` in `wrapHandle` and prune its `handleTable` entry on `Terminated`.
   - New `ThrottlerManagerSpec`: drives a real outbound association through the manager, completes the wrapped handle's `readHandlerPromise` with a throwing listener, sends a payload to fail the FSM, and asserts the child is stopped (not restarted), the original handle disassociated exactly once, and `ForceDisassociate` afterwards does not touch the pruned entry.
   
   ### Result
   A broken throttled association is torn down visibly instead of silently dropping payloads; dead handles no longer linger in the handle table.
   
   ### Tests
   - `sbt "remote / Test / testOnly org.apache.pekko.remote.transport.ThrottlerManagerSpec"`: passed
   - Directional check: reverted the fix, re-ran — failed as expected (child restarted, never terminated)
   - Regression: `ThrottlerTransportAdapterSpec`, `ThrottlerHandleSpec`: 8/8 passed
   - Full validation on CI
   
   ### References
   Fixes #3688


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org

Re: [PR] Stop failed ThrottledAssociation children and prune them from the handle table [pekko]

["pjfanning (via GitHub)" <gi...@apache.org>]

pjfanning commented on code in PR #3706:
URL: https://github.com/apache/pekko/pull/3706#discussion_r4367799420


##########
remote/src/test/scala/org/apache/pekko/remote/transport/ThrottlerManagerSpec.scala:
##########
@@ -0,0 +1,112 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.pekko.remote.transport
+
+import java.util.concurrent.atomic.AtomicInteger
+
+import scala.annotation.nowarn
+import scala.concurrent.Await
+import scala.concurrent.Future
+import scala.concurrent.Promise
+import scala.concurrent.duration._
+
+import org.apache.pekko
+import pekko.actor.Address
+import pekko.actor.Props
+import pekko.event.LoggingAdapter
+import pekko.remote.transport.ActorTransportAdapter.{ AssociateUnderlying, ListenerRegistered }
+import pekko.remote.transport.AssociationHandle.{ HandleEvent, HandleEventListener, InboundPayload }
+import pekko.remote.transport.ThrottlerTransportAdapter.{ ForceDisassociate, ForceDisassociateAck }
+import pekko.testkit.ImplicitSender
+import pekko.testkit.PekkoSpec
+import pekko.util.ByteString
+
+import com.typesafe.config.ConfigFactory
+
+@nowarn("msg=deprecated")
+class ThrottlerManagerSpec
+    extends PekkoSpec(ConfigFactory.parseString("""
+    pekko.actor.provider = remote
+    pekko.remote.artery.enabled = off
+    pekko.remote.classic.netty.tcp.port = 0
+  """))
+    with ImplicitSender {
+
+  private val addrLocal = Address("pekko", "systemA", "hostA", 1001)
+  private val addrRemote = Address("pekko", "systemB", "hostB", 1002)
+
+  private class TestAssociationHandle extends AssociationHandle {
+    val disassociations = new AtomicInteger(0)
+    override val localAddress: Address = addrLocal
+    override val remoteAddress: Address = addrRemote
+    override val readHandlerPromise: Promise[HandleEventListener] = Promise()
+    override def write(payload: ByteString): Boolean = throw new RuntimeException("write failed")
+    override def disassociate(reason: String, log: LoggingAdapter): Unit = {
+      disassociations.incrementAndGet()
+      ()
+    }
+  }
+
+  private class MockTransport(handle: AssociationHandle) extends Transport {
+    override def schemeIdentifier: String = "mock"
+    override def isResponsibleFor(address: Address): Boolean = true
+    override def maximumPayloadBytes: Int = 128000
+    override def listen: Future[(Address, Promise[Transport.AssociationEventListener])] =
+      Future.failed(new UnsupportedOperationException("not needed by this test"))
+    override def associate(remoteAddress: Address): Future[AssociationHandle] = Future.successful(handle)
+    override def shutdown(): Future[Boolean] = Future.successful(true)
+  }
+
+  "ThrottlerManager" should {
+
+    "stop a failed ThrottledAssociation and prune it from the handle table" in {

Review Comment:
   The case where a restart really does park and drop payloads is inbound (`InboundAssociation` -> `WaitExposedHandle`): `Handle` is never re-sent, so a restarted instance waits forever. Could we add a test for that path? It would fail without the `Stop` strategy, which this outbound test doesn't seem to (see the comment on line 101).



##########
remote/src/main/scala/org/apache/pekko/remote/transport/ThrottlerTransportAdapter.scala:
##########
@@ -289,6 +289,13 @@ private[transport] class ThrottlerManager(wrappedTransport: Transport)
   private var throttlingModes = Map[Address, (ThrottleMode, Direction)]()
   private var handleTable = List[(Address, ThrottlerHandle)]()
 
+  // A restarted ThrottledAssociation would re-enter its initial state, but Handle is

Review Comment:
   This only holds for inbound associations. An outbound `ThrottledAssociation`'s constructor completes `originalHandle.readHandlerPromise` (line 490), so a restart fails in the constructor and the actor is stopped rather than re-entering `WaitModeAndUpstreamListener`. Could the comment say it's the inbound case (`WaitExposedHandle`) that parks?



##########
remote/src/test/scala/org/apache/pekko/remote/transport/ThrottlerManagerSpec.scala:
##########
@@ -0,0 +1,112 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.pekko.remote.transport
+
+import java.util.concurrent.atomic.AtomicInteger
+
+import scala.annotation.nowarn
+import scala.concurrent.Await
+import scala.concurrent.Future
+import scala.concurrent.Promise
+import scala.concurrent.duration._
+
+import org.apache.pekko
+import pekko.actor.Address
+import pekko.actor.Props
+import pekko.event.LoggingAdapter
+import pekko.remote.transport.ActorTransportAdapter.{ AssociateUnderlying, ListenerRegistered }
+import pekko.remote.transport.AssociationHandle.{ HandleEvent, HandleEventListener, InboundPayload }
+import pekko.remote.transport.ThrottlerTransportAdapter.{ ForceDisassociate, ForceDisassociateAck }
+import pekko.testkit.ImplicitSender
+import pekko.testkit.PekkoSpec
+import pekko.util.ByteString
+
+import com.typesafe.config.ConfigFactory
+
+@nowarn("msg=deprecated")
+class ThrottlerManagerSpec
+    extends PekkoSpec(ConfigFactory.parseString("""
+    pekko.actor.provider = remote
+    pekko.remote.artery.enabled = off
+    pekko.remote.classic.netty.tcp.port = 0
+  """))
+    with ImplicitSender {
+
+  private val addrLocal = Address("pekko", "systemA", "hostA", 1001)
+  private val addrRemote = Address("pekko", "systemB", "hostB", 1002)
+
+  private class TestAssociationHandle extends AssociationHandle {
+    val disassociations = new AtomicInteger(0)
+    override val localAddress: Address = addrLocal
+    override val remoteAddress: Address = addrRemote
+    override val readHandlerPromise: Promise[HandleEventListener] = Promise()
+    override def write(payload: ByteString): Boolean = throw new RuntimeException("write failed")
+    override def disassociate(reason: String, log: LoggingAdapter): Unit = {
+      disassociations.incrementAndGet()
+      ()
+    }
+  }
+
+  private class MockTransport(handle: AssociationHandle) extends Transport {
+    override def schemeIdentifier: String = "mock"
+    override def isResponsibleFor(address: Address): Boolean = true
+    override def maximumPayloadBytes: Int = 128000
+    override def listen: Future[(Address, Promise[Transport.AssociationEventListener])] =
+      Future.failed(new UnsupportedOperationException("not needed by this test"))
+    override def associate(remoteAddress: Address): Future[AssociationHandle] = Future.successful(handle)
+    override def shutdown(): Future[Boolean] = Future.successful(true)
+  }
+
+  "ThrottlerManager" should {
+
+    "stop a failed ThrottledAssociation and prune it from the handle table" in {
+      val handle = new TestAssociationHandle
+      val manager = system.actorOf(Props(new ThrottlerManager(new MockTransport(handle))), "throttlerManager")
+
+      manager ! ListenerRegistered(new Transport.AssociationEventListener {
+        override def notify(ev: Transport.AssociationEvent): Unit = ()
+      })
+
+      val statusPromise = Promise[AssociationHandle]()
+      manager ! AssociateUnderlying(addrRemote, statusPromise)
+      val wrappedHandle = Await.result(statusPromise.future, 3.seconds).asInstanceOf[ThrottlerHandle]
+      val child = wrappedHandle.throttlerActor
+      watch(child)
+
+      // the upstream listener throws when the FSM forwards the payload to it,
+      // failing the ThrottledAssociation
+      wrappedHandle.readHandlerPromise.success(new HandleEventListener {
+        override def notify(ev: HandleEvent): Unit = throw new RuntimeException("listener failed")
+      })
+      child ! InboundPayload(ByteString("boom"))
+
+      // the failed association must be stopped rather than restarted: a restarted
+      // instance re-enters its initial state but Handle is never re-sent, so it
+      // would silently drop payloads forever
+      expectTerminated(child, max = 3.seconds)

Review Comment:
   I don't think this distinguishes `Stop` from the old default strategy on the outbound path. The outbound `ThrottledAssociation` constructor calls `originalHandle.readHandlerPromise.success(...)` (`ThrottlerTransportAdapter.scala:490`). Under the default strategy, the restart's `preRestart` runs `postStop` (disassociations = 1). The new instance's constructor then completes the same promise a second time and throws `IllegalStateException`, and the default decider stops on that `PostRestartException`. So with only the `supervisorStrategy` override reverted, `expectTerminated` and the `=== 1` check should still pass; only the final `ForceDisassociate` check, which tests the `handleTable` pruning, would fail.
   
   That doesn't match the directional check in the PR body ("child restarted, never terminated"). Could you re-run it with only the strategy reverted, keeping the watch and pruning?



##########
remote/src/test/scala/org/apache/pekko/remote/transport/ThrottlerManagerSpec.scala:
##########
@@ -0,0 +1,112 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.pekko.remote.transport
+
+import java.util.concurrent.atomic.AtomicInteger
+
+import scala.annotation.nowarn
+import scala.concurrent.Await
+import scala.concurrent.Future
+import scala.concurrent.Promise
+import scala.concurrent.duration._
+
+import org.apache.pekko
+import pekko.actor.Address
+import pekko.actor.Props
+import pekko.event.LoggingAdapter
+import pekko.remote.transport.ActorTransportAdapter.{ AssociateUnderlying, ListenerRegistered }
+import pekko.remote.transport.AssociationHandle.{ HandleEvent, HandleEventListener, InboundPayload }
+import pekko.remote.transport.ThrottlerTransportAdapter.{ ForceDisassociate, ForceDisassociateAck }
+import pekko.testkit.ImplicitSender
+import pekko.testkit.PekkoSpec
+import pekko.util.ByteString
+
+import com.typesafe.config.ConfigFactory
+
+@nowarn("msg=deprecated")
+class ThrottlerManagerSpec
+    extends PekkoSpec(ConfigFactory.parseString("""
+    pekko.actor.provider = remote
+    pekko.remote.artery.enabled = off
+    pekko.remote.classic.netty.tcp.port = 0
+  """))
+    with ImplicitSender {
+
+  private val addrLocal = Address("pekko", "systemA", "hostA", 1001)
+  private val addrRemote = Address("pekko", "systemB", "hostB", 1002)
+
+  private class TestAssociationHandle extends AssociationHandle {
+    val disassociations = new AtomicInteger(0)
+    override val localAddress: Address = addrLocal
+    override val remoteAddress: Address = addrRemote
+    override val readHandlerPromise: Promise[HandleEventListener] = Promise()
+    override def write(payload: ByteString): Boolean = throw new RuntimeException("write failed")
+    override def disassociate(reason: String, log: LoggingAdapter): Unit = {
+      disassociations.incrementAndGet()
+      ()
+    }
+  }
+
+  private class MockTransport(handle: AssociationHandle) extends Transport {
+    override def schemeIdentifier: String = "mock"
+    override def isResponsibleFor(address: Address): Boolean = true
+    override def maximumPayloadBytes: Int = 128000
+    override def listen: Future[(Address, Promise[Transport.AssociationEventListener])] =
+      Future.failed(new UnsupportedOperationException("not needed by this test"))
+    override def associate(remoteAddress: Address): Future[AssociationHandle] = Future.successful(handle)
+    override def shutdown(): Future[Boolean] = Future.successful(true)
+  }
+
+  "ThrottlerManager" should {
+
+    "stop a failed ThrottledAssociation and prune it from the handle table" in {
+      val handle = new TestAssociationHandle
+      val manager = system.actorOf(Props(new ThrottlerManager(new MockTransport(handle))), "throttlerManager")
+
+      manager ! ListenerRegistered(new Transport.AssociationEventListener {
+        override def notify(ev: Transport.AssociationEvent): Unit = ()
+      })
+
+      val statusPromise = Promise[AssociationHandle]()
+      manager ! AssociateUnderlying(addrRemote, statusPromise)
+      val wrappedHandle = Await.result(statusPromise.future, 3.seconds).asInstanceOf[ThrottlerHandle]

Review Comment:
   nit: `3.seconds` here and in `expectTerminated(child, max = 3.seconds)` aren't dilated, so they don't scale with `pekko.test.timefactor` on slow CI. `3.seconds.dilated`, or the testkit defaults, would avoid flakiness.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org