[serve] Push health: interleave health frames on open response streams

[johntaylor-cell]

Third PR of the push-health stack (stacked on #64912, which stacks on #64913).

Why

After P1 (health rides replica metric reports) and P2 (health rides handle metric
reports via response admission), one class of replica still falls back to heartbeat
RPCs: replicas that don't push metric reports and whose requests are held for a
long time — e.g. handle-owning pipeline replicas whose downstream calls run for
minutes. Admission-time piggyback (P2) only carries health when a request is
admitted; a replica sitting on long-running requests has no admissions, so its
carried health goes stale and the P1 heartbeat kicks in. At high replica counts
those heartbeats are exactly the standalone health RPCs this stack exists to
eliminate.

The observation: such a replica already has an open response stream to its
caller (the rejection-protocol generator). Health can ride that.

What

• ReplicaHealthFrame — a small frozen-dataclass system message (healthy,
  checked_at, consecutive_failures).
• Admission-time declaration: for actor-transport unary rejection-protocol
  calls, the caller sets RequestMetadata.supports_health_frames; the replica
  answers with will_send_health_frames in the accepted ReplicaQueueLengthInfo,
  binding the behavior for that request's lifetime.
• Replica side: a declared unary call runs through a wrapper generator that
  awaits the user task in health-period ticks and yields a frame whenever there is
  a newer self-check result — the final yield is always the user result.
• Router side: a lazy frame pump (armed only for declared requests, started
  only if the request is still pending after a short delay) consumes frames into
  the router's carried-health map, where P2 folds them into the handle metric
  reports it already sends. Frames never reach user code; the result path filters
  them.

Suppression (no double delivery)

Frames stop the moment health rides any other channel, re-checked every tick while
a request is held: if metric reports carry health (cadence ≤ health period) or the
replica's health was recently consumed into a handle report, the wrapper skips the
frame. Conversely the router arms a pump only when the replica declared frames —
at 10K+ fan-in, unconditionally-armed idle pumps cost real driver memory.

Limitations

Frames concentrate at the request owner: a single process holding requests on a
very large fleet receives that fleet's frames (measured ~1.6K frames/s on one
process at 16K replicas before declaration gating; gating removes the idle-pump
cost but a genuinely held-everywhere workload still fans in). For one-owner /
huge-fleet topologies the report-aggregation tier remains future work.

Testing

• New unit suite for the frame protocol: declaration binding, pump arming/lazy
  start/settling, per-tick suppression, frame filtering on the result path,
  newest-only dedupe (331 lines).
• End-to-end integration test: a single held request demonstrably advances the
  router's carried health past the admission-time entry.
• Full serve unit suites green on the stack tip (388 tests including P1 + P2
  suites, identical to the pre-split composition).

Benchmarks

Full-stack numbers (this PR + 64913 + 64912 on master + the rank-gate and id-walk-cache
PRs), HAProxy off, pinned replicas, 36 samples/cell:

Scale	Control loop	Loops/s	Health gap p99	Deadline misses
8,192	52.1 ms	5.60	9.5 s	23 of 1.09M (0.002%)
16,384	42.9 ms	4.99	9.5 s	0.55% (≈once per replica, at ramp)

Per-PR attribution (P2-only vs +P3 arms at 8K/16K) is being measured on the same
harness; numbers land here before review.

[cursor]

Cursor Bugbot has reviewed your changes using default effort and found 1 potential issue.

^{Reviewed by Cursor Bugbot for commit 0fa54bd. Configure here.}

[gemini-code-assist]

Code Review

This pull request implements a replica self-health push mechanism in Ray Serve to replace or defer pull-based probing, allowing replicas to piggyback health status on metric reports, response streams via ReplicaHealthFrame messages, or lightweight heartbeats. It also introduces a systemic stall guard to prevent probe storms during controller ingest lag, and optimizes controller performance by memoizing replica membership lookups and gating rank consistency checks. The review feedback identifies a critical deadlock risk in replica_result.py caused by holding a synchronous lock across await points, a potential performance issue in the health push registry pruning logic under high churn, and a bug where the push stall window remains uninitialized if target state info is missing.

[gemini-code-assist]

Holding the synchronous threading.Lock (self._object_ref_or_gen_sync_lock) across await points (lines 417 and 423) is a critical anti-pattern. If to_object_ref_async is suspended at an await point while holding this lock, any synchronous call to to_object_ref on the same thread will attempt to acquire the lock and block. Since the thread is blocked, the event loop cannot resume to_object_ref_async to release the lock, resulting in a guaranteed deadlock. To resolve this, release the lock before performing any asynchronous operations, and use an asyncio.Future to coordinate concurrent asynchronous callers.

            consumption_fut = getattr(self, "_async_consumption_future", None)
            if consumption_fut is not None:
                return await consumption_fut

            acquired = self._object_ref_or_gen_sync_lock.acquire(blocking=False)
            if acquired:
                try:
                    if self._obj_ref is not None:
                        return self._obj_ref
                    self._consumption_started = True
                    self._async_consumption_future = asyncio.get_running_loop().create_future()
                finally:
                    self._object_ref_or_gen_sync_lock.release()

                try:
                    while True:
                        obj_ref = await self._obj_ref_gen.__anext__()
                        if not self._health_frames_possible:
                            break
                        try:
                            value = await obj_ref
                        except Exception:
                            break
                        if isinstance(value, ReplicaHealthFrame):
                            self._record_health_frame(value)
                            continue
                        break
                    with self._object_ref_or_gen_sync_lock:
                        self._obj_ref = obj_ref
                    self._async_consumption_future.set_result(obj_ref)
                    return self._obj_ref
                except Exception as e:
                    self._async_consumption_future.set_exception(e)
                    raise

[gemini-code-assist]

If the number of active replicas or churned replicas exceeds _PRUNE_THRESHOLD (65536) and they are still within the _PRUNE_MAX_AGE_S (10 minutes) window, len(self._state) will remain above the threshold even after pruning. This causes every subsequent call to record to execute the O(N) dictionary comprehension, leading to high CPU usage and potential controller lag. Introduce a time-based guard to ensure pruning runs at most once per _PRUNE_MAX_AGE_S interval.

        now = time.time()
        if len(self._state) > self._PRUNE_THRESHOLD and now - getattr(self, "_last_prune_time", 0.0) > self._PRUNE_MAX_AGE_S:
            self._last_prune_time = now
            # Age by controller-clock arrival time (v[1]), immune to replica skew.
            cutoff = now - self._PRUNE_MAX_AGE_S
            self._state = {k: v for k, v in self._state.items() if v[1] >= cutoff}

[gemini-code-assist]

If self._target_state.info is None, self._push_stall_window_s is not updated and remains at its default value of 0.0. This will cause all pushed health reports to be treated as stale, potentially triggering false push stall verdicts. Fall back to DEFAULT_HEALTH_CHECK_PERIOD_S when self._target_state.info is None.

Suggested change

	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
	)
	period = (
	self._target_state.info.deployment_config.health_check_period_s
	if self._target_state.info is not None
	else DEFAULT_HEALTH_CHECK_PERIOD_S
	)
	self._push_stall_window_s = _push_freshness_window_s(period)

[johntaylor-cell]

Abandoning