[serve] Push-based replica health: self-checks ride replica metric reports

P1 of the push-health stack. Replicas run their own health check on a
periodic task (twice per health-check period) and push the result: on the
metric reports they already send when those are frequent enough to keep the
controller's view fresh, or on a lightweight heartbeat otherwise. The
controller keeps a push registry; pull probes become the fallback for stale
pushes, guarded by a per-deployment systemic-stall verdict (mass staleness
reads as controller ingest lag and defers probes, capped per episode) so an
overloaded controller cannot trigger a probe storm.

Review fixes folded in:

- Registry pruning is rate-limited to once per 30s; the over-threshold prune
  used to re-run an O(N) dict rebuild on every record() once all entries were
  fresh. The stall window defaults to the stock health-check period so
  deployments without target info (recovery, deletion) cannot misread every
  push as stale.
- An unhealthy self-check is never suppressed: suppression assumed a recent
  report carried current health, but a flip to unhealthy after the last carry
  would be delayed a full period.
- The push-stall tally accumulates across ticks rather than per dirty-set
  slice, which only engaged the guard above ~3200 replicas per deployment.
  Extract _reconcile_sweep_ticks (shared with the dirty set). Known follow-up:
  many small deployments can each stay under the per-deployment floor while
  the controller lags overall.
- Gap stats live in two windows instead of accumulating over the controller's
  lifetime, where the reported p50/p99/max went insensitive to current lag
  exactly when the numbers are worth reading; count stays cumulative.
- A timed-out self-check logs its cause, and marks itself unhealthy on
  cancellation: wait_for enforces the timeout by cancelling the check, and
  CancelledError is not an Exception, so the cached result stayed healthy and
  a fallback pull probe would answer healthy for a wedged check.
- A probe still in flight when a newer push is applied no longer overwrites it
  on resolving: its result predates the push, so it is dropped instead of
  flapping the failure count across pull-to-push transitions. ACTOR_CRASHED is
  exempt -- a crash is authoritative and a dead replica pushes nothing.
- Probe deferral no longer borrows _last_push_consume_time. That field now means
  strictly "a push was applied"; deferral carries its own deadline, supplied by
  the deployment as the stall episode's cap. Before, the gate honoured the
  borrowed stamp for a further freshness window past the cap (+1.5x the period,
  so 570s against a documented 120s at a 300s period), and the drop guard above
  read a deferral as a superseding observation and destroyed every probe result
  that resolved during an episode -- exactly when a probe is the only signal.
- The stall tally counts each replica once per window. The dirty set re-visits a
  replica with an in-flight probe every tick, and those are precisely the
  stale-push ones, so the undeduped tally could read a small failed cohort as
  fleet-wide ingest lag and defer probes for everyone. The MIN_TRACKED floor now
  means 64 distinct replicas, matching what it always claimed to mean.
- A replica with no registry entry is no longer deferred: probes are its only
  health signal and it cannot sway the verdict either way.
- health_check_failures_counter follows the observation actually acted on. Only
  the probe paths set the failure flag, so the counter had gone silent for
  push-detected failures while still counting probe results the controller
  discarded -- inverted precisely when push health is doing its job.
- Heartbeat suppression compares the metric-report interval against the
  configured health-check period, not the half-period evaluation cadence --
  under the stock config (10s period, 10s metric interval) the latter
  suppressed nothing, so every reporting replica heartbeated anyway.

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

Author: johntaylor-cell

python/ray/serve/_private/common.py

@@ -1061,6 +1061,11 @@ class ReplicaMetricReport:
     aggregated_metrics: Dict[str, float]
     metrics: Dict[str, TimeSeries]
     timestamp: float
+    # Replica-pushed self-health (None = sender does not push health; the
+    # controller then falls back to pull probes).
+    healthy: Optional[bool] = None
+    health_checked_at: Optional[float] = None
+    health_consecutive_failures: Optional[int] = None
 
 
 @dataclass

python/ray/serve/_private/controller.py

@@ -61,6 +61,7 @@
 from ray.serve._private.deployment_info import DeploymentInfo
 from ray.serve._private.deployment_state import (
     DeploymentStateManager,
+    ReplicaHealthPushRegistry,
 )
 from ray.serve._private.endpoint_state import EndpointState
 from ray.serve._private.exceptions import ExternalScalerDisabledError
@@ -254,13 +255,15 @@ async def __init__(
         ]
 
         self.autoscaling_state_manager = AutoscalingStateManager()
+        self._replica_health_push_registry = ReplicaHealthPushRegistry()
         self.deployment_state_manager = DeploymentStateManager(
             self.kv_store,
             self.long_poll_host,
             all_serve_actor_names,
             get_all_live_placement_group_names(),
             self.cluster_node_info_cache,
             self.autoscaling_state_manager,
+            health_push_registry=self._replica_health_push_registry,
         )
 
         # Manage all applications' state
@@ -391,10 +394,30 @@ def record_autoscaling_metrics_from_replica(
         )
         # Track in health metrics
         self._health_metrics_tracker.record_replica_metrics_delay(latency_ms)
+        if replica_metric_report.healthy is not None:
+            self._replica_health_push_registry.record(
+                replica_metric_report.replica_id.unique_id,
+                replica_metric_report.health_checked_at
+                or replica_metric_report.timestamp,
+                replica_metric_report.healthy,
+                replica_metric_report.health_consecutive_failures,
+            )
         self.autoscaling_state_manager.record_request_metrics_for_replica(
             replica_metric_report
         )
 
+    def record_replica_health(
+        self,
+        replica_unique_id: str,
+        checked_at: float,
+        healthy: bool,
+        consecutive_failures: Optional[int] = None,
+    ):
+        """Self-health heartbeat from replicas that do not push metric reports."""
+        self._replica_health_push_registry.record(
+            replica_unique_id, checked_at, healthy, consecutive_failures
+        )
+
     def record_autoscaling_metrics_from_handle(
         self, handle_metric_report: Union[HandleMetricReport, bytes]
     ):
@@ -619,6 +642,14 @@ async def run_control_loop_step(
             dsm_duration = time.time() - dsm_update_start_time
             self.dsm_update_duration_gauge_s.set(dsm_duration)
             self._health_metrics_tracker.record_dsm_update_duration(dsm_duration)
+            try:
+                _gs = self._replica_health_push_registry.gap_stats()
+                self._health_metrics_tracker.push_checks_recorded = _gs["count"]
+                self._health_metrics_tracker.push_check_gap_p50_s = _gs["p50_s"]
+                self._health_metrics_tracker.push_check_gap_p99_s = _gs["p99_s"]
+                self._health_metrics_tracker.push_check_gap_max_s = _gs["max_s"]
+            except Exception:
+                pass
             if not self.done_recovering_event.is_set() and not any_recovering:
                 self.done_recovering_event.set()
                 if num_loops > 0:

python/ray/serve/_private/controller_health_metrics_tracker.py

@@ -50,6 +50,13 @@ class ControllerHealthMetricsTracker:
     num_control_loops: int = 0
     last_control_loop_time: float = 0.0
 
+    # Replica self-check intervals observed via health pushes (set by the
+    # controller loop from the push registry).
+    push_checks_recorded: int = 0
+    push_check_gap_p50_s: float = 0.0
+    push_check_gap_p99_s: float = 0.0
+    push_check_gap_max_s: float = 0.0
+
     def record_loop_duration(self, duration: float):
         self.loop_durations.append(duration)
 
@@ -145,4 +152,8 @@ def collect_metrics(self) -> ControllerHealthMetrics:
             handle_metrics_delay_ms=handle_delay_stats,
             replica_metrics_delay_ms=replica_delay_stats,
             process_memory_mb=process_memory_mb,
+            push_checks_recorded=self.push_checks_recorded,
+            push_check_gap_p50_s=self.push_check_gap_p50_s,
+            push_check_gap_p99_s=self.push_check_gap_p99_s,
+            push_check_gap_max_s=self.push_check_gap_max_s,
         )