From 6523d8c60441eb15e25f80c4c51acec35ab3a639 Mon Sep 17 00:00:00 2001 From: CMGS Date: Mon, 5 Oct 2026 18:51:24 +0800 Subject: [PATCH 1/4] review: layout, one-line docs, and named func types across the post-#329 files Applies the style rules: relayEnds moves above the Server methods it split and relay.go's free funcs below auditTee's method set; KeptNode and FakeAgent move above the tests; func types spelled twice get names (bulkMaker, chunkSink); multi-fact godocs and module docstrings drop to one line; belowFloor no longer shadows comparable; tests reuse breakStore and waitFor and drop a helper comment that restated its name. --- protocol/wire/frame.go | 12 +-- sandboxd/engine/engine.go | 6 +- sandboxd/pool/archive_test.go | 4 +- sandboxd/pool/liveness_test.go | 1 - sandboxd/pool/pool_test.go | 9 +-- sandboxd/server/relay.go | 65 ++++++++-------- sdk/go/files.go | 3 +- sdk/go/utils.go | 6 +- sdk/python/tests/conftest.py | 3 +- sdk/python/tests/test_client.py | 28 +++---- sdk/python/tests/test_keepalive.py | 118 ++++++++++++++--------------- sdk/python/tests/test_stream.py | 3 +- sdk/python/tests/test_upgrade.py | 3 +- 13 files changed, 125 insertions(+), 136 deletions(-) diff --git a/protocol/wire/frame.go b/protocol/wire/frame.go index a8539f1d..aec57778 100644 --- a/protocol/wire/frame.go +++ b/protocol/wire/frame.go @@ -143,9 +143,11 @@ type Response interface{ RespType() string } type respDecoder func([]byte) (Response, error) +type bulkMaker func([]byte) Response + type bulkResponse struct { head []byte - mk func([]byte) Response + mk bulkMaker } // reqPtr and respPtr turn a mistyped decoder table entry into a compile error. @@ -159,15 +161,13 @@ type respPtr[T any] interface { Response } -// Exec starts a process; with Session set it runs inside that persistent -// shell instead. Detach is emitted even when false — it is part of the -// fixture corpus shape. +// Exec starts a process; with Session set it runs inside that persistent shell instead. type Exec struct { Argv []string `json:"argv"` Cwd string `json:"cwd,omitempty"` Env map[string]string `json:"env,omitempty"` User string `json:"user,omitempty"` - Detach bool `json:"detach"` + Detach bool `json:"detach"` // emitted even when false: the fixture corpus carries it Session string `json:"session,omitempty"` } @@ -716,7 +716,7 @@ func IsContinuation(line []byte) bool { return false } -func bulkDecoders(mks ...func([]byte) Response) map[string]bulkResponse { +func bulkDecoders(mks ...bulkMaker) map[string]bulkResponse { out := make(map[string]bulkResponse, len(mks)) for _, mk := range mks { tag := mk(nil).RespType() diff --git a/sandboxd/engine/engine.go b/sandboxd/engine/engine.go index 4e3e097c..7a5f5d68 100644 --- a/sandboxd/engine/engine.go +++ b/sandboxd/engine/engine.go @@ -101,7 +101,7 @@ func (e *Engine) VersionWarning(ctx context.Context) (version, warning string) { if err != nil { return "", fmt.Sprintf("cannot determine cocoon version (need >= %s): %v", RequiredCocoon, err) } - if below, comparable := belowFloor(v); comparable && below { + if below, known := belowFloor(v); known && below { return v, fmt.Sprintf("cocoon %s is below the required %s — upgrade cocoon", v, RequiredCocoon) } return v, "" @@ -493,8 +493,8 @@ func respFail(resp wire.Response) string { return "unexpected frame " + resp.RespType() } -// belowFloor reports whether v is below RequiredCocoon; comparable is false for a dev build. -func belowFloor(v string) (below, comparable bool) { +// belowFloor reports whether v is below RequiredCocoon; known is false for a dev build. +func belowFloor(v string) (below, known bool) { v = "v" + strings.TrimPrefix(v, "v") if semver.Canonical(v) != v { return false, false diff --git a/sandboxd/pool/archive_test.go b/sandboxd/pool/archive_test.go index a9726943..8b1b2c94 100644 --- a/sandboxd/pool/archive_test.go +++ b/sandboxd/pool/archive_test.go @@ -562,7 +562,7 @@ func TestArchiveRemovalCommitPinsCheckpoint(t *testing.T) { if !ckExists(t, m, ck) { t.Error("retry deleted the checkpoint before the claims commit") } - m.store.path = filepath.Join(t.TempDir(), "gone", "claims.json") + breakStore(t, m) m.store.writeMu.Unlock() locked = false if err := <-done; action == testArchiveRelease && err == nil { @@ -667,7 +667,7 @@ func TestArchiveDeleteRetryRechecksWakeRollback(t *testing.T) { defer m.recLocksMu.Unlock() return m.recRefs[ck] >= 2 }) - m.store.path = filepath.Join(t.TempDir(), "gone", "claims.json") + breakStore(t, m) m.store.writeMu.Unlock() locked = false if err := <-woke; err == nil { diff --git a/sandboxd/pool/liveness_test.go b/sandboxd/pool/liveness_test.go index ff0d61de..5a1a4747 100644 --- a/sandboxd/pool/liveness_test.go +++ b/sandboxd/pool/liveness_test.go @@ -524,7 +524,6 @@ func TestReconcileKeepsRunningClaimsTheHostLost(t *testing.T) { } } -// assertRestartFailsUntilBudget runs recoveries against a guest setup step that always fails. func assertRestartFailsUntilBudget(t *testing.T, m *Manager, eng *fakeEngine, sb *types.Sandbox) { t.Helper() for attempt := range vmmRestartBudget { diff --git a/sandboxd/pool/pool_test.go b/sandboxd/pool/pool_test.go index c2c519a8..68a0dbe9 100644 --- a/sandboxd/pool/pool_test.go +++ b/sandboxd/pool/pool_test.go @@ -426,14 +426,7 @@ func TestRefillServicesEveryPoolUnderTightBudget(t *testing.T) { infos, _ := m.Info() return len(infos) == 2 && infos[0].Warm == 3 && infos[1].Warm == 3 } - deadline := time.Now().Add(3 * time.Second) - for !both() && time.Now().Before(deadline) { - time.Sleep(5 * time.Millisecond) - } - if !both() { - infos, _ := m.Info() - t.Fatalf("one refill tick did not chain both pools to target: %+v", infos) - } + waitFor(t, both) }) } diff --git a/sandboxd/server/relay.go b/sandboxd/server/relay.go index ab8178aa..05916f1a 100644 --- a/sandboxd/server/relay.go +++ b/sandboxd/server/relay.go @@ -34,6 +34,11 @@ var ( silkdEnds = relayEnds{guest: func(c net.Conn) { _ = c.Close() }, client: drainClient} ) +// relayEnds is what a finished direction means to the peer that did not end it. +type relayEnds struct { + guest, client func(net.Conn) +} + // CloseRelays force-closes in-flight relays; http.Server.Shutdown does not track hijacked conns. func (s *Server) CloseRelays() { s.relayMu.Lock() @@ -94,12 +99,6 @@ func (s *Server) relay(ctx context.Context, id string, client net.Conn, clientBu s.splice(client, clientR, guest, switchingProtocols, silkdEnds) } -// relayEnds is what a finished direction means to the peer that did not end it. -type relayEnds struct { - guest func(net.Conn) - client func(net.Conn) -} - func (s *Server) splice(client net.Conn, clientR io.Reader, guest net.Conn, hello []byte, ends relayEnds) { release, ok := s.trackRelay(client, guest) if !ok { @@ -157,33 +156,6 @@ func (s *Server) trackRelay(client, guest net.Conn) (func(), bool) { }, true } -func keepAlive(client net.Conn) { - tcp, ok := client.(*net.TCPConn) - if !ok { - return - } - _ = tcp.SetKeepAliveConfig(net.KeepAliveConfig{ - Enable: true, - Idle: keepAliveIdle, - Interval: keepAliveInterval, - Count: keepAliveCount, - }) -} - -// drainClient half-closes; the read deadline is what unblocks the splice goroutine. -func drainClient(client net.Conn) { - utils.CloseWrite(client) - _ = client.SetReadDeadline(time.Now().Add(drainGrace)) -} - -// clientReader replays what the http server already buffered; reading the bufio past Buffered() would race the direct conn reads. -func clientReader(client net.Conn, clientBuf *bufio.Reader) io.Reader { - if n := clientBuf.Buffered(); n > 0 { - return io.MultiReader(io.LimitReader(clientBuf, int64(n)), client) - } - return client -} - // auditTee records each request line it relays; input frames and the bytes past the cap pass unrecorded. type auditTee struct { r io.Reader @@ -230,3 +202,30 @@ func (t *auditTee) endLine() { t.buf = t.buf[:0] t.skip = false } + +func keepAlive(client net.Conn) { + tcp, ok := client.(*net.TCPConn) + if !ok { + return + } + _ = tcp.SetKeepAliveConfig(net.KeepAliveConfig{ + Enable: true, + Idle: keepAliveIdle, + Interval: keepAliveInterval, + Count: keepAliveCount, + }) +} + +// drainClient half-closes; the read deadline is what unblocks the splice goroutine. +func drainClient(client net.Conn) { + utils.CloseWrite(client) + _ = client.SetReadDeadline(time.Now().Add(drainGrace)) +} + +// clientReader replays what the http server already buffered; reading the bufio past Buffered() would race the direct conn reads. +func clientReader(client net.Conn, clientBuf *bufio.Reader) io.Reader { + if n := clientBuf.Buffered(); n > 0 { + return io.MultiReader(io.LimitReader(clientBuf, int64(n)), client) + } + return client +} diff --git a/sdk/go/files.go b/sdk/go/files.go index b1ed4236..d6de2f7f 100644 --- a/sdk/go/files.go +++ b/sdk/go/files.go @@ -8,8 +8,7 @@ import ( "github.com/cocoonstack/sandbox/protocol/wire" ) -// WriteFile writes data to path in the sandbox, atomically (silkd renames a -// temp file into place). mode, when non-nil, sets the file's permission bits. +// WriteFile atomically writes data to path in the sandbox; a non-nil mode sets the file's permission bits. func (s *Sandbox) WriteFile(ctx context.Context, path string, data []byte, mode *uint32) error { return s.uploadRPC(ctx, &wire.FsWrite{Path: path, Mode: mode}, bytes.NewReader(data)) } diff --git a/sdk/go/utils.go b/sdk/go/utils.go index b05e8db1..32ec8a94 100644 --- a/sdk/go/utils.go +++ b/sdk/go/utils.go @@ -18,6 +18,8 @@ type respPtr[T any] interface { wire.Response } +type chunkSink func([]byte) error + func (s *Sandbox) doneRPC(ctx context.Context, req wire.Request) error { conn, l, err := s.call(ctx, req) if err != nil { @@ -63,7 +65,7 @@ func (s *Sandbox) uploadRPC(ctx context.Context, req wire.Request, r io.Reader) return l.done(<-terminal) } -func (s *Sandbox) downloadRPC(ctx context.Context, req wire.Request, sink func([]byte) error) error { +func (s *Sandbox) downloadRPC(ctx context.Context, req wire.Request, sink chunkSink) error { conn, l, err := s.call(ctx, req) if err != nil { return err @@ -162,7 +164,7 @@ func pumpStdio(ctx context.Context, conn *silkd.Conn, stdout, stderr io.Writer) } // drainData consumes Data frames into sink until Done; sink must not retain its argument. -func drainData(ctx context.Context, conn *silkd.Conn, sink func([]byte) error) error { +func drainData(ctx context.Context, conn *silkd.Conn, sink chunkSink) error { for { resp, err := recvTransient(ctx, conn) if err != nil { diff --git a/sdk/python/tests/conftest.py b/sdk/python/tests/conftest.py index 99244157..c8d1d892 100644 --- a/sdk/python/tests/conftest.py +++ b/sdk/python/tests/conftest.py @@ -1,5 +1,4 @@ -"""Fixtures and helpers shared by the suites: in-process fake nodes, addresses -that refuse a connection, and the relay-side upgrade handshake.""" +"""Shared fixtures: in-process fake nodes, refusing addresses, and the relay-side upgrade handshake.""" import json import os diff --git a/sdk/python/tests/test_client.py b/sdk/python/tests/test_client.py index 723ae7b8..90b799dc 100644 --- a/sdk/python/tests/test_client.py +++ b/sdk/python/tests/test_client.py @@ -10,6 +10,20 @@ from cocoonsandbox import APIError, Client, Template +class KeptNode(FakeNode): + protocol_version = "HTTP/1.1" + conns = 0 + drop_after_reply = False + + def setup(self): + super().setup() + KeptNode.conns += 1 + + def _reply(self, code, payload): + super()._reply(code, payload) + self.close_connection = self.drop_after_reply + + @pytest.fixture def node(): server = HTTPServer(("127.0.0.1", 0), FakeNode) @@ -368,17 +382,3 @@ def kept_node(): threading.Thread(target=server.serve_forever, daemon=True).start() yield f"127.0.0.1:{server.server_port}" server.shutdown() - - -class KeptNode(FakeNode): - protocol_version = "HTTP/1.1" - conns = 0 - drop_after_reply = False - - def setup(self): - super().setup() - KeptNode.conns += 1 - - def _reply(self, code, payload): - super()._reply(code, payload) - self.close_connection = self.drop_after_reply diff --git a/sdk/python/tests/test_keepalive.py b/sdk/python/tests/test_keepalive.py index 72f2fa14..05830729 100644 --- a/sdk/python/tests/test_keepalive.py +++ b/sdk/python/tests/test_keepalive.py @@ -16,6 +16,65 @@ INPUT_OPS = ("stdin", "stdin_close", "data", "data_end") +class FakeAgent: + """A relay-side silkd stand-in; proto 1 half-closes after one RPC and drains; slam closes as a failed relay does.""" + + def __init__(self, proto: int = KEEP_ALIVE_PROTO, hang_up_after: int = 0, slam: bool = False) -> None: + self.proto = proto + self.hang_up_after = hang_up_after + self.slam = slam + self.upgrades = 0 + self.hangups = 0 + self.closed = 0 + self._server = socket.create_server(("127.0.0.1", 0)) + self.addr = f"127.0.0.1:{self._server.getsockname()[1]}" + threading.Thread(target=self._accept, daemon=True).start() + + def stop(self) -> None: + self._server.close() + + def _accept(self) -> None: + with contextlib.suppress(OSError): + while True: + conn, _ = self._server.accept() + self.upgrades += 1 + threading.Thread(target=self._serve, args=(conn,), daemon=True).start() + + def _serve(self, conn: socket.socket) -> None: + with conn, contextlib.suppress(OSError): + reader = accept_upgrade(conn) + replies = 0 + while True: + line = reader.readline() + if not line: + break + op = json.loads(line)["op"] + if op in INPUT_OPS: + continue + for frame in self._answer(op): + conn.sendall(json.dumps(frame).encode() + b"\n") + replies += 1 + if self.slam: + break + if self.proto < KEEP_ALIVE_PROTO or replies == self.hang_up_after: + conn.shutdown(socket.SHUT_WR) + self.hangups += 1 + while reader.readline(): + pass + break + self.closed += 1 + + def _answer(self, op: str) -> list: + if op == "info": + return [{"type": "info", "version": "fake", "proto": self.proto, "uptime_secs": 0, "procs": 0}] + if op == "fs_stat": + return [{"type": "stat", "info": {"kind": "dir", "size": 0, "mode": 0o755, "mtime_epoch_secs": 0}}] + if op == "exec": + out = base64.b64encode(b"42\n").decode() + return [{"type": "started", "pid": 1}, {"type": "stdout", "data": out}, {"type": "exit", "code": 0}] + return [{"type": "error", "kind": "unimplemented", "message": op}] + + def test_calls_share_one_connection(fd_floor): agent = FakeAgent() sb = sandbox_at(agent.addr) @@ -140,62 +199,3 @@ def test_an_upload_an_old_daemon_rejects_raises_its_error_frame(): sb.write_file("/missing/f", bytes(64 << 20)) finally: agent.stop() - - -class FakeAgent: - """A relay-side silkd stand-in; proto 1 half-closes after one RPC and drains; slam closes as a failed relay does.""" - - def __init__(self, proto: int = KEEP_ALIVE_PROTO, hang_up_after: int = 0, slam: bool = False) -> None: - self.proto = proto - self.hang_up_after = hang_up_after - self.slam = slam - self.upgrades = 0 - self.hangups = 0 - self.closed = 0 - self._server = socket.create_server(("127.0.0.1", 0)) - self.addr = f"127.0.0.1:{self._server.getsockname()[1]}" - threading.Thread(target=self._accept, daemon=True).start() - - def stop(self) -> None: - self._server.close() - - def _accept(self) -> None: - with contextlib.suppress(OSError): - while True: - conn, _ = self._server.accept() - self.upgrades += 1 - threading.Thread(target=self._serve, args=(conn,), daemon=True).start() - - def _serve(self, conn: socket.socket) -> None: - with conn, contextlib.suppress(OSError): - reader = accept_upgrade(conn) - replies = 0 - while True: - line = reader.readline() - if not line: - break - op = json.loads(line)["op"] - if op in INPUT_OPS: - continue - for frame in self._answer(op): - conn.sendall(json.dumps(frame).encode() + b"\n") - replies += 1 - if self.slam: - break - if self.proto < KEEP_ALIVE_PROTO or replies == self.hang_up_after: - conn.shutdown(socket.SHUT_WR) - self.hangups += 1 - while reader.readline(): - pass - break - self.closed += 1 - - def _answer(self, op: str) -> list: - if op == "info": - return [{"type": "info", "version": "fake", "proto": self.proto, "uptime_secs": 0, "procs": 0}] - if op == "fs_stat": - return [{"type": "stat", "info": {"kind": "dir", "size": 0, "mode": 0o755, "mtime_epoch_secs": 0}}] - if op == "exec": - out = base64.b64encode(b"42\n").decode() - return [{"type": "started", "pid": 1}, {"type": "stdout", "data": out}, {"type": "exit", "code": 0}] - return [{"type": "error", "kind": "unimplemented", "message": op}] diff --git a/sdk/python/tests/test_stream.py b/sdk/python/tests/test_stream.py index 83332410..8b723009 100644 --- a/sdk/python/tests/test_stream.py +++ b/sdk/python/tests/test_stream.py @@ -1,5 +1,4 @@ -"""The client timeout bounds the dial and the upgrade, not the frames that -follow: a guest stream outlives it.""" +"""The client timeout bounds the dial and the upgrade, not the guest stream that follows.""" import contextlib import json diff --git a/sdk/python/tests/test_upgrade.py b/sdk/python/tests/test_upgrade.py index 235f80a6..0cf149e8 100644 --- a/sdk/python/tests/test_upgrade.py +++ b/sdk/python/tests/test_upgrade.py @@ -1,5 +1,4 @@ -"""The relay upgrade: over plain HTTP the request leaves with the first frame and -the 101 is read on the first recv; a refusal surfaces as an APIError.""" +"""Over plain HTTP the upgrade request leaves with the first frame; a refusal surfaces as an APIError.""" import contextlib import json From 6ad87962cd0e1446299fb5b11b7aec634217f7ab Mon Sep 17 00:00:00 2001 From: CMGS Date: Mon, 5 Oct 2026 18:55:22 +0800 Subject: [PATCH 2/4] cut: fold single-use decode and recv layers, collect purges in the reap scan DecodeResponse calls DecodeResponseInto with a nil scratch instead of both forwarding to a private decodeResponse, and decodeReq/decodeResp inline decodeAs. The SDK's three recv helpers become one transient read: every frame it reads is either streamed out before the next read or carries no bulk Data. reapOnce collects purge checkpoints in its locked scan, ReadFile reuses ReadFileTo, and archivesAtDeadline inlines archiveEnabledFor. Tests share one upload fake and the black_hole fixture. --- protocol/wire/frame.go | 52 ++++++++----------- sandboxd/pool/archive.go | 4 -- sandboxd/pool/claim.go | 15 ++---- sdk/go/files.go | 10 ++-- sdk/go/files_test.go | 87 ++++++++++++++------------------ sdk/go/pty.go | 2 +- sdk/go/utils.go | 17 ++----- sdk/python/tests/test_upgrade.py | 17 ++----- 8 files changed, 76 insertions(+), 128 deletions(-) diff --git a/protocol/wire/frame.go b/protocol/wire/frame.go index aec57778..eaf0a635 100644 --- a/protocol/wire/frame.go +++ b/protocol/wire/frame.go @@ -679,12 +679,25 @@ func DecodeRequest(line []byte) (Request, error) { // DecodeResponse parses one frame into its type's concrete Go type. Byte // fields are freshly allocated per frame, so callers may retain them. func DecodeResponse(line []byte) (Response, error) { - return decodeResponse(line, nil) + return DecodeResponseInto(line, nil) } // DecodeResponseInto is DecodeResponse for a streaming consumer: a bulk frame's bytes land in *scratch, grown as needed, and stay valid only until the next call with it. func DecodeResponseInto(line []byte, scratch *[]byte) (Response, error) { - return decodeResponse(line, scratch) + typ, err := frameTag(line, respTagHead, "type") + if err != nil { + return nil, fmt.Errorf("parse response frame: %w", err) + } + if bulk, ok := bulkResponses[string(typ)]; ok { + if data, ok := decodeBulk(line, bulk.head, scratch); ok { + return bulk.mk(data), nil + } + } + dec, ok := responseDecoders[string(typ)] + if !ok { + return nil, fmt.Errorf("unknown response type %q", typ) + } + return dec(line) } // NewFrameScanner wraps r for newline-delimited frames capped at MaxFrame. @@ -725,23 +738,6 @@ func bulkDecoders(mks ...bulkMaker) map[string]bulkResponse { return out } -func decodeResponse(line []byte, scratch *[]byte) (Response, error) { - typ, err := frameTag(line, respTagHead, "type") - if err != nil { - return nil, fmt.Errorf("parse response frame: %w", err) - } - if bulk, ok := bulkResponses[string(typ)]; ok { - if data, ok := decodeBulk(line, bulk.head, scratch); ok { - return bulk.mk(data), nil - } - } - dec, ok := responseDecoders[string(typ)] - if !ok { - return nil, fmt.Errorf("unknown response type %q", typ) - } - return dec(line) -} - // decodeBulk decodes a canonical bulk frame's payload into scratch, or a fresh buffer when scratch is nil; ok is false for any other shape. func decodeBulk(line, head []byte, scratch *[]byte) ([]byte, bool) { after, ok := bytes.CutPrefix(line, head) @@ -826,26 +822,18 @@ func scanTag(line []byte, key string) (string, error) { return "", nil } -func decodeAs[T any](line []byte) (*T, error) { - v := new(T) +func decodeReq[T any, PT reqPtr[T]](line []byte) (Request, error) { + v := PT(new(T)) if err := json.Unmarshal(line, v); err != nil { return nil, err } return v, nil } -func decodeReq[T any, PT reqPtr[T]](line []byte) (Request, error) { - v, err := decodeAs[T](line) - if err != nil { - return nil, err - } - return PT(v), nil -} - func decodeResp[T any, PT respPtr[T]](line []byte) (Response, error) { - v, err := decodeAs[T](line) - if err != nil { + v := PT(new(T)) + if err := json.Unmarshal(line, v); err != nil { return nil, err } - return PT(v), nil + return v, nil } diff --git a/sandboxd/pool/archive.go b/sandboxd/pool/archive.go index 451a6146..32123833 100644 --- a/sandboxd/pool/archive.go +++ b/sandboxd/pool/archive.go @@ -33,10 +33,6 @@ func (m *Manager) archiveDeleteFor(key types.PoolKey) time.Duration { return m.archiveDeleteDefault } -func (m *Manager) archiveEnabledFor(key types.PoolKey) bool { - return m.archiveAfterFor(key) > 0 -} - func (m *Manager) archiveOnce(ctx context.Context) { if !m.archiveSweep.CompareAndSwap(false, true) { return // the previous sweep's archives are still draining diff --git a/sandboxd/pool/claim.go b/sandboxd/pool/claim.go index cde7c084..a3c6dd12 100644 --- a/sandboxd/pool/claim.go +++ b/sandboxd/pool/claim.go @@ -502,7 +502,7 @@ func (m *Manager) reapOnce(ctx context.Context) { } m.mu.Lock() var expired []victim - var dropped []string + var dropped, purges []string for id, sb := range m.claimed { // a zero deadline means no expiry (an archived claim kept forever). if sb.Deadline.IsZero() || !now.After(sb.Deadline) { @@ -511,6 +511,7 @@ func (m *Manager) reapOnce(ctx context.Context) { switch { case sb.ArchiveCk != "": expired = append(expired, victim{action: reapPurge, id: id, ck: sb.ArchiveCk, tenant: sb.Tenant, sb: sb}) + purges = append(purges, sb.ArchiveCk) case m.archivesAtDeadline(sb): // archive instead of destroy, kept in m.claimed for archive() to move if _, busy := m.archiving[id]; busy { @@ -538,19 +539,13 @@ func (m *Manager) reapOnce(ctx context.Context) { } logger := log.WithFunc("pool.reapOnce") - var purges []string - for _, v := range expired { - if v.action == reapPurge { - purges = append(purges, v.ck) - } - } failed := m.markArchiveCks(purges) expired = slices.DeleteFunc(expired, func(v victim) bool { markErr, bad := failed[v.ck] - if bad && v.action == reapPurge { + if bad { logger.Errorf(ctx, markErr, "mark archive ck %s; keeping %s", v.ck, v.id) } - return bad && v.action == reapPurge + return bad }) if len(expired) == 0 { return @@ -629,7 +624,7 @@ func (m *Manager) reapOnce(ctx context.Context) { } func (m *Manager) archivesAtDeadline(sb *types.Sandbox) bool { - return sb.Failed == "" && !m.tenantGone(sb.Tenant) && (sb.OnExpire == types.ExpireArchive || sb.HibernateSnap != "" && m.archiveEnabledFor(sb.Key)) + return sb.Failed == "" && !m.tenantGone(sb.Tenant) && (sb.OnExpire == types.ExpireArchive || sb.HibernateSnap != "" && m.archiveAfterFor(sb.Key) > 0) } func (m *Manager) pauseExpired(ctx context.Context, sb *types.Sandbox) error { diff --git a/sdk/go/files.go b/sdk/go/files.go index d6de2f7f..bd98ac4d 100644 --- a/sdk/go/files.go +++ b/sdk/go/files.go @@ -15,15 +15,11 @@ func (s *Sandbox) WriteFile(ctx context.Context, path string, data []byte, mode // ReadFile returns the contents of path. func (s *Sandbox) ReadFile(ctx context.Context, path string) ([]byte, error) { - var out []byte - err := s.downloadRPC(ctx, &wire.FsRead{Path: path}, func(b []byte) error { - out = append(out, b...) - return nil - }) - if err != nil { + var buf bytes.Buffer + if err := s.ReadFileTo(ctx, path, &buf); err != nil { return nil, err } - return out, nil + return buf.Bytes(), nil } // ReadFileTo streams path into w; an error from w ends the read there. diff --git a/sdk/go/files_test.go b/sdk/go/files_test.go index ac4240ca..b2082da2 100644 --- a/sdk/go/files_test.go +++ b/sdk/go/files_test.go @@ -61,32 +61,7 @@ func TestFilesRoundTrip(t *testing.T) { } func TestAnUploadFillsEachFrameFromSmallReads(t *testing.T) { - frames := make(chan int, 1) - sb := testSandbox(t, newAgentServer(t, func(conn net.Conn) { - defer conn.Close() - sc := wire.NewFrameScanner(conn) - data := 0 - for sc.Scan() { - req, err := wire.DecodeRequest(sc.Bytes()) - if err != nil { - return - } - var resp wire.Response - switch req.(type) { - case *wire.Info: - resp = &wire.InfoResp{Proto: wire.KeepAliveProto} - case *wire.Data: - data++ - case *wire.DataEnd: - frames <- data - resp = wire.Done{} - } - if resp != nil { - line, _ := wire.EncodeResponse(resp) - _, _ = conn.Write(append(line, '\n')) - } - } - })) + sb, frames := uploadAgent(t) if err := sb.Push(t.Context(), "/d", iotest.OneByteReader(bytes.NewReader(make([]byte, 4096)))); err != nil { t.Fatalf("push: %v", err) } @@ -96,32 +71,13 @@ func TestAnUploadFillsEachFrameFromSmallReads(t *testing.T) { } func TestAnUploadStopsOnItsReadersError(t *testing.T) { - ended := make(chan bool, 1) - sb := testSandbox(t, newAgentServer(t, func(conn net.Conn) { - defer conn.Close() - sc := wire.NewFrameScanner(conn) - for sc.Scan() { - req, err := wire.DecodeRequest(sc.Bytes()) - if err != nil { - break - } - switch req.(type) { - case *wire.Info: - line, _ := wire.EncodeResponse(&wire.InfoResp{Proto: wire.KeepAliveProto}) - _, _ = conn.Write(append(line, '\n')) - case *wire.DataEnd: - ended <- true - return - } - } - ended <- false - })) + sb, frames := uploadAgent(t) truncated := io.MultiReader(bytes.NewReader(make([]byte, 4096)), iotest.ErrReader(io.ErrUnexpectedEOF)) if err := sb.Push(t.Context(), "/d", truncated); !errors.Is(err, io.ErrUnexpectedEOF) { t.Errorf("push = %v, want the reader's io.ErrUnexpectedEOF", err) } - if <-ended { - t.Error("a truncated stream was committed with data_end") + if got := <-frames; got != -1 { + t.Errorf("a truncated stream was committed with data_end after %d data frames", got) } } @@ -223,6 +179,41 @@ func TestUploadToAnOldDaemonReturnsItsRejection(t *testing.T) { } } +func uploadAgent(t *testing.T) (*Sandbox, <-chan int) { + t.Helper() + frames := make(chan int, 1) + sb := testSandbox(t, newAgentServer(t, func(conn net.Conn) { + defer conn.Close() + sc := wire.NewFrameScanner(conn) + data := 0 + for sc.Scan() { + req, err := wire.DecodeRequest(sc.Bytes()) + if err != nil { + break + } + var resp wire.Response + switch req.(type) { + case *wire.Info: + resp = &wire.InfoResp{Proto: wire.KeepAliveProto} + case *wire.Data: + data++ + case *wire.DataEnd: + frames <- data + resp = wire.Done{} + } + if resp != nil { + line, _ := wire.EncodeResponse(resp) + _, _ = conn.Write(append(line, '\n')) + } + if _, end := req.(*wire.DataEnd); end { + return + } + } + frames <- -1 + })) + return sb, frames +} + func fakeSandbox(t testing.TB) *Sandbox { t.Helper() fake := silkdtest.NewFake(t.TempDir()) diff --git a/sdk/go/pty.go b/sdk/go/pty.go index a46a05e5..117ab49c 100644 --- a/sdk/go/pty.go +++ b/sdk/go/pty.go @@ -68,7 +68,7 @@ func (p *Pty) ExitCode() (code int, ok bool) { func (p *Pty) drain(ctx context.Context, pw *io.PipeWriter, stop func()) { defer stop() for { - resp, err := recvTransient(ctx, p.conn) + resp, err := recv(ctx, p.conn) if err != nil { _ = pw.CloseWithError(err) return diff --git a/sdk/go/utils.go b/sdk/go/utils.go index 32ec8a94..7d414b98 100644 --- a/sdk/go/utils.go +++ b/sdk/go/utils.go @@ -136,7 +136,7 @@ func pumpStdio(ctx context.Context, conn *silkd.Conn, stdout, stderr io.Writer) stdout = cmp.Or(stdout, io.Discard) stderr = cmp.Or(stderr, io.Discard) for { - resp, err := recvTransient(ctx, conn) + resp, err := recv(ctx, conn) if err != nil { return pid, 0, false, err } @@ -166,7 +166,7 @@ func pumpStdio(ctx context.Context, conn *silkd.Conn, stdout, stderr io.Writer) // drainData consumes Data frames into sink until Done; sink must not retain its argument. func drainData(ctx context.Context, conn *silkd.Conn, sink chunkSink) error { for { - resp, err := recvTransient(ctx, conn) + resp, err := recv(ctx, conn) if err != nil { return err } @@ -205,18 +205,9 @@ func expect[T any, PT respPtr[T]](ctx context.Context, conn *silkd.Conn) (*T, er return nil, unexpected(resp) } +// recv reads one frame, translating a canceled ctx and an early EOF; a bulk frame's Data is valid until the next read. func recv(ctx context.Context, conn *silkd.Conn) (wire.Response, error) { - return recvWith(ctx, conn.Recv) -} - -// recvTransient is recv for a consumer done with a bulk frame's Data before its next read. -func recvTransient(ctx context.Context, conn *silkd.Conn) (wire.Response, error) { - return recvWith(ctx, conn.RecvTransient) -} - -// recvWith reads one frame, translating a canceled ctx and an early EOF. -func recvWith(ctx context.Context, next func() (wire.Response, error)) (wire.Response, error) { - resp, err := next() + resp, err := conn.RecvTransient() if err != nil { if ctxErr := ctx.Err(); ctxErr != nil { return nil, ctxErr diff --git a/sdk/python/tests/test_upgrade.py b/sdk/python/tests/test_upgrade.py index 0cf149e8..65bc8e65 100644 --- a/sdk/python/tests/test_upgrade.py +++ b/sdk/python/tests/test_upgrade.py @@ -74,18 +74,9 @@ def test_rejected_upgrade_is_an_api_error(): assert agent.early == ["fs_mkdir"], agent.early -def test_run_timeout_cuts_an_unanswered_upgrade(): - server = socket.create_server(("127.0.0.1", 0)) - addr = f"127.0.0.1:{server.getsockname()[1]}" - held: list[socket.socket] = [] - threading.Thread(target=lambda: held.append(server.accept()[0]), daemon=True).start() - sb = sandbox_at(addr, timeout=30, keep_alive=0) +def test_run_timeout_cuts_an_unanswered_upgrade(black_hole): + sb = sandbox_at(black_hole, timeout=30, keep_alive=0) started = time.monotonic() - try: - with pytest.raises(SandboxTimeout): - sb.run(["true"], timeout=0.2) - finally: - server.close() - for conn in held: - conn.close() + with pytest.raises(SandboxTimeout): + sb.run(["true"], timeout=0.2) assert time.monotonic() - started < 2, "the run deadline must cut a hung upgrade under a long client timeout" From 43279d5cbdef3628c49b34d0bdaab83ca1ffbb5e Mon Sep 17 00:00:00 2001 From: CMGS Date: Mon, 5 Oct 2026 18:56:20 +0800 Subject: [PATCH 3/4] review: resolve_program's doc states where it differs from execvp A candidate whose exec fails with ENOENT (a shebang interpreter that is gone) ends the spawn, where execvp would try the next PATH entry; the doc claimed full equivalence. --- silkd/src/exec.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/silkd/src/exec.rs b/silkd/src/exec.rs index abe2ee27..12ee4cd8 100644 --- a/silkd/src/exec.rs +++ b/silkd/src/exec.rs @@ -252,7 +252,7 @@ fn effective_path<'a>( }) } -/// Finds a bare program on path as execvp would, so std can posix_spawn an absolute path instead of forking; a relative entry leaves the search to execvp. +/// Finds a bare program on path like execvp, minus its fall-through past an entry whose exec fails, so std can posix_spawn an absolute path; a relative entry leaves the search to execvp. fn resolve_program(name: &str, path: Option<&str>) -> Option { if name.is_empty() || name.contains('/') { return None; From fc261087cb0c3f5d392db1c18a2bcd9ea8ab7f7a Mon Sep 17 00:00:00 2001 From: CMGS Date: Mon, 5 Oct 2026 19:03:37 +0800 Subject: [PATCH 4/4] cut: OpenPty takes wire.PtyOpen instead of a field-for-field PtyOpts copy PtyOpts duplicated wire.PtyOpen and was copied into it on every call; the SDK already returns wire types (DirEntry, FileInfo). Callers build wire.PtyOpen directly. --- docs/sdk.md | 2 +- e2e/cmd/smoke/main.go | 2 +- sdk/go/pty.go | 14 ++------------ sdk/go/pty_test.go | 8 ++++---- 4 files changed, 8 insertions(+), 18 deletions(-) diff --git a/docs/sdk.md b/docs/sdk.md index 303b6687..6f4fee36 100644 --- a/docs/sdk.md +++ b/docs/sdk.md @@ -645,7 +645,7 @@ error pointing at `Push`. ## Terminals ```go -pty, err := sb.OpenPty(ctx, sandbox.PtyOpts{Cols: 120, Rows: 40}) +pty, err := sb.OpenPty(ctx, wire.PtyOpen{Cols: 120, Rows: 40}) defer pty.Close() pty.Write([]byte("make test\n")) io.Copy(os.Stdout, pty) // EOF when the shell exits diff --git a/e2e/cmd/smoke/main.go b/e2e/cmd/smoke/main.go index daba46e1..ea77258d 100644 --- a/e2e/cmd/smoke/main.go +++ b/e2e/cmd/smoke/main.go @@ -417,7 +417,7 @@ func smokePromote(ctx context.Context, client *sandbox.Client, sb *sandbox.Sandb } func smokePty(ctx context.Context, sb *sandbox.Sandbox) error { - pty, err := sb.OpenPty(ctx, sandbox.PtyOpts{Cols: 80, Rows: 24}) + pty, err := sb.OpenPty(ctx, wire.PtyOpen{Cols: 80, Rows: 24}) if err != nil { return err } diff --git a/sdk/go/pty.go b/sdk/go/pty.go index 117ab49c..9079df0c 100644 --- a/sdk/go/pty.go +++ b/sdk/go/pty.go @@ -9,15 +9,6 @@ import ( "github.com/cocoonstack/sandbox/sdk/go/silkd" ) -// PtyOpts configures OpenPty. -type PtyOpts struct { - Cols uint16 - Rows uint16 - Cwd string - Env map[string]string - User string -} - // Pty is an open pseudo-terminal in the sandbox. type Pty struct { PID uint32 @@ -96,9 +87,8 @@ func (p *Pty) drain(ctx context.Context, pw *io.PipeWriter, stop func()) { } // OpenPty starts a shell whose lifetime is governed by ctx or Close. -func (s *Sandbox) OpenPty(ctx context.Context, opts PtyOpts) (*Pty, error) { - req := &wire.PtyOpen{Cols: opts.Cols, Rows: opts.Rows, Cwd: opts.Cwd, Env: opts.Env, User: opts.User} - conn, l, err := s.call(ctx, req) +func (s *Sandbox) OpenPty(ctx context.Context, opts wire.PtyOpen) (*Pty, error) { + conn, l, err := s.call(ctx, &opts) if err != nil { return nil, err } diff --git a/sdk/go/pty_test.go b/sdk/go/pty_test.go index 66920caf..08c4ec90 100644 --- a/sdk/go/pty_test.go +++ b/sdk/go/pty_test.go @@ -17,7 +17,7 @@ import ( func TestPtyEchoAndExit(t *testing.T) { sb := fakeSandbox(t) - pty, err := sb.OpenPty(t.Context(), PtyOpts{Cols: 80, Rows: 24}) + pty, err := sb.OpenPty(t.Context(), wire.PtyOpen{Cols: 80, Rows: 24}) if err != nil { t.Fatalf("OpenPty: %v", err) } @@ -67,7 +67,7 @@ func TestPtyReleasesTheRelayWhenTheShellExits(t *testing.T) { _, _ = r.ReadByte() close(released) }) - pty, err := legacySandbox(t, ts).OpenPty(t.Context(), PtyOpts{Cols: 80, Rows: 24}) + pty, err := legacySandbox(t, ts).OpenPty(t.Context(), wire.PtyOpen{Cols: 80, Rows: 24}) if err != nil { t.Fatalf("OpenPty: %v", err) } @@ -88,7 +88,7 @@ func TestPtyReleasesTheRelayWhenTheShellExits(t *testing.T) { func TestPtyCtxCancelIsTyped(t *testing.T) { sb := fakeSandbox(t) ctx, cancel := context.WithCancel(t.Context()) - pty, err := sb.OpenPty(ctx, PtyOpts{Cols: 80, Rows: 24}) + pty, err := sb.OpenPty(ctx, wire.PtyOpen{Cols: 80, Rows: 24}) if err != nil { t.Fatalf("OpenPty: %v", err) } @@ -103,7 +103,7 @@ func TestPtyCtxCancelIsTyped(t *testing.T) { func TestPtyCloseUnblocksUnreadOutput(t *testing.T) { sb := fakeSandbox(t) - pty, err := sb.OpenPty(t.Context(), PtyOpts{Cols: 80, Rows: 24}) + pty, err := sb.OpenPty(t.Context(), wire.PtyOpen{Cols: 80, Rows: 24}) if err != nil { t.Fatalf("OpenPty: %v", err) }