[TRTLLM-13409][fix] stop reporting disagg readiness for workers that have died

tensorrt_llm/serve/disagg_coordinator.py

@@ -79,6 +79,17 @@
 COORDINATOR_RESERVATION_TIMEOUT_DEFAULT_S = 180.0
 COORDINATOR_STATE_SYNC_INTERVAL_S = 3.0
 
+GEN_ONLY_BENCHMARK_ENV = "TRTLLM_DISAGG_BENCHMARK_GEN_ONLY"
+# Number of metadata refresh intervals `servers` may go un-refreshed before
+# readiness treats it as stale. Tolerates an occasional missed poll without
+# letting a stopped monitor look healthy.
+_MONITOR_STALENESS_POLL_MULTIPLIER = 3
+
+
+def gen_only_benchmark_enabled() -> bool:
+    """Generation-only benchmark mode, which configures no context servers."""
+    return os.getenv(GEN_ONLY_BENCHMARK_ENV) == "1"
+
 
 def coordinator_reservation_timeout() -> float:
     return float(
@@ -167,6 +178,14 @@ def __init__(
         )
         self._server_start_timeout_secs = server_start_timeout_secs
         self._health_check_interval_secs = health_check_interval_secs
+        # How long `servers` may go un-refreshed before readiness stops
+        # trusting it. A few poll intervals, so an occasional slow or failed
+        # poll does not flap /health, but a monitor that has stopped is caught.
+        self._monitor_staleness_secs = (
+            _MONITOR_STALENESS_POLL_MULTIPLIER * metadata_config.refresh_interval
+            if metadata_config
+            else None
+        )
         self._reservation_timeout_secs = (
             coordinator_reservation_timeout()
             if reservation_timeout_secs is None
@@ -324,7 +343,38 @@ async def is_ready(self) -> bool:
                 self._ctx_router.num_prepared_servers,
                 self._gen_router.num_prepared_servers,
             )
-        return True
+        # A dead worker must stop being reported as ready: otherwise a client
+        # polling /health waits out its whole timeout against a group that can
+        # never answer. Deliberately not sticky -- a metadata-driven
+        # deployment adds and removes workers routinely, so recovery shows up
+        # as recovery.
+        if not self._metadata_server:
+            # Static server list: no monitor, so the lists never shrink and
+            # there is nothing new to report.
+            return True
+
+        # Monitoring is what makes `servers` a statement about the cluster. If
+        # it has stopped, the list is frozen on its last value and must not be
+        # trusted -- report not-ready rather than failing open.
+        # None only if a metadata server exists without its config, which
+        # create_metadata_server() does not produce; skip rather than raise
+        # from a health path if that ever changes.
+        max_age = self._monitor_staleness_secs
+        if max_age is not None and (
+            self._ctx_router.monitoring_is_stale(max_age)
+            or self._gen_router.monitoring_is_stale(max_age)
+        ):
+            logger.warning("Server monitoring is stale or stopped; reporting not-ready")
+            return False
+
+        # Disaggregated serving needs at least one generation server, and at
+        # least one context server unless this is a generation-only benchmark
+        # run, which intentionally configures none.
+        if not self._gen_router.servers:
+            return False
+        if gen_only_benchmark_enabled():
+            return True
+        return bool(self._ctx_router.servers)
 
     async def cluster_info(self) -> Dict[str, Any]:
         info = {
@@ -347,9 +397,7 @@ async def cluster_info(self) -> Dict[str, Any]:
         return info
 
     async def _wait_for_all_servers_ready(self) -> None:
-        import os
-
-        gen_only = os.getenv("TRTLLM_DISAGG_BENCHMARK_GEN_ONLY") == "1"
+        gen_only = gen_only_benchmark_enabled()
 
         async def check_servers_ready():
             elapsed_time = 0

tensorrt_llm/serve/router.py

@@ -342,6 +342,15 @@ def __init__(
         self._server_role = server_role
         self._lock = asyncio.Lock()
         self._monitor_task = None
+        # Wall-clock of the last poll that completed without error, used by
+        # ``monitoring_is_stale()``. ``None`` means monitoring was never
+        # started (a static server list), which is not staleness.
+        self._last_successful_poll: Optional[float] = None
+        # When the monitor task was created. Stands in for
+        # ``_last_successful_poll`` until the first poll lands, so a monitor
+        # whose every poll has failed still ages into staleness instead of
+        # holding the initial server list healthy forever.
+        self._monitor_started_at: Optional[float] = None
         self._session = None
         self._health_check_timeout = metadata_server_cfg.health_check_timeout if metadata_server_cfg else None
         self._server_preparation_func = server_preparation_func
@@ -378,6 +387,37 @@ def servers(self) -> List[str]:
     def num_prepared_servers(self) -> int:
         return len(self._prepared_ready_servers)
 
+    def monitoring_is_stale(self, max_age_secs: float) -> bool:
+        """Whether metadata-driven monitoring has stopped keeping up.
+
+        True when the monitor task has ended, or when no poll has completed
+        within ``max_age_secs``. Both mean ``servers`` is no longer a
+        statement about the cluster, so a readiness probe must not trust it.
+
+        Before the first poll lands, the monitor's own start time is the
+        reference point. Otherwise a monitor whose every poll has failed since
+        startup would never age into staleness, and readiness would keep
+        trusting the initial server list -- the same fail-open this is meant
+        to close, just entered from startup rather than from a later death.
+
+        Always False when monitoring was never started -- a static server list
+        has no monitor to go stale, and its list is correct by construction.
+        """
+        if self._monitor_task is None:
+            return False
+        if self._monitor_task.done():
+            return True
+        # Explicit None checks throughout: 0.0 is a legitimate
+        # ``time.monotonic()`` value, and `or` would silently discard it.
+        reference = self._last_successful_poll
+        if reference is None:
+            reference = self._monitor_started_at
+        if reference is None:
+            # Task exists but was not created by ``start_server_monitoring()``.
+            # Nothing to measure against; do not invent staleness.
+            return False
+        return (time.monotonic() - reference) > max_age_secs
+
     @property
     def prepared_servers(self) -> set[str]:
         return set(self._prepared_ready_servers)
@@ -502,6 +542,10 @@ async def start_server_monitoring(self, poll_interval: float = 10.0):
 
         logger.info(
             f"Starting server monitoring for {self._server_role} servers")
+        # Set before the task exists so staleness has a reference point from
+        # the very first poll onwards, including the polls that fail.
+        self._monitor_started_at = time.monotonic()
+        self._last_successful_poll = None
         self._monitor_task = asyncio.create_task(
             self._monitor_servers(poll_interval))
 
@@ -532,10 +576,15 @@ async def _monitor_servers(self, poll_interval: float = 10.0):
                 role_specific_servers = self._filter_servers_by_role(
                     live_servers, server_key_map)
 
-                # Use filtered servers if available
+                # Use filtered servers if available. An empty list is a valid
+                # state -- every worker of this role is gone -- and must be
+                # published so readiness reflects it. Asserting here instead
+                # would kill the loop and leave a stale, healthy-looking list.
                 final_servers = role_specific_servers
-
-                assert final_servers, f"No {self._server_role} servers available"
+                if not final_servers:
+                    logger.warning(
+                        f"No live {self._server_role} servers; publishing an "
+                        "empty server list for this role")
 
                 # Update server list
                 async with self._lock:
@@ -562,17 +611,33 @@ async def _monitor_servers(self, poll_interval: float = 10.0):
                         logger.debug(
                             f"No change in {self._server_role} server list: {len(self._servers)} servers"
                         )
+                self._last_successful_poll = time.monotonic()
+            except asyncio.CancelledError:
+                raise
             except Exception as e:
+                # Keep polling. Re-raising ends the loop, and a monitor that is
+                # no longer running leaves ``self._servers`` frozen on its last
+                # value -- which readiness would then read as healthy forever.
+                # ``_last_successful_poll`` is deliberately not updated, so
+                # ``monitoring_is_stale()`` starts reporting the gap.
                 logger.error(f"Error in server monitoring: {e}")
-                raise
 
             # Wait before next poll
             await asyncio.sleep(poll_interval)
 
-    def _filter_servers_by_role(self, servers, server_key_map):
-        """Filter servers by role (context or generation)"""
+    def _filter_servers_by_role(self, servers: List[str],
+                                server_key_map: Dict[str, str]) -> List[str]:
+        """Filter servers by role (context or generation)
+
+        Returns an empty list when no server of this role is live. That is a
+        legitimate observation, not an error: callers such as a readiness
+        probe need to be able to see "this role has no workers left". Raising
+        here instead would kill the monitor loop and freeze ``self._servers``
+        on its last known-good value, which reports the role as healthy
+        forever.
+        """
         if not servers:
-            raise RuntimeError("No servers available")
+            return []
 
         filtered_servers = []
         # Invert to get {url: key} for lookup