fix(gms): honor configured timeout during initial weights admission

components/src/dynamo/vllm/tests/test_vllm_worker_factory.py

@@ -1857,3 +1857,74 @@ async def test_passes_the_disabling_default_through_unchanged(self) -> None:
         handler_cls = await self._create_encode_worker(0.0)
 
         assert handler_cls.call_args.kwargs["embedding_cache_capacity_gb"] == 0.0
+
+
+class TestShadowStandbyEntry:
+    """Cover the real ``_maybe_wait_for_failover_lock``.
+
+    Every other test in this file replaces it with an ``AsyncMock``, so the
+    bound on the pause would otherwise never be executed.
+    """
+
+    @staticmethod
+    def _handler_that_never_pauses():
+        class _NeverPauses:
+            async def pause(self, *_args: object) -> bool:
+                await asyncio.sleep(3600)
+                return True
+
+        return SimpleNamespace(_pause_controller=_NeverPauses())
+
+    @pytest.mark.timeout(30)
+    async def test_pause_that_never_returns_raises_naming_engine_and_stage(
+        self, monkeypatch
+    ):
+        monkeypatch.setenv("ENGINE_ID", "2")
+        monkeypatch.setenv("DYN_GMS_SHADOW_PAUSE_TIMEOUT_SECONDS", "0.2")
+        elect_and_wake = AsyncMock()
+        monkeypatch.setattr("dynamo.vllm.worker_factory.elect_and_wake", elect_and_wake)
+        runtime = Mock()
+
+        with pytest.raises(RuntimeError) as excinfo:
+            await _make_factory()._maybe_wait_for_failover_lock(
+                self._handler_that_never_pauses(),
+                runtime,
+                _make_config(gms_shadow_mode=True),
+            )
+
+        message = str(excinfo.value)
+        assert "engine-2" in message
+        assert "did not finish pausing for standby" in message
+        # The stalled stage has to be named, or a cluster log cannot tell a
+        # weights-admission stall from any other unfinished startup step.
+        assert "sleep" in message
+        # The bound has to print as configured: rounding a sub-second bound to
+        # "0s" would print the value that disables the bound.
+        assert "0.2s" in message
+        # A shadow that never paused holds no lock and cannot take a
+        # promotion, so it must not report itself Ready.
+        runtime.set_health_status.assert_not_called()
+        elect_and_wake.assert_not_awaited()
+
+    @pytest.mark.parametrize("raw_timeout", ["nan", "inf", "-inf"])
+    @pytest.mark.timeout(30)
+    async def test_non_finite_bound_is_refused_instead_of_waiting_forever(
+        self, monkeypatch, raw_timeout
+    ):
+        """float() takes "nan" and "inf", and both defeat the bound: NaN
+        compares false against 0 and falls to the unbounded branch, and
+        wait_for(inf) never fires. Either one restores the indefinite standby
+        wait this bound exists to remove."""
+        monkeypatch.setenv("ENGINE_ID", "2")
+        monkeypatch.setenv("DYN_GMS_SHADOW_PAUSE_TIMEOUT_SECONDS", raw_timeout)
+        elect_and_wake = AsyncMock()
+        monkeypatch.setattr("dynamo.vllm.worker_factory.elect_and_wake", elect_and_wake)
+
+        with pytest.raises(RuntimeError, match="finite number of seconds"):
+            await _make_factory()._maybe_wait_for_failover_lock(
+                self._handler_that_never_pauses(),
+                Mock(),
+                _make_config(gms_shadow_mode=True),
+            )
+
+        elect_and_wake.assert_not_awaited()

components/src/dynamo/vllm/worker_factory.py

@@ -67,6 +67,11 @@
 # serving nor error propagation may hang on it.
 WORKER_GC_STOP_TIMEOUT_SECONDS = 30.0
 
+# Bound for a shadow engine's pause before standby; generous because it spans a
+# real engine sleep. Override with the env var below; "0" waits unbounded.
+SHADOW_PAUSE_TIMEOUT_SECONDS = 900.0
+ENV_SHADOW_PAUSE_TIMEOUT_SECONDS = "DYN_GMS_SHADOW_PAUSE_TIMEOUT_SECONDS"
+
 # (engine_client, vllm_config, default_sampling_params, cleanup_resource, component_gauges)
 # component_gauges is None on the embedding-worker path: pooling engines
 # have no KV cache / scheduler gauges, so setup_vllm_engine() skips the
@@ -1155,7 +1160,42 @@ async def _maybe_wait_for_failover_lock(
         if config.gms_shadow_mode is not True:
             return False
 
-        await handler._pause_controller.pause(1)
+        engine_id = os.environ.get("ENGINE_ID", "0")
+        raw_timeout = os.environ.get(ENV_SHADOW_PAUSE_TIMEOUT_SECONDS)
+        timeout_s = float(raw_timeout) if raw_timeout else SHADOW_PAUSE_TIMEOUT_SECONDS
+        # float() accepts "nan" and "inf". Both defeat the bound: NaN compares
+        # false against 0 and takes the unbounded branch, and wait_for(inf)
+        # never fires. Reject them rather than silently waiting forever.
+        if not math.isfinite(timeout_s):
+            raise RuntimeError(
+                f"{ENV_SHADOW_PAUSE_TIMEOUT_SECONDS}={raw_timeout!r} is not a "
+                "finite number of seconds; use a positive value to bound the "
+                'pause, or "0" to wait unbounded'
+            )
+        # %g, not a fixed number of decimals: rounding a sub-second bound down
+        # to "0" would print the one value that means "wait forever".
+        logger.info(
+            "[Shadow] engine-%s pausing before standby (bound %gs)",
+            engine_id,
+            timeout_s,
+        )
+        try:
+            if timeout_s > 0:
+                await asyncio.wait_for(
+                    handler._pause_controller.pause(1), timeout=timeout_s
+                )
+            else:
+                await handler._pause_controller.pause(1)
+        except asyncio.TimeoutError as exc:
+            # Health is deliberately not published here: a shadow that never
+            # paused holds no lock and cannot take a promotion.
+            raise RuntimeError(
+                f"[Shadow] engine-{engine_id} did not finish pausing for standby "
+                f"within {timeout_s:g}s; the engine pause (pause_generation, then "
+                "the engine sleep, which asks GMS to unmap and abort its weights "
+                "and kv_cache mappings) never completed, so this engine cannot "
+                "enter standby"
+            ) from exc
         lock = await elect_and_wake(
             handler._pause_controller,
             runtime,

lib/gpu_memory_service/README.md

@@ -610,4 +610,9 @@ To bound how long a restored engine waits for the published weights layout befor
 --model-loader-extra-config '{"gms_ro_connect_timeout_ms": 300000}'
 ```
 
-The default is `null`, which waits indefinitely. Set an integer value to fail after that many milliseconds.
+The value bounds both weight admissions an engine performs: the first one, during device init, and the remap after a shadow is promoted. An expiry during the remap is fatal and exits the worker.
+
+An explicit value always wins. When the key is absent, the default depends on the engine:
+
+- a primary engine (`ENGINE_ID=0`) and any engine outside shadow mode default to `null`, which waits indefinitely;
+- a failover shadow (`ENGINE_ID` other than `0`, under `--gms-shadow-mode`) defaults to `900000` ms. GMS prefers writers, so with two or more shadows a reader can be refused admission behind a queued writer, and an untimed refusal parks the worker with no log and no traceback.

lib/gpu_memory_service/client/session.py

@@ -44,6 +44,10 @@
 
 logger = logging.getLogger(__name__)
 
+# Ceiling on the wait for the server's UDS socket to be listening. Separate
+# from any lock-admission deadline; see _GMSClientSession.__init__.
+_CONNECT_TIMEOUT_MS = 30_000
+
 
 class _GMSClientSession:
     """Connected GMS client session with granted lock state."""
@@ -59,21 +63,28 @@ def __init__(
         # Two distinct timeouts apply to this session, and they're asymmetric on
         # purpose:
         #
-        #   socket connect (here) — bounded by 30 s when the caller passes None.
-        #     This is the wait for the GMS server's UDS socket to be listening.
-        #     The server is a native sidecar in the same pod (intra-pod GMS) or a
-        #     sister pod under the same Grove gang-schedule (inter-pod GMS), so
-        #     the actual ready window is sub-second to a few seconds in the
-        #     normal case. A 30 s ceiling produces a clean ConnectionError on a
+        #   socket connect (here) — never waits longer than
+        #     _CONNECT_TIMEOUT_MS, whatever the caller asked for. This is the
+        #     wait for the GMS server's UDS socket to be listening. The server
+        #     is a native sidecar in the same pod (intra-pod GMS) or a sister
+        #     pod under the same Grove gang-schedule (inter-pod GMS), so the
+        #     actual ready window is sub-second to a few seconds in the normal
+        #     case. The ceiling produces a clean ConnectionError on a
         #     missing-server misconfiguration instead of an indefinite hang.
+        #     A caller may ask for less; a lock-admission deadline of minutes
+        #     must not stretch the wait for a socket that may never appear.
         #
         #   handshake / lock acquisition (below) — passes the caller's timeout_ms
         #     through unchanged, including None which means "wait indefinitely."
         #     The handshake wait depends on what other clients are doing
         #     (engine loading weights, loader committing, etc.) and can be
         #     minutes for large models. We deliberately don't impose a
         #     server-availability ceiling on a workload-shaped wait.
-        self._transport.connect(timeout_ms=30_000 if timeout_ms is None else timeout_ms)
+        self._transport.connect(
+            timeout_ms=_CONNECT_TIMEOUT_MS
+            if timeout_ms is None
+            else min(timeout_ms, _CONNECT_TIMEOUT_MS)
+        )
         try:
             response = self._transport.handshake(lock_type, timeout_ms)
         except Exception:

lib/gpu_memory_service/integrations/vllm/utils.py

@@ -44,6 +44,11 @@ def configure_gms_worker_logging() -> None:
     gms_root.propagate = False
 
 
+# Default weight-admission deadline for a shadow engine, in ms. Generous: it
+# sits behind a real writer's load, and must stay under the 2h startup probe.
+SHADOW_RO_CONNECT_TIMEOUT_MS = 900_000
+
+
 def configure_gms_lock_mode(engine_args) -> None:
     """Set gms_read_only in model_loader_extra_config based on ENGINE_ID.
 
@@ -52,6 +57,13 @@ def configure_gms_lock_mode(engine_args) -> None:
     deadlock: if multiple engines tried to acquire RW locks across TP ranks
     simultaneously, they could block each other indefinitely.
 
+    A non-primary engine also gets a default gms_ro_connect_timeout_ms when the
+    operator has not chosen one. GMS prefers writers, so with two or more
+    shadows a reader can be refused admission behind a queued writer; an
+    untimed refusal parks the worker silently. A single-engine deployment is
+    left alone and keeps waiting indefinitely, and an explicit value always
+    wins.
+
     Raises if user-specified gms_read_only conflicts with ENGINE_ID.
     """
     engine_id = os.environ.get("ENGINE_ID", "0")
@@ -72,6 +84,14 @@ def configure_gms_lock_mode(engine_args) -> None:
                 f"but gms_read_only=False was explicitly set."
             )
         extra["gms_read_only"] = True
+        if extra.get("gms_ro_connect_timeout_ms") is None:
+            extra["gms_ro_connect_timeout_ms"] = SHADOW_RO_CONNECT_TIMEOUT_MS
+            logger.info(
+                "[GMS] ENGINE_ID=%s is a shadow; defaulting "
+                "gms_ro_connect_timeout_ms to %d ms",
+                engine_id,
+                SHADOW_RO_CONNECT_TIMEOUT_MS,
+            )
 
     engine_args.model_loader_extra_config = extra
 

lib/gpu_memory_service/integrations/vllm/worker.py

@@ -175,11 +175,14 @@ def init_device(self) -> None:
         )
         mode = get_gms_lock_mode(extra)
         self.gms_ro_connect_timeout_ms = get_gms_ro_connect_timeout_ms(extra)
+        # Must honour the same deadline as the wake_up remap below: without it,
+        # a shadow refused admission blocks with no log and no traceback.
         get_or_create_gms_client_memory_manager(
             get_socket_path(device, "weights"),
             device,
             mode=mode,
             tag="weights",
+            timeout_ms=self.gms_ro_connect_timeout_ms,
         )
         # Parent will set device again (harmless) and do memory checks
         super().init_device()

lib/gpu_memory_service/tests/test_runtime_flows.py

@@ -32,6 +32,7 @@
 if HAS_PYNVML:
     import pynvml
 
+import gpu_memory_service.client.session as _session_module
 import gpu_memory_service.common.vmm as _vmm_module
 from _fake_vmm import FakeVMM
 from gpu_memory_service.client.memory_manager import (
@@ -72,6 +73,9 @@
 _GPU_MEMORY_RECOVERY_TIMEOUT_SECONDS = 30.0
 _FAST_POLL_INTERVAL_SECONDS = 0.01
 _SLOW_POLL_INTERVAL_SECONDS = 0.1
+# Room for retry jitter above the 150 ms ceiling the connect test installs,
+# while staying far below any lock-admission deadline a caller might pass.
+_CONNECT_CEILING_ASSERTION_SECONDS = 5.0
 
 
 def _gpu_memory_free_bytes(device: int = 0) -> int:
@@ -494,6 +498,22 @@ def open_writer() -> None:
         waiting_writer.close()
 
 
+@pytest.mark.timeout(_SOCKET_TEST_TIMEOUT_SECONDS)
+def test_a_long_lock_deadline_does_not_stretch_the_connect_wait(tmp_path, monkeypatch):
+    """A caller's lock-admission deadline is a workload-shaped wait; the wait
+    for the server socket to exist is not. A missing or misconfigured socket
+    must still fail against the availability ceiling."""
+    monkeypatch.setattr(_session_module, "_CONNECT_TIMEOUT_MS", 150)
+    missing_socket = str(tmp_path / "absent.sock")
+
+    started = time.monotonic()
+    with pytest.raises(ConnectionError, match="GMS server not running"):
+        _GMSClientSession(missing_socket, RequestedLockType.RO, 600_000)
+    elapsed = time.monotonic() - started
+
+    assert elapsed < _CONNECT_CEILING_ASSERTION_SECONDS
+
+
 @pytest.mark.timeout(_SOCKET_TEST_TIMEOUT_SECONDS)
 def test_rw_or_ro_times_out_while_writer_waits_behind_reader(running_gms):
     server, socket_path = running_gms