[Bugfix][Router] Reconcile the engine list when the K8s watch reconnects

src/tests/test_k8s_pod_ip_service_discovery.py

@@ -125,3 +125,47 @@ def test_modified_not_ready_with_ip_removes_registered():
     )
 
     assert "pod-c" not in d.available_engines
+
+
+def test_reconcile_drops_engine_whose_pod_is_gone():
+    """A DELETED event missed while the watch was down leaves a ghost.
+
+    The watch stream is the only source of removals, so an engine whose pod
+    disappeared during a disconnect is never dropped and keeps taking
+    traffic. Listing on reconnection must remove it.
+    """
+    d = _make_discovery()
+    d.label_selector = "environment=test"
+    _register(d, "pod-gone")
+    _register(d, "pod-alive")
+
+    still_running = MagicMock()
+    still_running.metadata.name = "pod-alive"
+    d.k8s_api = MagicMock()
+    d.k8s_api.list_namespaced_pod.return_value = MagicMock(items=[still_running])
+
+    d._reconcile_engines()
+
+    assert "pod-gone" not in d.available_engines
+    assert "pod-alive" in d.available_engines
+
+
+def test_reconcile_keeps_every_live_engine():
+    """The list must never drop an engine whose pod is still there."""
+    d = _make_discovery()
+    d.label_selector = "environment=test"
+    _register(d, "pod-a")
+    _register(d, "pod-b")
+
+    pods = []
+    for name in ("pod-a", "pod-b"):
+        pod = MagicMock()
+        pod.metadata.name = name
+        pods.append(pod)
+
+    d.k8s_api = MagicMock()
+    d.k8s_api.list_namespaced_pod.return_value = MagicMock(items=pods)
+
+    d._reconcile_engines()
+
+    assert set(d.available_engines) == {"pod-a", "pod-b"}

src/vllm_router/service_discovery.py

@@ -668,9 +668,35 @@ def _get_model_label(self, pod) -> Optional[str]:
             return None
         return pod.metadata.labels.get("model")
 
+    def _reconcile_engines(self):
+        """
+        Drop engines whose pod is gone from the cluster.
+
+        The watch stream is the only source of removals, so a DELETED event
+        that lands while the connection is down is lost for good and the
+        engine keeps receiving traffic. Listing on every (re)connection
+        closes that window: the watch keeps handling the steady state, the
+        list repairs whatever it missed.
+        """
+        pods = self.k8s_api.list_namespaced_pod(
+            namespace=self.namespace,
+            label_selector=self.label_selector,
+        )
+        live_pods = {pod.metadata.name for pod in pods.items}
+
+        with self.available_engines_lock:
+            stale = set(self.available_engines) - live_pods
+            for engine_name in stale:
+                logger.warning(
+                    f"Serving engine {engine_name} no longer exists but was "
+                    f"still registered: dropping it"
+                )
+                del self.available_engines[engine_name]
+
     def _watch_engines(self):
         while self.running:
             try:
+                self._reconcile_engines()
                 for event in self.k8s_watcher.stream(
                     self.k8s_api.list_namespaced_pod,
                     namespace=self.namespace,