diff --git a/docs/sdk-python.md b/docs/sdk-python.md index f71b6dd4..b12a8a83 100644 --- a/docs/sdk-python.md +++ b/docs/sdk-python.md @@ -145,8 +145,8 @@ another node with a lookup, also when that node is gone; releasing one already gone is not an error — double-release and reap races stay silent). Every other call follows a move too: a 404 or a refused connection from the owner runs `client.lookup` once, and the call is sent again to a different owner, since -every argument (stdin, file data, tar streams) is bytes; only `find_iter` -fails once before the next call reaches the new owner. An exception from the +every argument (stdin, file data, tar streams) is bytes. `find_iter` also retries +once before its first frame; it never replays matches already yielded. An exception from the caller's own `on_stdout`/`on_stderr` is raised as it is and never retried, and a released sandbox still raises a 404 `APIError`. Volume sandboxes cannot hibernate, fork, checkpoint, or promote. Egress-lane diff --git a/sdk/python/cocoonsandbox/sandbox.py b/sdk/python/cocoonsandbox/sandbox.py index fed19f15..cfe517da 100644 --- a/sdk/python/cocoonsandbox/sandbox.py +++ b/sdk/python/cocoonsandbox/sandbox.py @@ -287,16 +287,17 @@ def pull(self, path: str) -> bytes: with self._lease(self._connect("fs_pull", path=path)) as conn: return _drain_data(conn) - @_follows_move def find(self, path: str, pattern: str, glob: str = "") -> list[dict[str, Any]]: return list(self.find_iter(path, pattern, glob)) def find_iter(self, path: str, pattern: str, glob: str = "") -> Iterator[dict[str, Any]]: """Yields matches as they stream.""" - with self._lease(self._connect("fs_find", path=path, pattern=pattern, glob=glob or None)) as conn: - for f in conn.recv_until("done"): - if f["type"] == "match": - yield f + conn, frame = self._open_stream("fs_find", "", path=path, pattern=pattern, glob=glob or None) + with self._lease(conn): + while frame["type"] != "done": + if frame["type"] == "match": + yield frame + frame = conn.recv() @_follows_move def replace(self, files: list[str], pattern: str, replacement: str) -> list[dict[str, Any]]: @@ -551,7 +552,7 @@ def _lease(self, conn: Conn) -> Iterator[Conn]: def _open_stream(self, op: str, expect: str = "ready", **fields: object) -> tuple[Conn, dict[str, Any]]: conn = self._connect(op, **fields) try: - frame = _expect(conn, expect) + frame = _expect(conn, expect) if expect else conn.recv() except SilkdError: self._pool.park(conn) raise diff --git a/sdk/python/tests/test_keepalive.py b/sdk/python/tests/test_keepalive.py index 393b3866..00fb19bc 100644 --- a/sdk/python/tests/test_keepalive.py +++ b/sdk/python/tests/test_keepalive.py @@ -67,6 +67,8 @@ def _answer(self, op: str) -> list: return [{"type": "info", "version": "fake", "uptime_secs": 0, "procs": 0}] if op == "fs_stat": return [{"type": "stat", "info": {"kind": "dir", "size": 0, "mode": 0o755, "mtime_epoch_secs": 0}}] + if op == "fs_find": + return [{"type": "match", "path": "/work/f", "line": 1, "text": "TODO"}, {"type": "done"}] if op == "exec": out = base64.b64encode(b"42\n").decode() return [{"type": "started", "pid": 1}, {"type": "stdout", "data": out}, {"type": "exit", "code": 0}] @@ -211,6 +213,26 @@ def test_a_handle_follows_its_sandbox_off_a_retired_node(spawn_node, dead_addr): agent.stop() +@pytest.mark.parametrize("retired", [False, True], ids=["owner_404", "connection_refused"]) +@pytest.mark.parametrize("streamed", [True, False], ids=["find_iter", "find"]) +def test_find_retries_at_the_new_owner(spawn_node, dead_addr, relay, retired, streamed): + agent, _ = relay() + lookups = [] + + def locate(body, path): + lookups.append(path) + return 200, {"owner_addr": agent.addr} + + directory = spawn_node({("GET", "/v1/sandboxes/sb_1/owner"): locate}) + old = dead_addr if retired else spawn_node({}) + sb = Client(directory, keep_alive=0).attach(old, "sb_1", "tok") + matches = list(sb.find_iter("/work", "TODO")) if streamed else sb.find("/work", "TODO") + assert matches == [{"type": "match", "path": "/work/f", "line": 1, "text": "TODO"}] + assert sb.owner == agent.addr + assert len(lookups) == 1 + assert agent.upgrades == 1 + + def test_a_callback_error_tagged_for_this_sandbox_is_not_replayed(spawn_node): owner, other = FakeAgent(), FakeAgent() try: