diff --git a/architecture/gateway.md b/architecture/gateway.md index 1e3b761d2d..3c27fe1df9 100644 --- a/architecture/gateway.md +++ b/architecture/gateway.md @@ -66,6 +66,8 @@ it does not establish that a mutation is safe to repeat. SDKs retain the origina transport status, metadata, and unknown details alongside decoded fields. SDK deletion waits recognize missing-resource status through typed error wrappers without suppressing other failures. +Python SDK lifecycle waits use monotonic deadlines and bound each poll RPC and +sleep by the remaining wait budget, while retaining shorter per-call timeouts. Ordinary user-callable unary mutations explicitly opt into durable request admission when the client supplies a UUID. Typed adapters diff --git a/docs/sdk/python.mdx b/docs/sdk/python.mdx index afd1a9d399..a2f956988c 100644 --- a/docs/sdk/python.mdx +++ b/docs/sdk/python.mdx @@ -107,6 +107,11 @@ workspace for repeated exec, stop, start, and delete operations. List methods return lazy `Pager` instances; use `list_all()` only when you want to fetch the complete collection. +`wait_ready()`, `wait_stopped()`, and `wait_deleted()` apply `timeout_seconds` +to the whole wait, including lookup RPCs and polling delays. Each lookup also +respects a shorter client `timeout`. Changes to the system clock do not extend +or shorten the wait. + ## Next Steps - Review [Manage Sandboxes](/how-it-works/sandboxes/overview) for labels, templates, services, and Python SDK examples. diff --git a/python/openshell/sandbox.py b/python/openshell/sandbox.py index 20d4734443..6bdf8217fd 100644 --- a/python/openshell/sandbox.py +++ b/python/openshell/sandbox.py @@ -913,12 +913,15 @@ def sandbox_templates(self) -> SandboxTemplateClient: return SandboxTemplateClient(self._channel, timeout=self._timeout) def get(self, name: str, *, workspace: str) -> SandboxRef: + return self._get(name, workspace=workspace, timeout=self._timeout) + + def _get(self, name: str, *, workspace: str, timeout: float) -> SandboxRef: response = self._stub.GetSandbox( openshell_pb2.GetSandboxRequest( workspace_scope=_workspace_scope(workspace), name=name, ), - timeout=self._timeout, + timeout=timeout, ) return _sandbox_ref(response.sandbox) @@ -1075,20 +1078,28 @@ def wait_deleted( timeout_seconds: float = 60.0, expected_sandbox_id: str | None = None, ) -> None: - deadline = time.time() + timeout_seconds - while time.time() < deadline: + deadline = time.monotonic() + timeout_seconds + while (remaining := deadline - time.monotonic()) > 0: try: - current = self.get(name, workspace=workspace) + current = self._get( + name, workspace=workspace, timeout=min(self._timeout, remaining) + ) if ( expected_sandbox_id is not None and current.id != expected_sandbox_id ): return except grpc.RpcError as exc: - if getattr(exc, "code", lambda: None)() == grpc.StatusCode.NOT_FOUND: + code = getattr(exc, "code", lambda: None)() + if code == grpc.StatusCode.NOT_FOUND: return + if ( + code == grpc.StatusCode.DEADLINE_EXCEEDED + and time.monotonic() >= deadline + ): + break raise - time.sleep(1) + time.sleep(min(1, max(0, deadline - time.monotonic()))) raise SandboxError(f"sandbox {name} was not deleted within timeout") def wait_ready( @@ -1122,9 +1133,20 @@ def _wait_for_phase( target_name: str, timeout_seconds: float, ) -> SandboxRef: - deadline = time.time() + timeout_seconds - while time.time() < deadline: - sandbox = self.get(name, workspace=workspace) + deadline = time.monotonic() + timeout_seconds + while (remaining := deadline - time.monotonic()) > 0: + try: + sandbox = self._get( + name, workspace=workspace, timeout=min(self._timeout, remaining) + ) + except grpc.RpcError as exc: + if ( + getattr(exc, "code", lambda: None)() + == grpc.StatusCode.DEADLINE_EXCEEDED + and time.monotonic() >= deadline + ): + break + raise if sandbox.status.phase == target_phase: return sandbox if ( @@ -1139,7 +1161,7 @@ def _wait_for_phase( raise SandboxError(f"sandbox {name} stopped before becoming ready") if sandbox.status.phase == openshell_pb2.SANDBOX_PHASE_ERROR: raise SandboxError(f"sandbox {name} entered error phase") - time.sleep(1) + time.sleep(min(1, max(0, deadline - time.monotonic()))) raise SandboxError(f"sandbox {name} was not {target_name} within timeout") def exec_stream( diff --git a/python/openshell/sandbox_wait_test.py b/python/openshell/sandbox_wait_test.py new file mode 100644 index 0000000000..502d69ce35 --- /dev/null +++ b/python/openshell/sandbox_wait_test.py @@ -0,0 +1,164 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +from __future__ import annotations + +from concurrent import futures +from types import SimpleNamespace + +import grpc +import pytest + +import openshell.sandbox as sandbox_module +from openshell._proto import openshell_pb2 +from openshell.sandbox import SandboxClient, SandboxError + + +class Clock: + def __init__(self) -> None: + self.elapsed = 0.0 + self.wall_offset = 0.0 + + def monotonic(self) -> float: + return self.elapsed + + def time(self) -> float: + return self.elapsed + self.wall_offset + + def sleep(self, seconds: float) -> None: + assert seconds >= 0 + self.elapsed += seconds + + +@pytest.mark.parametrize("method", ["wait_ready", "wait_stopped", "wait_deleted"]) +@pytest.mark.parametrize("timeout_seconds", [0, -1]) +def test_expired_wait_does_not_poll(monkeypatch, method, timeout_seconds): + def get(_request, timeout): + pytest.fail(f"unexpected lookup with timeout {timeout}") + + with SandboxClient("localhost:1") as client: + monkeypatch.setattr(client, "_stub", SimpleNamespace(GetSandbox=get)) + with pytest.raises(SandboxError, match="within timeout"): + getattr(client, method)( + "job", workspace="team", timeout_seconds=timeout_seconds + ) + + +@pytest.mark.parametrize("method", ["wait_ready", "wait_stopped", "wait_deleted"]) +@pytest.mark.parametrize( + "code", [grpc.StatusCode.DEADLINE_EXCEEDED, grpc.StatusCode.PERMISSION_DENIED] +) +def test_wait_preserves_lookup_errors_before_deadline(monkeypatch, method, code): + class LookupError(grpc.RpcError): + def code(self): + return code + + error = LookupError() + + def get(_request, timeout): + assert 0 < timeout <= 0.1 + raise error + + with SandboxClient("localhost:1", timeout=0.1) as client: + monkeypatch.setattr(client, "_stub", SimpleNamespace(GetSandbox=get)) + with pytest.raises(LookupError) as caught: + getattr(client, method)("job", workspace="team", timeout_seconds=10) + assert caught.value is error + + +@pytest.mark.parametrize("method", ["wait_ready", "wait_stopped", "wait_deleted"]) +@pytest.mark.parametrize("client_timeout", [0.1, 30.0]) +def test_wait_bounds_rpc_and_sleep(monkeypatch, method, client_timeout): + clock = Clock() + monkeypatch.setattr(sandbox_module, "time", clock) + timeouts = [] + + def get(request, timeout): + assert request.name == "job" + assert request.workspace_scope.workspace == "team" + timeouts.append(timeout) + clock.elapsed += 0.05 + return openshell_pb2.SandboxResponse() + + with SandboxClient("localhost:1", timeout=client_timeout) as client: + monkeypatch.setattr(client, "_stub", SimpleNamespace(GetSandbox=get)) + with pytest.raises(SandboxError, match="within timeout"): + getattr(client, method)("job", workspace="team", timeout_seconds=1.25) + + assert timeouts == pytest.approx( + [min(client_timeout, 1.25), min(client_timeout, 0.2)] + ) + assert clock.elapsed == pytest.approx(1.25) + + +@pytest.mark.parametrize( + ("method", "phase"), + [ + ("wait_ready", openshell_pb2.SANDBOX_PHASE_READY), + ("wait_stopped", openshell_pb2.SANDBOX_PHASE_STOPPED), + ("wait_deleted", openshell_pb2.SANDBOX_PHASE_READY), + ], +) +def test_wait_ignores_wall_clock_changes(monkeypatch, method, phase): + clock = Clock() + monkeypatch.setattr(sandbox_module, "time", clock) + calls = 0 + + def get(_request, timeout): + nonlocal calls + assert timeout > 0 + calls += 1 + clock.wall_offset += 1000 + response = openshell_pb2.SandboxResponse() + response.sandbox.metadata.id = "original" if calls == 1 else "replacement" + if calls == 2: + response.sandbox.status.phase = phase + return response + + with SandboxClient("localhost:1") as client: + monkeypatch.setattr(client, "_stub", SimpleNamespace(GetSandbox=get)) + options = ( + {"expected_sandbox_id": "original"} if method == "wait_deleted" else {} + ) + getattr(client, method)("job", workspace="team", timeout_seconds=2, **options) + + assert calls == 2 + assert clock.elapsed == 1 + + +@pytest.mark.parametrize("method", ["wait_ready", "wait_stopped", "wait_deleted"]) +def test_wait_cancels_stalled_lookup_at_deadline(method): + import threading + import time + + def get(_request, context): + cancelled = threading.Event() + context.add_callback(cancelled.set) + cancelled.wait(5) + return openshell_pb2.SandboxResponse() + + server = grpc.server(futures.ThreadPoolExecutor(max_workers=1)) + server.add_generic_rpc_handlers( + ( + grpc.method_handlers_generic_handler( + "openshell.v1.OpenShell", + { + "GetSandbox": grpc.unary_unary_rpc_method_handler( + get, + request_deserializer=openshell_pb2.GetSandboxRequest.FromString, + response_serializer=openshell_pb2.SandboxResponse.SerializeToString, + ), + }, + ), + ) + ) + port = server.add_insecure_port("127.0.0.1:0") + server.start() + try: + with SandboxClient(f"127.0.0.1:{port}", timeout=2) as client: + started = time.monotonic() + with pytest.raises(SandboxError, match="within timeout"): + getattr(client, method)("job", workspace="team", timeout_seconds=0.1) + assert time.monotonic() - started < 1 + finally: + server.stop(0).wait()