diff --git a/python/simpler/remote_l3_session.py b/python/simpler/remote_l3_session.py index 66d42dbc2b..09317eb8a7 100644 --- a/python/simpler/remote_l3_session.py +++ b/python/simpler/remote_l3_session.py @@ -20,6 +20,7 @@ import hashlib import importlib import json +import math import os import signal import socket @@ -163,11 +164,24 @@ def _send_ready(fd: int, payload: dict[str, Any]) -> None: def _session_timeout_s(manifest: dict[str, Any]) -> float: timeout_s = float(manifest.get("session_timeout_s", 30.0)) - if timeout_s <= 0: - raise ValueError("manifest session_timeout_s must be positive") + if not (timeout_s > 0 and math.isfinite(timeout_s)): + raise ValueError("manifest session_timeout_s must be a positive finite number of seconds") return timeout_s +def _startup_remaining_s(manifest: dict[str, Any]) -> float: + # The parent's remaining slice of the single root startup budget. Absent only + # from a pre-P0.3 parent; fall back to the runtime command timeout then. The + # fallback is evaluated only when the key is absent, so a valid + # startup_remaining_s is not held hostage to an invalid session_timeout_s. + if "startup_remaining_s" not in manifest: + return _session_timeout_s(manifest) + remaining_s = float(manifest["startup_remaining_s"]) + if not (remaining_s > 0 and math.isfinite(remaining_s)): + raise ValueError("manifest startup_remaining_s must be a positive finite number of seconds") + return remaining_s + + def _health_loop(sock: socket.socket, stop: threading.Event, session_id: int, worker_id: int) -> None: conn: socket.socket | None = None sock.settimeout(0.2) @@ -911,6 +925,7 @@ def run_session(manifest: dict[str, Any], ready_fd: int) -> int: health_thread: threading.Thread | None = None try: session_timeout_s = _session_timeout_s(manifest) + startup_remaining_s = _startup_remaining_s(manifest) manifest_dispatch_registry = _install_manifest_dispatcher_registry(manifest) # Register the inner L3 callables before init() so they are frozen into # the eager startup snapshot and uploaded to the chip children before @@ -918,11 +933,14 @@ def run_session(manifest: dict[str, Any], ready_fd: int) -> int: # point: it forks and readies the whole inner L3->L2 tree, so the runner # reports ready (below) only once that subtree is up. manifest_inner_handles = _install_manifest_inner_registry(manifest, inner_worker) - # Bound the inner startup and mark it non-root so its chip/sub children - # inherit this runner's process group (see start_new_session in the - # daemon) rather than splitting into their own — the daemon reaps the - # whole L3->L2 subtree with one killpg on the runner. - inner_worker.init(_startup_deadline=time.monotonic() + session_timeout_s) + # Bound the inner startup by the parent's remaining startup budget + # (rebuilt against this host's own monotonic clock — the parent's + # absolute deadline is not comparable across machines), not a fresh full + # session_timeout_s. Mark it non-root so its chip/sub children inherit + # this runner's process group (see start_new_session in the daemon) + # rather than splitting into their own — the daemon reaps the whole + # L3->L2 subtree with one killpg on the runner. + inner_worker.init(_startup_deadline=time.monotonic() + startup_remaining_s) listen_host = str(manifest.get("listen_host", "127.0.0.1")) command_sock = _bind_listener(listen_host) diff --git a/python/simpler/remote_l3_worker.py b/python/simpler/remote_l3_worker.py index a51e939704..79d768bcf9 100644 --- a/python/simpler/remote_l3_worker.py +++ b/python/simpler/remote_l3_worker.py @@ -13,6 +13,7 @@ import argparse import contextlib import json +import math import os import select import signal @@ -67,11 +68,25 @@ def _validate_manifest(manifest: dict[str, Any]) -> None: def _session_timeout_s(manifest: dict[str, Any]) -> float: timeout_s = float(manifest.get("session_timeout_s", 30.0)) - if timeout_s <= 0: - raise ValueError("manifest session_timeout_s must be positive") + if not (timeout_s > 0 and math.isfinite(timeout_s)): + raise ValueError("manifest session_timeout_s must be a positive finite number of seconds") return timeout_s +def _startup_remaining_s(manifest: dict[str, Any]) -> float: + # The parent's remaining slice of the single root startup budget bounds how + # long the runner may take to bring up its subtree. Absent only from a + # pre-P0.3 parent; fall back to the runtime command timeout then. The + # fallback is evaluated only when the key is absent, so a valid + # startup_remaining_s is not held hostage to an invalid session_timeout_s. + if "startup_remaining_s" not in manifest: + return _session_timeout_s(manifest) + remaining_s = float(manifest["startup_remaining_s"]) + if not (remaining_s > 0 and math.isfinite(remaining_s)): + raise ValueError("manifest startup_remaining_s must be a positive finite number of seconds") + return remaining_s + + def _read_runner_ready(fd: int, timeout_s: float) -> dict[str, Any]: chunks = bytearray() deadline = time.monotonic() + timeout_s @@ -128,7 +143,9 @@ def _reap_session_runner(proc: subprocess.Popen[Any]) -> None: def _start_session(manifest: dict[str, Any]) -> dict[str, Any]: _validate_manifest(manifest) - timeout_s = _session_timeout_s(manifest) + # The runner must publish ready within the parent's remaining startup budget, + # not a fresh full command timeout. + timeout_s = _startup_remaining_s(manifest) ready_r, ready_w = os.pipe() manifest_path = "" proc: subprocess.Popen[Any] | None = None diff --git a/python/simpler/worker.py b/python/simpler/worker.py index b7f296618e..fcca0862bb 100644 --- a/python/simpler/worker.py +++ b/python/simpler/worker.py @@ -2133,8 +2133,8 @@ def _is_wildcard_session_host(host: str) -> bool: def _remote_session_timeout_s(self) -> float: timeout_s = float(self._config.get("remote_session_timeout_s", 30.0)) - if timeout_s <= 0: - raise ValueError("Worker remote_session_timeout_s must be positive") + if not (timeout_s > 0 and math.isfinite(timeout_s)): + raise ValueError("Worker remote_session_timeout_s must be a positive finite number of seconds") return timeout_s @staticmethod @@ -2182,7 +2182,9 @@ def _remote_dispatcher_entries_for_worker(self, worker_id: int) -> list[dict[str ) return entries - def _build_remote_manifest(self, *, spec: RemoteWorkerSpec, worker_id: int, session_id: int) -> dict[str, Any]: + def _build_remote_manifest( + self, *, spec: RemoteWorkerSpec, worker_id: int, session_id: int, startup_remaining_s: float + ) -> dict[str, Any]: daemon_host, _daemon_port = self._parse_remote_endpoint(spec.endpoint) listen_host = spec.session_listen_host or ("127.0.0.1" if daemon_host == "localhost" else daemon_host) if self._is_wildcard_session_host(listen_host) and not spec.allow_wildcard_session_bind: @@ -2198,7 +2200,11 @@ def _build_remote_manifest(self, *, spec: RemoteWorkerSpec, worker_id: int, sess "num_sub_workers": int(spec.num_sub_workers), "heap_ring_size": self._config.get("remote_heap_ring_size", None), "transport": spec.transport, + # session_timeout_s bounds the runtime command socket; startup_remaining_s + # bounds this session's slice of the single root startup budget. They are + # distinct: the remote must not spend runtime-command time as startup time. "session_timeout_s": self._remote_session_timeout_s(), + "startup_remaining_s": float(startup_remaining_s), "listen_host": listen_host, "connect_host": daemon_host, "remote_task_dispatcher": self._remote_dispatcher_entries_for_worker(worker_id), @@ -2207,10 +2213,12 @@ def _build_remote_manifest(self, *, spec: RemoteWorkerSpec, worker_id: int, sess } def _open_remote_session( - self, *, spec: RemoteWorkerSpec, worker_id: int, session_id: int, timeout_s: float + self, *, spec: RemoteWorkerSpec, worker_id: int, session_id: int, timeout_s: float, startup_remaining_s: float ) -> _RemoteSession: daemon_host, daemon_port = self._parse_remote_endpoint(spec.endpoint) - manifest = self._build_remote_manifest(spec=spec, worker_id=worker_id, session_id=session_id) + manifest = self._build_remote_manifest( + spec=spec, worker_id=worker_id, session_id=session_id, startup_remaining_s=startup_remaining_s + ) with socket.create_connection((daemon_host, daemon_port), timeout=timeout_s) as sock: sock.settimeout(timeout_s) self._send_remote_daemon_json(sock, manifest) @@ -3874,28 +3882,53 @@ def _init_hierarchical(self) -> None: else: self._worker = _Worker(self.level, int(heap_ring_size)) - # Open remote L3 sessions (the remote runner brings up its own L3->L2 - # tree and answers the daemon handshake). Opening only exchanges a - # transient socket message — it starts no local thread. Registering the - # socket into the C++ Worker (which spawns the RemoteL3Endpoint health - # thread) is deferred to _start_hierarchical, after the last local fork, - # to preserve the fork-before-thread invariant. - opened_remote_sessions: list[_RemoteSession] = [] - try: - for worker_id, spec in zip(self._remote_worker_ids, self._remote_worker_specs, strict=True): - session_id = uuid.uuid4().int & ((1 << 63) - 1) - if session_id == 0: - session_id = 1 - timeout_s = self._remote_session_timeout_s() - session = self._open_remote_session( - spec=spec, worker_id=worker_id, session_id=session_id, timeout_s=timeout_s - ) - opened_remote_sessions.append(session) - self._remote_sessions.append(session) - opened_remote_sessions.pop() - except BaseException: - self._close_remote_sessions(opened_remote_sessions) - raise + def _activate_remote_sessions(self, deadline: float) -> None: + """Open and register every remote L3 session within the shared startup budget. + + Called only from _start_hierarchical, after this process's last local + fork, so opening a session (which starts the remote subtree) and + registering its endpoint (which spawns the health thread) both stay + behind every local fork. All remotes draw from the single root startup + ``deadline``: each computes the remaining budget at the moment it opens, + propagates it as the manifest's ``startup_remaining_s`` so the remote + bounds its own subtree by this process's remaining time (measured on the + remote's own monotonic clock) instead of a fresh full timeout. Any + failure propagates to init()'s single rollback, which closes every + session recorded in ``self._remote_sessions``. + """ + session_timeout = self._remote_session_timeout_s() + for worker_id, spec in zip(self._remote_worker_ids, self._remote_worker_specs, strict=True): + remaining = deadline - time.monotonic() + if remaining <= 0: + raise RuntimeError("remote L3 session activation: startup deadline exceeded") + session_id = uuid.uuid4().int & ((1 << 63) - 1) + if session_id == 0: + session_id = 1 + # The handshake blocks until the remote subtree is READY, so the + # socket timeout must cover the startup budget granted below — not + # the (shorter) runtime command timeout. + session = self._open_remote_session( + spec=spec, + worker_id=worker_id, + session_id=session_id, + timeout_s=remaining, + startup_remaining_s=remaining, + ) + self._remote_sessions.append(session) + remaining = deadline - time.monotonic() + if remaining <= 0: + raise RuntimeError("remote L3 endpoint attach: startup deadline exceeded") + assert self._worker is not None + self._worker.add_remote_l3_socket( + session.worker_id, + session.session_id, + spec.transport, + session.command_host, + session.command_port, + session.health_host, + session.health_port, + min(session_timeout, remaining), + ) def _start_hierarchical(self) -> None: # noqa: PLR0912 -- three parallel fork loops (sub/chip/next) + bootstrap wait + scheduler register/init; branches track the fork order documented in the body """Fork every local child, await the subtree, register endpoints, start the scheduler. @@ -4065,25 +4098,11 @@ def _setup(inner=inner_worker): # failure, exit, or hang aborts startup here. self._await_children_ready(self._next_level_shms, self._next_level_pids, "next_level", deadline) - # Last local fork is done. Now — and only now — register remote L3 - # endpoints, which spawns the RemoteL3Endpoint health thread; doing it - # here keeps every local fork ahead of every C++/Python thread. Each - # remote attach consumes the shared startup budget. - for session, spec in zip(self._remote_sessions, self._remote_worker_specs, strict=True): - remaining = deadline - time.monotonic() - if remaining <= 0: - raise RuntimeError("remote L3 endpoint attach: startup deadline exceeded") - assert self._worker is not None - self._worker.add_remote_l3_socket( - session.worker_id, - session.session_id, - spec.transport, - session.command_host, - session.command_port, - session.health_host, - session.health_port, - min(self._remote_session_timeout_s(), remaining), - ) + # Last local fork is done. Now — and only now — open and register remote + # L3 sessions: opening starts the remote subtree and registering spawns + # the RemoteL3Endpoint health thread, so both must follow every local + # fork. Each remote consumes this process's remaining startup budget. + self._activate_remote_sessions(deadline) # _Worker was constructed in _init_hierarchical (pre-fork) so children # inherit the HeapRing MAP_SHARED mmap. Register PROCESS-mode workers via diff --git a/tests/ut/py/test_callable_identity.py b/tests/ut/py/test_callable_identity.py index 0dc45e2ba3..bd4c9ebecc 100644 --- a/tests/ut/py/test_callable_identity.py +++ b/tests/ut/py/test_callable_identity.py @@ -504,7 +504,7 @@ def close(self): def fake_worker_ctor(*args): return fake_c_worker - def fake_open_remote_session(self, *, spec, worker_id, session_id, timeout_s): + def fake_open_remote_session(self, *, spec, worker_id, session_id, timeout_s, startup_remaining_s): opened_worker_ids.append(worker_id) return worker_mod._RemoteSession( # noqa: SLF001 worker_id=worker_id, @@ -545,6 +545,7 @@ def test_remote_session_manifest_uses_endpoint_host_as_default_bind(): spec=RemoteWorkerSpec(endpoint="127.0.0.1:19073", platform="a2a3sim"), worker_id=0, session_id=1, + startup_remaining_s=30.0, ) assert loopback["listen_host"] == "127.0.0.1" assert loopback["connect_host"] == "127.0.0.1" @@ -553,6 +554,7 @@ def test_remote_session_manifest_uses_endpoint_host_as_default_bind(): spec=RemoteWorkerSpec(endpoint="10.0.0.8:19073", platform="a2a3sim"), worker_id=0, session_id=1, + startup_remaining_s=30.0, ) assert remote["listen_host"] == "10.0.0.8" assert remote["connect_host"] == "10.0.0.8" @@ -569,7 +571,7 @@ def test_remote_session_manifest_requires_wildcard_bind_opt_in(): session_listen_host="0.0.0.0", ) with pytest.raises(ValueError, match="wildcard session bind"): - worker._build_remote_manifest(spec=spec, worker_id=0, session_id=1) + worker._build_remote_manifest(spec=spec, worker_id=0, session_id=1, startup_remaining_s=30.0) opted_in = worker._build_remote_manifest( spec=RemoteWorkerSpec( @@ -580,6 +582,7 @@ def test_remote_session_manifest_requires_wildcard_bind_opt_in(): ), worker_id=0, session_id=1, + startup_remaining_s=30.0, ) assert opted_in["listen_host"] == "0.0.0.0" assert opted_in["connect_host"] == "10.0.0.8" @@ -1613,7 +1616,7 @@ def fake_worker_ctor(*args): calls = 0 - def fake_open_remote_session(self, *, spec, worker_id, session_id, timeout_s): + def fake_open_remote_session(self, *, spec, worker_id, session_id, timeout_s, startup_remaining_s): nonlocal calls calls += 1 if calls == 2: @@ -1670,7 +1673,7 @@ def close(self): def fake_worker_ctor(*args): return fake_c_worker - def fake_open_remote_session(self, *, spec, worker_id, session_id, timeout_s): + def fake_open_remote_session(self, *, spec, worker_id, session_id, timeout_s, startup_remaining_s): return opened_session def fake_close_remote_session(self, session): diff --git a/tests/ut/py/test_remote_l3_lifecycle.py b/tests/ut/py/test_remote_l3_lifecycle.py index 12cd4f10ca..1c272f7a55 100644 --- a/tests/ut/py/test_remote_l3_lifecycle.py +++ b/tests/ut/py/test_remote_l3_lifecycle.py @@ -10,6 +10,7 @@ import os import socket import threading +import time from typing import cast import pytest @@ -175,6 +176,63 @@ def close(self): os.close(ready_r) +def test_run_session_bounds_subtree_by_startup_remaining_not_session_timeout(monkeypatch): + """The inner subtree deadline comes from the parent's startup_remaining_s + (its slice of the single root startup budget), not the runtime command + session_timeout_s — the two are deliberately different here.""" + + captured = {} + + class FakeWorker: + def __init__(self, *args, **kwargs): + self.closed = False + + def init(self, *args, _startup_deadline=None, **kwargs): + captured["deadline"] = _startup_deadline + captured["at"] = time.monotonic() + + def close(self): + self.closed = True + + class FakeCommandSock: + def getsockname(self): + return ("127.0.0.1", 12345) + + def settimeout(self, timeout): + pass + + def accept(self): + raise socket.timeout("stop after init") + + def close(self): + pass + + class FakeHealthSock: + def getsockname(self): + return ("127.0.0.1", 12346) + + def close(self): + pass + + sockets = [FakeCommandSock(), FakeHealthSock()] + ready_r, ready_w = os.pipe() + + monkeypatch.setattr(remote_l3_session, "Worker", FakeWorker) + monkeypatch.setattr(remote_l3_session, "_install_manifest_dispatcher_registry", lambda manifest: {}) + monkeypatch.setattr(remote_l3_session, "_install_manifest_inner_registry", lambda manifest, worker: {}) + monkeypatch.setattr(remote_l3_session, "_bind_listener", lambda host: sockets.pop(0)) + monkeypatch.setattr(remote_l3_session, "_health_loop", lambda *args: None) + + try: + remote_l3_session.run_session(_manifest(session_timeout_s=0.01, startup_remaining_s=50.0), ready_w) + finally: + os.close(ready_r) + + budget = captured["deadline"] - captured["at"] + # ~50s startup budget, not the 0.01s runtime command timeout. + assert 40.0 < budget <= 50.0 + + def test_health_loop_closes_active_connection_on_stop(): stop = threading.Event() diff --git a/tests/ut/py/test_worker/test_remote_startup_budget.py b/tests/ut/py/test_worker/test_remote_startup_budget.py new file mode 100644 index 0000000000..92230b2cc6 --- /dev/null +++ b/tests/ut/py/test_worker/test_remote_startup_budget.py @@ -0,0 +1,208 @@ +# Copyright (c) PyPTO Contributors. +# This program is free software, you can redistribute it and/or modify it under the terms and conditions of +# CANN Open Software License Agreement Version 2.0 (the "License"). +# Please refer to the License for details. You may not use this file except in compliance with the License. +# THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND, EITHER EXPRESS OR IMPLIED, +# INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT, MERCHANTABILITY, OR FITNESS FOR A PARTICULAR PURPOSE. +# See LICENSE in the root of the software repository for the full text of the License. +# ----------------------------------------------------------------------------------------------------------- +"""P0.3 — simulation Remote L3 startup budget / activation correctness. + +Device-free tests that pin the single-root-deadline contract: + +- adding Remote L3 workers must not multiply the startup budget: every remote + draws from one root deadline via a decreasing remaining slice, propagated as + the manifest ``startup_remaining_s`` (distinct from the runtime + ``session_timeout_s``); +- sessions are opened and activated only in ``_activate_remote_sessions`` (after + the last local fork), never pre-fork in ``_init_hierarchical``; +- non-positive / non-finite startup and session timeouts fail before any + resource is created, on both the parent and the remote (untrusted-wire) side. +""" + +from __future__ import annotations + +import time +from unittest.mock import MagicMock + +import pytest +import simpler.remote_l3_session as session_mod +import simpler.remote_l3_worker as daemon_mod +from simpler.worker import RemoteWorkerSpec, Worker + + +def _spec(port: int = 19073) -> RemoteWorkerSpec: + return RemoteWorkerSpec(endpoint=f"127.0.0.1:{port}", platform="a2a3sim") + + +def _l4_with_remotes(n: int, **config) -> Worker: + w = Worker(level=4, num_sub_workers=0, **config) + for i in range(n): + w.add_remote_worker(_spec(19073 + i)) + return w + + +class TestManifestBudgetField: + def test_manifest_carries_startup_remaining_distinct_from_session_timeout(self): + w = Worker(level=4, num_sub_workers=0, remote_session_timeout_s=30.0) + try: + manifest = w._build_remote_manifest(spec=_spec(), worker_id=0, session_id=1, startup_remaining_s=12.5) + assert manifest["startup_remaining_s"] == 12.5 + assert manifest["session_timeout_s"] == 30.0 + # The startup budget must not be fixed to the runtime command timeout. + assert manifest["startup_remaining_s"] != manifest["session_timeout_s"] + finally: + w.close() + + +class TestParentTimeoutValidation: + @pytest.mark.parametrize("bad", [0.0, -1.0, float("inf"), float("nan")]) + def test_remote_session_timeout_rejected(self, bad): + w = Worker(level=4, num_sub_workers=0, remote_session_timeout_s=bad) + try: + with pytest.raises(ValueError, match="remote_session_timeout_s"): + w._remote_session_timeout_s() + finally: + w.close() + + def test_positive_finite_remote_session_timeout_accepted(self): + w = Worker(level=4, num_sub_workers=0, remote_session_timeout_s=17.0) + try: + assert w._remote_session_timeout_s() == 17.0 + finally: + w.close() + + +class TestRemoteSideValidation: + """Both remote entry points validate the wire budget before creating resources.""" + + @pytest.mark.parametrize("mod", [session_mod, daemon_mod]) + @pytest.mark.parametrize("bad", [0.0, -1.0, float("inf"), float("nan")]) + def test_startup_remaining_rejected(self, mod, bad): + with pytest.raises(ValueError, match="startup_remaining_s"): + mod._startup_remaining_s({"startup_remaining_s": bad, "session_timeout_s": 30.0}) + + @pytest.mark.parametrize("mod", [session_mod, daemon_mod]) + @pytest.mark.parametrize("bad", [0.0, -1.0, float("inf"), float("nan")]) + def test_session_timeout_rejected(self, mod, bad): + with pytest.raises(ValueError, match="session_timeout_s"): + mod._session_timeout_s({"session_timeout_s": bad}) + + @pytest.mark.parametrize("mod", [session_mod, daemon_mod]) + def test_startup_remaining_falls_back_to_session_timeout(self, mod): + # A pre-P0.3 parent omits startup_remaining_s; the remote must still bound + # itself by the (positive-finite) session timeout rather than crash. + assert mod._startup_remaining_s({"session_timeout_s": 25.0}) == 25.0 + + @pytest.mark.parametrize("mod", [session_mod, daemon_mod]) + def test_startup_remaining_used_when_present(self, mod): + assert mod._startup_remaining_s({"startup_remaining_s": 8.0, "session_timeout_s": 30.0}) == 8.0 + + +class TestActivationAfterFork: + def test_init_hierarchical_does_not_open_sessions(self, monkeypatch): + """Opening a session (which starts the remote subtree) is deferred out of + the pre-fork _init_hierarchical into post-fork _activate_remote_sessions.""" + + def _must_not_open(self, **_kwargs): + raise AssertionError("remote session opened before the last local fork") + + monkeypatch.setattr(Worker, "_open_remote_session", _must_not_open) + + w = _l4_with_remotes(2) + try: + w._init_hierarchical() + assert w._remote_sessions == [] + finally: + if w._worker is not None: + w._worker.close() + w._worker = None + w.close() + + +class TestSingleRootBudget: + def _drive_activation(self, monkeypatch, *, n_remotes, root_budget_s, per_open_delay_s=0.02): + granted: list[float] = [] + socket_timeouts: list[float] = [] + + def fake_open(self, *, spec, worker_id, session_id, timeout_s, startup_remaining_s): + granted.append(startup_remaining_s) + socket_timeouts.append(timeout_s) + time.sleep(per_open_delay_s) + return _FakeSession(worker_id, session_id) + + monkeypatch.setattr(Worker, "_open_remote_session", fake_open) + + w = _l4_with_remotes(n_remotes) + w._worker = MagicMock() + deadline = time.monotonic() + root_budget_s + try: + w._activate_remote_sessions(deadline) + finally: + w._worker = None + w.close() + return granted, socket_timeouts, w + + def test_budget_not_multiplied_by_remote_count(self, monkeypatch): + granted, socket_timeouts, _ = self._drive_activation(monkeypatch, n_remotes=3, root_budget_s=5.0) + assert len(granted) == 3 + # Every remote's granted startup budget fits inside the single root budget. + for g in granted: + assert 0 < g <= 5.0 + # And the budget shrinks per remote (shared deadline), never a fresh full + # timeout each — the multiplication bug. + assert granted[0] > granted[1] > granted[2] + # Socket handshake timeout tracks the same remaining budget. + for s, g in zip(socket_timeouts, granted, strict=True): + assert s == g + + def test_attach_called_once_per_remote(self, monkeypatch): + granted: list[float] = [] + + def fake_open(self, *, spec, worker_id, session_id, timeout_s, startup_remaining_s): + granted.append(startup_remaining_s) + return _FakeSession(worker_id, session_id) + + monkeypatch.setattr(Worker, "_open_remote_session", fake_open) + w = _l4_with_remotes(2) + mock_worker = MagicMock() + w._worker = mock_worker + try: + w._activate_remote_sessions(time.monotonic() + 5.0) + assert mock_worker.add_remote_l3_socket.call_count == 2 + assert len(w._remote_sessions) == 2 + finally: + w._worker = None + w.close() + + def test_expired_deadline_fails_fast_without_opening(self, monkeypatch): + opened = [] + + def fake_open(self, *, spec, worker_id, session_id, timeout_s, startup_remaining_s): + opened.append(worker_id) + return _FakeSession(worker_id, session_id) + + monkeypatch.setattr(Worker, "_open_remote_session", fake_open) + w = _l4_with_remotes(1) + w._worker = MagicMock() + try: + with pytest.raises(RuntimeError, match="startup deadline exceeded"): + w._activate_remote_sessions(time.monotonic() - 1.0) + assert opened == [] + assert w._remote_sessions == [] + finally: + w._worker = None + w.close() + + +class _FakeSession: + """Minimal stand-in for _RemoteSession that add_remote_l3_socket can read.""" + + def __init__(self, worker_id: int, session_id: int): + self.worker_id = worker_id + self.session_id = session_id + self.command_host = "127.0.0.1" + self.command_port = 1 + self.health_host = "127.0.0.1" + self.health_port = 2 + self.pid = 0