From 0a3b1c0f71feaaf102b272d484314beba6c232f8 Mon Sep 17 00:00:00 2001 From: CMGS Date: Thu, 3 Sep 2026 13:08:07 +0800 Subject: [PATCH 1/4] git.status caps its frame; find streams to the caller A status of a large tree rendered one frame past the cap and dropped the connection with no terminal frame. The entry list now stops at an eighth of the frame cap in estimated bytes and the result carries truncated: true, so the head of the list still arrives. The Go SDK gains FindSeq and the Python SDK find_iter: matches stream to the caller, and leaving the loop closes the connection, which ends the walk in the guest, so the caller bounds the result instead of accumulating it. Find keeps its slice API on top of the same stream. --- docs/sdk-python.md | 3 +- docs/sdk.md | 7 ++- docs/silkd.md | 2 +- .../v1/resp_git_status_result_truncated.json | 1 + protocol/wire/frame.go | 9 ++-- protocol/wire/frame_test.go | 4 +- sdk/go/find.go | 7 +++ sdk/go/find_test.go | 31 +++++++++++ sdk/go/utils.go | 53 +++++++++++++------ sdk/python/cocoonsandbox/sandbox.py | 9 +++- silkd/src/git.rs | 31 +++++++++++ silkd/src/proto.rs | 4 +- 12 files changed, 133 insertions(+), 28 deletions(-) create mode 100644 protocol/wire/fixtures/v1/resp_git_status_result_truncated.json diff --git a/docs/sdk-python.md b/docs/sdk-python.md index da6fcef9..f26ad6cc 100644 --- a/docs/sdk-python.md +++ b/docs/sdk-python.md @@ -367,6 +367,7 @@ path on the no-network lane. ```python matches = sb.find("/work", r"TODO|FIXME", glob="*.py") +for m in sb.find_iter("/work", r"TODO"): # streamed; leaving the loop ends the walk in the guest # [{"file", "line", "content"}]; glob is anchored *? wildcards on the file name results = sb.replace(["/work/main.py"], r"foo", "bar") @@ -394,7 +395,7 @@ when the relay drops, which `w.error` tells apart from a clean close (`None`). ```python sb.git_clone(url, "/work/repo", branch="main", depth=1, auth=token) # egress lane only -st = sb.git_status("/work/repo") # {"branch", "ahead", "behind", "files"} +st = sb.git_status("/work/repo") # {"branch", "ahead", "behind", "files", "truncated"?} sb.git_add("/work/repo", ["a.txt"]) sha = sb.git_commit("/work/repo", "message", "Dev ") sb.git_push("/work/repo", auth=token) # egress lane only diff --git a/docs/sdk.md b/docs/sdk.md index 37629df0..73c597a8 100644 --- a/docs/sdk.md +++ b/docs/sdk.md @@ -441,6 +441,11 @@ err = sb.Pull(ctx, "/work", tarWriter) // stream /work back as a tar ```go matches, err := sb.Find(ctx, "/work", `TODO|FIXME`, "*.go") // []wire.Match{File, Line, Content}; glob is anchored *? wildcards on the file name +for m, err := range sb.FindSeq(ctx, "/work", `TODO`, "") { // streamed; a break ends the walk in the guest + if err != nil || m.Line > 100 { + break + } +} results, err := sb.Replace(ctx, []string{"/work/main.go"}, `foo`, "bar") // []wire.Replaced{File, Replacements}; per-file atomic @@ -468,7 +473,7 @@ overflow instead of the stream silently dropping events. ```go err = sb.GitClone(ctx, url, "/work/repo", "main", 0, token) // egress lane only; depth > 0 = shallow -st, err := sb.GitStatus(ctx, "/work/repo") // Branch, Ahead, Behind, Files[] +st, err := sb.GitStatus(ctx, "/work/repo") // Branch, Ahead, Behind, Files[]; Truncated when the list was cut at ~1 MiB err = sb.GitAdd(ctx, "/work/repo", "a.txt") hash, err := sb.GitCommit(ctx, "/work/repo", "message", "Dev ") err = sb.GitPush(ctx, "/work/repo", token) // egress lane only diff --git a/docs/silkd.md b/docs/silkd.md index 4f096a58..20e2d0e2 100644 --- a/docs/silkd.md +++ b/docs/silkd.md @@ -38,7 +38,7 @@ a frame only one side can parse fails CI. | search | `fs_find {path, pattern, glob?}` → `match{file, line, content}`… → `done` / `fs_replace {files, pattern, replacement}` → `replaced{file, replacements}`… → `done` | regex as data, no shell quoting; `glob` is anchored `*`/`?` wildcards over file names; find skips binary and >8 MiB files, and replace skips the same 8 MiB bound with a zero count | | watch | `fs_watch {path, recursive?}` | `ready` once armed (events after it are guaranteed captured), then `event{kind, path}` until the client disconnects; watcher or delivery-queue overflow errors arrive as a terminal `error` instead of silently losing events | | pty | `pty_open {cols, rows, cwd?, env?, user?}` / `pty_resize {pid, cols, rows}` | a shell under a pseudo-terminal; output as `stdout` frames, input as `stdin` frames, exit terminal. PTYs register in the proc table like any exec | -| git | `git_clone {url, path, branch?, depth?, auth?}` / `git_status {path}` / `git_add {path, files}` / `git_commit {path, message, author}` / `git_push {path, auth?}` / `git_pull {path, auth?}` / `git_branch {path, action, name?}` | structured results (porcelain-v2 status, commit hash, branch list). `auth` is injected as an in-memory header, never written to guest disk | +| git | `git_clone {url, path, branch?, depth?, auth?}` / `git_status {path}` / `git_add {path, files}` / `git_commit {path, message, author}` / `git_push {path, auth?}` / `git_pull {path, auth?}` / `git_branch {path, action, name?}` | structured results (porcelain-v2 status, commit hash, branch list). A status past about 1 MiB of entries carries `truncated: true` with the head of the list, so one frame never exceeds the cap. `auth` is injected as an in-memory header, never written to guest disk | | port | `port_forward {port}` | relays guest TCP 127.0.0.1:port over this connection: `ready` once connected, then `data` both ways (`data_end` half-closes the guest socket); the server closing ends the stream with `done`. Works on both lanes — the no-network lane's only way in | | lsp | `lsp_start {language, root?}` / `lsp_request {server_id}` / `lsp_stop {server_id}` | a broker for the language server a flavor image ships: `lsp_start` spawns the argv named in `/etc/silkd/lsp.d/` (absent on the base image → `not_found` naming the flavor) and answers `lsp_started{server_id}`; `lsp_request` attaches the JSON-RPC byte stream (`ready`, then `data` both ways — silkd pipes bytes, it never parses LSP; `data_end` half-closes the server's stdin) and the stream ending reaps the server (v1 single-shot); `lsp_stop` kills it early, and a server nothing attaches within 5 minutes is reaped | | misc | `info` | `{version, proto, uptime_secs, procs, sessions}` — the in-guest readiness probe (sandboxd consumes it; no SDK surface). Distinct from the control plane's `GET /v1/info`, which reports node pools and claims | diff --git a/protocol/wire/fixtures/v1/resp_git_status_result_truncated.json b/protocol/wire/fixtures/v1/resp_git_status_result_truncated.json new file mode 100644 index 00000000..be64dcb7 --- /dev/null +++ b/protocol/wire/fixtures/v1/resp_git_status_result_truncated.json @@ -0,0 +1 @@ +{"type":"git_status_result","branch":"main","ahead":0,"behind":0,"files":[{"path":"a.txt","staged":"?","unstaged":"?"}],"truncated":true} diff --git a/protocol/wire/frame.go b/protocol/wire/frame.go index 24dcd9e7..499809f2 100644 --- a/protocol/wire/frame.go +++ b/protocol/wire/frame.go @@ -594,10 +594,11 @@ func (Event) RespType() string { return "event" } // GitStatusResult answers GitStatus. type GitStatusResult struct { - Branch string `json:"branch"` - Ahead uint32 `json:"ahead"` - Behind uint32 `json:"behind"` - Files []GitFileStatus `json:"files"` + Branch string `json:"branch"` + Ahead uint32 `json:"ahead"` + Behind uint32 `json:"behind"` + Files []GitFileStatus `json:"files"` + Truncated bool `json:"truncated,omitempty"` } func (GitStatusResult) RespType() string { return "git_status_result" } diff --git a/protocol/wire/frame_test.go b/protocol/wire/frame_test.go index 50eefd1e..4efef372 100644 --- a/protocol/wire/frame_test.go +++ b/protocol/wire/frame_test.go @@ -58,8 +58,8 @@ func TestFixtureCorpusRoundTrips(t *testing.T) { seen++ } - if seen != 60 { - t.Fatalf("fixture corpus: %d frames, want exactly 60", seen) + if seen != 61 { + t.Fatalf("fixture corpus: %d frames, want exactly 61", seen) } } diff --git a/sdk/go/find.go b/sdk/go/find.go index 9fd2c4bf..69b0c5b3 100644 --- a/sdk/go/find.go +++ b/sdk/go/find.go @@ -2,6 +2,7 @@ package sandbox import ( "context" + "iter" "github.com/cocoonstack/sandbox/protocol/wire" ) @@ -13,6 +14,12 @@ func (s *Sandbox) Find(ctx context.Context, path, pattern, glob string) ([]wire. return collectRPC[wire.Match](ctx, s, &wire.FsFind{Path: path, Pattern: pattern, Glob: glob}) } +// FindSeq streams the matches Find would collect; breaking out of the loop +// ends the walk in the guest, so the caller bounds the result. +func (s *Sandbox) FindSeq(ctx context.Context, path, pattern, glob string) iter.Seq2[wire.Match, error] { + return streamRPC[wire.Match](ctx, s, &wire.FsFind{Path: path, Pattern: pattern, Glob: glob}) +} + // Replace rewrites pattern (a regular expression) to replacement in each of // files, returning one result per file with its replacement count. func (s *Sandbox) Replace(ctx context.Context, files []string, pattern, replacement string) ([]wire.Replaced, error) { diff --git a/sdk/go/find_test.go b/sdk/go/find_test.go index 2b5861a0..bf681b5f 100644 --- a/sdk/go/find_test.go +++ b/sdk/go/find_test.go @@ -33,6 +33,37 @@ func TestFindStreamsMatches(t *testing.T) { } } +func TestFindSeqStopsAfterTheCallerBreaks(t *testing.T) { + sb := fakeSandbox(t) + ctx := t.Context() + if err := sb.Mkdir(ctx, "/work", true); err != nil { + t.Fatalf("mkdir: %v", err) + } + if err := sb.WriteFile(ctx, "/work/many.txt", []byte(strings.Repeat("TODO\n", 500)), nil); err != nil { + t.Fatalf("write: %v", err) + } + + var seen []wire.Match + for m, err := range sb.FindSeq(ctx, "/work", "TODO", "") { + if err != nil { + t.Fatalf("FindSeq: %v", err) + } + if seen = append(seen, m); len(seen) == 3 { + break + } + } + if len(seen) != 3 || seen[2].Line != 3 { + t.Fatalf("seen %+v, want the first three lines", seen) + } + all, err := sb.Find(ctx, "/work", "TODO", "") + if err != nil { + t.Fatalf("Find after an abandoned stream: %v", err) + } + if len(all) != 500 { + t.Errorf("Find returned %d matches, want 500", len(all)) + } +} + func TestFindBadPatternIsTypedError(t *testing.T) { sb := fakeSandbox(t) _, err := sb.Find(t.Context(), "/", "(", "") diff --git a/sdk/go/utils.go b/sdk/go/utils.go index c334e78d..1c8bf948 100644 --- a/sdk/go/utils.go +++ b/sdk/go/utils.go @@ -2,6 +2,7 @@ package sandbox import ( "context" + "iter" "fmt" "io" @@ -63,28 +64,46 @@ func oneShotRPC[T any, PT respPtr[T]](ctx context.Context, s *Sandbox, req wire. // collectRPC sends req and gathers every streamed frame of type T until Done. func collectRPC[T any, PT respPtr[T]](ctx context.Context, s *Sandbox, req wire.Request) ([]T, error) { - conn, done, err := s.call(ctx, req) - if err != nil { - return nil, err - } - defer done() var out []T - for { - resp, err := recv(ctx, conn) + for v, err := range streamRPC[T, PT](ctx, s, req) { if err != nil { return nil, err } - if v, ok := resp.(PT); ok { - out = append(out, *v) - continue + out = append(out, v) + } + return out, nil +} + +// streamRPC sends req and yields each streamed frame of type T until Done; breaking out closes the connection, which ends the guest-side producer. +func streamRPC[T any, PT respPtr[T]](ctx context.Context, s *Sandbox, req wire.Request) iter.Seq2[T, error] { + return func(yield func(T, error) bool) { + var zero T + conn, done, err := s.call(ctx, req) + if err != nil { + yield(zero, err) + return } - switch r := resp.(type) { - case *wire.Done: - return out, nil - case *wire.ErrorResp: - return nil, r - default: - return nil, unexpected(resp) + defer done() + for { + resp, err := recv(ctx, conn) + if err != nil { + yield(zero, err) + return + } + if v, ok := resp.(PT); ok { + if !yield(*v, nil) { + return + } + continue + } + switch r := resp.(type) { + case *wire.Done: + case *wire.ErrorResp: + yield(zero, r) + default: + yield(zero, unexpected(resp)) + } + return } } } diff --git a/sdk/python/cocoonsandbox/sandbox.py b/sdk/python/cocoonsandbox/sandbox.py index d536dde4..d0629673 100644 --- a/sdk/python/cocoonsandbox/sandbox.py +++ b/sdk/python/cocoonsandbox/sandbox.py @@ -157,9 +157,16 @@ def pull(self, path: str) -> bytes: return _drain_data(conn) def find(self, path: str, pattern: str, glob: str = "") -> list[dict]: + return list(self.find_iter(path, pattern, glob)) + + def find_iter(self, path: str, pattern: str, glob: str = "") -> Iterator[dict]: + """Yields matches as they stream; leaving the loop early closes the + connection, which ends the walk in the guest.""" with self._dial() as conn: conn.send("fs_find", path=path, pattern=pattern, glob=glob or None) - return [f for f in conn.recv_until("done") if f["type"] == "match"] + for f in conn.recv_until("done"): + if f["type"] == "match": + yield f def replace(self, files: list[str], pattern: str, replacement: str) -> list[dict]: with self._dial() as conn: diff --git a/silkd/src/git.rs b/silkd/src/git.rs index effd09a8..a0e32250 100644 --- a/silkd/src/git.rs +++ b/silkd/src/git.rs @@ -12,6 +12,10 @@ use tokio::process::Command; use crate::proto::{self, ErrorKind, GitBranchOp, GitFileStatus, Response}; use crate::sysutil; +/// Estimated JSON bytes of file entries one status frame carries before it truncates; an eighth of the frame cap leaves room for escaping. +const STATUS_FILES_BYTES: usize = proto::MAX_FRAME / 8; +const STATUS_ENTRY_OVERHEAD: usize = 40; + /// Clones `url` into `path` (optionally a branch, shallow depth), then Done. pub async fn clone( w: &mut W, @@ -220,12 +224,19 @@ fn parse_status(text: &str) -> Response { let mut branch = String::new(); let (mut ahead, mut behind) = (0, 0); let mut files = Vec::new(); + let mut bytes = 0; + let mut truncated = false; for line in text.lines() { if let Some(rest) = line.strip_prefix("# branch.head ") { branch = rest.to_string(); } else if let Some(rest) = line.strip_prefix("# branch.ab ") { (ahead, behind) = parse_ahead_behind(rest); } else if let Some(f) = parse_file_line(line) { + bytes += f.path.len() + STATUS_ENTRY_OVERHEAD; + if bytes > STATUS_FILES_BYTES { + truncated = true; + break; + } files.push(f); } } @@ -234,6 +245,7 @@ fn parse_status(text: &str) -> Response { ahead, behind, files, + truncated, } } @@ -309,6 +321,25 @@ fn split_author(author: &str) -> (&str, &str) { mod tests { use super::*; + #[test] + fn status_truncates_past_the_frame_budget() { + let text: String = (0..200_000).map(|i| format!("? f{i}\n")).collect(); + let Response::GitStatusResult { + files, truncated, .. + } = parse_status(&text) + else { + panic!("status frame") + }; + assert!(truncated); + assert!(!files.is_empty() && files.len() < 200_000); + assert!(serde_json::to_vec(&parse_status(&text)).unwrap().len() < proto::MAX_FRAME); + + let Response::GitStatusResult { truncated, .. } = parse_status("? one\n") else { + panic!("status frame") + }; + assert!(!truncated); + } + #[test] fn parses_ordinary_rename_untracked_and_conflict() { let ordinary = parse_file_line("1 .M N... 100644 100644 100644 h1 h2 src/main.rs").unwrap(); diff --git a/silkd/src/proto.rs b/silkd/src/proto.rs index f7218e9c..d462235d 100644 --- a/silkd/src/proto.rs +++ b/silkd/src/proto.rs @@ -279,6 +279,8 @@ pub enum Response { ahead: u32, behind: u32, files: Vec, + #[serde(default, skip_serializing_if = "std::ops::Not::not")] + truncated: bool, }, GitCommitResult { hash: String, @@ -820,7 +822,7 @@ mod tests { seen += 1; } assert_eq!( - seen, 60, + seen, 61, "fixture corpus: adding a verb means adding its fixture" ); } From 2f6f0200e421f4c4b905015cf373cb4fe5c09285 Mon Sep 17 00:00:00 2001 From: CMGS Date: Thu, 3 Sep 2026 13:14:20 +0800 Subject: [PATCH 2/4] sdk/go: sort the iter import --- sdk/go/utils.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/sdk/go/utils.go b/sdk/go/utils.go index 1c8bf948..00b943d0 100644 --- a/sdk/go/utils.go +++ b/sdk/go/utils.go @@ -2,9 +2,9 @@ package sandbox import ( "context" - "iter" "fmt" "io" + "iter" "github.com/cocoonstack/sandbox/protocol/wire" "github.com/cocoonstack/sandbox/sdk/go/silkd" From 30b9d18f7f7cb55297169aad0d43dca459f21fda Mon Sep 17 00:00:00 2001 From: CMGS Date: Thu, 3 Sep 2026 13:30:01 +0800 Subject: [PATCH 3/4] find: a client that goes away ends the walk A caller that stops reading after an early match left silkd scanning the rest of the tree whenever no further match tried to write; the writer now selects on the request side as watch does, so EOF or a stray frame ends the walk without a done frame. The find e2e tests keep their write half open until the terminal frame, the way the SDKs do. The SDK examples handle the error before the result-limit break, and the Python one closes the generator explicitly. --- docs/sdk-python.md | 5 +- docs/sdk.md | 5 +- sdk/python/cocoonsandbox/sandbox.py | 5 +- silkd/src/find.rs | 96 ++++++++++++++++++++++++----- silkd/src/server.rs | 2 +- silkd/tests/find_e2e.rs | 49 ++++++++++----- 6 files changed, 126 insertions(+), 36 deletions(-) diff --git a/docs/sdk-python.md b/docs/sdk-python.md index f26ad6cc..9be947dd 100644 --- a/docs/sdk-python.md +++ b/docs/sdk-python.md @@ -367,7 +367,10 @@ path on the no-network lane. ```python matches = sb.find("/work", r"TODO|FIXME", glob="*.py") -for m in sb.find_iter("/work", r"TODO"): # streamed; leaving the loop ends the walk in the guest +with closing(sb.find_iter("/work", r"TODO")) as stream: # streamed; closing ends the walk in the guest + for m in stream: + if m["line"] > 100: + break # [{"file", "line", "content"}]; glob is anchored *? wildcards on the file name results = sb.replace(["/work/main.py"], r"foo", "bar") diff --git a/docs/sdk.md b/docs/sdk.md index 73c597a8..339048b9 100644 --- a/docs/sdk.md +++ b/docs/sdk.md @@ -442,7 +442,10 @@ err = sb.Pull(ctx, "/work", tarWriter) // stream /work back as a tar matches, err := sb.Find(ctx, "/work", `TODO|FIXME`, "*.go") // []wire.Match{File, Line, Content}; glob is anchored *? wildcards on the file name for m, err := range sb.FindSeq(ctx, "/work", `TODO`, "") { // streamed; a break ends the walk in the guest - if err != nil || m.Line > 100 { + if err != nil { + return err + } + if m.Line > 100 { break } } diff --git a/sdk/python/cocoonsandbox/sandbox.py b/sdk/python/cocoonsandbox/sandbox.py index d0629673..ec933cd5 100644 --- a/sdk/python/cocoonsandbox/sandbox.py +++ b/sdk/python/cocoonsandbox/sandbox.py @@ -160,8 +160,9 @@ def find(self, path: str, pattern: str, glob: str = "") -> list[dict]: return list(self.find_iter(path, pattern, glob)) def find_iter(self, path: str, pattern: str, glob: str = "") -> Iterator[dict]: - """Yields matches as they stream; leaving the loop early closes the - connection, which ends the walk in the guest.""" + """Yields matches as they stream. Closing the generator (a `for` loop + that owns it, or contextlib.closing) closes the connection, which ends + the walk in the guest.""" with self._dial() as conn: conn.send("fs_find", path=path, pattern=pattern, glob=glob or None) for f in conn.recv_until("done"): diff --git a/silkd/src/find.rs b/silkd/src/find.rs index 635b8f9d..08b01ee1 100644 --- a/silkd/src/find.rs +++ b/silkd/src/find.rs @@ -8,7 +8,7 @@ use std::sync::Arc; use regex::Regex; use tokio::fs; -use tokio::io::AsyncWrite; +use tokio::io::{AsyncBufRead, AsyncBufReadExt, AsyncWrite}; use tokio::runtime::Handle; use tokio::sync::{Semaphore, SemaphorePermit, mpsc}; @@ -133,13 +133,18 @@ impl Walk<'_> { /// Streams `match` frames for every line under `path` matching `pattern`, /// terminated by `done`. `glob` narrows the walk to file names matching it /// (`*` and `?` wildcards); an invalid pattern is a bad-request error. -pub async fn find( +pub async fn find( + reader: &mut R, w: &mut W, path: String, pattern: String, glob: Option, -) -> std::io::Result<()> { - find_bounded(w, path, pattern, glob, MATCH_QUEUE_BYTES).await +) -> std::io::Result<()> +where + R: AsyncBufRead + Unpin, + W: AsyncWrite + Unpin, +{ + find_bounded(reader, w, path, pattern, glob, MATCH_QUEUE_BYTES).await } /// Rewrites every `pattern` match to `replacement` in each of `files`, @@ -191,13 +196,18 @@ pub async fn replace( proto::write_frame(w, &Response::Done).await } -async fn find_bounded( +async fn find_bounded( + reader: &mut R, w: &mut W, path: String, pattern: String, glob: Option, budget_bytes: usize, -) -> std::io::Result<()> { +) -> std::io::Result<()> +where + R: AsyncBufRead + Unpin, + W: AsyncWrite + Unpin, +{ let re = match Regex::new(&pattern) { Ok(re) => re, Err(e) => return proto::error_frame(w, ErrorKind::BadRequest, e.to_string()).await, @@ -223,18 +233,35 @@ async fn find_bounded( let mut batch = Vec::new(); let mut buf = Vec::new(); let mut failed = None; - while rx.recv_many(&mut batch, MATCH_QUEUE).await > 0 { - let sent = proto::write_frames(w, &mut buf, &batch).await; - budget.release(batch.iter().map(match_cost).sum()); - batch.clear(); - if let Err(e) = sent { - failed = Some(e); - break; + let mut gone = false; + loop { + tokio::select! { + biased; + // The client sends nothing during a find, so any readable state ends the walk. + _ = reader.fill_buf() => { + gone = true; + break; + } + n = rx.recv_many(&mut batch, MATCH_QUEUE) => { + if n == 0 { + break; + } + let sent = proto::write_frames(w, &mut buf, &batch).await; + budget.release(batch.iter().map(match_cost).sum()); + batch.clear(); + if let Err(e) = sent { + failed = Some(e); + break; + } + } } } drop(rx); budget.close(); let walked = walk.await.map_err(std::io::Error::other)?; + if gone { + return Ok(()); + } if let Some(e) = failed { if e.kind() == std::io::ErrorKind::InvalidData { return err_frame(w, &e, "find").await; @@ -278,7 +305,7 @@ fn glob_regex(glob: &str) -> Result { mod tests { use std::time::Duration; - use tokio::io::{AsyncReadExt, DuplexStream}; + use tokio::io::{AsyncReadExt, BufReader, DuplexStream}; use tokio::task::JoinHandle; use super::*; @@ -304,8 +331,22 @@ mod tests { (w, reader) } + fn silent_client() -> (DuplexStream, BufReader) { + let (peer, r) = tokio::io::duplex(64); + (peer, BufReader::new(r)) + } + async fn run(w: &mut DuplexStream, dir: &Path, budget: usize) -> std::io::Result<()> { - find_bounded(w, dir.display().to_string(), "needle".into(), None, budget).await + let (_peer, mut reader) = silent_client(); + find_bounded( + &mut reader, + w, + dir.display().to_string(), + "needle".into(), + None, + budget, + ) + .await } #[tokio::test(flavor = "multi_thread")] @@ -330,6 +371,31 @@ mod tests { assert!(found.is_err()); } + #[tokio::test(flavor = "multi_thread")] + async fn find_stops_without_done_when_the_client_disconnects() { + let dir = tree(2000, 1).await; + let (mut w, reader) = sink(); + let (peer, mut client) = silent_client(); + drop(peer); + let found = tokio::time::timeout( + Duration::from_secs(10), + find_bounded( + &mut client, + &mut w, + dir.path().display().to_string(), + "needle".into(), + None, + MATCH_QUEUE_BYTES, + ), + ) + .await + .expect("walk must end once the client is gone"); + found.expect("find"); + drop(w); + let out = reader.await.expect("join"); + assert!(!out.contains("\"type\":\"done\""), "{out}"); + } + #[tokio::test(flavor = "multi_thread")] async fn find_reports_an_oversized_match_as_an_error_frame() { let dir = tempfile::tempdir().expect("tempdir"); diff --git a/silkd/src/server.rs b/silkd/src/server.rs index b37236f5..563160b0 100644 --- a/silkd/src/server.rs +++ b/silkd/src/server.rs @@ -136,7 +136,7 @@ impl State { path, pattern, glob, - } => find::find(&mut writer, path, pattern, glob).await, + } => find::find(&mut reader, &mut writer, path, pattern, glob).await, Request::FsReplace { files, pattern, diff --git a/silkd/tests/find_e2e.rs b/silkd/tests/find_e2e.rs index 57f581ee..5ddd4857 100644 --- a/silkd/tests/find_e2e.rs +++ b/silkd/tests/find_e2e.rs @@ -3,13 +3,35 @@ #![allow(clippy::unwrap_used, clippy::expect_used)] mod common; +use std::sync::Arc; + +use serde_json::{Value, json}; +use silkd::server::State; +use tokio::io::AsyncWriteExt; + use common::{exchange, type_of}; -use serde_json::json; async fn write_file(dir: &std::path::Path, name: &str, body: &str) { tokio::fs::write(dir.join(name), body).await.unwrap(); } +async fn find_frames(req: Value) -> Vec { + let (mut cw, mut out, handle) = common::connect(&Arc::new(State::new())); + cw.write_all(format!("{req}\n").as_bytes()).await.unwrap(); + let mut frames = Vec::new(); + while let Some(l) = out.next_line().await.unwrap() { + let frame: Value = serde_json::from_str(&l).unwrap(); + let terminal = matches!(type_of(&frame), "done" | "error"); + frames.push(frame); + if terminal { + break; + } + } + drop(cw); + handle.await.unwrap().unwrap(); + frames +} + #[tokio::test] async fn find_streams_line_matches() { let dir = tempfile::tempdir().unwrap(); @@ -18,13 +40,12 @@ async fn find_streams_line_matches() { tokio::fs::create_dir(dir.path().join("sub")).await.unwrap(); write_file(&dir.path().join("sub"), "c.rs", "// TODO nested\n").await; - let frames = exchange(&[json!({ + let frames = find_frames(json!({ "op": "fs_find", "path": dir.path().to_str().unwrap(), "pattern": "TODO", "glob": "*.rs" - }) - .to_string()]) + })) .await; let matches: Vec<_> = frames.iter().filter(|f| type_of(f) == "match").collect(); @@ -45,13 +66,12 @@ async fn find_glob_is_a_real_glob_not_a_substring() { write_file(dir.path(), "a.rs", "TODO\n").await; write_file(dir.path(), "a.rs.bak", "TODO\n").await; - let frames = exchange(&[json!({ + let frames = find_frames(json!({ "op": "fs_find", "path": dir.path().to_str().unwrap(), "pattern": "TODO", "glob": "*.rs" - }) - .to_string()]) + })) .await; let files: Vec<_> = frames .iter() @@ -61,13 +81,12 @@ async fn find_glob_is_a_real_glob_not_a_substring() { assert_eq!(files.len(), 1, "{files:?}"); assert!(files[0].ends_with("a.rs")); - let frames = exchange(&[json!({ + let frames = find_frames(json!({ "op": "fs_find", "path": dir.path().to_str().unwrap(), "pattern": "TODO", "glob": "?.rs" - }) - .to_string()]) + })) .await; assert_eq!( frames.iter().filter(|f| type_of(f) == "match").count(), @@ -79,12 +98,11 @@ async fn find_glob_is_a_real_glob_not_a_substring() { #[tokio::test] async fn find_rejects_bad_pattern() { let dir = tempfile::tempdir().unwrap(); - let frames = exchange(&[json!({ + let frames = find_frames(json!({ "op": "fs_find", "path": dir.path().to_str().unwrap(), "pattern": "(" - }) - .to_string()]) + })) .await; assert_eq!(type_of(&frames[0]), "error"); assert_eq!(frames[0]["kind"], "bad_request"); @@ -159,12 +177,11 @@ async fn find_skips_a_file_over_the_size_bound() { .await .unwrap(); - let frames = exchange(&[json!({ + let frames = find_frames(json!({ "op": "fs_find", "path": dir.path().to_str().unwrap(), "pattern": "foo" - }) - .to_string()]) + })) .await; assert_eq!(frames.iter().filter(|f| type_of(f) == "match").count(), 0); From 56e03e8afa6ab428353e424e9fd0475c515044a0 Mon Sep 17 00:00:00 2001 From: CMGS Date: Thu, 3 Sep 2026 14:00:33 +0800 Subject: [PATCH 4/4] find: the walker checks for a gone receiver at every entry Ending the writer loop closed the budget, but a walker scanning a tail with no matches only noticed at its next send, so an abandoned find still read the whole tree. The walk now tests the channel before each entry and reports whether it completed. The Python example imports closing and the docstring names contextlib.closing as the portable cleanup. --- docs/sdk-python.md | 2 ++ sdk/python/cocoonsandbox/sandbox.py | 6 ++--- silkd/src/find.rs | 34 ++++++++++++++++++++++++----- 3 files changed, 33 insertions(+), 9 deletions(-) diff --git a/docs/sdk-python.md b/docs/sdk-python.md index 9be947dd..35f42bda 100644 --- a/docs/sdk-python.md +++ b/docs/sdk-python.md @@ -367,6 +367,8 @@ path on the no-network lane. ```python matches = sb.find("/work", r"TODO|FIXME", glob="*.py") +from contextlib import closing + with closing(sb.find_iter("/work", r"TODO")) as stream: # streamed; closing ends the walk in the guest for m in stream: if m["line"] > 100: diff --git a/sdk/python/cocoonsandbox/sandbox.py b/sdk/python/cocoonsandbox/sandbox.py index ec933cd5..99f7a472 100644 --- a/sdk/python/cocoonsandbox/sandbox.py +++ b/sdk/python/cocoonsandbox/sandbox.py @@ -160,9 +160,9 @@ def find(self, path: str, pattern: str, glob: str = "") -> list[dict]: return list(self.find_iter(path, pattern, glob)) def find_iter(self, path: str, pattern: str, glob: str = "") -> Iterator[dict]: - """Yields matches as they stream. Closing the generator (a `for` loop - that owns it, or contextlib.closing) closes the connection, which ends - the walk in the guest.""" + """Yields matches as they stream. Closing the generator closes the + connection and ends the walk in the guest; wrap it in contextlib.closing + for deterministic cleanup, since only CPython finalizes on refcount.""" with self._dial() as conn: conn.send("fs_find", path=path, pattern=pattern, glob=glob or None) for f in conn.recv_until("done"): diff --git a/silkd/src/find.rs b/silkd/src/find.rs index 08b01ee1..d8cc679b 100644 --- a/silkd/src/find.rs +++ b/silkd/src/find.rs @@ -67,9 +67,10 @@ struct Walk<'a> { } impl Walk<'_> { - /// Walks `root` depth-first, sending a `match` frame per matching line. A - /// failure on the root propagates; deeper ones skip only that directory. - fn run(&self, root: PathBuf) -> std::io::Result<()> { + /// Walks `root` depth-first, sending a `match` frame per matching line; false + /// means the receiver went away. A failure on the root propagates; deeper + /// ones skip only that directory. + fn run(&self, root: PathBuf) -> std::io::Result { let mut stack = vec![root]; let mut root = true; while let Some(dir) = stack.pop() { @@ -79,6 +80,9 @@ impl Walk<'_> { Err(_) => continue, }; for ent in rd { + if self.tx.is_closed() { + return Ok(false); + } let ent = match ent { Ok(ent) => ent, Err(e) if root => return Err(e), @@ -91,12 +95,12 @@ impl Walk<'_> { if ft.is_dir() { stack.push(p); } else if ft.is_file() && name_matches(&p, self.name_re) && !self.scan_file(&p) { - return Ok(()); + return Ok(false); } } root = false; } - Ok(()) + Ok(true) } /// Scans one file, reporting whether the receiver is still listening. The size @@ -269,7 +273,7 @@ where return Err(e); } match walked { - Ok(()) => proto::write_frame(w, &Response::Done).await, + Ok(_) => proto::write_frame(w, &Response::Done).await, Err(e) => err_frame(w, &e, "read_dir").await, } } @@ -396,6 +400,24 @@ mod tests { assert!(!out.contains("\"type\":\"done\""), "{out}"); } + #[tokio::test(flavor = "multi_thread")] + async fn walk_stops_at_the_next_entry_once_the_receiver_is_gone() { + let dir = tree(50, 1).await; + let re = Regex::new("needle").expect("regex"); + let (tx, rx) = mpsc::channel::(MATCH_QUEUE); + drop(rx); + let budget = MatchBudget::new(MATCH_QUEUE_BYTES, Handle::current()); + let walk = Walk { + re: &re, + name_re: None, + tx: &tx, + budget: &budget, + }; + let completed = + tokio::task::block_in_place(|| walk.run(dir.path().to_path_buf())).expect("walk"); + assert!(!completed); + } + #[tokio::test(flavor = "multi_thread")] async fn find_reports_an_oversized_match_as_an_error_frame() { let dir = tempfile::tempdir().expect("tempdir");