Accumulate push-stall tally across ticks, not per dirty-set slice

Review: the dirty-set health-checks only a slice (~N/sweep_ticks) each
tick, but the stall guard compared its 64-sample floor against that one
slice -- so it only engaged above ~3200 replicas per deployment.
Accumulate the tally across ticks and finalize once enough replicas are
sampled or a full sweep elapses; a large deployment whose one-tick slice
already exceeds the floor still finalizes next tick, so engagement stays
prompt where it matters. Extract _reconcile_sweep_ticks (shared with the
dirty set). Known follow-up: many small deployments can still each stay
under the per-deployment floor while the controller lags overall.

Signed-off-by: john.taylor <john.taylor@anyscale.com>

Author: johntaylor-cell

python/ray/serve/_private/deployment_state.py

@@ -3113,6 +3113,7 @@ def __init__(
         self._push_stall_since: Optional[float] = None
         self._push_stall_stale: int = 0
         self._push_stall_total: int = 0
+        self._push_stall_ticks: int = 0
         # Default to the stock period so pushes are not misread as stale on
         # deployments that have no target info yet (recovery, deletion).
         self._push_stall_window_s: float = _push_freshness_window_s(
@@ -4975,15 +4976,7 @@ def _dirty_set_active_pairs(self):
         self._outstanding_dirty_set = still
         n = container.count_state(ReplicaState.RUNNING)
         if n:
-            period = self._reconcile_sweep_period_s()
-            ticks = max(
-                1,
-                int(
-                    CONTROLLER_HEALTH_CHECK_RECONCILIATION_FRACTION
-                    * period
-                    / max(CONTROL_LOOP_INTERVAL_S, 1e-3)
-                ),
-            )
+            ticks = self._reconcile_sweep_ticks()
             slice_n = max(1, (n + ticks - 1) // ticks)
             start = self._dirty_set_rr_cursor % n
             sl = container.slice_state(ReplicaState.RUNNING, start, slice_n)
@@ -5017,10 +5010,42 @@ def _reconcile_sweep_period_s(self) -> float:
                 DEFAULT_HEALTH_CHECK_PERIOD_S, DEFAULT_REQUEST_ROUTING_STATS_PERIOD_S
             )
 
+    def _reconcile_sweep_ticks(self) -> int:
+        """Control-loop ticks the dirty-set takes to sweep the RUNNING bucket
+        once (see _dirty_set_active_pairs)."""
+        return max(
+            1,
+            int(
+                CONTROLLER_HEALTH_CHECK_RECONCILIATION_FRACTION
+                * self._reconcile_sweep_period_s()
+                / max(CONTROL_LOOP_INTERVAL_S, 1e-3)
+            ),
+        )
+
+    def _advance_push_stall_window(self) -> None:
+        """Finalize the systemic-stall verdict once enough replicas have been
+        sampled across ticks (or a full sweep elapsed), then reset the tally.
+
+        Each tick only health-checks a dirty-set slice (~N/sweep_ticks), well
+        below the guard's min-tracked floor for all but the largest deployments;
+        accumulating across ticks lets the guard reach a deployment-representative
+        sample. A large deployment whose one-tick slice already exceeds the floor
+        still finalizes next tick, so engagement stays prompt where it matters.
+        """
+        self._push_stall_ticks += 1
+        if (
+            self._push_stall_total >= HEALTH_PUSH_STALL_MIN_TRACKED
+            or self._push_stall_ticks >= self._reconcile_sweep_ticks()
+        ):
+            self._finalize_push_stall_verdict()
+            self._push_stall_stale = 0
+            self._push_stall_total = 0
+            self._push_stall_ticks = 0
+
     def _finalize_push_stall_verdict(self) -> None:
-        """Set next sweep's probe-deferral from this sweep's staleness tally.
+        """Set probe-deferral from the accumulated staleness tally.
 
-        Most of this deployment's pushed health going stale at once (by its own
+        Most of this deployment's sampled pushes going stale at once (by its own
         window) reads as controller-side ingest lag, so fallback probes are
         deferred -- bounded by HEALTH_PUSH_STALL_MAX_DEFER_S per episode so a
         genuine mass outage still falls through to probes.
@@ -5071,9 +5096,7 @@ def check_and_update_replicas(self):
         with state container from previous update() cycle to see if any state
         transition happened.
         """
-        self._finalize_push_stall_verdict()
-        self._push_stall_stale = 0
-        self._push_stall_total = 0
+        self._advance_push_stall_window()
         if self._target_state.info is not None:
             self._push_stall_window_s = _push_freshness_window_s(
                 self._target_state.info.deployment_config.health_check_period_s

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

@@ -9791,10 +9791,11 @@ def _shim(self, container):
         s._outstanding_dirty_set = set()
         s._dirty_set_rr_cursor = 0
         s._target_state = object()  # .info access raises -> default reconcile period
-        # Bind the real period helper so _dirty_set_active_pairs can call it on the shim.
+        # Bind the real period/ticks helpers so _dirty_set_active_pairs can call them.
         s._reconcile_sweep_period_s = DeploymentState._reconcile_sweep_period_s.__get__(
             s
         )
+        s._reconcile_sweep_ticks = DeploymentState._reconcile_sweep_ticks.__get__(s)
         return s
 
     def test_covers_all_running_within_one_sweep(self):
@@ -10246,6 +10247,40 @@ def test_majority_stale_defers_next_sweep(self):
         ds._apply_pushed_health(replica)
         replica.defer_push_fallback_probe.assert_called_once()
 
+    def test_accumulates_across_ticks_below_slice_threshold(self):
+        # Each tick samples only a small slice (< min tracked), but accumulation
+        # across ticks reaches the sample size, so systemic lag still engages.
+        ds = self._ds()
+        ds._push_stall_ticks = 0
+        ds._reconcile_sweep_ticks = lambda: 1000  # long sweep -> threshold path
+        for _ in range(8):  # 8 * 10 = 80 >= HEALTH_PUSH_STALL_MIN_TRACKED
+            self._seed(ds, 10, 10)
+            self._sweep(ds, 10)
+            ds._advance_push_stall_window()
+        assert ds._push_probes_deferred
+
+    def test_tiny_deployment_never_engages_even_accumulated(self):
+        # A full sweep accumulates < min tracked -> finalize bails, never defers.
+        ds = self._ds()
+        ds._push_stall_ticks = 0
+        ds._reconcile_sweep_ticks = lambda: 3
+        for _ in range(3):
+            self._seed(ds, 10, 10)
+            self._sweep(ds, 10)
+            ds._advance_push_stall_window()
+        assert not ds._push_probes_deferred
+
+    def test_large_slice_engages_in_one_tick(self):
+        # A single tick whose slice already exceeds the floor finalizes next
+        # tick -- prompt engagement preserved for large deployments.
+        ds = self._ds()
+        ds._push_stall_ticks = 0
+        ds._reconcile_sweep_ticks = lambda: 50
+        self._seed(ds, 100, 80)
+        self._sweep(ds, 100)
+        ds._advance_push_stall_window()  # total 100 >= floor -> finalized
+        assert ds._push_probes_deferred
+
     def test_below_min_tracked_never_defers(self):
         ds = self._ds()
         self._tally_and_verdict(ds, n=63, stale_n=63)