[Router][Bugfix] Defensive fallbacks in kvaware routing: 503 on no endpoints, session/QPS on unmapped instance

.gitignore

@@ -181,3 +181,6 @@ docs/book/src/docs
 
 # Sub charts
 helm/charts/*.tgz
+
+# local virtual environments
+.venv/

src/tests/test_kvaware_defensive_fallbacks.py

@@ -0,0 +1,99 @@
+"""Defensive fallbacks in KvawareRouter.route_request.
+
+Two crash paths observed/flagged in real deployments:
+- no endpoints available -> unguarded ``endpoints[0]`` raised IndexError
+  (LoadAwareRouter already answers 503; kvaware now mirrors it);
+- the KV lookup matches an instance the ip mapping cannot place (several
+  engines sharing one IP - QueryInstMsg keys instances by IP alone - or a
+  stale registration) -> unhandled KeyError failed the whole request with
+  a 500, even though session/QPS routing works fine without the cache hit.
+"""
+
+import pytest
+from fastapi import HTTPException
+
+from vllm_router.routers import routing_logic
+from vllm_router.routers.routing_logic import HashRing, KvawareRouter
+
+
+class _LookupMsg:
+    def __init__(self, tokens, event_id):
+        self.tokens = tokens
+        self.event_id = event_id
+
+
+class _QueryInstMsg:
+    def __init__(self, ip, event_id):
+        self.ip = ip
+        self.event_id = event_id
+
+
+@pytest.fixture(autouse=True)
+def _lmcache_message_stubs(monkeypatch):
+    # The lmcache import in routing_logic is guarded; the messages may be
+    # absent in the test environment - stub the two types the router builds.
+    monkeypatch.setattr(routing_logic, "LookupMsg", _LookupMsg, raising=False)
+    monkeypatch.setattr(routing_logic, "QueryInstMsg", _QueryInstMsg, raising=False)
+
+
+URL_A = "http://engine-a:8000"
+URL_B = "http://engine-b:8000"
+
+
+class Endpoint:
+    def __init__(self, url):
+        self.url = url
+        self.model_names = ["m"]
+
+
+class Tokenizer:
+    @staticmethod
+    def encode(prompt):
+        return [1] * 8
+
+
+class LookupRet:
+    def __init__(self, layout_info):
+        self.layout_info = layout_info
+
+
+class QueryRet:
+    # Real controllers answer QueryInstMsg with instance_id=None for IPs
+    # they cannot attribute - exactly how the broken mapping arises.
+    instance_id = None
+
+
+def _bare_router():
+    router = KvawareRouter.__new__(KvawareRouter)
+    router.tokenizer = Tokenizer()
+    router.threshold = 2000
+    router.instance_id_to_ip = {}
+    router.session_key = None
+    router.hash_ring = HashRing()
+    return router
+
+
+@pytest.mark.asyncio
+async def test_no_endpoints_answers_503_not_indexerror():
+    router = _bare_router()
+    with pytest.raises(HTTPException) as exc:
+        await router.route_request([], {}, {}, None, {"prompt": "hello"})
+    assert exc.value.status_code == 503
+
+
+@pytest.mark.asyncio
+async def test_unmapped_instance_falls_back_instead_of_500():
+    router = _bare_router()
+
+    async def query_manager(msg):
+        if isinstance(msg, _LookupMsg):
+            return LookupRet({"ghost-instance": ("LocalCPUBackend", 8)})
+        return QueryRet()
+
+    router.query_manager = query_manager
+    url = await router.route_request(
+        [Endpoint(URL_A), Endpoint(URL_B)], {}, {}, None, {"prompt": "hello"}
+    )
+    # QPS fallback with no stats picks the first endpoint - the request is
+    # served, just without the cache-hit placement.
+    assert url == URL_A

src/vllm_router/routers/routing_logic.py

@@ -361,6 +361,23 @@ def close(self):
                 pass
             self.lmcache_cluster_monitor_task = None
 
+    def fallback_url(
+        self,
+        endpoints: List[EndpointInfo],
+        request_stats: Dict[str, RequestStats],
+        request: Request,
+        request_json: Dict,
+    ) -> str:
+        """Route without cache information: session hash if any, else lowest
+        QPS. Shared by the lookup-miss and mapping-miss paths here and by
+        LoadAwareRouter."""
+        session_id = self.extract_session_id(request, request_json)
+        logger.debug(f"Fallback to using session id: {session_id}")
+        self._update_hash_ring(endpoints)
+        if session_id is None:
+            return self._qps_routing(endpoints, request_stats)
+        return self.hash_ring.get_node(session_id)
+
     async def route_request(
         self,
         endpoints: List[EndpointInfo],
@@ -384,7 +401,15 @@ async def route_request(
             request (Request): The incoming request
             request_json (Dict): The request body (needed for finding the
             longest prefix match)
+
+        Raises:
+            HTTPException: 503 if no endpoints are available.
         """
+        if not endpoints:
+            raise HTTPException(
+                status_code=503, detail="No backend endpoints available"
+            )
+
         token_ids = None
         # Local-first tokenization, fall back to remote "/tokenize" API on failure
         # TODO (Yuhan): Handle chat completions
@@ -423,17 +448,7 @@ async def route_request(
             or len(instance_id.layout_info) == 0
             or matched_tokens < max(len(token_ids) - self.threshold, 0)
         ):
-            session_id = self.extract_session_id(request, request_json)
-            logger.debug(f"Fallback to using session id: {session_id}")
-            # Update the hash ring with the current list of endpoints
-            self._update_hash_ring(endpoints)
-            if session_id is None:
-                # Route based on QPS if no session ID is present
-                url = self._qps_routing(endpoints, request_stats)
-            else:
-                # Use the hash ring to get the endpoint for the session ID
-                url = self.hash_ring.get_node(session_id)
-            return url
+            return self.fallback_url(endpoints, request_stats, request, request_json)
         else:
             queried_instance_ids = [info for info in instance_id.layout_info]
             if queried_instance_ids[0] not in self.instance_id_to_ip:
@@ -454,10 +469,25 @@ async def route_request(
                         endpoint.url
                     )
                 logger.info(f"Instance id to ip mapping: {self.instance_id_to_ip}")
+            url = self.instance_id_to_ip.get(queried_instance_ids[0])
+            if url is None:
+                # The lookup matched an instance the ip mapping cannot place
+                # (e.g. several engines sharing one IP - QueryInstMsg keys
+                # instances by IP alone - or a stale registration). Routing
+                # can still proceed without the cache hit; an unhandled
+                # KeyError here fails the whole request with a 500.
+                logger.warning(
+                    f"kvaware matched instance {queried_instance_ids[0]} has no "
+                    f"known endpoint mapping ({self.instance_id_to_ip}); falling "
+                    f"back to session/QPS routing"
+                )
+                return self.fallback_url(
+                    endpoints, request_stats, request, request_json
+                )
             logger.info(
                 f"Routing request to {queried_instance_ids[0]} found by kvaware router"
             )
-            return self.instance_id_to_ip[queried_instance_ids[0]]
+            return url
 
 
 class LoadAwareRouter(KvawareRouter):
@@ -674,22 +704,6 @@ async def tokenize_prompt(
             )
             return response.json()["tokens"]
 
-    def fallback_url(
-        self,
-        endpoints: List[EndpointInfo],
-        request_stats: Dict[str, RequestStats],
-        request: Request,
-        request_json: Dict,
-    ) -> str:
-        """Upstream's no-cache-info route: session hash if any, else lowest
-        QPS."""
-        session_id = self.extract_session_id(request, request_json)
-        logger.debug(f"Fallback to using session id: {session_id}")
-        self._update_hash_ring(endpoints)
-        if session_id is None:
-            return self._qps_routing(endpoints, request_stats)
-        return self.hash_ring.get_node(session_id)
-
     async def route_request(
         self,
         endpoints: List[EndpointInfo],