diff --git a/docs/sdk-python.md b/docs/sdk-python.md index af53f162..9147ed47 100644 --- a/docs/sdk-python.md +++ b/docs/sdk-python.md @@ -71,7 +71,9 @@ client = Client("10.0.0.5:7777", api_token="...", timeout=120.0) `api_token` is the node token — a root `api_token` (full access) or a tenant token (resource-creating verbs only; operator surfaces answer it 403). On a cluster every node shares the root token and the same tenants -set. `timeout` bounds every socket operation. +set. `timeout` bounds every control-plane request and the data-plane dial +and upgrade; a guest stream then lives until the guest ends it, as in the +Go SDK. **Clusters need nothing extra**: dial any node. On a warm miss the entry node answers with a redirect and `new` follows it transparently; the @@ -450,7 +452,6 @@ zero. `bad_request` / `not_found` / `unimplemented` / `internal` - `ExitError(code, stderr, stdout)` — non-zero exit from `exec` - `ProtocolError` — broken stream (EOF, oversized or undecodable frame) -- `StreamTimeout` — a socket operation hit the client `timeout`; retry the call ```python try: diff --git a/sdk/python/cocoonsandbox/__init__.py b/sdk/python/cocoonsandbox/__init__.py index af31cf0a..46513e08 100644 --- a/sdk/python/cocoonsandbox/__init__.py +++ b/sdk/python/cocoonsandbox/__init__.py @@ -11,7 +11,7 @@ from .checkpoint import Checkpoint from .client import Client -from .errors import APIError, ExitError, ProtocolError, SandboxError, SilkdError, StreamTimeout +from .errors import APIError, ExitError, ProtocolError, SandboxError, SilkdError from .sandbox import Lsp, PortConn, Pty, Sandbox, Session, Watcher from .template import Template @@ -28,7 +28,6 @@ "SandboxError", "Session", "SilkdError", - "StreamTimeout", "Template", "Watcher", ] diff --git a/sdk/python/cocoonsandbox/conn.py b/sdk/python/cocoonsandbox/conn.py index c9fe2e67..f267dd3e 100644 --- a/sdk/python/cocoonsandbox/conn.py +++ b/sdk/python/cocoonsandbox/conn.py @@ -8,7 +8,7 @@ from collections.abc import Iterator from typing import BinaryIO, TypeVar -from .errors import APIError, ProtocolError, SilkdError, StreamTimeout +from .errors import APIError, ProtocolError, SilkdError from .frames import MAX_FRAME, decode_response, encode_request _CloseableT = TypeVar("_CloseableT", bound="_Closeable") @@ -35,13 +35,11 @@ def send(self, op: str, **fields) -> None: self._sock.sendall(encode_request(op, **fields)) def recv(self) -> dict: - """Returns the next frame; raises SilkdError on an error frame, - StreamTimeout when the socket times out, and ProtocolError on EOF, an - oversized line, or an undecodable one — every failure is typed.""" + """Returns the next frame; raises SilkdError on an error frame and + ProtocolError on EOF, an oversized line, or an undecodable one — every + failure is typed.""" try: line = self._reader.readline(MAX_FRAME + 1) - except socket.timeout as exc: - raise StreamTimeout(f"read timed out: {exc}") from exc except OSError as exc: raise ProtocolError(f"read failed: {exc}") from exc if not line: @@ -78,7 +76,8 @@ def close(self) -> None: def dial_agent(addr: str, sandbox_id: str, token: str, timeout: float) -> Conn: """Opens the data-plane connection: TCP dial plus a hand-rolled HTTP - Upgrade, so nothing pools or proxies underneath the byte stream.""" + Upgrade, both bounded by timeout, so nothing pools or proxies underneath + the byte stream; the stream itself has no idle timeout.""" # id/token interpolate into the raw request line; CR/LF would inject headers. for name, value in (("sandbox id", sandbox_id), ("token", token)): if any(c in value for c in "\r\n\0"): @@ -122,6 +121,7 @@ def dial_agent(addr: str, sandbox_id: str, token: str, timeout: float) -> Conn: # A bogus content-length must not buffer unbounded bytes. body = reader.read(min(body_len, MAX_FRAME)).decode(errors="replace") if body_len else "" raise APIError("agent upgrade", code, body.strip() or status.strip()) + sock.settimeout(None) return Conn(sock, reader) except Exception: # makefile() holds a ref on the socket; close it too or the fd lingers. diff --git a/sdk/python/cocoonsandbox/errors.py b/sdk/python/cocoonsandbox/errors.py index 73da6dde..4b8f16eb 100644 --- a/sdk/python/cocoonsandbox/errors.py +++ b/sdk/python/cocoonsandbox/errors.py @@ -41,8 +41,3 @@ def __init__(self, code: int, stderr: str, stdout: str = ""): class ProtocolError(SandboxError): """The peer broke the frame protocol (unexpected frame, oversized line).""" - - -class StreamTimeout(SandboxError): - """A socket operation exceeded the client timeout; the connection is - desynced, so retry the call, not the frame.""" diff --git a/sdk/python/cocoonsandbox/sandbox.py b/sdk/python/cocoonsandbox/sandbox.py index f6dcf450..df912ff2 100644 --- a/sdk/python/cocoonsandbox/sandbox.py +++ b/sdk/python/cocoonsandbox/sandbox.py @@ -13,7 +13,7 @@ from .checkpoint import Checkpoint from .conn import Conn, _Closeable, dial_agent -from .errors import APIError, ExitError, ProtocolError, SandboxError, StreamTimeout +from .errors import APIError, ExitError, ProtocolError, SandboxError from .frames import BULK_CHUNK, FS_CHUNK from .template import Template @@ -450,7 +450,7 @@ def __iter__(self) -> Iterator[dict]: while True: try: frame = self._conn.recv() - except (ProtocolError, StreamTimeout) as e: + except ProtocolError as e: self.error = e return if frame["type"] == "event": diff --git a/sdk/python/tests/test_stream.py b/sdk/python/tests/test_stream.py new file mode 100644 index 00000000..b0873e21 --- /dev/null +++ b/sdk/python/tests/test_stream.py @@ -0,0 +1,60 @@ +"""The client timeout bounds the dial and the upgrade, not the frames that +follow: a guest stream outlives it.""" + +import json +import socket +import threading +import time + +import pytest + +from cocoonsandbox import Client, Sandbox + +TIMEOUT = 0.2 + + +def serve_port_forward(server: socket.socket, quiet: float) -> None: + conn, _ = server.accept() + reader = conn.makefile("rb") + while reader.readline() not in (b"\r\n", b""): + pass + conn.sendall(b"HTTP/1.1 101 Switching Protocols\r\n\r\n") + assert json.loads(reader.readline())["op"] == "port_forward" + conn.sendall(b'{"type":"ready"}\n') + time.sleep(quiet) + conn.sendall(b'{"type":"data","data":"bGF0ZQ=="}\n{"type":"done"}\n') + conn.close() + + +def serve_silence(server: socket.socket) -> None: + conn, _ = server.accept() + conn.recv(4096) + time.sleep(5 * TIMEOUT) + conn.close() + + +def test_port_stream_outlives_the_client_timeout(): + server = socket.create_server(("127.0.0.1", 0)) + addr = f"127.0.0.1:{server.getsockname()[1]}" + threading.Thread(target=serve_port_forward, args=(server, 3 * TIMEOUT), daemon=True).start() + sb = Sandbox(client=Client(addr, timeout=TIMEOUT), id="sb_1", token="tok", owner=addr) + try: + with sb.dial_port(5000) as port: + assert port.recv() == b"late" + assert port.recv() == b"" + finally: + server.close() + + +def test_dial_is_still_bounded_by_the_client_timeout(): + server = socket.create_server(("127.0.0.1", 0)) + addr = f"127.0.0.1:{server.getsockname()[1]}" + threading.Thread(target=serve_silence, args=(server,), daemon=True).start() + sb = Sandbox(client=Client(addr, timeout=TIMEOUT), id="sb_1", token="tok", owner=addr) + started = time.monotonic() + try: + with pytest.raises(OSError): + sb.dial_port(5000) + finally: + server.close() + assert time.monotonic() - started < 3 * TIMEOUT