sdk/python: keep the client timeout off a guest stream once it is up

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:

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",
 ]

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.

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."""

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":

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