diff --git a/docs/sdk-python.md b/docs/sdk-python.md index 99e3c75e..c944f136 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 577aa31d..720bf103 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..00b943d0 100644 --- a/sdk/go/utils.go +++ b/sdk/go/utils.go @@ -4,6 +4,7 @@ import ( "context" "fmt" "io" + "iter" "github.com/cocoonstack/sandbox/protocol/wire" "github.com/cocoonstack/sandbox/sdk/go/silkd" @@ -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 c3051513..9df8ea94 100644 --- a/silkd/src/proto.rs +++ b/silkd/src/proto.rs @@ -278,6 +278,8 @@ pub enum Response { ahead: u32, behind: u32, files: Vec, + #[serde(default, skip_serializing_if = "std::ops::Not::not")] + truncated: bool, }, GitCommitResult { hash: String, @@ -817,7 +819,7 @@ mod tests { seen += 1; } assert_eq!( - seen, 60, + seen, 61, "fixture corpus: adding a verb means adding its fixture" ); }