diff --git a/boot/init/src/boot.rs b/boot/init/src/boot.rs index 0da67981..622be227 100644 --- a/boot/init/src/boot.rs +++ b/boot/init/src/boot.rs @@ -233,11 +233,16 @@ fn scan_serials(ids: &[&str], found: &mut [Option]) { format!("/sys/block/{name}/serial"), format!("/sys/block/{name}/device/serial"), ]; + let device = format!("/dev/{name}"); for path in paths { let Ok(serial) = fs::read_to_string(&path) else { continue; }; - record_serial(ids, found, serial.trim_end(), &format!("/dev/{name}")); + if serial.trim_end().is_empty() { + continue; + } + record_serial(ids, found, serial.trim_end(), &device); + break; } } } diff --git a/boot/init/src/cfg.rs b/boot/init/src/cfg.rs index 23cfee09..a5e58fee 100644 --- a/boot/init/src/cfg.rs +++ b/boot/init/src/cfg.rs @@ -128,9 +128,7 @@ fn debug_token(val: &str) -> bool { val.is_empty() || val == "1" } -/// Kernel ip= fields: client:server:gw:netmask:hostname:device:autoconf[:dns0[:dns1]]. -/// Shorthand forms (ip=dhcp, ip=off) and malformed params are ignored — the -/// baked DHCP .network fallback then covers the NIC, like the old hook did. +/// Kernel ip= fields: client:server:gw:netmask:hostname:device:autoconf[:dns0[:dns1]]; anything else is ignored. fn parse_ip_param(val: &str) -> Option { let f: Vec<&str> = val.split(':').collect(); if f.len() < 7 || f[0].is_empty() || f[5].is_empty() { diff --git a/boot/init/src/main.rs b/boot/init/src/main.rs index 2672d36f..13a8921b 100644 --- a/boot/init/src/main.rs +++ b/boot/init/src/main.rs @@ -14,7 +14,6 @@ fn main() { boot::run() } -/// Keeps `cargo test` runnable on non-Linux dev hosts (cfg logic tests). #[cfg(not(target_os = "linux"))] fn main() { eprintln!("sandbox-init is Linux-only"); diff --git a/docs/deploy.md b/docs/deploy.md index 5ffe8288..d9795d9a 100644 --- a/docs/deploy.md +++ b/docs/deploy.md @@ -86,6 +86,7 @@ sandboxd reads one JSON file (`-config`, default | `cocoon_bin` | `cocoon` | cocoon CLI binary | | `restore_mode` | unset | clone and wake-restore memory mode: `copy`, `ondemand`, or `mmap`; use `mmap` for dense pools | | `no_direct_io` | false | use buffered writable disks for Cloud Hypervisor cold boots and clones; recommended for dense ephemeral pools to avoid direct-I/O CoW journal contention | +| `no_balloon` | false | boot pool and template VMs without the virtio-balloon (cocoon otherwise returns 25% of guest memory to the host); clones inherit it from the golden. A guest that thrashes before deflate-on-OOM fires — a 16G build tier running a large typecheck — needs its whole memory | | `advertise_addr` | = `listen` | the host:port clients reach this node at; returned as a claim's owner address and gossiped to peers. Must be routable when `listen` is a wildcard | | `bridges` / `networks` | unset | egress-lane attachment: a list of host bridge devices, or a list of CNI conflist names. Mutually exclusive; with neither set the node serves only the no-network lane. A Linux bridge holds at most 1024 ports (kernel `BR_MAX_PORTS`), so an N-entry list raises the node's egress ceiling to N×1024 — VMs spread over the list by a stable hash of the VM name, so size it with headroom (the spread is statistical, not exact). `bridges` keeps the raw TAP-on-bridge attachment (taps in the root netns, no per-VM network namespace or CNI plugin execution); `networks` runs the CNI chain per VM. [Guarded egress](egress.md) needs `bridges` and rejects a CNI network at load | | `volumes` | unset | node-local catalog of operator-managed dataset images: `[ {"name":"imagenet","path":"/srv/datasets/imagenet.img","directio":"off","tenants":["acme"]}, {"name":"scratch-db","path":"/srv/datasets/scratch.img","writable":true} ]`. Names match `^[a-z][a-z0-9_-]{0,19}$` and cannot start with `cocoon-`; paths are absolute; `directio` is `on`, `off`, or `auto` and defaults to `off` for both read-only and writable entries. `tenants` is an optional access list: empty means every authenticated scope, while every listed name must exist in the node's `tenants` config; root always has access. `writable` (default `false`) lets a claim request `mode: "rw"` on that entry — see [Dataset volumes](#dataset-volumes). The catalog is intentionally not part of the cluster digest | @@ -104,6 +105,7 @@ sandboxd reads one JSON file (`-config`, default | `checkpoint_ttl_hours` | 0 (keep forever) | ages out checkpoints older than this; the sweep runs hourly and at startup. Explicit deletes never wait for it. Must be nonzero and match fleet-wide when `checkpoint_peer_heal` is on — it is the expiry eligibility point for a healed replica a delete broadcast missed, after which its next successful hourly sweep removes it; persistent sweep failure extends retention until one succeeds, so it is not a hard ceiling | | `checkpoint_peer_heal` | false | on a cluster, lets a node pull a checkpoint it lacks from a peer — found via a live probe, not gossip — rather than failing the branch; see [placement lifecycle](cluster.md#checkpoints-on-a-cluster). Three requirements, all enforced at config load: a nonempty `api_token` (the blob transfer between peers authenticates with it; without one the raw record stream would be open), `mesh.cluster_key` set (the pull presents the fleet `api_token` to an address learned from the peer probe, so the gossip layer carrying that address must itself be authenticated), and `checkpoint_ttl_hours` nonzero (a replica a delete broadcast missed becomes eligible for expiry after it, and its next successful hourly sweep removes it — so it is the finite eligibility point, not an exact ceiling). A shared checkpoint store (`checkpoint_store` kind `s3`) ignores this setting — every node already resolves every checkpoint directly, so there is nothing to heal | | `warm_max` (pool entry) | 0 (static) | turns on the demand-adaptive watermark for that pool: the warm target rises from `warm` toward `warm_max` while claims arrive faster than the measured provision lead covers, and decays back over ~a minute of silence | +| `warmup` (pool entry) | unset | argv run in the golden VM after readiness and before its snapshot, so the files it touches are page-cache-resident in every clone — e.g. `["node", "-e", "0"]` on a Node flavor. It runs under the engine's 2-minute command timeout in silkd's base environment (`PATH`, `TERM`, and the node's proxy variables on the none lane); a non-zero exit or a timeout fails the golden build, so the pool stays unfilled until the config is fixed. Config-owned like `egress`: `PUT /v1/pools` rejects it, and a golden built with a different warmup is rebuilt | | `max_claims` | 0 (unlimited) | node-wide cap on live claims; claim/fork/branch requests beyond it answer 429 with the pool state unharmed (on a cluster, normal warm-candidate placement applies, with volume claims limited to candidates holding every requested volume) | | `audit_log` | false | append every relayed request frame's op + addressing fields (never payloads) to `/audit.jsonl`, size-rotated with one `.1` backup. Records are `{t, id, op}` plus whichever addressing fields the op carries (`argv`, `path`, `dest`, `from`, `to`, `url`, `session`, `port`), plus `decision` and `secret` (the ref name, never its value) on `egress` records; preview accesses record as op `preview`, one per request. A request frame whose first line exceeds 4 KiB is skipped, never truncated | | `idle_hibernate_seconds` | 0 (off) | node-wide idle policy for unpooled claims (template/checkpoint claims): a none-lane claim with no data-plane connection for this long is hibernated; the next call that reaches the guest wakes it transparently. Per-pool `idle_hibernate_seconds` does the same for that pool's claims; pooled keys ignore the node-wide value, and egress pools reject it because they cannot resume safely. Opt in deliberately: a wake costs latency and the snapshot, so callers with their own idle logic must not pay twice | diff --git a/docs/sandboxd-api.md b/docs/sandboxd-api.md index d9b89bd0..a6bd87c6 100644 --- a/docs/sandboxd-api.md +++ b/docs/sandboxd-api.md @@ -325,7 +325,8 @@ targets online — no restart, live claims untouched: Pools omitted from the list are drained: their unclaimed warm VMs are destroyed and the pool entry retires. `net`/`size` default like a claim's. Answers the fresh `GET /v1/info` payload. 400 bad key, negative warm/idle, -`warm_max` below `warm`, or duplicate pool; 401 bad api token; 409 egress +`warm_max` below `warm`, duplicate pool, or a config-owned `egress`/`warmup` +field; 401 bad api token; 409 egress pool on a node without an egress attachment. ## POST /v1/drain @@ -515,6 +516,28 @@ the connection is a byte-for-byte relay to the guest's silkd (one silkd RPC per connection — see [silkd](silkd.md)). 426 without the upgrade header, 404 unknown sandbox, 502 guest unreachable. +## POST /v1/sandboxes/{id}/exec + +Auth: the sandbox's own token. A buffered exec for clients that cannot hold +an upgraded connection — plain JSON in, plain JSON out, so it multiplexes over +HTTP/2 through a TLS proxy: + +```json +{"argv": ["node", "-v"], "cwd": "/work", "env": {"CI": "1"}, "timeout_seconds": 60} +``` + +→ `200 {"exit_code": 0, "stdout": "v22.23.2\n", "stderr": ""}`. The command +runs to completion (no stdin, no streaming, no detach — use the relay for +those); `timeout_seconds` 0 means no limit beyond the request itself. When +the node gives up on a started command — timeout, client gone, output cap — +it kills the child through silkd before answering, so nothing keeps running +behind a 504. Output comes back as JSON strings: bytes that are not valid +UTF-8 are replaced with U+FFFD, so binary output belongs on the relay. 400 +empty `argv`, negative `timeout_seconds`, an unknown field, or a silkd +`bad_request`; 401 missing bearer token; 404 unknown sandbox or wrong token; +413 when stdout+stderr exceed 8 MiB; 502 guest unreachable or any other +silkd error. A hibernated sandbox wakes transparently like on the relay. + ## GET /v1/sandboxes/{id}/owner Auth: the sandbox's own token. Answers `{"owner_addr": "host:port"}` when diff --git a/e2e/cmd/smoke/main.go b/e2e/cmd/smoke/main.go index b8392f14..4ff229b6 100644 --- a/e2e/cmd/smoke/main.go +++ b/e2e/cmd/smoke/main.go @@ -376,11 +376,11 @@ func smokeFork(ctx context.Context, sb *sandbox.Sandbox) error { } for i := range children { if _, err := children[1-i].Stat(ctx, fmt.Sprintf("/work/child-%d.txt", i)); err == nil { - return fmt.Errorf("child %d write visible in sibling — shared disk?", i) + return fmt.Errorf("child %d write visible in sibling: shared disk", i) } } if _, err := sb.Stat(ctx, "/work/child-0.txt"); err == nil { - return errors.New("child write visible in parent — shared disk?") + return errors.New("child write visible in parent: shared disk") } return nil } @@ -693,8 +693,8 @@ func want(got, exp string) error { } func isSilkdKind(err error, kind string) bool { - var er *wire.ErrorResp - return errors.As(err, &er) && er.Kind == kind + er, ok := errors.AsType[*wire.ErrorResp](err) + return ok && er.Kind == kind } func lspWrite(w io.Writer, body string) error { diff --git a/e2e/cmd/volumesmoke/main.go b/e2e/cmd/volumesmoke/main.go index a0882806..6dce910e 100644 --- a/e2e/cmd/volumesmoke/main.go +++ b/e2e/cmd/volumesmoke/main.go @@ -105,11 +105,11 @@ func run(addr, token, template, volume, rwVolume, probe string) error { } _, err = sbA.Exec(ctx, "touch", path.Join(mountA, ".sandboxd-write-probe")) - var exitErr *sandbox.ExitError if err == nil { return errors.New("write to read-only volume succeeded") } - if !errors.As(err, &exitErr) || !strings.Contains(strings.ToLower(exitErr.Stderr), "read-only file system") { + exitErr, ok := errors.AsType[*sandbox.ExitError](err) + if !ok || !strings.Contains(strings.ToLower(exitErr.Stderr), "read-only file system") { return fmt.Errorf("write failed without EROFS: %w", err) } diff --git a/e2e/fakeengine_test.go b/e2e/fakeengine_test.go index f5f05914..ca7909a3 100644 --- a/e2e/fakeengine_test.go +++ b/e2e/fakeengine_test.go @@ -30,7 +30,7 @@ type fakeEngine struct { func newFakeEngine(dir string) *fakeEngine { return &fakeEngine{ - real: engine.New("cocoon", nil, nil, false, ""), + real: engine.New("cocoon", nil, nil, false, false, ""), dir: dir, listeners: map[string]io.Closer{}, socks: map[string]string{}, @@ -110,6 +110,8 @@ func (f *fakeEngine) DialGuestPort(context.Context, string, uint16) (net.Conn, e func (f *fakeEngine) InstallCACert(context.Context, string, []byte) error { return nil } +func (f *fakeEngine) Warmup(context.Context, string, []string) error { return nil } + func (f *fakeEngine) DiskAttach(_ context.Context, vmName string, spec engine.VolumeSpec) error { f.mu.Lock() defer f.mu.Unlock() diff --git a/mcp/e2e.py b/mcp/e2e.py index 24d7d6cd..8f576374 100644 --- a/mcp/e2e.py +++ b/mcp/e2e.py @@ -13,7 +13,9 @@ class McpClient: def __init__(self, argv): - self.proc = subprocess.Popen(argv, stdin=subprocess.PIPE, stdout=subprocess.PIPE) + self.proc = subprocess.Popen( + argv, stdin=subprocess.PIPE, stdout=subprocess.PIPE + ) self.seq = 0 def call(self, method, params=None): @@ -52,12 +54,18 @@ def main() -> int: parser.add_argument("--template", default="rt:24.04") args = parser.parse_args() - mcp = McpClient([args.bin, "-addr", args.addr, "-token", args.token, "-template", args.template]) + mcp = McpClient( + [args.bin, "-addr", args.addr, "-token", args.token, "-template", args.template] + ) try: - init = mcp.call("initialize", {"protocolVersion": "2024-11-05", "capabilities": {}}) + init = mcp.call( + "initialize", {"protocolVersion": "2024-11-05", "capabilities": {}} + ) assert init["serverInfo"]["name"] == "sandbox-mcp", init tools = {t["name"] for t in mcp.call("tools/list")["tools"]} - assert {"create_sandbox", "exec", "checkpoint", "branch_checkpoint"} <= tools, tools + assert {"create_sandbox", "exec", "checkpoint", "branch_checkpoint"} <= tools, ( + tools + ) print(f" initialize + tools/list ok ({len(tools)} tools)") sandbox_id = mcp.tool("create_sandbox")["sandbox_id"] @@ -67,11 +75,15 @@ def main() -> int: mcp.tool("write_file", sandbox_id=sandbox_id, path="/root/m.txt", content="v1") assert mcp.tool("read_file", sandbox_id=sandbox_id, path="/root/m.txt") == "v1" - names = {e["name"] for e in mcp.tool("list_dir", sandbox_id=sandbox_id, path="/root")} + names = { + e["name"] for e in mcp.tool("list_dir", sandbox_id=sandbox_id, path="/root") + } assert "m.txt" in names, names print(" files ok") - ckpt = mcp.tool("checkpoint", sandbox_id=sandbox_id, name="mcp-step")["checkpoint_id"] + ckpt = mcp.tool("checkpoint", sandbox_id=sandbox_id, name="mcp-step")[ + "checkpoint_id" + ] mcp.tool("write_file", sandbox_id=sandbox_id, path="/root/m.txt", content="v2") branch = mcp.tool("branch_checkpoint", checkpoint_id=ckpt)["sandbox_id"] assert mcp.tool("read_file", sandbox_id=branch, path="/root/m.txt") == "v1" diff --git a/mcp/server.go b/mcp/server.go index 23647e6e..d50b3330 100644 --- a/mcp/server.go +++ b/mcp/server.go @@ -20,6 +20,24 @@ const ( defaultToolTTL = time.Hour ) +type rpcRequest struct { + ID json.RawMessage `json:"id"` + Method string `json:"method"` + Params json.RawMessage `json:"params"` +} + +type rpcResponse struct { + JSONRPC string `json:"jsonrpc"` + ID json.RawMessage `json:"id"` + Result any `json:"result,omitempty"` + Error *rpcErrorBody `json:"error,omitempty"` +} + +type rpcErrorBody struct { + Code int `json:"code"` + Message string `json:"message"` +} + // server owns one sandboxd client and the handles minted over this stdio // session: MCP tools address sandboxes and checkpoints by id, so the live // handles (with their tokens) stay here. @@ -179,24 +197,6 @@ func (s *server) dropCkpt(id string) { delete(s.ckpts, id) } -type rpcRequest struct { - ID json.RawMessage `json:"id"` - Method string `json:"method"` - Params json.RawMessage `json:"params"` -} - -type rpcResponse struct { - JSONRPC string `json:"jsonrpc"` - ID json.RawMessage `json:"id"` - Result any `json:"result,omitempty"` - Error *rpcErrorBody `json:"error,omitempty"` -} - -type rpcErrorBody struct { - Code int `json:"code"` - Message string `json:"message"` -} - func result(id json.RawMessage, v any) rpcResponse { return rpcResponse{JSONRPC: "2.0", ID: id, Result: v} } diff --git a/protocol/wire/frame.go b/protocol/wire/frame.go index 54345a79..74ba6233 100644 --- a/protocol/wire/frame.go +++ b/protocol/wire/frame.go @@ -182,7 +182,7 @@ func (Ps) Op() string { return "ps" } // Kill signals a process; nil Signal means SIGKILL. type Kill struct { PID uint32 `json:"pid"` - Signal *int32 `json:"signal,omitempty"` + Signal *int32 `json:"signal,omitzero"` } func (Kill) Op() string { return "kill" } @@ -238,7 +238,7 @@ func (StdinClose) Op() string { return "stdin_close" } // defaults. type FsWrite struct { Path string `json:"path"` - Mode *uint32 `json:"mode,omitempty"` + Mode *uint32 `json:"mode,omitzero"` } func (FsWrite) Op() string { return "fs_write" } @@ -366,7 +366,7 @@ type GitClone struct { URL string `json:"url"` Path string `json:"path"` Branch string `json:"branch,omitempty"` - Depth uint32 `json:"depth,omitempty"` + Depth uint32 `json:"depth,omitzero"` Auth string `json:"auth,omitempty"` } @@ -524,7 +524,7 @@ type ProcInfo struct { Argv []string `json:"argv"` Detached bool `json:"detached"` State string `json:"state"` - ExitCode *int32 `json:"exit_code,omitempty"` + ExitCode *int32 `json:"exit_code,omitzero"` StartedAtEpochSecs uint64 `json:"started_at_epoch_secs"` } @@ -630,7 +630,7 @@ type GitStatusResult struct { Ahead uint32 `json:"ahead"` Behind uint32 `json:"behind"` Files []GitFileStatus `json:"files"` - Truncated bool `json:"truncated,omitempty"` + Truncated bool `json:"truncated,omitzero"` } func (GitStatusResult) RespType() string { return "git_status_result" } diff --git a/sandboxd/ca.go b/sandboxd/ca.go index 7547b32c..789ec7d8 100644 --- a/sandboxd/ca.go +++ b/sandboxd/ca.go @@ -10,7 +10,6 @@ import ( "github.com/cocoonstack/sandbox/sandboxd/egress" ) -// runCA is the operator PKI tool: mint the cluster root, then per-node intermediates. func runCA(args []string) error { if len(args) == 0 { return fmt.Errorf("usage: sandboxd ca {init|issue-intermediate}") diff --git a/sandboxd/config/config.go b/sandboxd/config/config.go index 2d4cd98d..ed11592f 100644 --- a/sandboxd/config/config.go +++ b/sandboxd/config/config.go @@ -40,19 +40,22 @@ type PoolSpec struct { Warm int `json:"warm"` // WarmMax, when >0, lets the warm target rise from Warm toward it under demand. - WarmMax int `json:"warm_max,omitempty"` + WarmMax int `json:"warm_max,omitzero"` // Egress is this pool's allow-list, intersected with the tenant's; nil denies all egress. Egress *egress.Policy `json:"egress,omitempty"` + // Warmup runs in the golden VM before its snapshot, so every clone starts with its page cache. + Warmup []string `json:"warmup,omitempty"` + // IdleHibernateSeconds, when >0, hibernates idle claims after that many seconds. - IdleHibernateSeconds int `json:"idle_hibernate_seconds,omitempty"` + IdleHibernateSeconds int `json:"idle_hibernate_seconds,omitzero"` // ArchiveAfterSeconds, when >0, archives a hibernated claim; must exceed IdleHibernateSeconds. - ArchiveAfterSeconds int `json:"archive_after_seconds,omitempty"` + ArchiveAfterSeconds int `json:"archive_after_seconds,omitzero"` // ArchiveDeleteAfterSeconds, when >0, purges the checkpoint that long after archiving. - ArchiveDeleteAfterSeconds int `json:"archive_delete_after_seconds,omitempty"` + ArchiveDeleteAfterSeconds int `json:"archive_delete_after_seconds,omitzero"` } // ValidateLimits checks the warm/watermark/idle bounds shared by config and PUT /v1/pools. @@ -69,6 +72,9 @@ func (s PoolSpec) ValidateLimits() error { if s.Net == types.NetEgress && s.IdleHibernateSeconds > 0 { return fmt.Errorf("idle_hibernate_seconds is not supported for egress pools") } + if slices.Contains(s.Warmup, "") { + return fmt.Errorf("warmup must not contain an empty argument") + } return validateArchiveWindow(s.IdleHibernateSeconds, s.ArchiveAfterSeconds, s.ArchiveDeleteAfterSeconds) } @@ -94,7 +100,7 @@ func (e *EgressCAConfig) Set() bool { type TenantSpec struct { Name string `json:"name"` Token string `json:"token"` //nolint:gosec // config field, not a hardcoded credential - MaxClaims int `json:"max_claims,omitempty"` + MaxClaims int `json:"max_claims,omitzero"` // Egress is the tenant's allow-list (see PoolSpec.Egress). Egress *egress.Policy `json:"egress,omitempty"` @@ -105,7 +111,7 @@ type VolumeSpec struct { Name string `json:"name"` Path string `json:"path"` DirectIO string `json:"directio,omitempty"` - Writable bool `json:"writable,omitempty"` + Writable bool `json:"writable,omitzero"` Tenants []string `json:"tenants,omitempty"` } @@ -169,7 +175,10 @@ type Config struct { RestoreMode types.RestoreMode `json:"restore_mode,omitempty"` // NoDirectIO enables buffered writable disks for cold boots and clones. - NoDirectIO bool `json:"no_direct_io,omitempty"` + NoDirectIO bool `json:"no_direct_io,omitzero"` + + // NoBalloon boots VMs without the virtio-balloon, so a guest keeps its whole memory. + NoBalloon bool `json:"no_balloon,omitzero"` // APIToken, when set, guards claim and info. APIToken string `json:"api_token,omitempty"` //nolint:gosec // config field, not a hardcoded credential @@ -181,11 +190,11 @@ type Config struct { Secrets []egress.SecretSpec `json:"secrets,omitempty"` // IdleHibernateSeconds is the idle policy for unpooled claims; per-pool settings override. - IdleHibernateSeconds int `json:"idle_hibernate_seconds,omitempty"` + IdleHibernateSeconds int `json:"idle_hibernate_seconds,omitzero"` // ArchiveAfterSeconds and ArchiveDeleteAfterSeconds are the archive policy for unpooled keys. - ArchiveAfterSeconds int `json:"archive_after_seconds,omitempty"` - ArchiveDeleteAfterSeconds int `json:"archive_delete_after_seconds,omitempty"` + ArchiveAfterSeconds int `json:"archive_after_seconds,omitzero"` + ArchiveDeleteAfterSeconds int `json:"archive_delete_after_seconds,omitzero"` PreviewListen string `json:"preview_listen,omitempty"` PreviewSecret string `json:"preview_secret,omitempty"` //nolint:gosec // config field, not a hardcoded credential @@ -199,28 +208,28 @@ type Config struct { CheckpointStore *StoreConfig `json:"checkpoint_store,omitempty"` // CheckpointPeerHeal lets a node pull a checkpoint it lacks from a peer; off by default. - CheckpointPeerHeal bool `json:"checkpoint_peer_heal,omitempty"` + CheckpointPeerHeal bool `json:"checkpoint_peer_heal,omitzero"` // EgressInternalAllow re-admits CIDRs through the proxy's SSRF guard, node-wide. EgressInternalAllow []string `json:"egress_internal_allow,omitempty"` // CheckpointTTLHours ages out checkpoints; 0 keeps them forever. - CheckpointTTLHours int `json:"checkpoint_ttl_hours,omitempty"` + CheckpointTTLHours int `json:"checkpoint_ttl_hours,omitzero"` // MaxClaims caps live claims node-wide; 0 means unlimited. - MaxClaims int `json:"max_claims,omitempty"` + MaxClaims int `json:"max_claims,omitzero"` // AuditLog, when true, appends relayed request ops, never payloads, to audit.jsonl. - AuditLog bool `json:"audit_log,omitempty"` + AuditLog bool `json:"audit_log,omitzero"` // MaxForkCount caps children per fork call; each child is a full-RAM VM. - MaxForkCount int `json:"max_fork_count,omitempty"` + MaxForkCount int `json:"max_fork_count,omitzero"` // Volumes is the node-local catalog of operator-managed dataset images. Volumes []VolumeSpec `json:"volumes,omitempty"` // RefillConcurrency caps concurrent VM provisioning node-wide; 0 auto-scales with CPUs. - RefillConcurrency int `json:"refill_concurrency,omitempty"` + RefillConcurrency int `json:"refill_concurrency,omitzero"` // Mesh, when set, joins this node to a memberlist cluster; nil is a mesh of one. Mesh *MeshConfig `json:"mesh,omitempty"` diff --git a/sandboxd/egress/ca.go b/sandboxd/egress/ca.go index 0faa2259..2469b5ee 100644 --- a/sandboxd/egress/ca.go +++ b/sandboxd/egress/ca.go @@ -93,10 +93,7 @@ func (c *CA) SignLeaf(host string) (*tls.Certificate, error) { return nil, err } now := time.Now() - // Guests reject a chain whose leaf outlives or postdates its intermediate. - if now.After(c.interCert.NotAfter) { - return nil, fmt.Errorf("sign leaf %s: intermediate expired %s", host, c.interCert.NotAfter.Format(time.RFC3339)) - } + // Guests reject a chain whose leaf outlives its intermediate. notAfter := now.Add(leafValidity) if notAfter.After(c.interCert.NotAfter) { notAfter = c.interCert.NotAfter diff --git a/sandboxd/egress/ca_test.go b/sandboxd/egress/ca_test.go index 05e88f95..beb518f3 100644 --- a/sandboxd/egress/ca_test.go +++ b/sandboxd/egress/ca_test.go @@ -7,6 +7,7 @@ import ( "crypto/x509" "crypto/x509/pkix" "encoding/pem" + "slices" "testing" "time" ) @@ -134,7 +135,7 @@ func TestLoadCARootBundle(t *testing.T) { if err != nil { t.Fatalf("intermediate: %v", err) } - bundle := append(append([]byte{}, rootA...), rootB...) + bundle := slices.Concat(rootA, rootB) ca, err := LoadCA(bundle, interCert, interKey) if err != nil { t.Fatalf("LoadCA rotation bundle: %v", err) @@ -160,7 +161,7 @@ func TestLoadCARejectsNonCertBlockInRoot(t *testing.T) { if err != nil { t.Fatalf("intermediate: %v", err) } - polluted := append(append([]byte{}, rootCert...), rootKey...) + polluted := slices.Concat(rootCert, rootKey) if _, err := LoadCA(polluted, interCert, interKey); err == nil { t.Error("LoadCA accepted a non-certificate pem block in the root bundle") } diff --git a/sandboxd/egress/policy.go b/sandboxd/egress/policy.go index 49f175f7..a7d2a265 100644 --- a/sandboxd/egress/policy.go +++ b/sandboxd/egress/policy.go @@ -23,7 +23,7 @@ type Rule struct { Host string `json:"host"` Methods []string `json:"methods,omitempty"` Secret string `json:"secret,omitempty"` //nolint:gosec // reference name of a node-side secret, never a value - Intercept bool `json:"intercept,omitempty"` + Intercept bool `json:"intercept,omitzero"` } // matches expects host already lowercased by Eval. diff --git a/sandboxd/engine/cloneargs_test.go b/sandboxd/engine/cloneargs_test.go index fe333429..dc4e9b48 100644 --- a/sandboxd/engine/cloneargs_test.go +++ b/sandboxd/engine/cloneargs_test.go @@ -23,7 +23,7 @@ func TestCloneArgsRestoreMode(t *testing.T) { } for _, tc := range cases { t.Run(tc.name, func(t *testing.T) { - e := New("cocoon", []string{"br0"}, nil, false, tc.mode) + e := New("cocoon", []string{"br0"}, nil, false, false, tc.mode) for _, args := range [][]string{ e.cloneArgs("/goldens/g1", "sbx-1", tc.key), e.cloneSnapArgs("ck_1", "sbx-1", tc.key), @@ -45,7 +45,7 @@ func TestLifecycleArgsApplyDirectIOPolicy(t *testing.T) { key := types.PoolKey{Template: "rt:24.04", Net: types.NetNone, Size: types.SizeMedium} for _, noDirectIO := range []bool{false, true} { t.Run(strconv.FormatBool(noDirectIO), func(t *testing.T) { - e := New("cocoon", nil, nil, noDirectIO, "") + e := New("cocoon", nil, nil, noDirectIO, false, "") want := "--no-direct-io=" + strconv.FormatBool(noDirectIO) cold := e.runColdArgs("sbx-1", key) for _, args := range [][]string{ @@ -71,8 +71,8 @@ func TestEgressVMsSpreadOverEveryConfiguredShard(t *testing.T) { e *Engine flag string }{ - "networks": {New("cocoon", nil, shards, false, ""), "--network"}, - "bridges": {New("cocoon", shards, nil, false, ""), "--bridge"}, + "networks": {New("cocoon", nil, shards, false, false, ""), "--network"}, + "bridges": {New("cocoon", shards, nil, false, false, ""), "--bridge"}, } { t.Run(name, func(t *testing.T) { counts := map[string]int{} @@ -107,13 +107,28 @@ func TestNetArgsHonorsTheLaneAndTheAttachment(t *testing.T) { none := types.PoolKey{Template: "rt:24.04", Net: types.NetNone, Size: types.SizeMedium} egress := types.PoolKey{Template: "rt:24.04", Net: types.NetEgress, Size: types.SizeMedium} - if args := New("cocoon", nil, []string{"cni"}, false, "").netArgs("sbx-1", none, false); len(args) != 0 { + if args := New("cocoon", nil, []string{"cni"}, false, false, "").netArgs("sbx-1", none, false); len(args) != 0 { t.Errorf("none lane took an attachment: %v", args) } - if args := New("cocoon", []string{"br0"}, nil, false, "").netArgs("sbx-1", egress, false); !slices.Equal(args, []string{"--bridge", "br0"}) { + if args := New("cocoon", []string{"br0"}, nil, false, false, "").netArgs("sbx-1", egress, false); !slices.Equal(args, []string{"--bridge", "br0"}) { t.Errorf("bridge lane args = %v", args) } - if args := New("cocoon", nil, []string{"cocoon-dhcp"}, false, "").netArgs("sbx-1", egress, false); !slices.Equal(args, []string{"--network", "cocoon-dhcp"}) { + if args := New("cocoon", nil, []string{"cocoon-dhcp"}, false, false, "").netArgs("sbx-1", egress, false); !slices.Equal(args, []string{"--network", "cocoon-dhcp"}) { t.Errorf("single-network args = %v", args) } } + +func TestNoBalloonReachesColdBootsOnly(t *testing.T) { + key := types.PoolKey{Template: "rt:24.04", Net: types.NetNone, Size: types.SizeSmall} + for _, noBalloon := range []bool{false, true} { + t.Run(strconv.FormatBool(noBalloon), func(t *testing.T) { + e := New("cocoon", nil, nil, false, noBalloon, "") + if got := slices.Contains(e.runColdArgs("sbx-1", key), "--no-balloon"); got != noBalloon { + t.Errorf("cold args carry --no-balloon = %v, want %v", got, noBalloon) + } + if slices.Contains(e.cloneArgs("/goldens/g1", "sbx-1", key), "--no-balloon") { + t.Error("clone args carry --no-balloon; clones inherit it from the golden") + } + }) + } +} diff --git a/sandboxd/engine/engine.go b/sandboxd/engine/engine.go index 330264d2..e3491833 100644 --- a/sandboxd/engine/engine.go +++ b/sandboxd/engine/engine.go @@ -65,12 +65,13 @@ type Engine struct { bridges []string networks []string noDirectIO bool + noBalloon bool restoreMode types.RestoreMode } // New returns a cocoon engine with node-wide network and disk policy. -func New(bin string, bridges, networks []string, noDirectIO bool, restoreMode types.RestoreMode) *Engine { - return &Engine{bin: bin, bridges: bridges, networks: networks, noDirectIO: noDirectIO, restoreMode: restoreMode} +func New(bin string, bridges, networks []string, noDirectIO, noBalloon bool, restoreMode types.RestoreMode) *Engine { + return &Engine{bin: bin, bridges: bridges, networks: networks, noDirectIO: noDirectIO, noBalloon: noBalloon, restoreMode: restoreMode} } // Version reports cocoon's version string: a "vX.Y.Z" release or a "master-" dev build. @@ -275,11 +276,7 @@ func (e *Engine) DialGuestPort(ctx context.Context, vsockSocket string, port uin } stop := context.AfterFunc(ctx, func() { _ = conn.Close() }) defer stop() - req, err := wire.EncodeRequest(wire.PortForward{Port: port}) - if err != nil { - _ = conn.Close() - return nil, err - } + req, _ := wire.EncodeRequest(wire.PortForward{Port: port}) if _, err := conn.Write(append(req, '\n')); err != nil { _ = conn.Close() return nil, fmt.Errorf("write port_forward: %w", err) @@ -338,6 +335,9 @@ func (e *Engine) restoreArgs() []string { func (e *Engine) runColdArgs(name string, key types.PoolKey) []string { spec, _ := key.Size.Spec() args := []string{"vm", "run", argName, name, argOutput, formatJSON, "--cpu", strconv.Itoa(spec.CPU), "--memory", spec.Memory, e.directIOArg()} + if e.noBalloon { + args = append(args, "--no-balloon") + } args = append(args, e.netArgs(name, key, true)...) return append(args, key.Template) } @@ -376,10 +376,7 @@ func (e *Engine) infoRoundTrip(ctx context.Context, vsockSocket string) error { defer func() { _ = conn.Close() }() stop := context.AfterFunc(ctx, func() { _ = conn.Close() }) defer stop() - probe, err := wire.EncodeRequest(wire.Info{}) - if err != nil { - return fmt.Errorf("encode info: %w", err) - } + probe, _ := wire.EncodeRequest(wire.Info{}) if _, err = conn.Write(append(probe, '\n')); err != nil { return fmt.Errorf("write info: %w", err) } diff --git a/sandboxd/engine/engine_test.go b/sandboxd/engine/engine_test.go index 790a9bfb..7143625a 100644 --- a/sandboxd/engine/engine_test.go +++ b/sandboxd/engine/engine_test.go @@ -23,7 +23,7 @@ func TestDialSilkdConsumesOnlyHandshake(t *testing.T) { listenMuxer(t, path, "OK 2048\nX") - conn, err := New("cocoon", nil, nil, false, "").DialSilkd(t.Context(), path) + conn, err := New("cocoon", nil, nil, false, false, "").DialSilkd(t.Context(), path) if err != nil { t.Fatalf("DialSilkd: %v", err) } @@ -41,7 +41,7 @@ func TestDialSilkdRejectedHandshake(t *testing.T) { path := sockPath(t) listenMuxer(t, path, "ERR no guest listener\n") - _, err := New("cocoon", nil, nil, false, "").DialSilkd(t.Context(), path) + _, err := New("cocoon", nil, nil, false, false, "").DialSilkd(t.Context(), path) if err == nil || !strings.Contains(err.Error(), "no guest listener") { t.Errorf("got %v, want handshake rejection", err) } @@ -51,7 +51,7 @@ func TestProbeSucceeds(t *testing.T) { path := sockPath(t) listenMuxer(t, path, "OK 2048\n", infoFrame) - if err := New("cocoon", nil, nil, false, "").Probe(t.Context(), path, 2*time.Second); err != nil { + if err := New("cocoon", nil, nil, false, false, "").Probe(t.Context(), path, 2*time.Second); err != nil { t.Errorf("Probe: %v", err) } } @@ -60,7 +60,7 @@ func TestInfoRoundTripRejectsErrorFrame(t *testing.T) { path := sockPath(t) listenMuxer(t, path, "OK 2048\n", errFrame) - err := New("cocoon", nil, nil, false, "").infoRoundTrip(t.Context(), path) + err := New("cocoon", nil, nil, false, false, "").infoRoundTrip(t.Context(), path) if err == nil || !strings.Contains(err.Error(), `info reply type "error"`) { t.Errorf("got %v, want error-frame rejection", err) } @@ -70,7 +70,7 @@ func TestProbeRetriesPastFailures(t *testing.T) { path := sockPath(t) listenMuxer(t, path, "OK 2048\n", errFrame, errFrame, infoFrame) - if err := New("cocoon", nil, nil, false, "").Probe(t.Context(), path, 2*time.Second); err != nil { + if err := New("cocoon", nil, nil, false, false, "").Probe(t.Context(), path, 2*time.Second); err != nil { t.Errorf("Probe: %v", err) } } @@ -79,7 +79,7 @@ func TestProbeRetriesUntilListenerAppears(t *testing.T) { path := sockPath(t) done := make(chan error, 1) go func() { - done <- New("cocoon", nil, nil, false, "").Probe(t.Context(), path, 2*time.Second) + done <- New("cocoon", nil, nil, false, false, "").Probe(t.Context(), path, 2*time.Second) }() time.Sleep(60 * time.Millisecond) @@ -117,7 +117,7 @@ func TestDialGuestPortCtxCancel(t *testing.T) { ctx, cancel := context.WithTimeout(t.Context(), 150*time.Millisecond) defer cancel() - if _, err := New("cocoon", nil, nil, false, "").DialGuestPort(ctx, path, 8080); !errors.Is(err, context.DeadlineExceeded) { + if _, err := New("cocoon", nil, nil, false, false, "").DialGuestPort(ctx, path, 8080); !errors.Is(err, context.DeadlineExceeded) { t.Errorf("got %v, want context.DeadlineExceeded", err) } } @@ -125,7 +125,7 @@ func TestDialGuestPortCtxCancel(t *testing.T) { func TestProbeTimeout(t *testing.T) { path := sockPath(t) - err := New("cocoon", nil, nil, false, "").Probe(t.Context(), path, 150*time.Millisecond) + err := New("cocoon", nil, nil, false, false, "").Probe(t.Context(), path, 150*time.Millisecond) if err == nil || !strings.Contains(err.Error(), "silkd probe") { t.Errorf("got %v, want probe timeout", err) } diff --git a/sandboxd/engine/installca.go b/sandboxd/engine/installca.go index 439cebc1..e0d8876a 100644 --- a/sandboxd/engine/installca.go +++ b/sandboxd/engine/installca.go @@ -14,7 +14,7 @@ const ( caCertGuestPath = "/usr/local/share/ca-certificates/sandbox-egress.crt" caBundlePath = "/etc/ssl/certs/ca-certificates.crt" // guestExecPATH is set because silkd starts the guest command with an empty environment. - guestExecPATH = "/usr/sbin:/usr/bin:/sbin:/bin" + guestExecPATH = "/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin" ) // InstallCACert makes the guest trust the cluster root without update-ca-certificates. diff --git a/sandboxd/engine/installca_test.go b/sandboxd/engine/installca_test.go index 8c502c43..c9589499 100644 --- a/sandboxd/engine/installca_test.go +++ b/sandboxd/engine/installca_test.go @@ -18,7 +18,7 @@ import ( func TestInstallCACertWritesCertAndUpdates(t *testing.T) { path := sockPath(t) fake := serveFakeSilkd(t, path) - if err := New("cocoon", nil, nil, false, "").InstallCACert(t.Context(), path, []byte("CERT-PEM")); err != nil { + if err := New("cocoon", nil, nil, false, false, "").InstallCACert(t.Context(), path, []byte("CERT-PEM")); err != nil { t.Fatalf("InstallCACert: %v", err) } fake.mu.Lock() @@ -48,7 +48,7 @@ func TestInstallCACertNonzeroExitFails(t *testing.T) { path := sockPath(t) fake := serveFakeSilkd(t, path) fake.execCode = 3 - err := New("cocoon", nil, nil, false, "").InstallCACert(t.Context(), path, []byte("x")) + err := New("cocoon", nil, nil, false, false, "").InstallCACert(t.Context(), path, []byte("x")) if err == nil || !strings.Contains(err.Error(), "exit code 3") { t.Errorf("got %v, want exit code 3 failure", err) } @@ -58,7 +58,7 @@ func TestInstallCACertWriteErrorFrameFails(t *testing.T) { path := sockPath(t) fake := serveFakeSilkd(t, path) fake.writeErr = "disk full" - err := New("cocoon", nil, nil, false, "").InstallCACert(t.Context(), path, []byte("x")) + err := New("cocoon", nil, nil, false, false, "").InstallCACert(t.Context(), path, []byte("x")) if err == nil || !strings.Contains(err.Error(), "disk full") { t.Errorf("got %v, want fs_write error-frame failure", err) } diff --git a/sandboxd/engine/volume.go b/sandboxd/engine/volume.go index 16fa51d5..a5fc1a7e 100644 --- a/sandboxd/engine/volume.go +++ b/sandboxd/engine/volume.go @@ -155,6 +155,6 @@ func (e *Engine) diskAttachArgs(vmName string, spec VolumeSpec) ([]string, error } func isNotFound(err error) bool { - var respErr *wire.ErrorResp - return errors.As(err, &respErr) && respErr.Kind == wire.KindNotFound + respErr, ok := errors.AsType[*wire.ErrorResp](err) + return ok && respErr.Kind == wire.KindNotFound } diff --git a/sandboxd/engine/volume_test.go b/sandboxd/engine/volume_test.go index 687605e1..4aee874a 100644 --- a/sandboxd/engine/volume_test.go +++ b/sandboxd/engine/volume_test.go @@ -24,7 +24,7 @@ func TestDiskAttachArgsModeAndDirectIO(t *testing.T) { {"writable", "", true, "off"}, } { t.Run(tt.name, func(t *testing.T) { - e := New("cocoon", nil, nil, false, "") + e := New("cocoon", nil, nil, false, false, "") args, err := e.diskAttachArgs("sbx-1", VolumeSpec{ Name: "imagenet", Path: "/srv/datasets/imagenet.img", DirectIO: tt.directIO, RW: tt.rw, }) @@ -48,7 +48,7 @@ func TestDiskAttachArgsModeAndDirectIO(t *testing.T) { } func TestDiskAttachArgsRejectBadOptions(t *testing.T) { - _, err := New("cocoon", nil, nil, false, "").diskAttachArgs("sbx-1", VolumeSpec{DirectIO: "maybe"}) + _, err := New("cocoon", nil, nil, false, false, "").diskAttachArgs("sbx-1", VolumeSpec{DirectIO: "maybe"}) if err == nil || !strings.Contains(err.Error(), "on, off, or auto") { t.Errorf("got %v, want directio validation error", err) } @@ -67,7 +67,7 @@ func TestMountVolumeUsesSysfsAndRequestedMode(t *testing.T) { path := sockPath(t) fake := serveFakeSilkd(t, path) configureVolumeDevices(fake) - if err := New("cocoon", nil, nil, false, "").MountVolume( + if err := New("cocoon", nil, nil, false, false, "").MountVolume( t.Context(), path, "imagenet", "/datasets/training", tt.rw, ); err != nil { t.Fatalf("MountVolume: %v", err) @@ -103,7 +103,7 @@ func TestUnmountVolumeExecsBoundedUmount(t *testing.T) { } path := sockPath(t) fake := serveFakeSilkd(t, path) - e := New("cocoon", nil, nil, false, "") + e := New("cocoon", nil, nil, false, false, "") if err := e.UnmountVolume(t.Context(), path, "/datasets/training"); err != nil { t.Fatalf("UnmountVolume: %v", err) } @@ -124,7 +124,7 @@ func TestUnmountVolumeExecsBoundedUmount(t *testing.T) { func TestSyncGuestExecsPlainSync(t *testing.T) { path := sockPath(t) fake := serveFakeSilkd(t, path) - e := New("cocoon", nil, nil, false, "") + e := New("cocoon", nil, nil, false, false, "") if err := e.SyncGuest(t.Context(), path); err != nil { t.Fatalf("SyncGuest: %v", err) } @@ -148,7 +148,7 @@ func TestMountVolumeWaitsForDelayedSysfsSerial(t *testing.T) { fake.mu.Lock() fake.readMisses["/sys/block/vdc/serial"] = 1 fake.mu.Unlock() - if err := New("cocoon", nil, nil, false, "").MountVolume( + if err := New("cocoon", nil, nil, false, false, "").MountVolume( t.Context(), path, "imagenet", "/datasets/training", false, ); err != nil { t.Fatalf("MountVolume: %v", err) @@ -171,7 +171,7 @@ func TestMountVolumeWaitsForDelayedDevNode(t *testing.T) { fake.mu.Lock() fake.statMisses["/dev/vdc"] = 3 fake.mu.Unlock() - if err := New("cocoon", nil, nil, false, "").MountVolume( + if err := New("cocoon", nil, nil, false, false, "").MountVolume( t.Context(), path, "imagenet", "/datasets/training", false, ); err != nil { t.Fatalf("MountVolume: %v", err) @@ -207,7 +207,7 @@ func TestMountVolumeStopsAtFailedStage(t *testing.T) { fake.mu.Lock() tt.prepare(fake) fake.mu.Unlock() - err := New("cocoon", nil, nil, false, "").MountVolume( + err := New("cocoon", nil, nil, false, false, "").MountVolume( t.Context(), path, "imagenet", "/datasets/training", false, ) if err == nil || !strings.Contains(err.Error(), tt.wantErr) { @@ -231,7 +231,7 @@ func TestMountVolumeDeviceProbeIsBoundedAndCancelable(t *testing.T) { configureVolumeDevices(fake) ctx, cancel := context.WithCancel(t.Context()) cancel() - err := New("cocoon", nil, nil, false, "").MountVolume( + err := New("cocoon", nil, nil, false, false, "").MountVolume( ctx, path, "missing", "/datasets/training", false, ) if !errors.Is(err, context.Canceled) { @@ -246,7 +246,7 @@ func TestMountVolumeDevNodeNeverAppearsTimesOut(t *testing.T) { fake.mu.Lock() delete(fake.stat, "/dev/vdc") fake.mu.Unlock() - err := New("cocoon", nil, nil, false, "").MountVolume( + err := New("cocoon", nil, nil, false, false, "").MountVolume( t.Context(), path, "imagenet", "/datasets/training", false, ) if !errors.Is(err, context.DeadlineExceeded) || !strings.Contains(err.Error(), "wait for volume device imagenet") { diff --git a/sandboxd/engine/warmup.go b/sandboxd/engine/warmup.go new file mode 100644 index 00000000..800f8446 --- /dev/null +++ b/sandboxd/engine/warmup.go @@ -0,0 +1,10 @@ +package engine + +import "context" + +// Warmup runs argv in the guest so the golden snapshot carries what it touches. +func (e *Engine) Warmup(ctx context.Context, vsockSocket string, argv []string) error { + ctx, cancel := context.WithTimeout(ctx, cmdTimeout) + defer cancel() + return e.silkdExec(ctx, vsockSocket, argv...) +} diff --git a/sandboxd/main.go b/sandboxd/main.go index 44ddff2d..c7559bdd 100644 --- a/sandboxd/main.go +++ b/sandboxd/main.go @@ -74,7 +74,7 @@ func main() { logger.Warnf(ctx, "volume %s path %s unavailable at startup: %v", volume.Name, volume.Path, statErr) } } - eng := engine.New(cfg.CocoonBin, cfg.Bridges, cfg.Networks, cfg.NoDirectIO, cfg.RestoreMode) + eng := engine.New(cfg.CocoonBin, cfg.Bridges, cfg.Networks, cfg.NoDirectIO, cfg.NoBalloon, cfg.RestoreMode) if v, warn := eng.VersionWarning(ctx); warn != "" { logger.Warn(ctx, warn) } else { diff --git a/sandboxd/mesh/state_test.go b/sandboxd/mesh/state_test.go index 17a9dbd4..42b40b4e 100644 --- a/sandboxd/mesh/state_test.go +++ b/sandboxd/mesh/state_test.go @@ -106,7 +106,7 @@ func TestUpdateSelfConcurrentDropsNothing(t *testing.T) { wg.Go(func() { m.UpdateSelf(t.Context(), b, nil, nil) }) wg.Wait() - want := base + uint64(2*(i+1)) //nolint:gosec // loop index is small and positive + want := base + uint64(2*(i+1)) if m.self.Epoch != want { t.Fatalf("round %d: epoch %d, want %d (an update was dropped)", i, m.self.Epoch, want) } diff --git a/sandboxd/pool/archive_test.go b/sandboxd/pool/archive_test.go index 220a7f05..284d2440 100644 --- a/sandboxd/pool/archive_test.go +++ b/sandboxd/pool/archive_test.go @@ -921,7 +921,7 @@ func mustArchive(t *testing.T, m *Manager, sb *types.Sandbox) { func ckExists(t *testing.T, m *Manager, ck string) bool { t.Helper() - _, _, _, release, err := m.ckpts.Fetch(t.Context(), ck) //nolint:dogsled // existence only needs Fetch success + _, _, _, release, err := m.ckpts.Fetch(t.Context(), ck) //nolint:dogsled if err != nil { return false } diff --git a/sandboxd/pool/journal.go b/sandboxd/pool/journal.go index 9f32d159..e1fc1b52 100644 --- a/sandboxd/pool/journal.go +++ b/sandboxd/pool/journal.go @@ -14,7 +14,7 @@ const journalMaxBytes = 64 * 1024 * 1024 // usageEvent is one line of usage.jsonl, the billing stream the platform collector folds. type usageEvent struct { Time time.Time `json:"t"` - Event string `json:"ev"` // claim|hibernate|wake|fork|promote|checkpoint|release|reap + Event string `json:"ev"` ID string `json:"id"` VMName string `json:"vm,omitempty"` KeyHash string `json:"key,omitempty"` // claim only @@ -83,8 +83,5 @@ func (j *journal) rotate() error { j.size = st.Size() } } - if closeErr := old.Close(); closeErr != nil && !errors.Is(closeErr, os.ErrClosed) { - renameErr = errors.Join(renameErr, closeErr) - } - return renameErr + return errors.Join(renameErr, old.Close()) } diff --git a/sandboxd/pool/pool.go b/sandboxd/pool/pool.go index 3a6c5136..a7db4648 100644 --- a/sandboxd/pool/pool.go +++ b/sandboxd/pool/pool.go @@ -71,7 +71,8 @@ const ( vmStateRunning = "running" vmStateCreating = "creating" - caSidecarSuffix = ".cafp" + caSidecarSuffix = ".cafp" + warmupSidecarSuffix = ".warmup" ) var ( @@ -114,6 +115,7 @@ type Engine interface { Probe(ctx context.Context, vsockSocket string, timeout time.Duration) error DialGuestPort(ctx context.Context, vsockSocket string, port uint16) (net.Conn, error) InstallCACert(ctx context.Context, vsockSocket string, certPEM []byte) error + Warmup(ctx context.Context, vsockSocket string, argv []string) error DiskAttach(ctx context.Context, vmName string, spec engine.VolumeSpec) error MountVolume(ctx context.Context, vsockSocket, name, mount string, rw bool) error UnmountVolume(ctx context.Context, vsockSocket, mount string) error @@ -126,7 +128,7 @@ type SandboxSummary struct { Key types.PoolKey `json:"key"` Deadline time.Time `json:"deadline"` Hibernated bool `json:"hibernated"` - Archived bool `json:"archived,omitempty"` + Archived bool `json:"archived,omitzero"` FromCheckpoint string `json:"from_checkpoint,omitempty"` Volumes []types.Volume `json:"volumes,omitempty"` // ClaimRef echoes the caller reference; empty for fork and checkpoint-branch claims. @@ -273,6 +275,7 @@ type Manager struct { tenantLive map[string]int tenantEgress map[string]*egress.Policy // per-tenant allow-list; nil = no tenant policy poolEgress map[types.PoolKey]*egress.Policy + poolWarmups map[types.PoolKey][]string usage *journal audit *journal counters counters @@ -413,6 +416,7 @@ func NewManager(ctx context.Context, cfg *config.Config, eng Engine, secrets *eg m.tenantMax = make(map[string]int, len(cfg.Tenants)) m.tenantEgress = make(map[string]*egress.Policy, len(cfg.Tenants)) m.poolEgress = make(map[types.PoolKey]*egress.Policy, len(cfg.Pools)) + m.poolWarmups = make(map[types.PoolKey][]string, len(cfg.Pools)) for _, tn := range cfg.Tenants { m.tenantMax[tn.Name] = tn.MaxClaims if tn.Egress != nil { @@ -439,6 +443,9 @@ func NewManager(ctx context.Context, cfg *config.Config, eng Engine, secrets *eg m.poolEgress[spec.PoolKey] = spec.Egress m.guardedEgress = true } + if len(spec.Warmup) > 0 { + m.poolWarmups[spec.PoolKey] = spec.Warmup + } } if slices.ContainsFunc(cfg.Pools, func(s config.PoolSpec) bool { return s.Egress.Intercepts() }) { ca, err := loadEgressCA(cfg.EgressCA) diff --git a/sandboxd/pool/pool_test.go b/sandboxd/pool/pool_test.go index 3a8780bb..6a45df17 100644 --- a/sandboxd/pool/pool_test.go +++ b/sandboxd/pool/pool_test.go @@ -794,6 +794,7 @@ type fakeEngine struct { snapSaves, exports, snapshots []string exportContent []byte caInstalls []string + warmups [][]string staleReconciles []string installCAErr error diskAttachErr error @@ -1030,6 +1031,16 @@ func (f *fakeEngine) InstallCACert(_ context.Context, vsockSocket string, _ []by return f.installCAErr } +func (f *fakeEngine) Warmup(_ context.Context, _ string, argv []string) error { + f.mu.Lock() + defer f.mu.Unlock() + if len(f.snapSaves) > 0 { + return fmt.Errorf("warmup after snapshot") + } + f.warmups = append(f.warmups, argv) + return nil +} + func (f *fakeEngine) DiskAttach(_ context.Context, _ string, spec engine.VolumeSpec) error { f.mu.Lock() f.volumeSpecs = append(f.volumeSpecs, spec) diff --git a/sandboxd/pool/poolstore.go b/sandboxd/pool/poolstore.go index a9db122a..c8e642dc 100644 --- a/sandboxd/pool/poolstore.go +++ b/sandboxd/pool/poolstore.go @@ -112,11 +112,12 @@ func (m *Manager) adoptPersistedPools(ctx context.Context) error { return nil } -// poolSeedHash digests a pool set's warm-target shape, order-independent and egress-excluded. +// poolSeedHash digests a pool set's warm-target shape, order-independent, without the config-owned egress and warmup. func poolSeedHash(specs []config.PoolSpec) string { shaped := slices.Clone(specs) for i := range shaped { shaped[i].Egress = nil + shaped[i].Warmup = nil } slices.SortFunc(shaped, func(a, b config.PoolSpec) int { return strings.Compare(a.Hash(), b.Hash()) }) raw, _ := json.Marshal(shaped) diff --git a/sandboxd/pool/reconcile.go b/sandboxd/pool/reconcile.go index d67155b0..a4bb8bcd 100644 --- a/sandboxd/pool/reconcile.go +++ b/sandboxd/pool/reconcile.go @@ -223,9 +223,6 @@ func (m *Manager) quarantineClaim(ctx context.Context, sb *types.Sandbox) bool { // readoptEgressTap records and returns a live egress claim's tap, "" when there is none. func (m *Manager) readoptEgressTap(sb *types.Sandbox, live map[string]types.VMRecord) string { - if sb.Key.Net != types.NetEgress { - return "" - } rec, ok := live[sb.VMName] tap := rec.TapDevice() if !ok || tap == "" { diff --git a/sandboxd/pool/refill.go b/sandboxd/pool/refill.go index 82ea628e..a3d712b6 100644 --- a/sandboxd/pool/refill.go +++ b/sandboxd/pool/refill.go @@ -194,33 +194,44 @@ func (m *Manager) buildGoldenSteps(ctx context.Context, key types.PoolKey, name, return fmt.Errorf("install egress ca: %w", err) } } + warmup := m.poolWarmup(key) + if len(warmup) > 0 { + if err := m.eng.Warmup(ctx, sock, warmup); err != nil { + return fmt.Errorf("warmup: %w", err) + } + } if err := m.eng.SnapshotSave(ctx, name, snap); err != nil { return err } if err := m.exportGolden(ctx, snap, final); err != nil { return err } - return m.writeGoldenCASidecar(final, caBaked) + if err := m.writeGoldenCASidecar(final, caBaked); err != nil { + return err + } + return writeGoldenSidecar(final+warmupSidecarSuffix, warmupStamp(warmup)) +} + +func (m *Manager) poolWarmup(key types.PoolKey) []string { + m.mu.Lock() + defer m.mu.Unlock() + return m.poolWarmups[key] } // writeGoldenCASidecar records or clears the baked-CA fingerprint; a rotated CA forces a rebuild. func (m *Manager) writeGoldenCASidecar(final string, caBaked bool) error { - path := final + caSidecarSuffix - if !caBaked { - if err := os.Remove(path); err != nil && !errors.Is(err, os.ErrNotExist) { - return fmt.Errorf("clear ca sidecar: %w", err) - } - return nil - } - if err := os.WriteFile(path, []byte(m.egressCA.Fingerprint()), 0o644); err != nil { //nolint:gosec // public fingerprint - return fmt.Errorf("write ca sidecar: %w", err) + var stamp string + if caBaked { + stamp = m.egressCA.Fingerprint() } - return nil + return writeGoldenSidecar(final+caSidecarSuffix, stamp) } -// adoptGolden points p at an on-disk golden whose baked-CA state still fits. +// adoptGolden points p at an on-disk golden whose baked-CA state and warmup still fit. func (m *Manager) adoptGolden(p *pool) { - if g := filepath.Join(m.goldensDir(), p.hash); dirExists(g) && m.goldenCAMatches(g, m.poolEgress[p.key].Intercepts()) { + g := filepath.Join(m.goldensDir(), p.hash) + if dirExists(g) && m.goldenCAMatches(g, m.poolEgress[p.key].Intercepts()) && + goldenSidecarMatches(g+warmupSidecarSuffix, warmupStamp(m.poolWarmups[p.key])) { p.goldenDir = g } } @@ -228,11 +239,9 @@ func (m *Manager) adoptGolden(p *pool) { // goldenCAMatches reports whether a golden's baked-CA state fits the pool. func (m *Manager) goldenCAMatches(final string, caNeeded bool) bool { if !caNeeded { - _, err := os.Stat(final + caSidecarSuffix) - return errors.Is(err, os.ErrNotExist) + return goldenSidecarMatches(final+caSidecarSuffix, "") } - fp, err := os.ReadFile(final + caSidecarSuffix) //nolint:gosec // node-local golden path - return err == nil && m.egressCA != nil && string(fp) == m.egressCA.Fingerprint() + return m.egressCA != nil && goldenSidecarMatches(final+caSidecarSuffix, m.egressCA.Fingerprint()) } // exportGolden exports snap into final through a unique sibling *.tmp dir. @@ -440,3 +449,28 @@ func (m *Manager) dropSnap(ctx context.Context, snap string) { log.WithFunc("pool.dropSnap").Warnf(ctx, "drop snapshot %s: %v", snap, err) } } + +func writeGoldenSidecar(path, stamp string) error { + if stamp == "" { + if err := os.Remove(path); err != nil && !errors.Is(err, os.ErrNotExist) { + return fmt.Errorf("clear golden sidecar %s: %w", filepath.Base(path), err) + } + return nil + } + if err := os.WriteFile(path, []byte(stamp), 0o644); err != nil { //nolint:gosec // public stamp + return fmt.Errorf("write golden sidecar %s: %w", filepath.Base(path), err) + } + return nil +} + +func goldenSidecarMatches(path, stamp string) bool { + got, err := os.ReadFile(path) //nolint:gosec // node-local golden path + if stamp == "" { + return errors.Is(err, os.ErrNotExist) + } + return err == nil && string(got) == stamp +} + +func warmupStamp(argv []string) string { + return strings.Join(argv, "\x00") +} diff --git a/sandboxd/pool/setpools.go b/sandboxd/pool/setpools.go index 1a2fa9f9..e327bca7 100644 --- a/sandboxd/pool/setpools.go +++ b/sandboxd/pool/setpools.go @@ -14,7 +14,6 @@ import ( // SetPools replaces the node's desired warm targets; existing claims are unaffected. func (m *Manager) SetPools(ctx context.Context, specs []config.PoolSpec) error { desired := make(map[types.PoolKey]config.PoolSpec, len(specs)) - hashes := make(map[string]types.PoolKey, len(specs)) for _, spec := range specs { spec = normalizePoolSpec(spec) if err := m.validate(spec.PoolKey); err != nil { @@ -27,13 +26,12 @@ func (m *Manager) SetPools(ctx context.Context, specs []config.PoolSpec) error { if spec.Egress != nil { return fmt.Errorf("%w: pool %q: egress is set in the config file, not via the API", ErrBadKey, spec.Template) } - if existing, ok := hashes[spec.Hash()]; ok && existing != spec.PoolKey { - return fmt.Errorf("%w: pool key hash collision between %q and %q", ErrBadKey, existing.Template, spec.Template) + if spec.Warmup != nil { + return fmt.Errorf("%w: pool %q: warmup is set in the config file, not via the API", ErrBadKey, spec.Template) } if _, ok := desired[spec.PoolKey]; ok { return fmt.Errorf("%w: duplicate pool %q", ErrBadKey, spec.Template) } - hashes[spec.Hash()] = spec.PoolKey desired[spec.PoolKey] = spec } diff --git a/sandboxd/pool/telemetry.go b/sandboxd/pool/telemetry.go index b7e46270..d5e34d39 100644 --- a/sandboxd/pool/telemetry.go +++ b/sandboxd/pool/telemetry.go @@ -55,9 +55,9 @@ type auditFrame struct { To string `json:"to,omitempty"` URL string `json:"url,omitempty"` Session string `json:"session,omitempty"` - Port uint16 `json:"port,omitempty"` - Decision string `json:"decision,omitempty"` // egress: allow|deny - Secret string `json:"secret,omitempty"` //nolint:gosec // the secret's ref name, never its value + Port uint16 `json:"port,omitzero"` + Decision string `json:"decision,omitempty"` + Secret string `json:"secret,omitempty"` //nolint:gosec // the secret's ref name, never its value } // Counters snapshots the monotonic telemetry counters. diff --git a/sandboxd/pool/template.go b/sandboxd/pool/template.go index 4fa6a334..ce653f5c 100644 --- a/sandboxd/pool/template.go +++ b/sandboxd/pool/template.go @@ -38,7 +38,7 @@ func (m *Manager) Promote(ctx context.Context, id string, cred Cred, template, t return types.PoolKey{}, "", ErrNoEgressFork } key := types.PoolKey{Template: template, Net: sb.Key.Net, Size: sb.Key.Size} - if m.pooledHash(key.Hash()) { + if m.pooled(key) { // a configured pool owns this key; promoting over it would change what refills produce return types.PoolKey{}, "", ErrPooledTemplate } @@ -73,7 +73,7 @@ func (m *Manager) DeleteTemplate(ctx context.Context, key types.PoolKey, tenant if err := m.validate(key); err != nil { return err } - if m.pooledHash(key.Hash()) { + if m.pooled(key) { return ErrPooledTemplate } id := store.TemplateID(key.Hash()) @@ -141,7 +141,7 @@ func (m *Manager) HasPoolGolden(key types.PoolKey) bool { // HasPromotedTemplate is resolveGolden's test exactly, so routing never promises a refused golden. func (m *Manager) HasPromotedTemplate(ctx context.Context, key types.PoolKey, tenant string) bool { - if m.pooledHash(key.Hash()) { + if m.pooled(key) { return false } id := store.TemplateID(key.Hash()) @@ -163,16 +163,11 @@ func (m *Manager) HasPromotedTemplate(ctx context.Context, key types.PoolKey, te return owner == "" || tenantOwns(tenant, owner) } -// pooledHash guards on the hash, not the key: a colliding key would reach a pool's golden dir. -func (m *Manager) pooledHash(hash string) bool { +func (m *Manager) pooled(key types.PoolKey) bool { m.mu.Lock() defer m.mu.Unlock() - for _, p := range m.pools { - if p.hash == hash { - return true - } - } - return false + _, ok := m.pools[key] + return ok } // recLock takes the per-record lock and a live reference (pair with recDone); clones and wakes hold it shared so a delete or re-publish never runs under an in-flight read. diff --git a/sandboxd/pool/warmup_test.go b/sandboxd/pool/warmup_test.go new file mode 100644 index 00000000..63620e77 --- /dev/null +++ b/sandboxd/pool/warmup_test.go @@ -0,0 +1,89 @@ +package pool + +import ( + "errors" + "os" + "path/filepath" + "slices" + "strings" + "testing" + + "github.com/cocoonstack/sandbox/sandboxd/config" +) + +func TestGoldenBuildRunsWarmupBeforeSnapshot(t *testing.T) { + eng := newFakeEngine() + argv := []string{"node", "-e", "0"} + m := newTestManager(t, eng, config.PoolSpec{PoolKey: testKey, Warm: 1, Warmup: argv}) + final := filepath.Join(m.goldensDir(), testKey.Hash()) + if err := m.buildGoldenSteps(t.Context(), testKey, "sbx-gb", "snap", final); err != nil { + t.Fatalf("buildGoldenSteps: %v", err) + } + if len(eng.warmups) != 1 || !slices.Equal(eng.warmups[0], argv) { + t.Fatalf("warmups = %v, want [%v]", eng.warmups, argv) + } + stamp, err := os.ReadFile(final + warmupSidecarSuffix) + if err != nil { + t.Fatalf("read warmup sidecar: %v", err) + } + if string(stamp) != warmupStamp(argv) { + t.Errorf("sidecar = %q, want %q", stamp, warmupStamp(argv)) + } +} + +func TestGoldenBuildSkipsWarmupWhenUnset(t *testing.T) { + eng := newFakeEngine() + m := newTestManager(t, eng, config.PoolSpec{PoolKey: testKey, Warm: 1}) + final := filepath.Join(m.goldensDir(), testKey.Hash()) + if err := m.buildGoldenSteps(t.Context(), testKey, "sbx-gb", "snap", final); err != nil { + t.Fatalf("buildGoldenSteps: %v", err) + } + if len(eng.warmups) != 0 { + t.Errorf("Warmup called %d times for a pool without one", len(eng.warmups)) + } + if _, err := os.Stat(final + warmupSidecarSuffix); !errors.Is(err, os.ErrNotExist) { + t.Errorf("warmup sidecar present for a pool without one: %v", err) + } +} + +func TestAdoptGoldenRequiresMatchingWarmup(t *testing.T) { + m := newTestManager(t, newFakeEngine(), config.PoolSpec{PoolKey: testKey, Warm: 1, Warmup: []string{"node", "-e", "0"}}) + final := filepath.Join(m.goldensDir(), testKey.Hash()) + if err := os.MkdirAll(final, 0o755); err != nil { + t.Fatalf("mkdir golden: %v", err) + } + p := m.pools[testKey] + m.adoptGolden(p) + if p.goldenDir != "" { + t.Error("adopted a golden built without the warmup") + } + if err := os.WriteFile(final+warmupSidecarSuffix, []byte(warmupStamp([]string{"python3", "-c", "0"})), 0o644); err != nil { + t.Fatalf("write sidecar: %v", err) + } + m.adoptGolden(p) + if p.goldenDir != "" { + t.Error("adopted a golden built with a different warmup") + } + if err := os.WriteFile(final+warmupSidecarSuffix, []byte(warmupStamp([]string{"node", "-e", "0"})), 0o644); err != nil { + t.Fatalf("write sidecar: %v", err) + } + m.adoptGolden(p) + if p.goldenDir != final { + t.Errorf("goldenDir = %q, want %q", p.goldenDir, final) + } +} + +func TestSetPoolsRejectsWarmup(t *testing.T) { + m := newTestManager(t, newFakeEngine(), config.PoolSpec{PoolKey: testKey, Warm: 1}) + err := m.SetPools(t.Context(), []config.PoolSpec{{PoolKey: testKey, Warm: 1, Warmup: []string{"true"}}}) + if !errors.Is(err, ErrBadKey) || !strings.Contains(err.Error(), "warmup is set in the config file") { + t.Errorf("SetPools error = %v, want ErrBadKey naming warmup as config-owned", err) + } +} + +func TestPoolSpecRejectsEmptyWarmupArgument(t *testing.T) { + spec := config.PoolSpec{PoolKey: testKey, Warm: 1, Warmup: []string{"node", ""}} + if err := spec.ValidateLimits(); err == nil || !strings.Contains(err.Error(), "warmup") { + t.Errorf("ValidateLimits error = %v, want an empty-argument rejection", err) + } +} diff --git a/sandboxd/server/exec.go b/sandboxd/server/exec.go new file mode 100644 index 00000000..648520d9 --- /dev/null +++ b/sandboxd/server/exec.go @@ -0,0 +1,176 @@ +package server + +import ( + "cmp" + "context" + "errors" + "io" + "net" + "net/http" + "time" + + "github.com/projecteru2/core/log" + + "github.com/cocoonstack/sandbox/protocol/wire" +) + +const ( + execOutputCap = 8 << 20 + execKillWait = 5 * time.Second +) + +var errExecOutputCap = errors.New("command output exceeds the buffered exec cap") + +// ExecRequest is a buffered exec: one command whose whole output comes back in the response. +type ExecRequest struct { + Argv []string `json:"argv"` + Cwd string `json:"cwd,omitempty"` + Env map[string]string `json:"env,omitempty"` + TimeoutSeconds int `json:"timeout_seconds,omitzero"` +} + +// ExecResponse is the exit code and complete output of a buffered exec. +type ExecResponse struct { + ExitCode int32 `json:"exit_code"` + Stdout string `json:"stdout"` + Stderr string `json:"stderr"` +} + +func (s *Server) handleExec(w http.ResponseWriter, r *http.Request) { + token, ok := sandboxToken(w, r) + if !ok { + return + } + req, ok := decodeBodyStrict[ExecRequest](w, r) + if !ok { + return + } + if len(req.Argv) == 0 { + writeErr(w, http.StatusBadRequest, "argv must not be empty") + return + } + if req.TimeoutSeconds < 0 { + writeErr(w, http.StatusBadRequest, "timeout_seconds must not be negative") + return + } + ctx := r.Context() + id := r.PathValue("id") + guest, ok := s.wakeGuest(ctx, w, id, token) + if !ok { + return + } + if req.TimeoutSeconds > 0 { + var cancel context.CancelFunc + ctx, cancel = context.WithTimeout(ctx, time.Duration(req.TimeoutSeconds)*time.Second) + defer cancel() + } + stop := context.AfterFunc(ctx, func() { _ = guest.Close() }) + defer func() { + stop() + _ = guest.Close() + }() + frame, _ := wire.EncodeRequest(wire.Exec{Argv: req.Argv, Cwd: req.Cwd, Env: req.Env}) + frame = append(frame, '\n') + if s.mgr.AuditEnabled() { + s.mgr.Audit(ctx, id, frame) + } + stdinClose, _ := wire.EncodeRequest(wire.StdinClose{}) + frame = append(frame, stdinClose...) + frame = append(frame, '\n') + if _, err := guest.Write(frame); err != nil { + writeErr(w, http.StatusBadGateway, "guest agent unreachable") + return + } + resp, pid, err := collectExec(guest) + if err != nil && pid != 0 { + // a dropped connection only reaches a child that writes; a silent one needs the kill + if killErr := s.killExec(ctx, id, token, pid); killErr != nil { + log.WithFunc("server.handleExec").Errorf(ctx, killErr, "kill exec pid %d in %s", pid, id) + } + } + silkdErr, isSilkd := errors.AsType[*wire.ErrorResp](err) + switch { + case err == nil: + writeJSON(w, http.StatusOK, resp) + case errors.Is(ctx.Err(), context.DeadlineExceeded): + writeErr(w, http.StatusGatewayTimeout, "command timed out") + case errors.Is(ctx.Err(), context.Canceled): + return + case isSilkd && silkdErr.Kind == wire.KindBadRequest: + writeErr(w, http.StatusBadRequest, silkdErr.Message) + case isSilkd: + writeErr(w, http.StatusBadGateway, silkdErr.Error()) + case errors.Is(err, errExecOutputCap): + writeErr(w, http.StatusRequestEntityTooLarge, err.Error()) + default: + log.WithFunc("server.handleExec").Errorf(ctx, err, "exec relay for %s", id) + writeErr(w, http.StatusBadGateway, "guest agent closed before the command exited") + } +} + +func (s *Server) killExec(ctx context.Context, id, token string, pid uint32) error { + ctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), execKillWait) + defer cancel() + sock, err := s.mgr.WakeAgentSocket(ctx, id, token) + if err != nil { + return err + } + guest, err := s.dialer.DialSilkd(ctx, sock) + if err != nil { + return err + } + stop := context.AfterFunc(ctx, func() { _ = guest.Close() }) + defer func() { + stop() + _ = guest.Close() + }() + frame, _ := wire.EncodeRequest(wire.Kill{PID: pid}) + if _, err = guest.Write(append(frame, '\n')); err != nil { + return err + } + sc := wire.NewFrameScanner(guest) + if !sc.Scan() { + return cmp.Or(sc.Err(), io.ErrUnexpectedEOF) + } + reply, err := wire.DecodeResponse(sc.Bytes()) + if err != nil { + return err + } + if silkdErr, ok := reply.(*wire.ErrorResp); ok && silkdErr.Kind != wire.KindNotFound { + return silkdErr + } + return nil +} + +func collectExec(guest net.Conn) (resp ExecResponse, pid uint32, err error) { + var stdout, stderr []byte + sc := wire.NewFrameScanner(guest) + for sc.Scan() { + frame, err := wire.DecodeResponse(sc.Bytes()) + if err != nil { + return ExecResponse{}, pid, err + } + switch f := frame.(type) { + case *wire.Started: + pid = f.PID + case *wire.Stdout: + if len(stdout)+len(stderr)+len(f.Data) > execOutputCap { + return ExecResponse{}, pid, errExecOutputCap + } + stdout = append(stdout, f.Data...) + case *wire.Stderr: + if len(stdout)+len(stderr)+len(f.Data) > execOutputCap { + return ExecResponse{}, pid, errExecOutputCap + } + stderr = append(stderr, f.Data...) + case *wire.Exit: + return ExecResponse{ExitCode: f.Code, Stdout: string(stdout), Stderr: string(stderr)}, 0, nil + case *wire.ErrorResp: + return ExecResponse{}, 0, f + } + } + if err := sc.Err(); err != nil { + return ExecResponse{}, pid, err + } + return ExecResponse{}, pid, io.ErrUnexpectedEOF +} diff --git a/sandboxd/server/exec_test.go b/sandboxd/server/exec_test.go new file mode 100644 index 00000000..89c3f5b5 --- /dev/null +++ b/sandboxd/server/exec_test.go @@ -0,0 +1,207 @@ +package server + +import ( + "bufio" + "bytes" + "context" + "encoding/base64" + "encoding/json" + "io" + "net" + "net/http" + "net/http/httptest" + "strings" + "sync/atomic" + "testing" + "time" + + "github.com/cocoonstack/sandbox/protocol/wire" +) + +func TestExecReturnsBufferedOutput(t *testing.T) { + var gotExec, gotStdinClose wire.Request + ts, _ := newRelayServer(t, func(conn net.Conn) { + defer conn.Close() + r := bufio.NewReader(conn) + line, err := r.ReadBytes('\n') + if err != nil { + return + } + gotExec, _ = wire.DecodeRequest(bytes.TrimSpace(line)) + line, err = r.ReadBytes('\n') + if err != nil { + return + } + gotStdinClose, _ = wire.DecodeRequest(bytes.TrimSpace(line)) + _, _ = io.WriteString(conn, `{"type":"started","pid":7}`+"\n") + _, _ = io.WriteString(conn, `{"type":"stdout","data":"`+base64.StdEncoding.EncodeToString([]byte("v22\n"))+`"}`+"\n") + _, _ = io.WriteString(conn, `{"type":"stderr","data":"`+base64.StdEncoding.EncodeToString([]byte("warn\n"))+`"}`+"\n") + _, _ = io.WriteString(conn, `{"type":"exit","code":3}`+"\n") + }) + status, body := postExec(t, ts, `{"argv":["node","-v"],"cwd":"/work","env":{"A":"1"},"timeout_seconds":5}`) + if status != http.StatusOK { + t.Fatalf("status %d, want 200: %s", status, body) + } + var out ExecResponse + if err := json.Unmarshal(body, &out); err != nil { + t.Fatalf("decode: %v", err) + } + if out.ExitCode != 3 || out.Stdout != "v22\n" || out.Stderr != "warn\n" { + t.Errorf("response = %+v", out) + } + exec, ok := gotExec.(*wire.Exec) + if !ok || strings.Join(exec.Argv, " ") != "node -v" || exec.Cwd != "/work" || exec.Env["A"] != "1" || exec.Detach { + t.Errorf("guest request = %#v", gotExec) + } + if _, ok := gotStdinClose.(*wire.StdinClose); !ok { + t.Errorf("guest stdin request = %#v, want stdin_close", gotStdinClose) + } +} + +func TestExecMapsSilkdErrors(t *testing.T) { + for _, tt := range []struct { + kind string + want int + }{ + {"bad_request", http.StatusBadRequest}, + {"internal", http.StatusBadGateway}, + } { + t.Run(tt.kind, func(t *testing.T) { + ts, _ := newRelayServer(t, func(conn net.Conn) { + defer conn.Close() + _, _ = bufio.NewReader(conn).ReadBytes('\n') + _, _ = io.WriteString(conn, `{"type":"error","kind":"`+tt.kind+`","message":"spawn failed"}`+"\n") + }) + if status, _ := postExec(t, ts, `{"argv":["true"]}`); status != tt.want { + t.Errorf("status %d, want %d", status, tt.want) + } + }) + } +} + +func TestExecTimeoutKillsTheCommand(t *testing.T) { + var conns atomic.Int32 + killed := make(chan uint32, 1) + closed := make(chan struct{}) + ts, _ := newRelayServer(t, func(conn net.Conn) { + r := bufio.NewReader(conn) + if conns.Add(1) == 1 { + defer close(closed) + _, _ = r.ReadBytes('\n') + _, _ = io.WriteString(conn, `{"type":"started","pid":7}`+"\n") + _, _ = r.ReadBytes('\n') + _, _ = r.ReadBytes('\n') + return + } + defer conn.Close() + line, _ := r.ReadBytes('\n') + if req, _ := wire.DecodeRequest(bytes.TrimSpace(line)); req != nil { + if kill, ok := req.(*wire.Kill); ok { + killed <- kill.PID + } + } + _, _ = io.WriteString(conn, `{"type":"done"}`+"\n") + }) + if status, _ := postExec(t, ts, `{"argv":["sleep","60"],"timeout_seconds":1}`); status != http.StatusGatewayTimeout { + t.Fatalf("status %d, want 504", status) + } + select { + case <-closed: + case <-time.After(3 * time.Second): + t.Fatal("guest conn still open after the timeout") + } + select { + case pid := <-killed: + if pid != 7 { + t.Errorf("killed pid %d, want 7", pid) + } + case <-time.After(3 * time.Second): + t.Fatal("no kill reached the guest after the timeout") + } +} + +func TestExecRejectsUnknownFields(t *testing.T) { + ts, _ := newRelayServer(t, func(conn net.Conn) { _ = conn.Close() }) + if status, _ := postExec(t, ts, `{"argv":["whoami"],"user":"nobody"}`); status != http.StatusBadRequest { + t.Errorf("status %d, want 400", status) + } +} + +func TestExecRejectsEmptyArgv(t *testing.T) { + ts, _ := newRelayServer(t, func(conn net.Conn) { _ = conn.Close() }) + if status, _ := postExec(t, ts, `{"argv":[]}`); status != http.StatusBadRequest { + t.Errorf("status %d, want 400", status) + } +} + +func TestExecGuestClosingEarlyIs502(t *testing.T) { + ts, _ := newRelayServer(t, func(conn net.Conn) { + _, _ = bufio.NewReader(conn).ReadBytes('\n') + _ = conn.Close() + }) + if status, _ := postExec(t, ts, `{"argv":["true"]}`); status != http.StatusBadGateway { + t.Errorf("status %d, want 502", status) + } +} + +func TestExecTimeoutStartsAfterWake(t *testing.T) { + wakeHadCommandDeadline := make(chan bool, 1) + mgr := &execManager{ + fakeManager: &fakeManager{}, + wake: func(ctx context.Context, _, _ string) (string, error) { + deadline, ok := ctx.Deadline() + wakeHadCommandDeadline <- ok && time.Until(deadline) < 2*time.Second + return "/v/sock", nil + }, + } + dialer := &fakeDialer{dial: func(context.Context, string) (net.Conn, error) { + relayEnd, guestEnd := net.Pipe() + go func() { + defer guestEnd.Close() + r := bufio.NewReader(guestEnd) + _, _ = r.ReadBytes('\n') + _, _ = r.ReadBytes('\n') + _, _ = io.WriteString(guestEnd, `{"type":"exit","code":0}`+"\n") + }() + return relayEnd, nil + }} + srv := New("", nil, "node:7777", mgr, dialer, nil, nil, nil, nil) + ts := httptest.NewServer(srv.Handler()) + t.Cleanup(func() { ts.Close(); srv.CloseRelays() }) + + if status, body := postExec(t, ts, `{"argv":["true"],"timeout_seconds":1}`); status != http.StatusOK { + t.Fatalf("status %d, want 200: %s", status, body) + } + if <-wakeHadCommandDeadline { + t.Error("command timeout was active during wake") + } +} + +type execManager struct { + *fakeManager + wake func(context.Context, string, string) (string, error) +} + +func (m *execManager) WakeAgentSocket(ctx context.Context, id, token string) (string, error) { + return m.wake(ctx, id, token) +} + +func postExec(t *testing.T, ts *httptest.Server, body string) (int, []byte) { + t.Helper() + req, err := http.NewRequestWithContext(t.Context(), http.MethodPost, ts.URL+"/v1/sandboxes/sb_1/exec", strings.NewReader(body)) + if err != nil { + t.Fatalf("new request: %v", err) + } + req.Header.Set("Authorization", "Bearer tok") + req.Header.Set("Content-Type", "application/json") + resp, err := ts.Client().Do(req) + if err != nil { + t.Fatalf("post exec: %v", err) + } + defer resp.Body.Close() + out, err := io.ReadAll(resp.Body) + if err != nil { + t.Fatalf("read response: %v", err) + } + return resp.StatusCode, out +} diff --git a/sandboxd/server/preview_test.go b/sandboxd/server/preview_test.go index 19755c5e..dccf2f00 100644 --- a/sandboxd/server/preview_test.go +++ b/sandboxd/server/preview_test.go @@ -176,6 +176,6 @@ func newGuestServer(t *testing.T, body func(r *http.Request) string) string { } func mintToken(ps *PreviewServer, id string, port uint16, ttl time.Duration) string { - url := ps.Mint(id, port, ttl) - return strings.TrimSuffix(url[strings.Index(url, "/p/")+3:], "/") + _, token, _ := strings.Cut(ps.Mint(id, port, ttl), "/p/") + return strings.TrimSuffix(token, "/") } diff --git a/sandboxd/server/relay.go b/sandboxd/server/relay.go index 1b96a7fb..c1748a98 100644 --- a/sandboxd/server/relay.go +++ b/sandboxd/server/relay.go @@ -44,20 +44,8 @@ func (s *Server) handleAgent(w http.ResponseWriter, r *http.Request) { writeErr(w, http.StatusUpgradeRequired, "upgrade to "+upgradeProto+" required") return } - // WakeAgentSocket restores a hibernated VM first, making the relay the transparent wake path - sock, err := s.mgr.WakeAgentSocket(r.Context(), r.PathValue("id"), token) - switch { - case writePoolErr(w, err): - return - case err != nil: - log.WithFunc("server.handleAgent").Errorf(r.Context(), err, "agent socket for %s", r.PathValue("id")) - writeErr(w, http.StatusInternalServerError, "sandbox lookup failed") - return - } - guest, err := s.dialer.DialSilkd(r.Context(), sock) - if err != nil { - log.WithFunc("server.handleAgent").Errorf(r.Context(), err, "dial silkd for %s", r.PathValue("id")) - writeErr(w, http.StatusBadGateway, "guest agent unreachable") + guest, ok := s.wakeGuest(r.Context(), w, r.PathValue("id"), token) + if !ok { return } client, bufrw, err := http.NewResponseController(w).Hijack() @@ -70,6 +58,27 @@ func (s *Server) handleAgent(w http.ResponseWriter, r *http.Request) { s.relay(r.Context(), r.PathValue("id"), client, bufrw.Reader, guest) } +// wakeGuest resolves the sandbox's agent socket, waking a hibernated VM first, and dials silkd; +// on failure it has already answered the request. +func (s *Server) wakeGuest(ctx context.Context, w http.ResponseWriter, id, token string) (net.Conn, bool) { + sock, err := s.mgr.WakeAgentSocket(ctx, id, token) + switch { + case writePoolErr(w, err): + return nil, false + case err != nil: + log.WithFunc("server.wakeGuest").Errorf(ctx, err, "agent socket for %s", id) + writeErr(w, http.StatusInternalServerError, "sandbox lookup failed") + return nil, false + } + guest, err := s.dialer.DialSilkd(ctx, sock) + if err != nil { + log.WithFunc("server.wakeGuest").Errorf(ctx, err, "dial silkd for %s", id) + writeErr(w, http.StatusBadGateway, "guest agent unreachable") + return nil, false + } + return guest, true +} + // relay writes the 101 and splices the conns until silkd closes or the client vanishes. func (s *Server) relay(ctx context.Context, id string, client net.Conn, clientBuf *bufio.Reader, guest net.Conn) { s.relayMu.Lock() diff --git a/sandboxd/server/server.go b/sandboxd/server/server.go index 83f5f42a..1de98eb7 100644 --- a/sandboxd/server/server.go +++ b/sandboxd/server/server.go @@ -126,10 +126,10 @@ type InfoResponse struct { Claimed int `json:"claimed"` Hibernated int `json:"hibernated"` Archived int `json:"archived"` - Draining bool `json:"draining,omitempty"` + Draining bool `json:"draining,omitzero"` Peers []string `json:"peers,omitempty"` // AtCapacity marks refill parked because the node refused another VM. - AtCapacity bool `json:"at_capacity,omitempty"` + AtCapacity bool `json:"at_capacity,omitzero"` AtCapacityReason string `json:"at_capacity_reason,omitempty"` } @@ -198,6 +198,7 @@ func (s *Server) Handler() http.Handler { mux.HandleFunc("POST /v1/drain", s.requireRoot(s.handleDrain)) mux.HandleFunc("DELETE /v1/drain", s.requireRoot(s.handleUncordon)) mux.HandleFunc("GET /v1/sandboxes/{id}/agent", s.handleAgent) + mux.HandleFunc("POST /v1/sandboxes/{id}/exec", s.handleExec) mux.HandleFunc("GET /v1/sandboxes/{id}/owner", s.handleOwner) mux.HandleFunc("GET /v1/info", s.requireRoot(s.handleInfo)) mux.HandleFunc("GET /v1/peers", s.requireToken(s.handlePeers)) @@ -244,7 +245,7 @@ func (s *Server) handleClaim(w http.ResponseWriter, r *http.Request) { func (s *Server) handleVolumeClaim(w http.ResponseWriter, r *http.Request, req types.ClaimRequest, key types.PoolKey, hash, tenant string) { volumes, err := types.ValidateVolumes(req.Volumes, req.VolumesAttachOnly) if err != nil { - writeErr(w, http.StatusBadRequest, fmt.Errorf("%w: %v", pool.ErrBadVolume, err).Error()) + writeErr(w, http.StatusBadRequest, fmt.Sprintf("%v: %v", pool.ErrBadVolume, err)) return } req.Volumes = volumes diff --git a/sandboxd/server/server_test.go b/sandboxd/server/server_test.go index a3a1738b..1f828629 100644 --- a/sandboxd/server/server_test.go +++ b/sandboxd/server/server_test.go @@ -1506,7 +1506,8 @@ func TestPreviewHandlerZeroDeadlineMintsLiveToken(t *testing.T) { if err := json.NewDecoder(resp.Body).Decode(&out); err != nil { t.Fatalf("decode: %v", err) } - token := strings.TrimSuffix(out.URL[strings.Index(out.URL, "/p/")+3:], "/") + _, token, _ := strings.Cut(out.URL, "/p/") + token = strings.TrimSuffix(token, "/") claims, ok := ps.verify(token) if !ok { t.Fatalf("minted token %q does not verify", token) diff --git a/sandboxd/store/peer/broadcast.go b/sandboxd/store/peer/broadcast.go index 38d1ddc0..34ae397f 100644 --- a/sandboxd/store/peer/broadcast.go +++ b/sandboxd/store/peer/broadcast.go @@ -1,6 +1,7 @@ package peer import ( + "cmp" "context" "fmt" "io" @@ -30,10 +31,7 @@ func (b *Broadcaster) Delete(ctx context.Context, id string) { if len(addrs) == 0 { return } - client := b.Client - if client == nil { - client = &http.Client{Timeout: defaultDeleteClientTimeout} - } + client := cmp.Or(b.Client, &http.Client{Timeout: defaultDeleteClientTimeout}) logger := log.WithFunc("peer.Broadcaster.Delete") var wg sync.WaitGroup for _, addr := range addrs { diff --git a/sandboxd/store/peer/probe.go b/sandboxd/store/peer/probe.go index 8fcbdc2e..0745cd38 100644 --- a/sandboxd/store/peer/probe.go +++ b/sandboxd/store/peer/probe.go @@ -94,10 +94,7 @@ func (p *HTTPProber) fanOut(ctx context.Context, id string, maxOwners int, grace if len(addrs) == 0 { return nil } - client := p.Client - if client == nil { - client = &http.Client{Timeout: defaultProbeClientTimeout} - } + client := cmp.Or(p.Client, &http.Client{Timeout: defaultProbeClientTimeout}) ctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), defaultProbeClientTimeout) defer cancel() diff --git a/sandboxd/store/peer/transport.go b/sandboxd/store/peer/transport.go index 50dcb97d..09e72c38 100644 --- a/sandboxd/store/peer/transport.go +++ b/sandboxd/store/peer/transport.go @@ -59,10 +59,7 @@ func (p *HTTPPuller) Pull(ctx context.Context, addr, id, dst string) error { req.Header.Set("Authorization", "Bearer "+p.Token) } - client := p.Client - if client == nil { - client = http.DefaultClient - } + client := cmp.Or(p.Client, http.DefaultClient) resp, err := client.Do(req) //nolint:gosec // addr comes from the mesh's own member view if err != nil { return fmt.Errorf("pull %s from %s: %w", id, addr, err) diff --git a/sandboxd/store/s3/s3.go b/sandboxd/store/s3/s3.go index 3df01115..9e3ad615 100644 --- a/sandboxd/store/s3/s3.go +++ b/sandboxd/store/s3/s3.go @@ -40,7 +40,7 @@ type Config struct { Prefix string `json:"prefix,omitempty"` Endpoint string `json:"endpoint,omitempty"` Region string `json:"region,omitempty"` - ForcePathStyle bool `json:"force_path_style,omitempty"` + ForcePathStyle bool `json:"force_path_style,omitzero"` } type publishFile struct { @@ -197,8 +197,8 @@ func (s *Store) readMeta(ctx context.Context, id string) ([]byte, string, error) Bucket: &s.bucket, Key: aws.String(s.key(id, store.MetaFile)), }) if err != nil { - var apiErr smithy.APIError - if errors.As(err, &apiErr) && (apiErr.ErrorCode() == "NoSuchKey" || apiErr.ErrorCode() == "NotFound") { //nolint:goconst // AWS API error codes, compared as literals + apiErr, ok := errors.AsType[smithy.APIError](err) + if ok && (apiErr.ErrorCode() == "NoSuchKey" || apiErr.ErrorCode() == "NotFound") { //nolint:goconst // AWS API error codes, compared as literals return nil, "", store.ErrNotFound } return nil, "", fmt.Errorf("record %s: %w", id, err) diff --git a/sandboxd/store/s3/s3_test.go b/sandboxd/store/s3/s3_test.go index f80f3fc4..5ae9cca3 100644 --- a/sandboxd/store/s3/s3_test.go +++ b/sandboxd/store/s3/s3_test.go @@ -132,7 +132,7 @@ func TestFetchLegacyExportLayout(t *testing.T) { t.Fatalf("Fetch: %v", err) } defer release() - got, err := os.ReadFile(filepath.Join(dir, "disk.img")) //nolint:gosec // test path + got, err := os.ReadFile(filepath.Join(dir, "disk.img")) if err != nil || string(got) != "legacy-bytes" { t.Fatalf("fetched legacy export: %q, %v", got, err) } @@ -296,7 +296,7 @@ func TestPublishDigestedFailurePreservesCommittedGeneration(t *testing.T) { t.Fatalf("Fetch old generation: %v", err) } defer release() - content, err := os.ReadFile(filepath.Join(dir, "disk.img")) //nolint:gosec // test path + content, err := os.ReadFile(filepath.Join(dir, "disk.img")) if err != nil { t.Fatalf("read old generation: %v", err) } @@ -314,7 +314,7 @@ func TestPublishDigestedFailurePreservesCommittedGeneration(t *testing.T) { t.Fatalf("Fetch replacement: %v", err) } defer release() - content, err = os.ReadFile(filepath.Join(dir, "disk.img")) //nolint:gosec // test path + content, err = os.ReadFile(filepath.Join(dir, "disk.img")) if err != nil { t.Fatalf("read replacement: %v", err) } @@ -339,7 +339,7 @@ func TestRepublishRetainsGenerationSelectedByAnotherStore(t *testing.T) { if err = reader.populate(t.Context(), id, selected, gen); err != nil { t.Fatalf("fetch selected first generation after re-publish: %v", err) } - got, err := os.ReadFile(filepath.Join(gen, store.ExportDir, "disk.img")) //nolint:gosec // test path + got, err := os.ReadFile(filepath.Join(gen, store.ExportDir, "disk.img")) if err != nil || string(got) != "first" { t.Fatalf("selected generation bytes: %q, %v, want first", got, err) } diff --git a/sandboxd/types/api.go b/sandboxd/types/api.go index 3fbe93e9..2b3ca854 100644 --- a/sandboxd/types/api.go +++ b/sandboxd/types/api.go @@ -6,7 +6,7 @@ import ( // TTLField is the shared requested-lease field; zero means the server default. type TTLField struct { - TTLSeconds int `json:"ttl_seconds,omitempty"` + TTLSeconds int `json:"ttl_seconds,omitzero"` } // TTL converts the wire seconds to a duration. @@ -21,11 +21,11 @@ type ClaimRequest struct { Size Size `json:"size,omitempty"` Volumes []Volume `json:"volumes,omitempty"` // VolumesAttachOnly attaches every requested volume without mounting it. - VolumesAttachOnly bool `json:"volumes_attach_only,omitempty"` + VolumesAttachOnly bool `json:"volumes_attach_only,omitzero"` TTLField - NoRedirect bool `json:"no_redirect,omitempty"` + NoRedirect bool `json:"no_redirect,omitzero"` // RequirePromoted makes the target refuse a cold-image fallback. - RequirePromoted bool `json:"require_promoted,omitempty"` + RequirePromoted bool `json:"require_promoted,omitzero"` // ClaimRef is an opaque caller reference recorded on the claim. ClaimRef string `json:"claim_ref,omitempty"` } @@ -50,7 +50,7 @@ type ClaimResponse struct { Redirect []string `json:"redirect,omitempty"` // RequirePromoted tells a redirecting client to preserve that requirement on retry. - RequirePromoted bool `json:"require_promoted,omitempty"` + RequirePromoted bool `json:"require_promoted,omitzero"` } // VolumeInfo is the caller-visible, host-path-free catalog projection. @@ -61,7 +61,7 @@ type VolumeInfo struct { Available bool `json:"available"` Nodes int `json:"nodes"` // Writable reports whether the operator allows rw claims of this name. - Writable bool `json:"writable,omitempty"` + Writable bool `json:"writable,omitzero"` } // VolumeListResponse is the wire reply of GET /v1/volumes. @@ -96,7 +96,7 @@ type CheckpointResponse struct { type CheckpointClaimRequest struct { TTLField // NoRedirect makes the retry resolve locally instead of bouncing between two nodes. - NoRedirect bool `json:"no_redirect,omitempty"` + NoRedirect bool `json:"no_redirect,omitzero"` } // CheckpointListResponse is the wire reply of GET /v1/checkpoints. diff --git a/sandboxd/types/types.go b/sandboxd/types/types.go index 74fc4747..7128721c 100644 --- a/sandboxd/types/types.go +++ b/sandboxd/types/types.go @@ -193,7 +193,7 @@ type Checkpoint struct { CreatedAt time.Time `json:"created_at"` // Archive marks a lifecycle-internal wake image: hidden from listings and undeletable. - Archive bool `json:"archive,omitempty"` + Archive bool `json:"archive,omitzero"` } // VMNetConfig is the per-NIC host tap the egress-lane nft lock binds. diff --git a/sandboxd/utils/utils.go b/sandboxd/utils/utils.go index 733a6ea1..f38b9e23 100644 --- a/sandboxd/utils/utils.go +++ b/sandboxd/utils/utils.go @@ -57,11 +57,11 @@ func WriteFileSync(path string, data []byte, perm os.FileMode) error { if err != nil { return fmt.Errorf("open dir: %w", err) } - defer func() { _ = d.Close() }() if err = d.Sync(); err != nil { + _ = d.Close() return fmt.Errorf("sync dir: %w", err) } - return nil + return d.Close() } // RemoveDirEntries removes dir's entries for which match returns true; nil matches all. diff --git a/sdk/go/checkpoint.go b/sdk/go/checkpoint.go index ab6497a5..cf452213 100644 --- a/sdk/go/checkpoint.go +++ b/sdk/go/checkpoint.go @@ -123,8 +123,8 @@ type checkpointResponse struct { } type checkpointClaimRequest struct { - TTLSeconds int `json:"ttl_seconds,omitempty"` - NoRedirect bool `json:"no_redirect,omitempty"` + TTLSeconds int `json:"ttl_seconds,omitzero"` + NoRedirect bool `json:"no_redirect,omitzero"` } type checkpointListResponse struct { diff --git a/sdk/go/client.go b/sdk/go/client.go index 7ca8920c..89823837 100644 --- a/sdk/go/client.go +++ b/sdk/go/client.go @@ -47,6 +47,23 @@ type Client struct { hc *http.Client } +// Connect returns a client for a sandboxd node. addr accepts a +// comma-separated seed list for forward compatibility; v0 uses the first +// entry. Calls are bounded by their ctx — checkpoint and promote run as long +// as the snapshot takes, so the client sets no blanket deadline. +func Connect(addr string, opts ...ClientOption) (*Client, error) { + first, _, _ := strings.Cut(addr, ",") + first = strings.TrimSpace(first) + if first == "" { + return nil, fmt.Errorf("empty sandboxd address") + } + c := &Client{addr: first, hc: &http.Client{}} + for _, opt := range opts { + opt(c) + } + return c, nil +} + // New claims a sandbox for template. Without options the node serves its // defaults: the no-network lane and the smallest size tier. New returns when // the sandbox's silkd is reachable. Against a cluster, a warm miss redirects @@ -198,23 +215,6 @@ func (c *Client) roundTrip(ctx context.Context, method, addr, path string, body return c.hc.Do(req) //nolint:gosec // dialing the caller-configured node is the SDK's purpose } -// Connect returns a client for a sandboxd node. addr accepts a -// comma-separated seed list for forward compatibility; v0 uses the first -// entry. Calls are bounded by their ctx — checkpoint and promote run as long -// as the snapshot takes, so the client sets no blanket deadline. -func Connect(addr string, opts ...ClientOption) (*Client, error) { - first, _, _ := strings.Cut(addr, ",") - first = strings.TrimSpace(first) - if first == "" { - return nil, fmt.Errorf("empty sandboxd address") - } - c := &Client{addr: first, hc: &http.Client{}} - for _, opt := range opts { - opt(c) - } - return c, nil -} - // doJSON issues one control-plane request and decodes a 200 reply into T; // any other status maps through apiError under verb. The shared plumbing // behind every decode-a-reply verb in this file and its siblings. @@ -276,8 +276,8 @@ func tryEach(candidates []string, call func(addr string) error, retry func(error // retryMiss retries a miss (the next candidate may own the record) or a // transport failure (dead peer); a served error is real and stops the walk. func retryMiss(err error) bool { - var he *APIError - return !errors.As(err, &he) || he.Status == http.StatusNotFound + he, ok := errors.AsType[*APIError](err) + return !ok || he.Status == http.StatusNotFound } // retryAny retries a redirect candidate's failure unconditionally: one @@ -292,8 +292,8 @@ func retryAny(error) bool { return true } // request, a forbidden token, or an egress conflict is definitive: the // origin would fail the same way. func retryTransient(err error) bool { - var he *APIError - if !errors.As(err, &he) { + he, ok := errors.AsType[*APIError](err) + if !ok { return true } switch he.Status { @@ -441,10 +441,10 @@ type claimRequest struct { Net string `json:"net,omitempty"` Size string `json:"size,omitempty"` Volumes []Volume `json:"volumes,omitempty"` - VolumesAttachOnly bool `json:"volumes_attach_only,omitempty"` - TTLSeconds int `json:"ttl_seconds,omitempty"` - NoRedirect bool `json:"no_redirect,omitempty"` - RequirePromoted bool `json:"require_promoted,omitempty"` + VolumesAttachOnly bool `json:"volumes_attach_only,omitzero"` + TTLSeconds int `json:"ttl_seconds,omitzero"` + NoRedirect bool `json:"no_redirect,omitzero"` + RequirePromoted bool `json:"require_promoted,omitzero"` ClaimRef string `json:"claim_ref,omitempty"` } @@ -480,7 +480,7 @@ type claimResponse struct { TemplateDigest string `json:"template_digest,omitempty"` Volumes []Volume `json:"volumes,omitempty"` Redirect []string `json:"redirect,omitempty"` - RequirePromoted bool `json:"require_promoted,omitempty"` + RequirePromoted bool `json:"require_promoted,omitzero"` } type volumeListResponse struct { @@ -490,7 +490,7 @@ type volumeListResponse struct { type forkRequest struct { Token string `json:"token"` Count int `json:"count"` - TTLSeconds int `json:"ttl_seconds,omitempty"` + TTLSeconds int `json:"ttl_seconds,omitzero"` } type forkResponse struct { diff --git a/sdk/go/client_test.go b/sdk/go/client_test.go index e2232b58..57ede843 100644 --- a/sdk/go/client_test.go +++ b/sdk/go/client_test.go @@ -495,8 +495,8 @@ func TestAPIErrorDrainsOversizedBody(t *testing.T) { body := strings.NewReader(`{"error":"` + strings.Repeat("x", 3*4096) + `"}`) err := apiError("claim", &http.Response{StatusCode: http.StatusTooManyRequests, Body: io.NopCloser(body)}) - var he *APIError - if !errors.As(err, &he) || he.Status != http.StatusTooManyRequests { + he, ok := errors.AsType[*APIError](err) + if !ok || he.Status != http.StatusTooManyRequests { t.Fatalf("apiError = %v, want a 429 APIError", err) } if body.Len() != 0 { diff --git a/sdk/go/find_test.go b/sdk/go/find_test.go index bf681b5f..68a3acd4 100644 --- a/sdk/go/find_test.go +++ b/sdk/go/find_test.go @@ -67,8 +67,8 @@ func TestFindSeqStopsAfterTheCallerBreaks(t *testing.T) { func TestFindBadPatternIsTypedError(t *testing.T) { sb := fakeSandbox(t) _, err := sb.Find(t.Context(), "/", "(", "") - var e *wire.ErrorResp - if !errors.As(err, &e) || e.Kind != wire.KindBadRequest { + e, ok := errors.AsType[*wire.ErrorResp](err) + if !ok || e.Kind != wire.KindBadRequest { t.Errorf("got %v, want bad_request error frame", err) } } diff --git a/sdk/go/info.go b/sdk/go/info.go index 156356d3..c0f44ff1 100644 --- a/sdk/go/info.go +++ b/sdk/go/info.go @@ -16,9 +16,9 @@ type NodeInfo struct { Claimed int `json:"claimed"` Hibernated int `json:"hibernated"` Archived int `json:"archived"` - Draining bool `json:"draining,omitempty"` + Draining bool `json:"draining,omitzero"` Peers []string `json:"peers,omitempty"` - AtCapacity bool `json:"at_capacity,omitempty"` + AtCapacity bool `json:"at_capacity,omitzero"` AtCapacityReason string `json:"at_capacity_reason,omitempty"` } @@ -45,7 +45,7 @@ type SandboxSummary struct { Key PoolKey `json:"key"` Deadline time.Time `json:"deadline"` Hibernated bool `json:"hibernated"` - Archived bool `json:"archived,omitempty"` + Archived bool `json:"archived,omitzero"` FromCheckpoint string `json:"from_checkpoint,omitempty"` Volumes []Volume `json:"volumes,omitempty"` ClaimRef string `json:"claim_ref,omitempty"` diff --git a/sdk/go/options.go b/sdk/go/options.go index dde7eb63..a24f65bc 100644 --- a/sdk/go/options.go +++ b/sdk/go/options.go @@ -42,7 +42,7 @@ type VolumeInfo struct { SizeBytes int64 `json:"size_bytes"` Available bool `json:"available"` Nodes int `json:"nodes"` - Writable bool `json:"writable,omitempty"` + Writable bool `json:"writable,omitzero"` } // Option configures a New claim. diff --git a/sdk/go/pools.go b/sdk/go/pools.go index 4f61767d..ad6c9ad6 100644 --- a/sdk/go/pools.go +++ b/sdk/go/pools.go @@ -15,10 +15,10 @@ type PoolSpec struct { Net NetShape `json:"net,omitempty"` Size Size `json:"size,omitempty"` Warm int `json:"warm"` - WarmMax int `json:"warm_max,omitempty"` - IdleHibernateSeconds int `json:"idle_hibernate_seconds,omitempty"` - ArchiveAfterSeconds int `json:"archive_after_seconds,omitempty"` - ArchiveDeleteAfterSeconds int `json:"archive_delete_after_seconds,omitempty"` + WarmMax int `json:"warm_max,omitzero"` + IdleHibernateSeconds int `json:"idle_hibernate_seconds,omitzero"` + ArchiveAfterSeconds int `json:"archive_after_seconds,omitzero"` + ArchiveDeleteAfterSeconds int `json:"archive_delete_after_seconds,omitzero"` } // PoolResult is one node's outcome from SetPoolsCluster. diff --git a/sdk/go/port.go b/sdk/go/port.go index 7c0a50ab..b0ffd909 100644 --- a/sdk/go/port.go +++ b/sdk/go/port.go @@ -182,7 +182,7 @@ func closeWrite(conn net.Conn) { type previewRequest struct { Token string `json:"token"` Port uint16 `json:"port"` - TTLSeconds int `json:"ttl_seconds,omitempty"` + TTLSeconds int `json:"ttl_seconds,omitzero"` } type previewResponse struct { diff --git a/sdk/go/sandbox.go b/sdk/go/sandbox.go index ba2cc85a..ddf5418d 100644 --- a/sdk/go/sandbox.go +++ b/sdk/go/sandbox.go @@ -202,7 +202,7 @@ func (s *Sandbox) Hibernate(ctx context.Context) error { func (s *Sandbox) Close() error { ctx, cancel := context.WithTimeout(context.Background(), releaseTimeout) defer cancel() - resp, err := s.post(ctx, "release", nil) + resp, err := s.c.roundTrip(ctx, http.MethodPost, s.owner, "/v1/sandboxes/"+s.ID+"/release", nil, s.token) if err != nil { return err } @@ -239,12 +239,6 @@ func (s *Sandbox) call(ctx context.Context, req wire.Request) (*silkd.Conn, func return conn, done, nil } -// post sends a sandbox-scoped verb to the owning node (which holds the -// claim); a non-nil body is JSON. -func (s *Sandbox) post(ctx context.Context, verb string, body io.Reader) (*http.Response, error) { - return s.c.roundTrip(ctx, http.MethodPost, s.owner, "/v1/sandboxes/"+s.ID+"/"+verb, body, s.token) -} - // pumpStdin chunks the reader into stdin frames; Send's own locking keeps // the pump safe against the caller's concurrent frames. func pumpStdin(conn *silkd.Conn, r io.Reader) { diff --git a/sdk/go/sandbox_test.go b/sdk/go/sandbox_test.go index d879c9af..02b2994d 100644 --- a/sdk/go/sandbox_test.go +++ b/sdk/go/sandbox_test.go @@ -33,8 +33,8 @@ func TestExecNonZeroExit(t *testing.T) { sb := testSandbox(t, newAgentServer(t, silkdtest.ServeConn)) _, err := sb.Exec(t.Context(), "false") - var exitErr *ExitError - if !errors.As(err, &exitErr) || exitErr.Code != 1 { + exitErr, ok := errors.AsType[*ExitError](err) + if !ok || exitErr.Code != 1 { t.Errorf("got %v, want ExitError code 1", err) } } @@ -43,8 +43,8 @@ func TestExecSurfacesErrorFrame(t *testing.T) { sb := testSandbox(t, newAgentServer(t, silkdtest.ServeConn)) _, err := sb.Exec(t.Context(), "no-such-binary") - var silkdErr *wire.ErrorResp - if !errors.As(err, &silkdErr) || silkdErr.Kind != wire.KindNotFound { + silkdErr, ok := errors.AsType[*wire.ErrorResp](err) + if !ok || silkdErr.Kind != wire.KindNotFound { t.Errorf("got %v, want silkd not_found error", err) } } diff --git a/sdk/go/silkd/silkdtest/fake.go b/sdk/go/silkd/silkdtest/fake.go index eb424aa8..0560852a 100644 --- a/sdk/go/silkd/silkdtest/fake.go +++ b/sdk/go/silkd/silkdtest/fake.go @@ -117,12 +117,7 @@ func (f *Fake) fsRead(conn net.Conn, path string) { errFrame(conn, wire.KindNotFound, err.Error()) return } - for len(data) > 0 { - n := min(readChunk, len(data)) - send(conn, &wire.DataResp{Data: data[:n]}) - data = data[n:] - } - send(conn, wire.Done{}) + sendChunked(conn, data) } func (f *Fake) fsList(conn net.Conn, path string) { @@ -243,12 +238,7 @@ func (f *Fake) fsPull(conn net.Conn, path string) { } _ = tw.Close() data := buf.Bytes() - for len(data) > 0 { - n := min(readChunk, len(data)) - send(conn, &wire.DataResp{Data: data[:n]}) - data = data[n:] - } - send(conn, wire.Done{}) + sendChunked(conn, data) } func (f *Fake) sessionCreate(conn net.Conn, req *wire.SessionCreate) { @@ -481,3 +471,12 @@ func drainUpload(r *bufio.Reader) ([]byte, error) { } } } + +func sendChunked(conn net.Conn, data []byte) { + for len(data) > 0 { + n := min(readChunk, len(data)) + send(conn, &wire.DataResp{Data: data[:n]}) + data = data[n:] + } + send(conn, wire.Done{}) +} diff --git a/sdk/go/watch_test.go b/sdk/go/watch_test.go index ed229d72..a87229e7 100644 --- a/sdk/go/watch_test.go +++ b/sdk/go/watch_test.go @@ -40,8 +40,8 @@ func TestWatchDeliversEventsUntilClose(t *testing.T) { func TestWatchMissingPathFailsSynchronously(t *testing.T) { sb := fakeSandbox(t) _, err := sb.Watch(t.Context(), "/nope", false) - var e *wire.ErrorResp - if !errors.As(err, &e) || e.Kind != wire.KindNotFound { + e, ok := errors.AsType[*wire.ErrorResp](err) + if !ok || e.Kind != wire.KindNotFound { t.Errorf("Watch = %v, want synchronous not_found (no ready frame)", err) } } diff --git a/sdk/langchain/cocoonsandbox_langchain/toolkit.py b/sdk/langchain/cocoonsandbox_langchain/toolkit.py index 9d533b66..5d36a113 100644 --- a/sdk/langchain/cocoonsandbox_langchain/toolkit.py +++ b/sdk/langchain/cocoonsandbox_langchain/toolkit.py @@ -39,8 +39,15 @@ class CocoonToolkit: prepared state. """ - def __init__(self, addr: str, api_token: str = "", template: str = "rt:24.04", - net: str = "", ttl_seconds: int = 0, from_checkpoint: str = ""): + def __init__( + self, + addr: str, + api_token: str = "", + template: str = "rt:24.04", + net: str = "", + ttl_seconds: int = 0, + from_checkpoint: str = "", + ): self._client = Client(addr, api_token=api_token, timeout=CALL_TIMEOUT) self._template = template self._net = net @@ -59,28 +66,40 @@ def __exit__(self, *exc) -> None: def get_tools(self) -> list[StructuredTool]: """The sandbox tool set; sync-native (_run), async via to_thread.""" return [ - self._tool("sandbox_exec", - "Run a shell command in the sandbox and wait for it to exit. " - "Returns stdout; a non-empty stderr is appended as a 'stderr:' " - "line and a non-zero status as an 'exit code: N' line; a " - "command that prints nothing and exits 0 returns '(no output)'. " - "The call is cut off after 5 minutes. " - "Files and installed packages persist across calls; environment " - "variables and the working directory do not.", - ExecInput, self._exec), - self._tool("sandbox_write_file", - "Write text to a file in the sandbox, replacing any existing " - "file atomically. The parent directory must already exist.", - WriteFileInput, self._write_file), - self._tool("sandbox_read_file", - "Return the whole content of a file in the sandbox as text " - "(undecodable bytes are replaced); a missing path is a tool " - "error. Prefer sandbox_exec with head or tail for large files.", - PathInput, self._read_file), - self._tool("sandbox_list_dir", - "List one directory (not recursive) as a JSON array of " - "{name, kind, size}; kind is file, dir, symlink, or other.", - PathInput, self._list_dir), + self._tool( + "sandbox_exec", + "Run a shell command in the sandbox and wait for it to exit. " + "Returns stdout; a non-empty stderr is appended as a 'stderr:' " + "line and a non-zero status as an 'exit code: N' line; a " + "command that prints nothing and exits 0 returns '(no output)'. " + "The call is cut off after 5 minutes. " + "Files and installed packages persist across calls; environment " + "variables and the working directory do not.", + ExecInput, + self._exec, + ), + self._tool( + "sandbox_write_file", + "Write text to a file in the sandbox, replacing any existing " + "file atomically. The parent directory must already exist.", + WriteFileInput, + self._write_file, + ), + self._tool( + "sandbox_read_file", + "Return the whole content of a file in the sandbox as text " + "(undecodable bytes are replaced); a missing path is a tool " + "error. Prefer sandbox_exec with head or tail for large files.", + PathInput, + self._read_file, + ), + self._tool( + "sandbox_list_dir", + "List one directory (not recursive) as a JSON array of " + "{name, kind, size}; kind is file, dir, symlink, or other.", + PathInput, + self._list_dir, + ), ] def close(self) -> None: @@ -111,14 +130,14 @@ def _tool(self, name: str, description: str, schema: type[BaseModel], func) -> S async def arun(**kwargs): return await asyncio.to_thread(func, **kwargs) - return StructuredTool.from_function(func=func, coroutine=arun, name=name, - description=description, args_schema=schema) + return StructuredTool.from_function( + func=func, coroutine=arun, name=name, description=description, args_schema=schema + ) def _exec(self, command: str, cwd: str = "") -> str: out: list[bytes] = [] errs: list[bytes] = [] - code = self.sandbox().run(["sh", "-c", command], cwd=cwd, - on_stdout=out.append, on_stderr=errs.append) + code = self.sandbox().run(["sh", "-c", command], cwd=cwd, on_stdout=out.append, on_stderr=errs.append) stdout = b"".join(out).decode(errors="replace") stderr = b"".join(errs).decode(errors="replace") result = stdout diff --git a/sdk/langchain/tests/test_toolkit.py b/sdk/langchain/tests/test_toolkit.py index 17bf8dca..08e8c1f8 100644 --- a/sdk/langchain/tests/test_toolkit.py +++ b/sdk/langchain/tests/test_toolkit.py @@ -75,7 +75,7 @@ def test_file_tools_round_trip(monkeypatch): def test_close_releases_once(monkeypatch): kit, fake = hooked(monkeypatch) - kit.get_tools()[0].invoke({"command": "x"}) # forces the claim + kit.get_tools()[0].invoke({"command": "x"}) kit.close() kit.close() assert fake.closed == 1 diff --git a/sdk/openai/cocoonsandbox_openai/__init__.py b/sdk/openai/cocoonsandbox_openai/__init__.py index 9b83b65f..9911c959 100644 --- a/sdk/openai/cocoonsandbox_openai/__init__.py +++ b/sdk/openai/cocoonsandbox_openai/__init__.py @@ -1,11 +1,11 @@ """OpenAI Agents SDK sandbox provider for cocoon microVMs. - from agents.sandbox import SandboxRunConfig # SDK side - from cocoonsandbox_openai import CocoonSandboxClient, CocoonSandboxClientOptions +from agents.sandbox import SandboxRunConfig # SDK side +from cocoonsandbox_openai import CocoonSandboxClient, CocoonSandboxClientOptions - client = CocoonSandboxClient() - session = await client.create(options=CocoonSandboxClientOptions( - addr="10.0.0.5:7777", api_token="...", template="ghcr.io/cocoonstack/sandbox/rt:24.04")) +client = CocoonSandboxClient() +session = await client.create(options=CocoonSandboxClientOptions( + addr="10.0.0.5:7777", api_token="...", template="ghcr.io/cocoonstack/sandbox/rt:24.04")) """ from .adapter import ( diff --git a/sdk/openai/cocoonsandbox_openai/adapter.py b/sdk/openai/cocoonsandbox_openai/adapter.py index ae99ae42..9ea611b3 100644 --- a/sdk/openai/cocoonsandbox_openai/adapter.py +++ b/sdk/openai/cocoonsandbox_openai/adapter.py @@ -145,9 +145,7 @@ async def create( options: CocoonSandboxClientOptions, ) -> SandboxSession: client = Client(options.addr, api_token=options.api_token) - sb = await asyncio.to_thread( - client.new, options.template, net=options.net, ttl_seconds=options.ttl_seconds - ) + sb = await asyncio.to_thread(client.new, options.template, net=options.net, ttl_seconds=options.ttl_seconds) if manifest is None: manifest = Manifest(root="/workspace") session_id = uuid.uuid4() diff --git a/sdk/openai/tests/test_adapter.py b/sdk/openai/tests/test_adapter.py index 72049faa..cd44f448 100644 --- a/sdk/openai/tests/test_adapter.py +++ b/sdk/openai/tests/test_adapter.py @@ -57,7 +57,6 @@ async def go(): inner = session._inner assert inner.state.sandbox_id == "sb_1" and inner.state.sandbox_token == "tok" - # serialize → deserialize must reattach to the same claim. payload = client.serialize_session_state(inner.state) restored = client.deserialize_session_state(payload) assert isinstance(restored, CocoonSandboxSessionState) diff --git a/sdk/python/README.md b/sdk/python/README.md index 1dbce801..74b1849e 100644 --- a/sdk/python/README.md +++ b/sdk/python/README.md @@ -11,7 +11,7 @@ client = Client("10.0.0.5:7777", api_token="...") with client.new("ghcr.io/cocoonstack/sandbox/rt:24.04") as sb: print(sb.exec("echo", "hello")) ckpt = sb.checkpoint("after-setup") - branch = ckpt.new() # a fresh sandbox at the captured moment + branch = ckpt.new() # a fresh sandbox at the captured moment ``` stdlib-only and synchronous — `pip install cocoonstack-sandbox` brings no diff --git a/sdk/python/cocoonsandbox/checkpoint.py b/sdk/python/cocoonsandbox/checkpoint.py index 5dce1d2d..a8550f0e 100644 --- a/sdk/python/cocoonsandbox/checkpoint.py +++ b/sdk/python/cocoonsandbox/checkpoint.py @@ -27,22 +27,8 @@ def new(self, ttl_seconds: int = 0) -> Sandbox: redirect to the node that actually holds it; if every candidate fails transiently, the claim falls back to the origin once so it heals (pulls the checkpoint) locally.""" - # local import: a top-level one closes the client -> sandbox -> checkpoint cycle. - from .client import _redirect_fallback - claim = {"ttl_seconds": ttl_seconds} if ttl_seconds else {} - path = f"/v1/checkpoints/{self.id}/claim" - reply = self._client._post_json(self._addr, path, claim, "claim checkpoint") - redirect = reply.get("redirect") or [] - if not redirect: - return self._client._handle_from(self._addr, reply) - claim["no_redirect"] = True - - def post(peer): - return self._client._post_json(peer, path, claim, "claim checkpoint") - - addr, reply = _redirect_fallback(self._addr, redirect, post, "claim checkpoint") - return self._client._handle_from(addr, reply) + return self._client._claim_from(self._addr, claim, f"/v1/checkpoints/{self.id}/claim", "claim checkpoint") def delete(self) -> None: """Removes the checkpoint and broadcasts the drop; cleanup is diff --git a/sdk/python/cocoonsandbox/client.py b/sdk/python/cocoonsandbox/client.py index 435dad58..44d8e8cc 100644 --- a/sdk/python/cocoonsandbox/client.py +++ b/sdk/python/cocoonsandbox/client.py @@ -125,8 +125,8 @@ def info(self) -> dict: """The node's pool/claim counters, as served by GET /v1/info.""" return self._request(self.addr, "GET", "/v1/info", None, "info") - def _claim_from(self, addr: str, claim: dict) -> Sandbox: - reply = self._post_json(addr, "/v1/claim", claim, "claim") + def _claim_from(self, addr: str, claim: dict, path: str = "/v1/claim", verb: str = "claim") -> Sandbox: + reply = self._post_json(addr, path, claim, verb) redirect = reply.get("redirect") or [] if not redirect: return self._handle_from(addr, reply) @@ -135,9 +135,9 @@ def _claim_from(self, addr: str, claim: dict) -> Sandbox: claim["require_promoted"] = True def post(peer): - return self._post_json(peer, "/v1/claim", claim, "claim") + return self._post_json(peer, path, claim, verb) - owner, reply = _redirect_fallback(addr, redirect, post, "claim") + owner, reply = _redirect_fallback(addr, redirect, post, verb) return self._handle_from(owner, reply) def _peers(self) -> list: diff --git a/sdk/python/cocoonsandbox/sandbox.py b/sdk/python/cocoonsandbox/sandbox.py index 997da7ad..7470c983 100644 --- a/sdk/python/cocoonsandbox/sandbox.py +++ b/sdk/python/cocoonsandbox/sandbox.py @@ -50,7 +50,7 @@ def __enter__(self) -> Sandbox: def __exit__(self, *exc) -> None: try: self.close() - except Exception: + except APIError: if exc[0] is None: # a clean block surfaces a real release failure raise diff --git a/sdk/python/tests/test_client.py b/sdk/python/tests/test_client.py index 77c43e9d..08130f74 100644 --- a/sdk/python/tests/test_client.py +++ b/sdk/python/tests/test_client.py @@ -21,6 +21,9 @@ def do_GET(self): def do_DELETE(self): self._dispatch("DELETE") + def log_message(self, *args): + pass + def _dispatch(self, method): length = int(self.headers.get("Content-Length") or 0) body = json.loads(self.rfile.read(length)) if length else {} @@ -39,9 +42,6 @@ def _reply(self, code, payload): self.end_headers() self.wfile.write(raw) - def log_message(self, *args): - pass - @pytest.fixture def node(): diff --git a/sdk/python/tests/test_frames.py b/sdk/python/tests/test_frames.py index 9f043488..ed40136a 100644 --- a/sdk/python/tests/test_frames.py +++ b/sdk/python/tests/test_frames.py @@ -21,16 +21,15 @@ def test_nested_data_field_not_shadowed(): assert frame["meta"] == {"data": "WFhY"} -def test_trailing_bytes_rejected(): +@pytest.mark.parametrize( + "raw", + [ + b'{"type":"stdout","data":"aGk="}garbage', + b'{"type":"data","data":"QUJD\r\nREVG"}', + b'{"type":"stdout","data":"aGk="', + ], + ids=["trailing_bytes", "control_bytes_in_base64", "unterminated_frame"], +) +def test_malformed_frame_rejected(raw): with pytest.raises(json.JSONDecodeError): - frames.decode_response(b'{"type":"stdout","data":"aGk="}garbage') - - -def test_control_bytes_in_base64_rejected(): - with pytest.raises(json.JSONDecodeError): - frames.decode_response(b'{"type":"data","data":"QUJD\r\nREVG"}') - - -def test_unterminated_frame_rejected(): - with pytest.raises(json.JSONDecodeError): - frames.decode_response(b'{"type":"stdout","data":"aGk="') + frames.decode_response(raw) diff --git a/sdk/python/tests/test_proc.py b/sdk/python/tests/test_proc.py index 394a8d4a..2f54b452 100644 --- a/sdk/python/tests/test_proc.py +++ b/sdk/python/tests/test_proc.py @@ -37,6 +37,26 @@ def close(self): pass +class BlockingStdinConn(FakeConn): + """A guest that stops draining stdin until its output is read: send blocks + past the buffer, exactly as the real socket pair does.""" + + def __init__(self, frames, buffer_frames): + super().__init__(frames) + self._room = threading.Semaphore(buffer_frames) + self._read = False + + def send(self, op, **fields): + if op == "stdin" and not self._read and not self._room.acquire(blocking=False): + assert self._room.acquire(timeout=5), "stdin send deadlocked" + super().send(op, **fields) + + def recv(self): + self._read = True + self._room.release() + return super().recv() + + def fake_sandbox(monkeypatch, frames): sb = Sandbox(client=Client("127.0.0.1:1"), id="sb_1", token="tok", owner="127.0.0.1:1") conn = FakeConn(frames) @@ -79,26 +99,6 @@ def test_attach_returns_exit_code(monkeypatch): assert out == [b"late"] -class BlockingStdinConn(FakeConn): - """A guest that stops draining stdin until its output is read: send blocks - past the buffer, exactly as the real socket pair does.""" - - def __init__(self, frames, buffer_frames): - super().__init__(frames) - self._room = threading.Semaphore(buffer_frames) - self._read = False - - def send(self, op, **fields): - if op == "stdin" and not self._read and not self._room.acquire(blocking=False): - assert self._room.acquire(timeout=5), "stdin send deadlocked" - super().send(op, **fields) - - def recv(self): - self._read = True - self._room.release() - return super().recv() - - def test_run_pumps_stdin_while_reading_output(monkeypatch): sb, conn = fake_sandbox(monkeypatch, [{"type": "exit", "code": 0}]) blocking = BlockingStdinConn([{"type": "exit", "code": 0}], buffer_frames=1) diff --git a/sdk/python/tests/test_wire_binding.py b/sdk/python/tests/test_wire_binding.py index ca03884c..f7e1de60 100644 --- a/sdk/python/tests/test_wire_binding.py +++ b/sdk/python/tests/test_wire_binding.py @@ -88,6 +88,32 @@ UNSENT = {"req_session_create": {"id"}} +class BranchActionConn: + """Records the action each git_branch verb puts on the wire.""" + + def __init__(self, sent): + self.sent = sent + self.action = "" + + def __enter__(self): + return self + + def __exit__(self, *exc): + pass + + def send(self, op, **fields): + self.action = fields.get("action") + self.sent.append(self.action) + + def recv(self): + if self.action == "list": + return {"type": "git_branches", "branches": [], "current": ""} + return {"type": "done"} + + def recv_until(self, *terminal): + yield self.recv() + + def _pty_stub(sb, pid): return Pty(sb, None, pid) @@ -130,32 +156,6 @@ def guest(): return sb, sent, thread -class BranchActionConn: - """Records the action each git_branch verb puts on the wire.""" - - def __init__(self, sent): - self.sent = sent - self.action = "" - - def __enter__(self): - return self - - def __exit__(self, *exc): - pass - - def send(self, op, **fields): - self.action = fields.get("action") - self.sent.append(self.action) - - def recv(self): - if self.action == "list": - return {"type": "git_branches", "branches": [], "current": ""} - return {"type": "done"} - - def recv_until(self, *terminal): - yield self.recv() - - @pytest.mark.parametrize("stem,replies,invoke", CASES, ids=lambda c: c if isinstance(c, str) else "") def test_call_site_matches_fixture(monkeypatch, stem, replies, invoke): fixture = json.loads((FIXTURES / f"{stem}.json").read_text()) diff --git a/silkd/src/find.rs b/silkd/src/find.rs index f1c84808..ca054fc1 100644 --- a/silkd/src/find.rs +++ b/silkd/src/find.rs @@ -68,6 +68,7 @@ impl Walk<'_> { fn run(&self, root: PathBuf) -> std::io::Result { let mut stack = vec![root]; let mut root = true; + let mut body = String::new(); while let Some(dir) = stack.pop() { let rd = match std::fs::read_dir(&dir) { Ok(rd) => rd, @@ -89,7 +90,10 @@ impl Walk<'_> { let p = ent.path(); if ft.is_dir() { stack.push(p); - } else if ft.is_file() && name_matches(&p, self.name_re) && !self.scan_file(&p) { + } else if ft.is_file() + && name_matches(&p, self.name_re) + && !self.scan_file(&p, &mut body) + { return Ok(false); } } @@ -99,15 +103,15 @@ impl Walk<'_> { } /// Scans one file; the size bound comes off the open handle, so check and read see one file. - fn scan_file(&self, path: &Path) -> bool { + fn scan_file(&self, path: &Path, body: &mut String) -> bool { let Ok(mut file) = std::fs::File::open(path) else { return true; }; if file.metadata().is_ok_and(|m| m.len() > FIND_MAX_FILE) { return true; } - let mut body = String::new(); - if file.read_to_string(&mut body).is_err() { + body.clear(); + if file.read_to_string(body).is_err() { return true; } let name: Arc = path.to_string_lossy().into(); @@ -156,7 +160,7 @@ pub async fn replace( }; for file in files { let mut count: u64 = 0; - if !oversized(&file).await { + if !is_oversized(&file).await { let body = match fs::read_to_string(&file).await { Ok(body) => body, Err(e) => return err_frame(w, &e, "read").await, @@ -264,7 +268,7 @@ where } } -async fn oversized(file: &str) -> bool { +async fn is_oversized(file: &str) -> bool { fs::metadata(file) .await .is_ok_and(|m| m.len() > FIND_MAX_FILE) diff --git a/silkd/src/session.rs b/silkd/src/session.rs index 29845fc3..3b63db03 100644 --- a/silkd/src/session.rs +++ b/silkd/src/session.rs @@ -43,7 +43,6 @@ impl Table { _ => format!("sess-{}", sysutil::tmp_suffix()), }; let mut child = Command::new("bash") - // same sanitized baseline as exec/pty; the request's cwd/env layer on via the init line. .env_clear() .envs(sysutil::base_env()) .stdin(Stdio::piped()) diff --git a/silkd/src/sysutil.rs b/silkd/src/sysutil.rs index f8ba0ae1..21f20878 100644 --- a/silkd/src/sysutil.rs +++ b/silkd/src/sysutil.rs @@ -49,7 +49,7 @@ pub fn rand_token() -> String { /// SIGKILLs the group led by `pgid`, so a session's external command dies with its shell. pub fn kill_group(pgid: u32) { // kill(-0) would target silkd's own group; a synthetic id passes the guard and misses with ESRCH. - if !valid_pid(pgid) { + if !is_valid_pid(pgid) { return; } // SAFETY: kill(2) takes no pointers; the guards above keep the pid_t cast @@ -60,7 +60,7 @@ pub fn kill_group(pgid: u32) { /// Sends `sig` to `pid`, ignoring ESRCH against a just-exited pid. pub fn signal_pid(pid: u32, sig: i32) { // pid 0 means silkd's whole process group to kill(2). - if !valid_pid(pid) { + if !is_valid_pid(pid) { return; } // SAFETY: kill(2) takes no pointers; the guards above keep the pid_t cast @@ -213,7 +213,7 @@ fn fill_random(b: &mut [u8]) -> bool { } /// Rejects pid 0 and anything that would go negative through the pid_t cast. -fn valid_pid(id: u32) -> bool { +fn is_valid_pid(id: u32) -> bool { id != 0 && id <= i32::MAX as u32 } diff --git a/silkd/tests/exec_e2e.rs b/silkd/tests/exec_e2e.rs index fb4e5b9e..89c04f59 100644 --- a/silkd/tests/exec_e2e.rs +++ b/silkd/tests/exec_e2e.rs @@ -6,9 +6,21 @@ mod common; use std::sync::Arc; use std::time::Duration; -use common::{decode, exchange, one, roundtrip, stdout_body, type_of}; use silkd::server::State; +use common::{decode, exchange, one, roundtrip, stdout_body, type_of}; + +async fn detached_pid(state: &Arc, script: &str) -> u64 { + let argv = serde_json::json!(["/bin/sh", "-c", script]); + let started = one( + state, + &serde_json::json!({"op":"exec","argv":argv,"detach":true}).to_string(), + ) + .await; + assert_eq!(type_of(&started[0]), "started"); + started[0]["pid"].as_u64().unwrap() +} + #[tokio::test] async fn exec_streams_stdout_then_exit() { let frames = roundtrip(r#"{"op":"exec","argv":["/bin/echo","-n","hello"]}"#).await; @@ -100,17 +112,6 @@ async fn daemonizer_exit_is_the_last_frame() { ); } -async fn detached_pid(state: &Arc, script: &str) -> u64 { - let argv = serde_json::json!(["/bin/sh", "-c", script]); - let started = one( - state, - &serde_json::json!({"op":"exec","argv":argv,"detach":true}).to_string(), - ) - .await; - assert_eq!(type_of(&started[0]), "started"); - started[0]["pid"].as_u64().unwrap() -} - #[tokio::test] async fn attach_streams_live_output_then_exit() { let state = Arc::new(State::new()); diff --git a/silkd/tests/find_e2e.rs b/silkd/tests/find_e2e.rs index 5ddd4857..45f1a99a 100644 --- a/silkd/tests/find_e2e.rs +++ b/silkd/tests/find_e2e.rs @@ -73,10 +73,10 @@ async fn find_glob_is_a_real_glob_not_a_substring() { "glob": "*.rs" })) .await; - let files: Vec<_> = frames + let files: Vec<&str> = frames .iter() .filter(|f| type_of(f) == "match") - .map(|m| m["file"].as_str().unwrap().to_string()) + .map(|m| m["file"].as_str().unwrap()) .collect(); assert_eq!(files.len(), 1, "{files:?}"); assert!(files[0].ends_with("a.rs")); diff --git a/silkd/tests/forward_e2e.rs b/silkd/tests/forward_e2e.rs index c959d721..24316d7c 100644 --- a/silkd/tests/forward_e2e.rs +++ b/silkd/tests/forward_e2e.rs @@ -8,12 +8,11 @@ use std::time::Duration; use base64::Engine; use base64::engine::general_purpose::STANDARD; +use silkd::server::State; use tokio::io::{AsyncReadExt, AsyncWriteExt}; use tokio::net::TcpListener; use tokio::time::timeout; -use silkd::server::State; - const DEADLINE: Duration = Duration::from_secs(30); async fn echo_listener() -> u16 { @@ -39,6 +38,12 @@ async fn echo_listener() -> u16 { port } +fn data_payload(line: &str) -> Option<&str> { + let start = line.find("\"data\":\"")? + "\"data\":\"".len(); + let rest = &line[start..]; + Some(&rest[..rest.find('"')?]) +} + #[tokio::test] async fn forward_round_trips_and_done_on_server_close() { timeout(DEADLINE, async { @@ -116,12 +121,6 @@ async fn forward_bidirectional_bulk_no_deadlock() { .expect("test deadline"); } -fn data_payload(line: &str) -> Option<&str> { - let start = line.find("\"data\":\"")? + "\"data\":\"".len(); - let rest = &line[start..]; - Some(&rest[..rest.find('"')?]) -} - #[tokio::test] async fn forward_refused_port_is_not_found() { timeout(DEADLINE, async { diff --git a/silkd/tests/fs_e2e.rs b/silkd/tests/fs_e2e.rs index ff660959..917c5960 100644 --- a/silkd/tests/fs_e2e.rs +++ b/silkd/tests/fs_e2e.rs @@ -3,9 +3,10 @@ #![allow(clippy::unwrap_used, clippy::expect_used)] mod common; -use common::{b64, exchange, type_of}; use serde_json::json; +use common::{b64, exchange, type_of}; + #[tokio::test] async fn write_then_read_roundtrips_bytes() { let dir = tempfile::tempdir().unwrap(); diff --git a/silkd/tests/git_e2e.rs b/silkd/tests/git_e2e.rs index de36626f..3d63ce6d 100644 --- a/silkd/tests/git_e2e.rs +++ b/silkd/tests/git_e2e.rs @@ -3,9 +3,10 @@ #![allow(clippy::unwrap_used, clippy::expect_used)] mod common; -use common::{exchange, type_of}; use serde_json::{Value, json}; +use common::{exchange, type_of}; + fn git(dir: &std::path::Path, args: &[&str]) { let ok = std::process::Command::new("git") .arg("-C") diff --git a/silkd/tests/lsp_e2e.rs b/silkd/tests/lsp_e2e.rs index fbadd53f..276a538b 100644 --- a/silkd/tests/lsp_e2e.rs +++ b/silkd/tests/lsp_e2e.rs @@ -1,18 +1,18 @@ //! LSP broker integration against a fake language server, each test under a deadline so a relay deadlock fails CI. #![allow(clippy::unwrap_used, clippy::expect_used)] +mod common; + use std::io::Write; use std::sync::Arc; use std::time::Duration; use serde_json::json; +use silkd::server::State; use tempfile::TempDir; use tokio::io::AsyncWriteExt; use tokio::time::timeout; -use silkd::server::State; - -mod common; use common::{b64, connect, decode, one, type_of}; const DEADLINE: Duration = Duration::from_secs(30); @@ -90,10 +90,7 @@ async fn lsp_broker_relays_to_the_server() { &json!({"op":"lsp_start","language":"faketest"}).to_string(), ) .await; - let server_id = start.last().unwrap()["server_id"] - .as_str() - .unwrap() - .to_string(); + let server_id = start.last().unwrap()["server_id"].as_str().unwrap(); let (mut cw, mut out, handle) = connect(&state); cw.write_all( @@ -160,10 +157,7 @@ async fn lsp_stop_kills_an_idle_server() { &json!({"op":"lsp_start","language":"idletest"}).to_string(), ) .await; - let server_id = start.last().unwrap()["server_id"] - .as_str() - .unwrap() - .to_string(); + let server_id = start.last().unwrap()["server_id"].as_str().unwrap(); let stop = one( &state, &json!({"op":"lsp_stop","server_id":server_id}).to_string(), @@ -193,10 +187,7 @@ async fn lsp_request_reaps_when_the_client_vanishes_before_ready() { &json!({"op":"lsp_start","language":"gonetest"}).to_string(), ) .await; - let server_id = start.last().unwrap()["server_id"] - .as_str() - .unwrap() - .to_string(); + let server_id = start.last().unwrap()["server_id"].as_str().unwrap(); let (mut client, server) = tokio::io::duplex(1 << 20); let request = json!({"op":"lsp_request","server_id":server_id}).to_string(); diff --git a/silkd/tests/pty_e2e.rs b/silkd/tests/pty_e2e.rs index 08ff0750..3b329276 100644 --- a/silkd/tests/pty_e2e.rs +++ b/silkd/tests/pty_e2e.rs @@ -6,11 +6,12 @@ mod common; use std::sync::Arc; use std::time::Duration; -use common::{FrameLines, FrameWriter}; use serde_json::{Value, json}; use silkd::server::State; use tokio::io::AsyncWriteExt; +use common::{FrameLines, FrameWriter}; + async fn send(cw: &mut FrameWriter, frame: Value) { cw.write_all(frame.to_string().as_bytes()).await.unwrap(); cw.write_all(b"\n").await.unwrap(); diff --git a/silkd/tests/session_e2e.rs b/silkd/tests/session_e2e.rs index ac0dce5f..c3e30b2b 100644 --- a/silkd/tests/session_e2e.rs +++ b/silkd/tests/session_e2e.rs @@ -4,15 +4,19 @@ mod common; use std::sync::Arc; +use std::time::Duration; -use common::{one, stdout_body, type_of}; use serde_json::{Value, json}; use silkd::server::State; +use common::{one, stdout_body, type_of}; + async fn create(state: &Arc, extra: Value) -> String { let mut req = json!({"op":"session_create"}); - for (k, v) in extra.as_object().unwrap() { - req[k] = v.clone(); + if let Value::Object(map) = extra { + for (k, v) in map { + req[k] = v; + } } let f = one(state, &req.to_string()).await; assert_eq!(type_of(&f[0]), "session_created", "create failed: {f:?}"); @@ -141,7 +145,6 @@ async fn forged_sentinel_in_output_does_not_desync() { #[tokio::test] async fn reap_idle_removes_idle_sessions() { - use std::time::Duration; let state = Arc::new(State::new()); let id = create(&state, json!({})).await; let reaped = state.sessions.reap_idle(Duration::ZERO); @@ -156,7 +159,6 @@ async fn reap_idle_removes_idle_sessions() { #[tokio::test] async fn reap_skips_a_session_running_a_command() { - use std::time::Duration; let state = Arc::new(State::new()); let id = create(&state, json!({})).await; let busy = { diff --git a/silkd/tests/tree_e2e.rs b/silkd/tests/tree_e2e.rs index 4fc45f83..234cf7b4 100644 --- a/silkd/tests/tree_e2e.rs +++ b/silkd/tests/tree_e2e.rs @@ -6,9 +6,10 @@ mod common; use std::path::Path; use std::process::Command; -use common::{b64, data_frames, exchange, payload, type_of}; use serde_json::json; +use common::{b64, data_frames, exchange, payload, type_of}; + fn sys_tar_create(dir: &Path) -> Vec { let out = Command::new("tar") .arg("-c")