From 29503eec426315739279b1acd0ab765463d89ad2 Mon Sep 17 00:00:00 2001 From: CMGS Date: Fri, 6 Nov 2026 18:13:50 +0800 Subject: [PATCH 01/11] pool: log failed reap marks before taking the manager lock Folding the failed-mark filter into the keep loop moved one Error line per expiring archive under m.mu; with the checkpoint store down that is thousands of lines on every reap tick, held against the claim path. --- sandboxd/pool/reap.go | 12 ++++++++---- 1 file changed, 8 insertions(+), 4 deletions(-) diff --git a/sandboxd/pool/reap.go b/sandboxd/pool/reap.go index e2eea3214..9ce89de44 100644 --- a/sandboxd/pool/reap.go +++ b/sandboxd/pool/reap.go @@ -4,6 +4,7 @@ import ( "cmp" "context" "net/netip" + "slices" "time" "github.com/projecteru2/core/log" @@ -79,14 +80,17 @@ func (m *Manager) reapOnce(ctx context.Context) { logger := log.WithFunc("pool.reapOnce") failed := m.ckMarks.markAll(purges) + expired = slices.DeleteFunc(expired, func(v victim) bool { + markErr, bad := failed[v.ck] + if bad { + logger.Errorf(ctx, markErr, "mark archive ck %s; keeping %s", v.ck, v.id) + } + return bad + }) m.mu.Lock() keep := expired[:0] var purgeCks, purged []string for _, v := range expired { - if markErr, bad := failed[v.ck]; bad { - logger.Errorf(ctx, markErr, "mark archive ck %s; keeping %s", v.ck, v.id) - continue - } if v.action == reapPurge { if m.claimed[v.id] != v.sb || v.sb.ArchiveCk != v.ck { continue From 6ca7fb8261df3d5f2168ba54d1fb9161d2e8e055 Mon Sep 17 00:00:00 2001 From: CMGS Date: Fri, 6 Nov 2026 18:18:30 +0800 Subject: [PATCH 02/11] pool: retire stale readers through the bounded maintenance fan-out dropStaleReaders ran cocoon vm rm for each stale reader on the Run loop itself, so a wave of wakes with open reader connections stalled refill and liveness behind serialized engine removals. --- sandboxd/pool/reader.go | 16 +++++++--------- sandboxd/pool/reader_test.go | 32 +++++++++++++++++++++++++++++++- 2 files changed, 38 insertions(+), 10 deletions(-) diff --git a/sandboxd/pool/reader.go b/sandboxd/pool/reader.go index 22f4a0887..4e8a29eed 100644 --- a/sandboxd/pool/reader.go +++ b/sandboxd/pool/reader.go @@ -138,20 +138,18 @@ func (m *Manager) dropReader(ctx context.Context, r *reader, force bool) { m.destroyReader(ctx, r) } -func (m *Manager) dropStaleReaders(ctx context.Context) { +func (m *Manager) dropStaleReaders(ctx context.Context) *sync.WaitGroup { rs := m.readers rs.mu.Lock() live := slices.Collect(maps.Values(rs.live)) rs.mu.Unlock() - for _, r := range live { - m.mu.Lock() + m.mu.Lock() + stale := slices.DeleteFunc(live, func(r *reader) bool { sb := m.claimed[r.claim] - stale := sb == nil || sleepSource(sb) != r.source - m.mu.Unlock() - if stale { - m.dropReader(ctx, r, true) - } - } + return sb != nil && sleepSource(sb) == r.source + }) + m.mu.Unlock() + return m.runBounded(ctx, len(stale), func(ctx context.Context, i int) { m.dropReader(ctx, stale[i], true) }) } func (m *Manager) ensureReader(ctx context.Context, sb *types.Sandbox) error { diff --git a/sandboxd/pool/reader_test.go b/sandboxd/pool/reader_test.go index bfebed0cd..8f946ef6f 100644 --- a/sandboxd/pool/reader_test.go +++ b/sandboxd/pool/reader_test.go @@ -663,7 +663,7 @@ func TestReaderWithAConnectionOpenIsRetiredOnceItsClaimWakesOrLeaves(t *testing. if err != nil { t.Fatalf("leave=%v: %v", leave, err) } - m.dropStaleReaders(t.Context()) + m.dropStaleReaders(t.Context()).Wait() if removes, _ := readerLifecycle(eng); !slices.Contains(removes, "sbx-reader-1") { t.Errorf("leave=%v: removes=%v, want the reader destroyed although its connection is open", leave, removes) } @@ -671,6 +671,36 @@ func TestReaderWithAConnectionOpenIsRetiredOnceItsClaimWakesOrLeaves(t *testing. } } +func TestStaleReaderTeardownNeverBlocksTheCaller(t *testing.T) { + eng := newFakeEngine() + m := newReaderManager(t, eng) + sb := mustClaim(t, m, testKey) + hibernateClaim(t, m, sb) + seedReaderVMs(m, testKey, "sbx-reader-1") + open, _, err := m.DialReader(t.Context(), sb.ID, sb.Token) + if err != nil { + t.Fatalf("DialReader: %v", err) + } + defer func() { _ = open.Close() }() + if _, _, _, err = m.WakeAgentSocket(t.Context(), sb.ID, sb.Token); err != nil { + t.Fatalf("wake: %v", err) + } + stall := make(chan struct{}) + eng.mu.Lock() + eng.removeStall = stall + eng.mu.Unlock() + started := make(chan *sync.WaitGroup, 1) + go func() { started <- m.dropStaleReaders(t.Context()) }() + select { + case wg := <-started: + close(stall) + wg.Wait() + case <-time.After(5 * time.Second): + close(stall) + t.Fatal("dropStaleReaders waited for the engine removal") + } +} + func TestFailedPreHibernateSyncSendsReadsToTheWakePath(t *testing.T) { for _, tt := range []struct { name string From 3ed2ff87b8c5759426fa7e33f30c43e475b120df Mon Sep 17 00:00:00 2001 From: CMGS Date: Fri, 6 Nov 2026 18:18:30 +0800 Subject: [PATCH 03/11] sdk/python: treat any connect-phase failure of a redirect candidate as not delivered _claim_refused accepted only a refused connection or a DNS failure, so a candidate on a dead host (EHOSTUNREACH, connect timeout) stopped the redirect walk while the Go SDK moved on; the client now connects explicitly and tags any connect failure, TLS included, as a dial error. --- sdk/python/cocoonsandbox/client.py | 11 ++++++++--- sdk/python/cocoonsandbox/errors.py | 4 ++++ sdk/python/tests/test_fault_matrix.py | 17 +++++++++++++++++ 3 files changed, 29 insertions(+), 3 deletions(-) diff --git a/sdk/python/cocoonsandbox/client.py b/sdk/python/cocoonsandbox/client.py index 0790dc76d..469c0f37d 100644 --- a/sdk/python/cocoonsandbox/client.py +++ b/sdk/python/cocoonsandbox/client.py @@ -5,7 +5,6 @@ import http.client import json import queue -import socket import ssl import threading import time @@ -16,7 +15,7 @@ from .checkpoint import Checkpoint from .conn import Conn, _Closeable, dial_agent, is_quiet, remaining_timeout from .endpoint import _endpoint_url -from .errors import APIError, SandboxTimeout, _error_message, _require_field +from .errors import APIError, SandboxTimeout, _DialError, _error_message, _require_field from .sandbox import Sandbox from .template import _claim_body, _template_query from .volumes import Volume @@ -249,6 +248,12 @@ def _roundtrip( ) -> tuple[int, bytes]: key = (url.scheme, url.netloc) conn = self._take(key, timeout) + if conn.sock is None: + try: + conn.connect() + except OSError as exc: + conn.close() + raise _DialError(str(exc)) from exc status = 0 try: conn.request(method, url.path + (f"?{url.query}" if url.query else ""), body=data, headers=headers) @@ -317,7 +322,7 @@ def _retry_transient(exc: APIError) -> bool: def _claim_refused(exc: APIError) -> bool: if exc.status: return exc.status not in (502, 504) - return exc.__cause__ is None or isinstance(exc.__cause__, (ConnectionRefusedError, socket.gaierror)) + return exc.__cause__ is None or isinstance(exc.__cause__, _DialError) def _redirect_fallback( diff --git a/sdk/python/cocoonsandbox/errors.py b/sdk/python/cocoonsandbox/errors.py index 221bf30ff..3c7a612c1 100644 --- a/sdk/python/cocoonsandbox/errors.py +++ b/sdk/python/cocoonsandbox/errors.py @@ -64,6 +64,10 @@ class ProtocolError(SandboxError): """The peer broke the frame protocol (unexpected frame, oversized line).""" +class _DialError(OSError): + pass + + def _require_field(reply: dict[str, Any], key: str, verb: str) -> Any: if key not in reply: raise APIError(verb, 0, f"reply without {key}") diff --git a/sdk/python/tests/test_fault_matrix.py b/sdk/python/tests/test_fault_matrix.py index 48b4c72bb..6e5790c1b 100644 --- a/sdk/python/tests/test_fault_matrix.py +++ b/sdk/python/tests/test_fault_matrix.py @@ -1,5 +1,7 @@ """Fault matrix for the cluster redirect follow and the lookup scatter.""" +import errno +import http.client import socket import threading import time @@ -23,6 +25,21 @@ def test_claim_redirect_skips_a_failed_candidate(spawn_node, dead_addr, status, assert sb.id == "sb_ok" and sb.owner == good +def test_claim_redirect_skips_an_unreachable_candidate(spawn_node, monkeypatch): + good = spawn_node({("POST", "/v1/claim"): lambda body, path: (200, {"id": "sb_ok", "token": "t"})}) + entry = spawn_node({("POST", "/v1/claim"): lambda body, path: (200, {"redirect": ["10.0.0.9:7777", good]})}) + connect = http.client.HTTPConnection.connect + + def unreachable(conn): + if conn.host == "10.0.0.9": + raise OSError(errno.EHOSTUNREACH, "no route to host") + connect(conn) + + monkeypatch.setattr(http.client.HTTPConnection, "connect", unreachable) + sb = Client(entry).new("rt:24.04") + assert sb.id == "sb_ok" and sb.owner == good + + def test_claim_redirect_stops_when_a_candidate_may_have_claimed(spawn_node): server = socket.create_server(("127.0.0.1", 0)) From 333177dbcc067772f0b23ef0493150b3fcab6ad7 Mon Sep 17 00:00:00 2001 From: CMGS Date: Fri, 6 Nov 2026 18:14:45 +0800 Subject: [PATCH 04/11] sdk/go: rebind a failed upload once WriteFile's retryMoved and uploadRPC's own deferred rebind each ran a full Lookup when the owner was gone, doubling the failure latency; only Push, which never retries, rebinds after its upload now. --- sdk/go/client_test.go | 23 +++++++++++++++++++++++ sdk/go/tree.go | 4 +++- sdk/go/utils.go | 3 +-- 3 files changed, 27 insertions(+), 3 deletions(-) diff --git a/sdk/go/client_test.go b/sdk/go/client_test.go index 4a44bf5e2..d6d40832a 100644 --- a/sdk/go/client_test.go +++ b/sdk/go/client_test.go @@ -204,6 +204,29 @@ func TestAMovedSandboxRetriesAStreamCallAtItsNewOwner(t *testing.T) { } } +func TestAWriteToAGoneSandboxLooksItsOwnerUpOnce(t *testing.T) { + var lookups atomic.Int32 + node := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + switch r.URL.Path { + case "/v1/peers": + _ = json.MarshalWrite(w, map[string]any{"peers": []string{}}) + case "/v1/sandboxes/sb_1/owner": + lookups.Add(1) + w.WriteHeader(http.StatusNotFound) + default: + w.WriteHeader(http.StatusNotFound) + } + })) + t.Cleanup(node.Close) + sb := testSandbox(t, node) + if err := sb.WriteFile(t.Context(), "/f", []byte("x"), nil); err == nil { + t.Fatal("a write to a released sandbox succeeded") + } + if n := lookups.Load(); n != 1 { + t.Errorf("owner probed %d times, want one lookup", n) + } +} + func TestAMovedSandboxDoesNotReplayARunThatStreamedStdin(t *testing.T) { sb, adopter := movedSandbox(t) if _, err := sb.Run(t.Context(), Cmd{Argv: []string{"cat"}, Stdin: strings.NewReader("once")}); err == nil { diff --git a/sdk/go/tree.go b/sdk/go/tree.go index 1f0376a43..280837abc 100644 --- a/sdk/go/tree.go +++ b/sdk/go/tree.go @@ -9,7 +9,9 @@ import ( // Push extracts a tar stream under dest in the sandbox, the no-network lane's way to move a project in. func (s *Sandbox) Push(ctx context.Context, dest string, tarStream io.Reader) error { - return s.uploadRPC(ctx, &wire.FsPush{Dest: dest}, tarStream) + err := s.uploadRPC(ctx, &wire.FsPush{Dest: dest}, tarStream) + s.rebind(ctx, err) + return err } // Pull streams the tree at path back as a tar archive, written to tarStream. diff --git a/sdk/go/utils.go b/sdk/go/utils.go index eb7f053bb..7009c6b7c 100644 --- a/sdk/go/utils.go +++ b/sdk/go/utils.go @@ -33,8 +33,7 @@ func (s *Sandbox) doneRPC(ctx context.Context, req wire.Request) error { } // uploadRPC streams r as full Data frames after req and expects Done; an early terminal frame is the guest rejecting the upload and stops the stream. -func (s *Sandbox) uploadRPC(ctx context.Context, req wire.Request, r io.Reader) (err error) { - defer func() { s.rebind(ctx, err) }() +func (s *Sandbox) uploadRPC(ctx context.Context, req wire.Request, r io.Reader) error { conn, l, err := s.call(ctx, req) if err != nil { return err From 516480e368487290c9b40f3795dbbd0a3e44c606 Mon Sep 17 00:00:00 2001 From: CMGS Date: Fri, 6 Nov 2026 18:17:46 +0800 Subject: [PATCH 05/11] review: silkd pty_name below the public helpers and tokio Child imported (exported above unexported) --- silkd/src/pty.rs | 6 +++--- silkd/src/sysutil.rs | 26 +++++++++++++------------- 2 files changed, 16 insertions(+), 16 deletions(-) diff --git a/silkd/src/pty.rs b/silkd/src/pty.rs index 3c6afd975..e06f2fc63 100644 --- a/silkd/src/pty.rs +++ b/silkd/src/pty.rs @@ -7,7 +7,7 @@ use std::time::Duration; use tokio::io::AsyncWrite; use tokio::io::unix::AsyncFd; -use tokio::process::Command; +use tokio::process::{Child, Command}; use tokio::sync::{mpsc, oneshot}; use tokio::time::{Instant, sleep_until, timeout}; @@ -120,7 +120,7 @@ async fn pump( proc: &Proc, client: mpsc::Receiver, out: &mut W, - child: &mut tokio::process::Child, + child: &mut Child, ) -> i32 { // stdin rides its own task so a slow write cannot stall output or reaping. let (disc_tx, mut disc_rx) = oneshot::channel::<()>(); @@ -176,7 +176,7 @@ async fn pump( } } -fn hang_up(child: &tokio::process::Child) -> Instant { +fn hang_up(child: &Child) -> Instant { if let Some(pid) = child.id() { sysutil::signal_group(pid, libc::SIGHUP); } diff --git a/silkd/src/sysutil.rs b/silkd/src/sysutil.rs index 9174ce853..e6af01fad 100644 --- a/silkd/src/sysutil.rs +++ b/silkd/src/sysutil.rs @@ -151,19 +151,6 @@ pub fn openpty(cols: u16, rows: u16) -> std::io::Result<(OwnedFd, OwnedFd)> { Ok((master, slave)) } -#[cfg(target_os = "linux")] -fn pty_name(master: &nix::pty::PtyMaster) -> nix::Result { - nix::pty::ptsname_r(master) -} - -#[cfg(not(target_os = "linux"))] -fn pty_name(master: &nix::pty::PtyMaster) -> nix::Result { - static PTSNAME: Mutex<()> = Mutex::new(()); - let _serial = lock(&PTSNAME); - // SAFETY: PTSNAME serializes every ptsname call in silkd, so its static buffer is copied before another call reuses it. - unsafe { nix::pty::ptsname(master) } -} - pub fn watch_fd(fd: T) -> std::io::Result> { // SAFETY: AsFd's I/O-safety contract keeps the raw fd open while `fd`, and so the AsyncFd, lives. unsafe { AsyncFd::register(fd) }.map_err(|e| e.into_parts().1) @@ -382,6 +369,19 @@ fn is_valid_pid(id: u32) -> bool { id != 0 && id <= i32::MAX as u32 } +#[cfg(target_os = "linux")] +fn pty_name(master: &nix::pty::PtyMaster) -> nix::Result { + nix::pty::ptsname_r(master) +} + +#[cfg(not(target_os = "linux"))] +fn pty_name(master: &nix::pty::PtyMaster) -> nix::Result { + static PTSNAME: Mutex<()> = Mutex::new(()); + let _serial = lock(&PTSNAME); + // SAFETY: PTSNAME serializes every ptsname call in silkd, so its static buffer is copied before another call reuses it. + unsafe { nix::pty::ptsname(master) } +} + fn send_signal(target: libc::pid_t, sig: i32) { // SAFETY: kill(2) takes no pointers; every caller vets the target with is_valid_pid first. unsafe { libc::kill(target, sig) }; From f1ac98387dc7755b3ac14789fe09e2fabbc80084 Mon Sep 17 00:00:00 2001 From: CMGS Date: Fri, 6 Nov 2026 18:17:46 +0800 Subject: [PATCH 06/11] mcp, config: name the repeated sandbox_id description once, and give PoolSpec one receiver kind --- mcp/tools.go | 28 +++++++++++++++------------- sandboxd/config/config.go | 2 +- 2 files changed, 16 insertions(+), 14 deletions(-) diff --git a/mcp/tools.go b/mcp/tools.go index c76e09cd1..9921d3b1a 100644 --- a/mcp/tools.go +++ b/mcp/tools.go @@ -13,6 +13,8 @@ import ( sandbox "github.com/cocoonstack/sandbox/sdk/go" ) +const sandboxIDDesc = "id returned by create_sandbox, fork, or branch_checkpoint" + var ( errOutputCap = errors.New("output past the cap") @@ -23,43 +25,43 @@ var ( }, { "exec", "Run a shell command in a sandbox, wait for it to exit, and return stdout, stderr, and the exit code as JSON. The call is cut off after 5 minutes; for servers or long jobs use spawn instead. A failed or cut-off call still returns the output collected so far next to an error field. Each stream keeps its first 1 MiB and sets truncated beyond that. A hibernated sandbox wakes transparently on this call.", - schema(props{"sandbox_id": str("id returned by create_sandbox, fork, or branch_checkpoint"), "command": str("shell command, run via sh -c"), "cwd": str("working directory; empty runs in the guest's default")}, "sandbox_id", "command"), toolExec, + schema(props{"sandbox_id": str(sandboxIDDesc), "command": str("shell command, run via sh -c"), "cwd": str("working directory; empty runs in the guest's default")}, "sandbox_id", "command"), toolExec, }, { "spawn", "Start a shell command detached in a sandbox and return its guest pid immediately. The process keeps running across later tool calls; its output goes to a per-process ring buffer that keeps the newest 256 KiB, which logs replays. Use exec instead when you need the result now.", - schema(props{"sandbox_id": str("id returned by create_sandbox, fork, or branch_checkpoint"), "command": str("shell command, run via sh -c"), "cwd": str("working directory; empty runs in the guest's default")}, "sandbox_id", "command"), toolSpawn, + schema(props{"sandbox_id": str(sandboxIDDesc), "command": str("shell command, run via sh -c"), "cwd": str("working directory; empty runs in the guest's default")}, "sandbox_id", "command"), toolSpawn, }, { "ps", "List the sandbox's tracked processes (exec, spawn, and pty commands) as JSON: pid, argv, detached, state (running or exited), exit_code once exited, and start time. Processes started inside the guest by other means are not listed.", - schema(props{"sandbox_id": str("id returned by create_sandbox, fork, or branch_checkpoint")}, "sandbox_id"), toolPs, + schema(props{"sandbox_id": str(sandboxIDDesc)}, "sandbox_id"), toolPs, }, { "kill", "Send a signal to a tracked process. Killing a process that already exited is a no-op success; the guest never re-signals a reaped pid.", - schema(props{"sandbox_id": str("id returned by create_sandbox, fork, or branch_checkpoint"), "pid": integer("guest pid from spawn or ps"), "signal": integer("signal number, e.g. 15 for SIGTERM; 0 sends SIGKILL")}, "sandbox_id", "pid"), toolKill, + schema(props{"sandbox_id": str(sandboxIDDesc), "pid": integer("guest pid from spawn or ps"), "signal": integer("signal number, e.g. 15 for SIGTERM; 0 sends SIGKILL")}, "sandbox_id", "pid"), toolKill, }, { "logs", "Return a tracked process's buffered stdout and stderr, plus exit_code once it has ended. The buffer keeps the newest 256 KiB per process, so redirect long output to a file when it must be complete.", - schema(props{"sandbox_id": str("id returned by create_sandbox, fork, or branch_checkpoint"), "pid": integer("guest pid from spawn or ps")}, "sandbox_id", "pid"), toolLogs, + schema(props{"sandbox_id": str(sandboxIDDesc), "pid": integer("guest pid from spawn or ps")}, "sandbox_id", "pid"), toolLogs, }, { "write_file", "Write text content to a file in a sandbox, replacing any existing file. The write is atomic (temp file plus rename); an existing file keeps its permission bits. The parent directory must already exist.", - schema(props{"sandbox_id": str("id returned by create_sandbox, fork, or branch_checkpoint"), "path": str("absolute path of the file"), "content": str("full file content; written verbatim")}, "sandbox_id", "path", "content"), toolWriteFile, + schema(props{"sandbox_id": str(sandboxIDDesc), "path": str("absolute path of the file"), "content": str("full file content; written verbatim")}, "sandbox_id", "path", "content"), toolWriteFile, }, { "read_file", "Return the whole content of a regular file up to 1 MiB in a sandbox as text; bytes that are not valid UTF-8 are replaced, so binary content is lossy. A larger file, a directory, or a device is an error: use exec with head, tail, or a checksum instead. A missing path is an error.", - schema(props{"sandbox_id": str("id returned by create_sandbox, fork, or branch_checkpoint"), "path": str("absolute path of the file")}, "sandbox_id", "path"), toolReadFile, + schema(props{"sandbox_id": str(sandboxIDDesc), "path": str("absolute path of the file")}, "sandbox_id", "path"), toolReadFile, }, { "list_dir", "List one directory in a sandbox (not recursive) as JSON entries with name, kind (file, dir, symlink, or other), and size in bytes.", - schema(props{"sandbox_id": str("id returned by create_sandbox, fork, or branch_checkpoint"), "path": str("absolute path of the directory")}, "sandbox_id", "path"), toolListDir, + schema(props{"sandbox_id": str(sandboxIDDesc), "path": str("absolute path of the directory")}, "sandbox_id", "path"), toolListDir, }, { "fork", "Clone a sandbox into N independent children that start from its exact memory and disk state, including running processes; each child gets its own id and lives one hour. N is capped by the node's max_fork_count (default 16). All-or-nothing: on failure no child survives. The parent keeps running. An egress-lane sandbox cannot fork.", - schema(props{"sandbox_id": str("id returned by create_sandbox, fork, or branch_checkpoint"), "count": integer("number of children, 1 to the node's max_fork_count (default 16)")}, "sandbox_id", "count"), toolFork, + schema(props{"sandbox_id": str(sandboxIDDesc), "count": integer("number of children, 1 to the node's max_fork_count (default 16)")}, "sandbox_id", "count"), toolFork, }, { "checkpoint", "Capture a sandbox's full state (memory, disk, running processes) without stopping it and return a checkpoint id. branch_checkpoint claims new sandboxes from that moment any number of times; a checkpoint outlives this session until it is deleted or expires under the node's checkpoint TTL. An egress-lane sandbox cannot checkpoint.", - schema(props{"sandbox_id": str("id returned by create_sandbox, fork, or branch_checkpoint"), "name": str("optional human label")}, "sandbox_id"), toolCheckpoint, + schema(props{"sandbox_id": str(sandboxIDDesc), "name": str("optional human label")}, "sandbox_id"), toolCheckpoint, }, { "branch_checkpoint", "Claim a fresh sandbox that resumes from a checkpoint's exact captured moment and return its id; it lives one hour. The checkpoint is unchanged and can be branched again.", @@ -69,13 +71,13 @@ var ( {"delete_checkpoint", "Delete the node's copy of a checkpoint and tell peers to drop theirs; a peer replica that misses that broadcast stays branchable until its own TTL sweep. Sandboxes already branched from it are unaffected.", schema(props{"checkpoint_id": str("id returned by checkpoint or list_checkpoints")}, "checkpoint_id"), toolDeleteCheckpoint}, { "hibernate", "Snapshot a sandbox and stop its VM, freeing memory while keeping its id, files, processes, and shell state. The next call that reaches the guest (exec, spawn, ps, kill, logs, the file tools) wakes it transparently in tens of milliseconds; fork, checkpoint, promote and release act on the snapshot without waking it. An egress-lane sandbox cannot hibernate.", - schema(props{"sandbox_id": str("id returned by create_sandbox, fork, or branch_checkpoint")}, "sandbox_id"), toolHibernate, + schema(props{"sandbox_id": str(sandboxIDDesc)}, "sandbox_id"), toolHibernate, }, { "promote", "Publish the sandbox's current state as a named template on its node; later create_sandbox calls with that name (and the same net and size) start from it. Re-promoting to the same name replaces the template. An egress-lane sandbox cannot promote.", - schema(props{"sandbox_id": str("id returned by create_sandbox, fork, or branch_checkpoint"), "template_name": str("template name to publish as")}, "sandbox_id", "template_name"), toolPromote, + schema(props{"sandbox_id": str(sandboxIDDesc), "template_name": str("template name to publish as")}, "sandbox_id", "template_name"), toolPromote, }, - {"release", "Destroy a sandbox and free its resources; files and processes inside it are lost. This session forgets the id, so a second release of it is rejected as unknown.", schema(props{"sandbox_id": str("id returned by create_sandbox, fork, or branch_checkpoint")}, "sandbox_id"), toolRelease}, + {"release", "Destroy a sandbox and free its resources; files and processes inside it are lost. This session forgets the id, so a second release of it is rejected as unknown.", schema(props{"sandbox_id": str(sandboxIDDesc)}, "sandbox_id"), toolRelease}, {"node_info", "Report the connected node's warm pools, promoted templates, live claims, drain state, capacity, and mesh peers as JSON. Needs the operator token; a tenant token is refused.", schema(props{}), toolNodeInfo}, } ) diff --git a/sandboxd/config/config.go b/sandboxd/config/config.go index 64c14c220..dcbd9e895 100644 --- a/sandboxd/config/config.go +++ b/sandboxd/config/config.go @@ -97,7 +97,7 @@ func (s *PoolSpec) UnmarshalJSONFrom(dec *jsontext.Decoder) error { } // ValidateLimits checks the warm/watermark/idle bounds shared by config and PUT /v1/pools. -func (s PoolSpec) ValidateLimits() error { +func (s *PoolSpec) ValidateLimits() error { if s.Warm < 0 { return fmt.Errorf("warm must not be negative") } From 7cc410143fba037a73600b487b01a8fa1189d212 Mon Sep 17 00:00:00 2001 From: CMGS Date: Fri, 6 Nov 2026 18:17:46 +0800 Subject: [PATCH 07/11] e2e, mcp, pool, sdk/go: one-line package docs for the drivers and drop godocs that restate their code --- e2e/cmd/androidsmoke/main.go | 5 +---- e2e/cmd/browsersmoke/main.go | 4 +--- e2e/cmd/coldproof/main.go | 5 +---- e2e/cmd/credsmoke/main.go | 4 +--- e2e/cmd/crossnode/main.go | 6 +----- e2e/cmd/demo/main.go | 3 +-- e2e/cmd/guestserver/main.go | 8 +------- e2e/cmd/interceptsmoke/main.go | 6 +----- e2e/cmd/lifecycle/main.go | 6 +----- e2e/cmd/meshsmoke/main.go | 6 +----- e2e/cmd/portsmoke/main.go | 5 +---- e2e/cmd/pullbench/main.go | 4 +--- e2e/cmd/pushbench/main.go | 3 +-- e2e/cmd/qaab/main.go | 3 +-- e2e/cmd/ringsmoke/main.go | 4 +--- e2e/cmd/smoke/main.go | 5 +---- e2e/cmd/sockssmoke/main.go | 8 +------- e2e/cmd/volumesmoke/main.go | 3 +-- mcp/main.go | 5 +---- sandboxd/pool/watermark.go | 3 --- sdk/go/port.go | 1 - 21 files changed, 19 insertions(+), 78 deletions(-) diff --git a/e2e/cmd/androidsmoke/main.go b/e2e/cmd/androidsmoke/main.go index e5d9aaeb4..8efa5af0d 100644 --- a/e2e/cmd/androidsmoke/main.go +++ b/e2e/cmd/androidsmoke/main.go @@ -1,7 +1,4 @@ -// androidsmoke is the android-flavor acceptance: claim (none/xlarge) → Android -// init tree over the relay → adb CNXN handshake on the adb port → -// checkpoint/branch of the booted guest. No session step: the guest -// ships no bash. +// androidsmoke checks the android flavor: claim, the init tree over the relay, the adb handshake, and checkpoint/branch. package main import ( diff --git a/e2e/cmd/browsersmoke/main.go b/e2e/cmd/browsersmoke/main.go index fe9afe4b3..b735ba531 100644 --- a/e2e/cmd/browsersmoke/main.go +++ b/e2e/cmd/browsersmoke/main.go @@ -1,6 +1,4 @@ -// browsersmoke is the browser-flavor acceptance: claim the browser flavor → CDP -// /json/version over the relay → open a target → checkpoint/branch of the -// warmed browser. +// browsersmoke checks the browser flavor: claim, CDP /json/version over the relay, a new target, and checkpoint/branch. package main import ( diff --git a/e2e/cmd/coldproof/main.go b/e2e/cmd/coldproof/main.go index a1d4f26cd..3d2651dea 100644 --- a/e2e/cmd/coldproof/main.go +++ b/e2e/cmd/coldproof/main.go @@ -1,7 +1,4 @@ -// coldproof measures a cold claim end-to-end and proves the guest is live: -// it claims an unpooled template (RunCold path), then execs inside the guest -// and prints the guest's own /proc/uptime and kernel — the guest-side clock -// is the evidence that claim-ready means a booted, exec-serving system. +// coldproof times a cold claim and prints the guest uptime and kernel as proof it booted. package main import ( diff --git a/e2e/cmd/credsmoke/main.go b/e2e/cmd/credsmoke/main.go index c1e1b94af..b968e7a1b 100644 --- a/e2e/cmd/credsmoke/main.go +++ b/e2e/cmd/credsmoke/main.go @@ -1,6 +1,4 @@ -// credsmoke is the claim-credential acceptance: claims each send their own env-injected header, -// query or body credential through the proxy (-inject-size: on an "inject" pool), the guest never -// holds the value, and rotation, refusal, clearing and forks behave as the API says. +// credsmoke checks claim credentials end to end through the proxy; -inject-size names an "inject" pool. package main import ( diff --git a/e2e/cmd/crossnode/main.go b/e2e/cmd/crossnode/main.go index c3a00ff30..92a1bcddc 100644 --- a/e2e/cmd/crossnode/main.go +++ b/e2e/cmd/crossnode/main.go @@ -1,8 +1,4 @@ -// crossnode is the two-node store acceptance: a checkpoint created on node -// A is claimed on node B through the shared checkpoint store; run it with a -// template absent from B's image store so the claim exercises the engine's -// clone --pull path (base blobs resolved by digest). The marker written on -// A proves the memory state crossed nodes. +// crossnode claims on node B a checkpoint made on node A through the shared store; use a template absent from B's image store. package main import ( diff --git a/e2e/cmd/demo/main.go b/e2e/cmd/demo/main.go index b172d864e..05424d8c0 100644 --- a/e2e/cmd/demo/main.go +++ b/e2e/cmd/demo/main.go @@ -1,5 +1,4 @@ -// demo exercises a live sandboxd through the SDK and prints per-iteration -// latencies — the bare-metal e2e script drives it on a real node. +// demo claims, execs and releases through the SDK and prints per-iteration latencies. package main import ( diff --git a/e2e/cmd/guestserver/main.go b/e2e/cmd/guestserver/main.go index 66bb4f6f4..efaf6de58 100644 --- a/e2e/cmd/guestserver/main.go +++ b/e2e/cmd/guestserver/main.go @@ -1,10 +1,4 @@ -// guestserver is the in-guest listener the port-relay e2e drives: a loopback -// HTTP server that reports the protocol it was reached over. It stands in for -// any guest daemon during hardware tests — the relay carries bytes, so proving -// HTTP/1.1 and h2c both survive it needs a server that speaks both. -// -// It is uploaded into a claimed sandbox by portsmoke; nothing in the stock -// image provides an HTTP listener. +// guestserver is the in-guest HTTP/1.1 and h2c listener portsmoke uploads to test the port relay. package main import ( diff --git a/e2e/cmd/interceptsmoke/main.go b/e2e/cmd/interceptsmoke/main.go index 6ee7f5817..0701ccad1 100644 --- a/e2e/cmd/interceptsmoke/main.go +++ b/e2e/cmd/interceptsmoke/main.go @@ -1,8 +1,4 @@ -// interceptsmoke is the HTTPS-interception acceptance: a none-lane sandbox whose -// golden had the cluster root baked reaches an HTTPS echo through the proxy; the -// leaf the node's intermediate signed validates against that baked root, the TLS -// is terminated (so the guest sees our issuer, not the origin's), and the secret -// is injected into the HTTPS request — the guarantees plaintext already had. +// interceptsmoke checks HTTPS interception: a none-lane guest with the baked cluster root reaches an HTTPS echo with the secret injected. package main import ( diff --git a/e2e/cmd/lifecycle/main.go b/e2e/cmd/lifecycle/main.go index 69758f817..4f41251ab 100644 --- a/e2e/cmd/lifecycle/main.go +++ b/e2e/cmd/lifecycle/main.go @@ -1,7 +1,4 @@ -// lifecycle exercises sandboxd's auto-archive: a claim idles into hibernation, -// then to a store checkpoint with its local VM dropped, and the next access -// wakes it from the store with the same id and guest state intact. The node -// must run a pool with idle_hibernate_seconds + archive_after_seconds set. +// lifecycle checks that an idle claim archives and wakes with its id and guest state; the pool needs idle_hibernate_seconds and archive_after_seconds. package main import ( @@ -82,7 +79,6 @@ func run(addr, token, template, netShape string, wait time.Duration) error { return nil } -// waitArchived returns how long the reaper's idle→hibernate→archive ladder took to archive the claim. func waitArchived(ctx context.Context, client *sandbox.Client, timeout time.Duration) (time.Duration, error) { start := time.Now() deadline := start.Add(timeout) diff --git a/e2e/cmd/meshsmoke/main.go b/e2e/cmd/meshsmoke/main.go index ead451382..a2d4a08b5 100644 --- a/e2e/cmd/meshsmoke/main.go +++ b/e2e/cmd/meshsmoke/main.go @@ -1,8 +1,4 @@ -// meshsmoke proves cluster template routing on a real two-node mesh: -// a claim at the entry node redirects to the peer that pools the key, a -// promote there is reachable by name from the entry node, and a name-based -// delete follows gossip to the owner. Run against node A of a two-node -// cluster where only node B pools the template. +// meshsmoke checks template routing on a two-node mesh; run it against node A where only node B pools the template. package main import ( diff --git a/e2e/cmd/portsmoke/main.go b/e2e/cmd/portsmoke/main.go index 67f0b3ebd..affe92cc4 100644 --- a/e2e/cmd/portsmoke/main.go +++ b/e2e/cmd/portsmoke/main.go @@ -1,7 +1,4 @@ -// portsmoke proves sandboxd's guest-port relay on real hardware: it starts a -// listener inside a claimed microVM and drives -// GET /v1/sandboxes/{id}/ports/{port} end to end — the path an edge proxy -// takes to reach a sandbox that has no NIC. Run by scripts/port-e2e.sh. +// portsmoke checks the guest-port relay end to end against a claimed microVM; scripts/port-e2e.sh runs it. package main import ( diff --git a/e2e/cmd/pullbench/main.go b/e2e/cmd/pullbench/main.go index e71fbdceb..c4be508b6 100644 --- a/e2e/cmd/pullbench/main.go +++ b/e2e/cmd/pullbench/main.go @@ -1,6 +1,4 @@ -// pullbench measures fs_pull throughput end to end — the hardware A/B for -// silkd's chunk-frame render path: writes size MiB of random bytes in the -// guest, then times n sequential Pulls. +// pullbench times n sequential fs_pull transfers of size MiB from the guest. package main import ( diff --git a/e2e/cmd/pushbench/main.go b/e2e/cmd/pushbench/main.go index 17543bffe..a1fa7945c 100644 --- a/e2e/cmd/pushbench/main.go +++ b/e2e/cmd/pushbench/main.go @@ -1,5 +1,4 @@ -// pushbench measures fs_write throughput end to end — the hardware A/B for -// the SDK's data-frame encode path: writes size MiB to the guest n times. +// pushbench times n fs_write transfers of size MiB to the guest. package main import ( diff --git a/e2e/cmd/qaab/main.go b/e2e/cmd/qaab/main.go index d001f4598..76a174853 100644 --- a/e2e/cmd/qaab/main.go +++ b/e2e/cmd/qaab/main.go @@ -1,5 +1,4 @@ -// qaab measures one A/B arm against a live sandboxd: a concurrent cold-claim -// burst, then a concurrent in-guest write burst, reported as one JSON row. +// qaab measures one A/B arm: a concurrent cold-claim burst, then a concurrent in-guest write burst, as one JSON row. package main import ( diff --git a/e2e/cmd/ringsmoke/main.go b/e2e/cmd/ringsmoke/main.go index e2e0a316f..5263d3a8e 100644 --- a/e2e/cmd/ringsmoke/main.go +++ b/e2e/cmd/ringsmoke/main.go @@ -1,6 +1,4 @@ -// ringsmoke drives two silkd paths the other smokes leave alone: a detached -// process that overflows its output ring one byte at a time, whose logs must -// replay exactly the newest 256 KiB, and exec as a named user. +// ringsmoke checks the replay of a detached process's 256 KiB output ring and exec as a named user. package main import ( diff --git a/e2e/cmd/smoke/main.go b/e2e/cmd/smoke/main.go index 4d8bbe3cd..a2da9747d 100644 --- a/e2e/cmd/smoke/main.go +++ b/e2e/cmd/smoke/main.go @@ -1,7 +1,4 @@ -// smoke drives the full SDK v2 surface against one live sandbox: files, -// sessions, find/replace, watch, git, and pty — the bare-metal proof that -// silkd v2 works end to end through the relay. Run by scripts/sandboxd-e2e.sh -// on a real node. +// smoke drives the SDK surface (files, sessions, find/replace, watch, git, pty) against one live sandbox; scripts/sandboxd-e2e.sh runs it. package main import ( diff --git a/e2e/cmd/sockssmoke/main.go b/e2e/cmd/sockssmoke/main.go index a2c5c8e01..f362db16b 100644 --- a/e2e/cmd/sockssmoke/main.go +++ b/e2e/cmd/sockssmoke/main.go @@ -1,10 +1,4 @@ -// Command sockssmoke proves the SOCKS5 egress channel from inside a claimed -// sandbox: an allowed host tunnels over 127.0.0.1:1080, a GET-only host is -// reachable on the HTTP proxy but refused on SOCKS5, an unlisted host and an -// unlisted port are refused, IMAPS rides the tunnel on its listed port, the -// HTTP door refuses unlisted ports on CONNECT and forward requests, and no -// tunnel opens from a pool without a policy or from one whose policy did not -// opt in. +// sockssmoke checks the SOCKS5 egress door from a claimed sandbox against the allow, refuse and opt-in rules of its pool policy. package main import ( diff --git a/e2e/cmd/volumesmoke/main.go b/e2e/cmd/volumesmoke/main.go index f667a1fe1..ec4021df9 100644 --- a/e2e/cmd/volumesmoke/main.go +++ b/e2e/cmd/volumesmoke/main.go @@ -1,5 +1,4 @@ -// volumesmoke validates shared read-only dataset mounts, and optionally one -// writable dataset mount and live volume changes, on a live Cloud Hypervisor node. +// volumesmoke checks read-only dataset mounts and, optionally, a writable mount and live volume changes. package main import ( diff --git a/mcp/main.go b/mcp/main.go index 32960183f..47df6b041 100644 --- a/mcp/main.go +++ b/mcp/main.go @@ -1,7 +1,4 @@ -// sandbox-mcp is a Model Context Protocol server over stdio: it exposes the -// sandbox surface (claim, exec, files, fork, checkpoint, promote, hibernate) -// as MCP tools. One process serves one sandboxd endpoint, configured by -// flags or environment. +// sandbox-mcp serves the sandbox surface as Model Context Protocol tools over stdio, one sandboxd endpoint per process. package main import ( diff --git a/sandboxd/pool/watermark.go b/sandboxd/pool/watermark.go index 5b5815d9c..5aa34c670 100644 --- a/sandboxd/pool/watermark.go +++ b/sandboxd/pool/watermark.go @@ -13,7 +13,6 @@ import ( ) const ( - // ewmaAlpha weights the newest observation in the arrival-rate and provision-lead averages ewmaAlpha = 0.3 rateBin = time.Second // rateDecayTau halves the observed arrival rate roughly every 40s of silence @@ -153,7 +152,6 @@ func (p *pool) noteLead(d time.Duration) { p.lead = time.Duration(ewmaAlpha*float64(d) + (1-ewmaAlpha)*float64(p.lead)) } -// effectiveTarget raises the floor toward warmMax by rate × lead. func (p *pool) effectiveTarget(now time.Time) int { if p.warmMax <= 0 { return p.floor @@ -172,7 +170,6 @@ func (p *pool) effectiveTarget(now time.Time) int { return max(p.floor, min(dynamic, p.warmMax)) } -// shrink trims the pool to its target once the warm count has stayed above it for warmShrinkDwell, returning the trimmed VM names. func (p *pool) shrink(now time.Time) []string { target := p.effectiveTarget(now) switch { diff --git a/sdk/go/port.go b/sdk/go/port.go index 4365d6468..680f3b6ea 100644 --- a/sdk/go/port.go +++ b/sdk/go/port.go @@ -110,7 +110,6 @@ func (s *Sandbox) ProxyPort(ctx context.Context, localAddr string, port uint16) return l, nil } -// openStream turns a call into a raw byte stream. func (s *Sandbox) openStream(ctx context.Context, req wire.Request) (*PortConn, error) { var p *PortConn err := s.retryMoved(ctx, func() error { From decb16792a2c3e31550cc14bf9918edac91d7020 Mon Sep 17 00:00:00 2001 From: CMGS Date: Fri, 6 Nov 2026 18:35:36 +0800 Subject: [PATCH 08/11] sdk/python: a refused owner still counts as gone once connect failures are tagged Wrapping connect failures in _DialError hid the ConnectionRefusedError that _owner_answer checks, so fork, renew, hibernate, checkpoint, promote, preview and volume calls stopped following a sandbox off a retired node. --- sdk/python/cocoonsandbox/sandbox.py | 4 +++- sdk/python/tests/test_fault_matrix.py | 10 +++++++--- 2 files changed, 10 insertions(+), 4 deletions(-) diff --git a/sdk/python/cocoonsandbox/sandbox.py b/sdk/python/cocoonsandbox/sandbox.py index ee50e49e3..8bbc11618 100644 --- a/sdk/python/cocoonsandbox/sandbox.py +++ b/sdk/python/cocoonsandbox/sandbox.py @@ -21,6 +21,7 @@ SandboxTimeout, SilkdError, VolumeError, + _DialError, _mark_gone, _require_field, ) @@ -849,6 +850,7 @@ def _drain_data(conn: Conn) -> bytes: def _owner_answer(exc: APIError, tag: tuple[str, str]) -> APIError: - if exc.status == 404 or isinstance(exc.__cause__, ConnectionRefusedError): + cause = exc.__cause__.__cause__ if isinstance(exc.__cause__, _DialError) else exc.__cause__ + if exc.status == 404 or isinstance(cause, ConnectionRefusedError): return _mark_gone(exc, tag) return exc diff --git a/sdk/python/tests/test_fault_matrix.py b/sdk/python/tests/test_fault_matrix.py index 6e5790c1b..5ee137f03 100644 --- a/sdk/python/tests/test_fault_matrix.py +++ b/sdk/python/tests/test_fault_matrix.py @@ -228,15 +228,19 @@ def release(body, path): assert released -def test_a_moved_sandbox_retries_a_control_call_at_its_new_owner(spawn_node): +@pytest.mark.parametrize("retired", [False, True], ids=["old_node_404", "old_node_retired"]) +def test_a_moved_sandbox_retries_a_control_call_at_its_new_owner(spawn_node, dead_addr, retired): adopter = spawn_node( { ("GET", "/v1/sandboxes/sb_1/owner"): lambda body, path: (200, {}), ("POST", "/v1/sandboxes/sb_1/renew"): lambda body, path: (200, {"deadline": "2026-11-06T12:00:00Z"}), } ) - old = spawn_node({("GET", "/v1/peers"): lambda body, path: (200, {"peers": [adopter]})}) - sb = Client(old).attach(old, "sb_1", "tok") + if retired: + entry, old = adopter, dead_addr + else: + entry = old = spawn_node({("GET", "/v1/peers"): lambda body, path: (200, {"peers": [adopter]})}) + sb = Client(entry).attach(old, "sb_1", "tok") assert sb.renew(60) == "2026-11-06T12:00:00Z" assert sb.owner == adopter From c7c554f5614b3d244261f138def634f669ed0513 Mon Sep 17 00:00:00 2001 From: CMGS Date: Fri, 6 Nov 2026 18:37:03 +0800 Subject: [PATCH 09/11] silkd: share one compiled regex across a replace and reuse fs.read's buffers fs.replace cloned the Regex per file, and a regex clone builds a fresh cache pool, so every file restarted the lazy DFA cold; an Arc shares one pool across the blocking tasks. fs.read dropped its full head buffer and frame and stream_data_frames allocated new ones, about 600 KiB per large read; both are passed in now. --- silkd/src/find.rs | 8 ++++++-- silkd/src/fs.rs | 9 ++++++--- silkd/src/proto.rs | 8 ++++---- silkd/src/tree.rs | 6 ++++-- 4 files changed, 20 insertions(+), 11 deletions(-) diff --git a/silkd/src/find.rs b/silkd/src/find.rs index 66e5e0c70..6b4d63bca 100644 --- a/silkd/src/find.rs +++ b/silkd/src/find.rs @@ -168,12 +168,16 @@ pub async fn replace( replacement: String, ) -> std::io::Result<()> { let re = match compile(pattern).await { - Ok(re) => re, + Ok(re) => Arc::new(re), Err(e) => return proto::error_frame(w, ErrorKind::BadRequest, e).await, }; let replacement: Arc = replacement.into(); for file in files { - let (re, replacement, path) = (re.clone(), Arc::clone(&replacement), PathBuf::from(&file)); + let (re, replacement, path) = ( + Arc::clone(&re), + Arc::clone(&replacement), + PathBuf::from(&file), + ); let replaced = tokio::task::spawn_blocking(move || { let mut body = String::new(); if !read_bounded(&path, &mut body)? { diff --git a/silkd/src/fs.rs b/silkd/src/fs.rs index 24e0265e2..0d45db712 100644 --- a/silkd/src/fs.rs +++ b/silkd/src/fs.rs @@ -67,7 +67,7 @@ pub async fn read(w: &mut W, path: String) -> io::Result< }) .await .map_err(io::Error::other)?; - let (file, head) = match opened { + let (file, mut head) = match opened { Ok(Some((file, Ok(head)))) => (file, head), Ok(Some((_, Err(e)))) => return err_frame(w, &e, "read").await, Ok(None) => { @@ -75,11 +75,14 @@ pub async fn read(w: &mut W, path: String) -> io::Result< } Err(e) => return err_frame(w, &e, "open").await, }; + let mut frame = Vec::new(); if !head.is_empty() { - proto::write_chunk_frame(w, &mut Vec::new(), "data", &head).await?; + proto::write_chunk_frame(w, &mut frame, "data", &head).await?; } if head.len() == BULK_CHUNK - && let Err(e) = proto::stream_data_frames(&mut fs::File::from_std(file), w).await? + && let Err(e) = + proto::stream_data_frames(&mut fs::File::from_std(file), w, &mut head, &mut frame) + .await? { return err_frame(w, &e, "read").await; } diff --git a/silkd/src/proto.rs b/silkd/src/proto.rs index f9f4da1a1..5a282ab54 100644 --- a/silkd/src/proto.rs +++ b/silkd/src/proto.rs @@ -511,17 +511,17 @@ pub async fn error_frame( pub async fn stream_data_frames( reader: &mut R, w: &mut W, + buf: &mut [u8], + frame: &mut Vec, ) -> io::Result> where R: AsyncRead + Unpin, W: AsyncWrite + Unpin, { - let mut buf = vec![0u8; BULK_CHUNK]; - let mut frame = Vec::new(); loop { - match reader.read(&mut buf).await { + match reader.read(buf).await { Ok(0) => return Ok(Ok(())), - Ok(n) => write_chunk_frame(w, &mut frame, "data", &buf[..n]).await?, + Ok(n) => write_chunk_frame(w, frame, "data", &buf[..n]).await?, Err(e) => return Ok(Err(e)), } } diff --git a/silkd/src/tree.rs b/silkd/src/tree.rs index 7fd1b8ded..7e846b28c 100644 --- a/silkd/src/tree.rs +++ b/silkd/src/tree.rs @@ -12,7 +12,7 @@ use nix::unistd::{self, UnlinkatFlags}; use tokio::io::{AsyncBufRead, AsyncReadExt, AsyncWrite}; use tokio::process::Command; -use crate::proto::{self, ErrorKind, Response, err_frame}; +use crate::proto::{self, BULK_CHUNK, ErrorKind, Response, err_frame}; use crate::sysutil; const STDERR_CAP: usize = 16 * 1024; @@ -73,7 +73,9 @@ pub async fn pull(w: &mut W, path: &str) -> io::Result<() Err(e) => return err_frame(w, &e, "spawn tar").await, }; - if let Err(e) = proto::stream_data_frames(&mut out, w).await? { + if let Err(e) = + proto::stream_data_frames(&mut out, w, &mut vec![0u8; BULK_CHUNK], &mut Vec::new()).await? + { let _ = child.wait().await; return err_frame(w, &e, "read tar").await; } From 79af8d2d040c30e068541cead6db8c7ab608610e Mon Sep 17 00:00:00 2001 From: CMGS Date: Fri, 6 Nov 2026 18:38:37 +0800 Subject: [PATCH 10/11] review: import the tokio process types silkd spells out, drop restating comments, trim the smoke doc and the single-lookup test --- e2e/cmd/smoke/main.go | 2 +- sandboxd/pool/watermark.go | 1 - sdk/go/client_test.go | 9 ++------- silkd/src/exec.rs | 11 ++++------- silkd/src/pty.rs | 1 - silkd/src/sysutil.rs | 4 ++-- silkd/src/tree.rs | 8 +++----- 7 files changed, 12 insertions(+), 24 deletions(-) diff --git a/e2e/cmd/smoke/main.go b/e2e/cmd/smoke/main.go index a2da9747d..1c45e08f6 100644 --- a/e2e/cmd/smoke/main.go +++ b/e2e/cmd/smoke/main.go @@ -1,4 +1,4 @@ -// smoke drives the SDK surface (files, sessions, find/replace, watch, git, pty) against one live sandbox; scripts/sandboxd-e2e.sh runs it. +// smoke drives the SDK surface against one live sandbox; scripts/sandboxd-e2e.sh runs it. package main import ( diff --git a/sandboxd/pool/watermark.go b/sandboxd/pool/watermark.go index 5aa34c670..6cc539056 100644 --- a/sandboxd/pool/watermark.go +++ b/sandboxd/pool/watermark.go @@ -130,7 +130,6 @@ func (p *pool) trimWarm(target int) []string { return trim } -// noteArrival counts one claim; a bin that has spanned rateBin folds its arrivals per second into the rate EWMA. func (p *pool) noteArrival(now time.Time) { switch elapsed := now.Sub(p.binStart); { case p.binStart.IsZero(): diff --git a/sdk/go/client_test.go b/sdk/go/client_test.go index d6d40832a..5755ade51 100644 --- a/sdk/go/client_test.go +++ b/sdk/go/client_test.go @@ -207,15 +207,10 @@ func TestAMovedSandboxRetriesAStreamCallAtItsNewOwner(t *testing.T) { func TestAWriteToAGoneSandboxLooksItsOwnerUpOnce(t *testing.T) { var lookups atomic.Int32 node := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { - switch r.URL.Path { - case "/v1/peers": - _ = json.MarshalWrite(w, map[string]any{"peers": []string{}}) - case "/v1/sandboxes/sb_1/owner": + if r.URL.Path == "/v1/sandboxes/sb_1/owner" { lookups.Add(1) - w.WriteHeader(http.StatusNotFound) - default: - w.WriteHeader(http.StatusNotFound) } + w.WriteHeader(http.StatusNotFound) })) t.Cleanup(node.Close) sb := testSandbox(t, node) diff --git a/silkd/src/exec.rs b/silkd/src/exec.rs index b421fa22b..251e92ce8 100644 --- a/silkd/src/exec.rs +++ b/silkd/src/exec.rs @@ -9,7 +9,7 @@ use std::time::Duration; use nix::unistd::{AccessFlags, access}; use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt}; -use tokio::process::Command; +use tokio::process::{ChildStderr, ChildStdin, ChildStdout, Command}; use tokio::sync::mpsc; use tokio::time::timeout; @@ -169,8 +169,8 @@ where /// Reads both streams in this one task, so a nested spawn cannot outlive the pump abort. async fn pump_out( proc: Arc, - stdout: Option, - stderr: Option, + stdout: Option, + stderr: Option, fg: Option>, ) { let fg = fg.as_ref(); @@ -210,10 +210,7 @@ where } } -async fn pump_stdin( - stdin: Option, - mut client: mpsc::Receiver, -) { +async fn pump_stdin(stdin: Option, mut client: mpsc::Receiver) { let Some(mut sink) = stdin else { return }; while let Some(frame) = client.recv().await { match frame { diff --git a/silkd/src/pty.rs b/silkd/src/pty.rs index e06f2fc63..1ce2e9f62 100644 --- a/silkd/src/pty.rs +++ b/silkd/src/pty.rs @@ -205,7 +205,6 @@ async fn pump_stdin( } } -/// Reads the master's buffered tail after the shell exits. async fn drain( master: &AsyncFd, proc: &Proc, diff --git a/silkd/src/sysutil.rs b/silkd/src/sysutil.rs index e6af01fad..5a683ba10 100644 --- a/silkd/src/sysutil.rs +++ b/silkd/src/sysutil.rs @@ -14,7 +14,7 @@ use std::sync::{LazyLock, Mutex}; use nix::fcntl::{FcntlArg, OFlag, fcntl}; use tokio::io::unix::AsyncFd; -use tokio::process::Command; +use tokio::process::{Child, Command}; static TMP_SEQ: AtomicU64 = AtomicU64::new(0); @@ -195,7 +195,7 @@ pub fn exit_code(status: ExitStatus) -> i32 { } /// Reaps the child and maps its status via `exit_code`; -1 when the wait fails. -pub async fn wait_code(child: &mut tokio::process::Child) -> i32 { +pub async fn wait_code(child: &mut Child) -> i32 { child.wait().await.map_or(-1, exit_code) } diff --git a/silkd/src/tree.rs b/silkd/src/tree.rs index 7e846b28c..3ac93e19f 100644 --- a/silkd/src/tree.rs +++ b/silkd/src/tree.rs @@ -10,7 +10,7 @@ use nix::fcntl::{self, AtFlags, OFlag}; use nix::sys::stat::{self, Mode, SFlag}; use nix::unistd::{self, UnlinkatFlags}; use tokio::io::{AsyncBufRead, AsyncReadExt, AsyncWrite}; -use tokio::process::Command; +use tokio::process::{Child, ChildStderr, Command}; use crate::proto::{self, BULK_CHUNK, ErrorKind, Response, err_frame}; use crate::sysutil; @@ -132,9 +132,7 @@ where } /// Spawns tar draining stderr on its own task, so a full stderr pipe cannot deadlock. -fn spawn_tar( - mut cmd: Command, -) -> io::Result<(tokio::process::Child, tokio::task::JoinHandle>)> { +fn spawn_tar(mut cmd: Command) -> io::Result<(Child, tokio::task::JoinHandle>)> { let mut child = cmd.stderr(Stdio::piped()).kill_on_drop(true).spawn()?; let stderr = sysutil::take_pipe(child.stderr.take(), "stderr")?; Ok((child, tokio::spawn(drain(stderr)))) @@ -186,7 +184,7 @@ fn is_dir(mode: stat::mode_t) -> bool { } /// Reads a child's stderr up to STDERR_CAP; tar's first error is the useful one. -async fn drain(mut stderr: tokio::process::ChildStderr) -> Vec { +async fn drain(mut stderr: ChildStderr) -> Vec { let mut out = Vec::new(); let mut buf = [0u8; 4096]; while let Ok(n) = stderr.read(&mut buf).await { From 47ea429973cdc7629f4519bffde6f900e8505d9c Mon Sep 17 00:00:00 2001 From: CMGS Date: Fri, 6 Nov 2026 18:56:26 +0800 Subject: [PATCH 11/11] silkd: wait for the forked child to leave the group before the client-gone stdin test hangs up The test dropped the client after the first stdout frame; on a loaded runner the group SIGKILL could land before perl ran setpgrp, killing the child that writes the result, and the test waited out its 30 s deadline. Reproduced by delaying setpgrp 0.5 s; the child now marks itself ready after setpgrp and the test waits for that mark. --- silkd/tests/exec_e2e.rs | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/silkd/tests/exec_e2e.rs b/silkd/tests/exec_e2e.rs index e7027e9aa..946714883 100644 --- a/silkd/tests/exec_e2e.rs +++ b/silkd/tests/exec_e2e.rs @@ -410,7 +410,7 @@ async fn a_client_gone_mid_output_releases_the_stdin_a_forked_child_holds() { &state, json!({ "op": "exec", - "argv": ["/bin/sh", "-c", "exec 3<&0; perl -e 'setpgrp; exec @ARGV' /bin/sh -c 'sleep 1; wc -c <&3 >\"$OUT.tmp\"; mv \"$OUT.tmp\" \"$OUT\"' >/dev/null 2>&1 & while :; do echo x; sleep 0.05; done"], + "argv": ["/bin/sh", "-c", "exec 3<&0; perl -e 'setpgrp; exec @ARGV' /bin/sh -c ': >\"$OUT.ready\"; sleep 1; wc -c <&3 >\"$OUT.tmp\"; mv \"$OUT.tmp\" \"$OUT\"' >/dev/null 2>&1 & while :; do echo x; sleep 0.05; done"], "env": {"OUT": out.to_str().unwrap()} }), ) @@ -420,6 +420,7 @@ async fn a_client_gone_mid_output_releases_the_stdin_a_forked_child_holds() { send(&mut cw, json!({"op":"stdin","data":chunk})).await; } frames_until(&mut lines, |f| type_of(f) == "stdout").await; + wait_for_path(&dir.path().join("count.ready")).await; drop(lines); drop(cw);