sdk/python: rebind find_iter after an owner moves

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

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

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: