[serve] Gate rank-consistency check on replica membership changes (#64911)

## [serve] Gate rank-consistency check on replica membership changes

### Why
The Ray Serve controller runs on the head node, continuously reconciling
deployments and
making autoscaling decisions.
`check_rank_consistency_and_reassign_minimally` runs at the
tail of every `DeploymentState.update()` — once per deployment per
control-loop tick,
which at measured loop rates is several times a second.

The pass is O(N) in replicas with large constant factors: it
materializes the active-key
set and the rank-key set, takes set differences both ways, copies the
entire `_ranks`
dict, tallies per-rank counts across every active key, and sorts all
rank values — then
the manager above it regroups replicas per node. In steady state every
one of those
passes re-derives an identical answer. At 16K, py-spy showed it
monopolizing the event
loop and starving handle-report ingest for 20 s+.

Ranks can only become inconsistent when replica membership changes, so
gate the pass on
the active replica-id set: run it when the set differs from the last
checked one, skip
when it does not. Transitions are unaffected — membership churn during
rollouts and
autoscaling changes the set, so the pass still runs exactly when it can
matter. The
guards that were already there (deployment not HEALTHY, or any STARTING
replica) run
first, before the replica list is materialized.

`RAY_SERVE_FAIL_ON_RANK_ERROR` defaults off, so in production the pass
can log an error
and return a safe default. When that happens the membership is
deliberately **not**
cached and the check is retried next tick, rather than latching a
membership the pass
never validated. The error flag is read before the reassignment
reconfigure, which calls
back into the rank manager and resets it.

The gate's own cost is the remaining O(N) here: building the id set
measures 1.9 ms at
16,384 replicas, ~4.5% of the optimized loop. #64910, stacked on this
PR, replaces the
set comparison with a `ReplicaStateContainer` mutation-version integer,
which removes it.

### What
- Extract the pass into `_maybe_check_rank_consistency()`.
- Gate it on the set of active replica ids: if the set is unchanged
since the pass last
  ran, skip it.
- Only record a membership as checked when the pass did not swallow an
error.
`RAY_SERVE_FAIL_ON_RANK_ERROR` is off by default, so the pass can catch
an exception and
return `[]`; caching that would mean a stable-but-inconsistent
deployment never retries.
  `DeploymentRankManager` now records whether the last rank op errored.
- The existing guards are preserved verbatim: only when the deployment
is HEALTHY, and
never while STARTING replicas exist (the node-migration case documented
in the original
  comment).

### Why this is safe
Rank consistency can only be violated by a membership change — replicas
added, removed, or
replaced — and every such change alters the active id set, so the pass
still runs on
exactly the ticks where it can find work. Rank release and removal from
`_replicas` happen
in the same `pop(states=[STOPPING])` branch, so the id set and the rank
table cannot drift
apart across ticks. The per-node and node-rank passes group off
`_replica_to_node`, which
is only mutated by rank assign/recover/release, so a fixed replica-id
set implies a fixed
node grouping.

Set comparison is exact — there is no fingerprint collision to reason
about. The residual
cost is the O(N) set construction itself (~1.9 ms/tick at 16K replicas);
the follow-up
#64910 replaces it with an exact `ReplicaStateContainer` mutation
counter, making the gate
O(1).

### Results
- Follow-up profile of the same 16K workload: the rank pass no longer
appears in
controller CPU samples; report ingest resumes (stale-report drops stop).
- Composite (with the rest of the optimization stack): 16K steady-state
control loop
  352.6 ms → 42.9 ms.

### Testing
- 6 new unit tests. Four over the gate itself: runs once then skips on
stable
membership; re-runs on membership change; a swallowed rank error is not
cached and
retries next tick; skips entirely (without recording a membership) while
STARTING
replicas exist. Two over a real
`DeploymentRankManager(fail_on_rank_error=False)` —
the production default — covering that a swallowed error is not cached,
and that the
error flag is read before `_reconfigure_replicas_with_new_ranks` can
clear it.
- Full `test_deployment_state.py` suite green (237 tests).
<img width="1248" height="728" alt="perf_A6_loop_haproxy_on"
src="https://github.com/user-attachments/assets/34758e31-6c04-46f9-958c-d6b34d3eae32"
/>

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

Author: johntaylor-cell

python/ray/serve/_private/deployment_state.py

@@ -2934,6 +2934,7 @@ def __init__(
         self._rank_manager = DeploymentRankManager(
             fail_on_rank_error=RAY_SERVE_FAIL_ON_RANK_ERROR
         )
+        self._last_rank_membership_ids: Optional[Set[str]] = None
 
         self.replica_average_ongoing_requests: Dict[str, float] = {}
 
@@ -4972,24 +4973,41 @@ def check_and_update_replicas(self):
         # if we delay the rank reassignment, the rank system will be in an invalid state
         # for a longer period of time. Abrar made this decision because he is not confident
         # about how rollouts work in the deployment state machine.
-        active_replicas = self._replicas.get()
+        self._maybe_check_rank_consistency()
+
+    def _maybe_check_rank_consistency(self) -> None:
+        """Run the rank-consistency pass only when replica membership changed.
+
+        The pass is O(N) with heavy constants and used to run on every
+        control-loop tick in steady state -- at 10K+ replicas it monopolizes
+        the controller loop. Rank consistency can only be violated by
+        membership changes, so the active replica-id set gates it.
+        """
+        # O(1) guards first -- `get()` below copies the whole replica list, and these
+        # two rule out the busiest ticks (rollouts, migrations) without paying for it.
         if (
-            active_replicas
-            and self._curr_status_info.status == DeploymentStatus.HEALTHY
+            self._curr_status_info.status != DeploymentStatus.HEALTHY
             # Skip consistency check if there are STARTING replicas. During node
             # migration, new replicas are created in STARTING state (without ranks)
             # after the status is set to HEALTHY. Running the consistency check
             # with STARTING replicas causes "active keys without ranks" error.
-            and self._replicas.count(states=[ReplicaState.STARTING]) == 0
+            or self._replicas.count(states=[ReplicaState.STARTING]) != 0
         ):
-            replicas_to_reconfigure = (
-                self._rank_manager.check_rank_consistency_and_reassign_minimally(
-                    active_replicas,
-                )
+            return
+        active_replicas = self._replicas.get()
+        if not active_replicas:
+            return
+        active_replica_ids = {r.replica_id.unique_id for r in active_replicas}
+        if active_replica_ids == self._last_rank_membership_ids:
+            return
+        replicas_to_reconfigure = (
+            self._rank_manager.check_rank_consistency_and_reassign_minimally(
+                active_replicas,
             )
-
-            # Reconfigure replicas that had their ranks reassigned
-            self._reconfigure_replicas_with_new_ranks(replicas_to_reconfigure)
+        )
+        # Reconfigure replicas that had their ranks reassigned
+        self._reconfigure_replicas_with_new_ranks(replicas_to_reconfigure)
+        self._last_rank_membership_ids = active_replica_ids
 
     def _handle_deployment_actor_failed_health_check(
         self,

python/ray/serve/tests/unit/test_deployment_state.py

@@ -10585,5 +10585,115 @@ def _make_app(name):
     assert dsm.is_ready_for_shutdown()
 
 
+def _scale_to(dsm, ds, num_replicas, version="1", ticks=14):
+    """Change membership via the public deploy path and settle back to HEALTHY."""
+    info, _ = deployment_info(num_replicas=num_replicas, version=version)
+    dsm.deploy(TEST_DEPLOYMENT_ID, info)
+    for _ in range(ticks):
+        dsm.update()
+        for replica in ds._replicas.get([ReplicaState.STARTING]):
+            replica._actor.set_ready()
+        if (
+            ds._curr_status_info.status == DeploymentStatus.HEALTHY
+            and ds._replicas.count(states=[ReplicaState.STARTING]) == 0
+        ):
+            dsm.update()
+            return
+    raise AssertionError(
+        "deployment never settled: status=%s running=%d starting=%d"
+        % (
+            ds._curr_status_info.status,
+            ds._replicas.count(states=[ReplicaState.RUNNING]),
+            ds._replicas.count(states=[ReplicaState.STARTING]),
+        )
+    )
+
+
+class CountingRankManager(ds_mod.DeploymentRankManager):
+    """Real rank manager that records how often the consistency pass runs.
+
+    Injected at the same seam the fixture uses for actor wrappers, so the rank logic under
+    test stays real; only the call count is added.
+    """
+
+    instances = []
+
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+        self.consistency_calls = 0
+        CountingRankManager.instances.append(self)
+
+    def check_rank_consistency_and_reassign_minimally(self, active_replicas):
+        self.consistency_calls += 1
+        return super().check_rank_consistency_and_reassign_minimally(active_replicas)
+
+
+@pytest.fixture
+def rank_gate_dsm(mock_deployment_state_manager, monkeypatch):
+    """A running deployment whose rank manager counts consistency passes."""
+    CountingRankManager.instances = []
+    monkeypatch.setattr(ds_mod, "DeploymentRankManager", CountingRankManager)
+
+    create_dsm, timer, _, _ = mock_deployment_state_manager
+    dsm: DeploymentStateManager = create_dsm()
+    info, v1 = deployment_info(num_replicas=3, version="1")
+    assert dsm.deploy(TEST_DEPLOYMENT_ID, info)
+    ds = dsm._deployment_states[TEST_DEPLOYMENT_ID]
+
+    dsm.update()
+    for replica in ds._replicas.get():
+        replica._actor.set_ready()
+    dsm.update()  # STARTING -> RUNNING
+    check_counts(ds, total=3, by_state=[(ReplicaState.RUNNING, 3, v1)])
+
+    # Settle: the deployment reaches HEALTHY a tick or two after the replicas do, and the
+    # gate legitimately runs once for this membership. Tests measure from after that.
+    for _ in range(8):
+        dsm.update()
+    assert ds._curr_status_info.status == DeploymentStatus.HEALTHY
+    assert (
+        ds._rank_manager.consistency_calls >= 1
+    ), "gate never ran for a new membership"
+    return dsm, ds, ds._rank_manager, timer
+
+
+class TestRankConsistencyMembershipGate:
+    """The rank-consistency pass runs only when replica membership changes."""
+
+    def test_skips_while_membership_unchanged(self, rank_gate_dsm):
+        dsm, _, rank_manager, _ = rank_gate_dsm
+        after_startup = rank_manager.consistency_calls
+        for _ in range(5):
+            dsm.update()
+        assert rank_manager.consistency_calls == after_startup
+
+    def test_reruns_when_a_replica_leaves(self, rank_gate_dsm):
+        dsm, ds, rank_manager, timer = rank_gate_dsm
+        before = rank_manager.consistency_calls
+
+        # Membership change through the public path: scale up adds a new replica id.
+        _scale_to(dsm, ds, 4)
+
+        assert rank_manager.consistency_calls > before
+
+    def test_starting_replicas_skip_the_pass(
+        self, mock_deployment_state_manager, monkeypatch
+    ):
+        """A STARTING replica has no rank yet, so running the pass would raise
+        "active keys without ranks"; the guard must skip before that."""
+        CountingRankManager.instances = []
+        monkeypatch.setattr(ds_mod, "DeploymentRankManager", CountingRankManager)
+        create_dsm, _, _, _ = mock_deployment_state_manager
+        dsm: DeploymentStateManager = create_dsm()
+        info, _ = deployment_info(num_replicas=2, version="1")
+        assert dsm.deploy(TEST_DEPLOYMENT_ID, info)
+        ds = dsm._deployment_states[TEST_DEPLOYMENT_ID]
+
+        dsm.update()  # replicas are STARTING, never marked ready
+        dsm.update()
+        check_counts(ds, total=2, by_state=[(ReplicaState.STARTING, 2, None)])
+        assert ds._rank_manager.consistency_calls == 0
+
+
 if __name__ == "__main__":
     sys.exit(pytest.main(["-v", "-s", __file__]))