[serve] Cache per-tick replica id walks behind a container membership fingerprint

The per-tick id walks (alive-actor-ids, active-node-ids, running-ids, the autoscaler
id-set rebuild) recompute identical results on every steady-state tick. Memoize each on
a ReplicaStateContainer membership fingerprint, and disarm while STARTING or RECOVERING
replicas exist, since their actor_id/actor_node_id materialize in place without touching
the container.

The fingerprint is the summed per-replica hash of the container's {replica id -> state}
content rather than a mutation counter, because the health and migration passes pop a
whole bucket and re-add every replica to the state it came from. A counter reads that
churn as a membership change, and one node draining anywhere in the cluster is enough to
trigger it: at 4 replicas it bumps 5 times per otherwise-idle tick, which invalidates all
three memos and re-runs the O(N) rank-consistency pass on every tick. Gang deployments
health-check through that same pop-and-re-add path on every tick, so for them a counter
never holds at all. Summing per-replica terms makes pop-and-re-add cancel exactly, while
a real arrival, departure or state transition still changes it.

The same key replaces the id-set key of the rank-consistency gate, which keeps that gate
O(1) per tick without losing the content-based semantics that let it hold under churn. It
changes strictly more often than the id set it replaces -- it includes state -- so the
gate can never skip a pass the id set would have run. It is paired with the exact replica
count, because unlike an id-set comparison it is a hash: memberships of the same size
collide with probability ~2^-64, and while a stale memo self-corrects on the next real
change, a skipped rank pass would not.

Caching disarms while STARTING, UPDATING or RECOVERING replicas exist: the startup check
calls check_ready() for all three, and that rewrites actor_id/node_id in place without a
membership change.

One delta from the gate as it stands in the base PR: the empty-membership early return
now caches before returning.

Manager-level getters return frozensets so the memo cannot be mutated by callers (with
set() at the one mutating caller), and AutoscalingContext.running_replicas is copied to a
list at the public boundary. The autoscaler's identity skip is safe because
_running_replicas and _cached_running_replica_strs are only ever written together and both
start empty, so the pair cannot desync -- note that re-registering a deployment reuses the
existing DeploymentAutoscalingState rather than presenting a new list object.

Signed-off-by: john.taylor <john.taylor@anyscale.com>
Co-Authored-By: Claude <noreply@anthropic.com>

Author: johntaylor-cell

python/ray/serve/_private/autoscaling_state.py

@@ -3,7 +3,18 @@
 import math
 import time
 from collections import defaultdict
-from typing import TYPE_CHECKING, Any, Callable, Dict, List, Optional, Set, Tuple, Union
+from typing import (
+    TYPE_CHECKING,
+    Any,
+    Callable,
+    Dict,
+    List,
+    Optional,
+    Sequence,
+    Set,
+    Tuple,
+    Union,
+)
 
 from ray.serve._private.common import (
     RUNNING_REQUESTS_KEY,
@@ -93,7 +104,7 @@ def __init__(self, deployment_id: DeploymentID):
         # user defined policy returns a dictionary of state that is persisted between autoscaling decisions
         # content of the dictionary is determined by the user defined policy
         self._policy_state: Optional[Dict[str, Any]] = None
-        self._running_replicas: List[ReplicaID] = []
+        self._running_replicas: Sequence[ReplicaID] = []
         self._cached_running_replica_strs: Set[str] = set()
         self._target_capacity: Optional[float] = None
         self._target_capacity_direction: Optional[TargetCapacityDirection] = None
@@ -208,8 +219,12 @@ def get_num_replicas_upper_bound(self) -> int:
             self._target_capacity,
         )
 
-    def update_running_replica_ids(self, running_replicas: List[ReplicaID]):
+    def update_running_replica_ids(self, running_replicas: Sequence[ReplicaID]):
         """Update cached set of running replica IDs for this deployment."""
+        if running_replicas is self._running_replicas:
+            # Identity means DeploymentState handed back its memoized tuple, so the
+            # running-replica ids are unchanged and the derived str set still matches.
+            return
         self._running_replicas = running_replicas
         self._cached_running_replica_strs = {
             r.to_full_id_str() for r in running_replicas
@@ -395,7 +410,9 @@ def get_autoscaling_context(
             app_name=self._deployment_id.app_name,
             current_num_replicas=len(self._running_replicas),
             target_num_replicas=curr_target_num_replicas,
-            running_replicas=self._running_replicas,
+            # Copy: _running_replicas holds the memoized tuple from
+            # get_running_replica_ids(), and this is a stable public List[ReplicaID].
+            running_replicas=list(self._running_replicas),
             total_num_requests=self.get_total_num_requests,
             capacity_adjusted_min_replicas=self.get_num_replicas_lower_bound(),
             capacity_adjusted_max_replicas=self.get_num_replicas_upper_bound(),
@@ -1013,7 +1030,7 @@ def get_decision_num_replicas(
             }
 
     def update_running_replica_ids(
-        self, deployment_id: DeploymentID, running_replicas: List[ReplicaID]
+        self, deployment_id: DeploymentID, running_replicas: Sequence[ReplicaID]
     ):
         self._deployment_autoscaling_states[deployment_id].update_running_replica_ids(
             running_replicas
@@ -1183,7 +1200,7 @@ def should_autoscale_deployment(self, deployment_id: DeploymentID):
         )
 
     def update_running_replica_ids(
-        self, deployment_id: DeploymentID, running_replicas: List[ReplicaID]
+        self, deployment_id: DeploymentID, running_replicas: Sequence[ReplicaID]
     ):
         app_state = self._app_autoscaling_states.get(deployment_id.app_name)
         if app_state:

python/ray/serve/_private/controller.py

@@ -533,7 +533,9 @@ def _update_proxy_nodes(self):
         Controller decides where proxy actors should run
         (head node and nodes with deployment replicas).
         """
-        new_proxy_nodes = self.deployment_state_manager.get_active_node_ids()
+        # Copy: get_active_node_ids() returns a shared frozenset, and `frozenset - set`
+        # stays frozen, so the .add() below would raise without this.
+        new_proxy_nodes = set(self.deployment_state_manager.get_active_node_ids())
         new_proxy_nodes = new_proxy_nodes - set(
             self.cluster_node_info_cache.get_draining_nodes()
         )