(pekko) branch fix/adjust-pool-size-bounds created (now 990888892e)

[he...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

He-Pin pushed a change to branch fix/adjust-pool-size-bounds
in repository https://gitbox.apache.org/repos/asf/pekko.git


      at 990888892e fix: enforce resizer bounds and minimum pool size for AdjustPoolSize

This branch includes the following new commits:

     new 990888892e fix: enforce resizer bounds and minimum pool size for AdjustPoolSize

The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails.  The revisions
listed as "add" were already present in the repository and have only
been added to this reference.



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

(pekko) 01/01: fix: enforce resizer bounds and minimum pool size for AdjustPoolSize

[he...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

He-Pin pushed a commit to branch fix/adjust-pool-size-bounds
in repository https://gitbox.apache.org/repos/asf/pekko.git

commit 990888892ea6c23430d387d46e22e8edffcc0c7b
Author: 虎鸣 <he...@alibaba-inc.com>
AuthorDate: Tue Jul 28 14:32:24 2026 +0800

    fix: enforce resizer bounds and minimum pool size for AdjustPoolSize
    
    Motivation:
    AdjustPoolSize bypasses pool configuration bounds. When a resizer is
    configured, the pool can grow beyond upperBound or shrink below
    lowerBound. Without a resizer, the pool can shrink to zero routees.
    
    Modification:
    Add a clamp method to the Resizer trait (default: min 1) and override
    it in DefaultResizer and DefaultOptimalSizeExploringResizer to enforce
    their lowerBound/upperBound. ResizablePoolActor now intercepts
    AdjustPoolSize and clamps the resulting size. RouterPoolActor enforces
    a minimum of 1 routee for non-resizer pools.
    
    Result:
    AdjustPoolSize respects resizer bounds when configured, and never
    shrinks a pool below 1 routee otherwise.
    
    Tests:
    - sbt "actor-tests / Test / testOnly org.apache.pekko.routing.RoundRobinSpec" passed
    - sbt "actor-tests / Test / testOnly org.apache.pekko.routing.ResizerSpec" passed
    - sbt "actor / mimaReportBinaryIssues" passed
    
    References:
    Fixes #3253
---
 .../org/apache/pekko/routing/ResizerSpec.scala     | 35 ++++++++++++++++++++++
 .../org/apache/pekko/routing/RoundRobinSpec.scala  | 10 +++++++
 .../routing/OptimalSizeExploringResizer.scala      |  3 ++
 .../scala/org/apache/pekko/routing/Resizer.scala   | 25 ++++++++++++++++
 .../org/apache/pekko/routing/RoutedActorCell.scala | 12 ++++----
 .../org/apache/pekko/routing/RouterConfig.scala    |  9 ++----
 6 files changed, 83 insertions(+), 11 deletions(-)

diff --git a/actor-tests/src/test/scala/org/apache/pekko/routing/ResizerSpec.scala b/actor-tests/src/test/scala/org/apache/pekko/routing/ResizerSpec.scala
index 198b7144cf..52284eec42 100644
--- a/actor-tests/src/test/scala/org/apache/pekko/routing/ResizerSpec.scala
+++ b/actor-tests/src/test/scala/org/apache/pekko/routing/ResizerSpec.scala
@@ -249,6 +249,41 @@ class ResizerSpec extends PekkoSpec(ResizerSpec.config) with DefaultTimeout with
 
     }
 
+    "clamp AdjustPoolSize growth to upperBound" in {
+      val resizer = DefaultResizer(lowerBound = 2, upperBound = 5, messagesPerResize = 100)
+      val router = system.actorOf(
+        RoundRobinPool(nrOfInstances = 0, resizer = Some(resizer)).props(Props[TestActor]()))
+
+      val latch = new TestLatch(2)
+      router ! latch
+      router ! latch
+      Await.ready(latch, remainingOrDefault)
+
+      routeeSize(router) should ===(2)
+
+      router ! AdjustPoolSize(10)
+      routeeSize(router) should ===(5)
+    }
+
+    "clamp AdjustPoolSize shrink to lowerBound" in {
+      val resizer = DefaultResizer(lowerBound = 2, upperBound = 5, messagesPerResize = 100)
+      val router = system.actorOf(
+        RoundRobinPool(nrOfInstances = 0, resizer = Some(resizer)).props(Props[TestActor]()))
+
+      val latch = new TestLatch(2)
+      router ! latch
+      router ! latch
+      Await.ready(latch, remainingOrDefault)
+
+      routeeSize(router) should ===(2)
+
+      router ! AdjustPoolSize(2)
+      routeeSize(router) should ===(4)
+
+      router ! AdjustPoolSize(-10)
+      routeeSize(router) should ===(2)
+    }
+
   }
 
 }
diff --git a/actor-tests/src/test/scala/org/apache/pekko/routing/RoundRobinSpec.scala b/actor-tests/src/test/scala/org/apache/pekko/routing/RoundRobinSpec.scala
index 2e47538c5d..a7d80b29e5 100644
--- a/actor-tests/src/test/scala/org/apache/pekko/routing/RoundRobinSpec.scala
+++ b/actor-tests/src/test/scala/org/apache/pekko/routing/RoundRobinSpec.scala
@@ -125,6 +125,16 @@ class RoundRobinSpec extends PekkoSpec with DefaultTimeout with ImplicitSender {
       actor ! RemoveRoutee(other)
       routeeSize(actor) should ===(5)
     }
+
+    "not shrink below 1 routee with AdjustPoolSize" in {
+      val actor = system.actorOf(RoundRobinPool(3).props(routeeProps = Props(new Actor {
+          def receive = Actor.emptyBehavior
+        })), "round-robin-min-one")
+
+      routeeSize(actor) should ===(3)
+      actor ! AdjustPoolSize(-10)
+      routeeSize(actor) should ===(1)
+    }
   }
 
   "round robin group" must {
diff --git a/actor/src/main/scala/org/apache/pekko/routing/OptimalSizeExploringResizer.scala b/actor/src/main/scala/org/apache/pekko/routing/OptimalSizeExploringResizer.scala
index de1e5bf192..fb33004c59 100644
--- a/actor/src/main/scala/org/apache/pekko/routing/OptimalSizeExploringResizer.scala
+++ b/actor/src/main/scala/org/apache/pekko/routing/OptimalSizeExploringResizer.scala
@@ -284,6 +284,9 @@ case class DefaultOptimalSizeExploringResizer(
     Math.max(lowerBound, Math.min(proposedChange + currentSize, upperBound)) - currentSize
   }
 
+  override def clamp(proposedSize: Int): Int =
+    Math.max(lowerBound, Math.min(proposedSize, upperBound))
+
   private def optimize(currentSize: PoolSize): Int = {
 
     val adjacentDispatchWaits: Map[PoolSize, Duration] = {
diff --git a/actor/src/main/scala/org/apache/pekko/routing/Resizer.scala b/actor/src/main/scala/org/apache/pekko/routing/Resizer.scala
index 3208cd099d..1e4d0e93d9 100644
--- a/actor/src/main/scala/org/apache/pekko/routing/Resizer.scala
+++ b/actor/src/main/scala/org/apache/pekko/routing/Resizer.scala
@@ -63,6 +63,16 @@ trait Resizer {
    */
   def resize(currentRoutees: immutable.IndexedSeq[Routee]): Int
 
+  /**
+   * Constrain a proposed total pool size to the bounds enforced by this resizer.
+   * Used by [[AdjustPoolSize]] management messages to respect resizer bounds.
+   * The default implementation only ensures the size is at least 1.
+   *
+   * @param proposedSize the proposed total number of routees
+   * @return the clamped pool size
+   */
+  def clamp(proposedSize: Int): Int = math.max(1, proposedSize)
+
 }
 
 object Resizer {
@@ -188,6 +198,9 @@ case class DefaultResizer(
     else delta
   }
 
+  override def clamp(proposedSize: Int): Int =
+    math.max(lowerBound, math.min(proposedSize, upperBound))
+
   /**
    * Number of routees considered busy, or above 'pressure level'.
    *
@@ -347,6 +360,18 @@ private[pekko] class ResizablePoolActor(supervisorStrategy: SupervisorStrategy)
     ({
       case Resize =>
         resizerCell.resize(initial = false)
+      case AdjustPoolSize(change: Int) =>
+        val currentRoutees = cell.router.routees
+        val currentSize = currentRoutees.size
+        val clampedSize = resizerCell.resizer.clamp(currentSize + change)
+        val effectiveChange = clampedSize - currentSize
+        if (effectiveChange > 0) {
+          val newRoutees = Vector.fill(effectiveChange)(pool.newRoutee(cell.routeeProps, context))
+          cell.addRoutees(newRoutees)
+        } else if (effectiveChange < 0) {
+          val abandon = currentRoutees.drop(currentRoutees.length + effectiveChange)
+          cell.removeRoutees(abandon, stopChild = true)
+        }
     }: Actor.Receive).orElse(super.receive)
 
 }
diff --git a/actor/src/main/scala/org/apache/pekko/routing/RoutedActorCell.scala b/actor/src/main/scala/org/apache/pekko/routing/RoutedActorCell.scala
index 5150037ef5..9344b22578 100644
--- a/actor/src/main/scala/org/apache/pekko/routing/RoutedActorCell.scala
+++ b/actor/src/main/scala/org/apache/pekko/routing/RoutedActorCell.scala
@@ -203,12 +203,14 @@ private[pekko] class RouterPoolActor(override val supervisorStrategy: Supervisor
   override def receive =
     ({
       case AdjustPoolSize(change: Int) =>
-        if (change > 0) {
-          val newRoutees = Vector.fill(change)(pool.newRoutee(cell.routeeProps, context))
+        val currentRoutees = cell.router.routees
+        val currentSize = currentRoutees.size
+        val effectiveChange = math.max(1, currentSize + change) - currentSize
+        if (effectiveChange > 0) {
+          val newRoutees = Vector.fill(effectiveChange)(pool.newRoutee(cell.routeeProps, context))
           cell.addRoutees(newRoutees)
-        } else if (change < 0) {
-          val currentRoutees = cell.router.routees
-          val abandon = currentRoutees.drop(currentRoutees.length + change)
+        } else if (effectiveChange < 0) {
+          val abandon = currentRoutees.drop(currentRoutees.length + effectiveChange)
           cell.removeRoutees(abandon, stopChild = true)
         }
     }: Actor.Receive).orElse(super.receive)
diff --git a/actor/src/main/scala/org/apache/pekko/routing/RouterConfig.scala b/actor/src/main/scala/org/apache/pekko/routing/RouterConfig.scala
index 2dbd45a14a..e579de965f 100644
--- a/actor/src/main/scala/org/apache/pekko/routing/RouterConfig.scala
+++ b/actor/src/main/scala/org/apache/pekko/routing/RouterConfig.scala
@@ -444,12 +444,9 @@ final case class RemoveRoutee(routee: Routee) extends RouterManagementMesssage
  *
  * Positive `change` will add that number of routees to the [[Pool]].
  * Negative `change` will remove that number of routees from the [[Pool]].
- * This is a direct management operation, so `change` is not constrained by the
- * pool's initial `nrOfInstances` or by bounds configured on a [[Resizer]].
- * A resizer may adjust the pool back within its bounds the next time it runs.
- *
- * If a resizable pool is reduced to zero routees, the ordinary message that
- * triggers the next resize may be lost before new routees are added.
+ * If a [[Resizer]] is configured, the resulting pool size is clamped to the
+ * resizer's bounds (see [[Resizer.clamp]]). Without a resizer, the pool will
+ * not shrink below 1 routee.
  *
  * Routees are stopped by sending a [[pekko.actor.PoisonPill]] to the routee.
  * Precautions are taken reduce the risk of dropping messages that are concurrently


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