diff --git a/.github/workflows/release.yml b/.github/workflows/release.yml index f1e6d8a1..5bd00b78 100644 --- a/.github/workflows/release.yml +++ b/.github/workflows/release.yml @@ -25,7 +25,7 @@ jobs: - name: Build silkd (musl static) run: | sudo apt-get update && sudo apt-get install -y --no-install-recommends musl-tools - (cd silkd && cargo build --release --target x86_64-unknown-linux-musl) + (cd silkd && cargo build --release --locked --target x86_64-unknown-linux-musl) strip silkd/target/x86_64-unknown-linux-musl/release/silkd mkdir -p dist-extra tar -C silkd/target/x86_64-unknown-linux-musl/release \ diff --git a/.github/workflows/sandboxd.yml b/.github/workflows/sandboxd.yml index cb05e862..1f9466c4 100644 --- a/.github/workflows/sandboxd.yml +++ b/.github/workflows/sandboxd.yml @@ -13,6 +13,8 @@ on: - "mcp/**" - "e2e/**" - "protocol/**" + - "Makefile" + - ".golangci.yml" - ".github/workflows/sandboxd.yml" pull_request: paths: @@ -22,6 +24,8 @@ on: - "mcp/**" - "e2e/**" - "protocol/**" + - "Makefile" + - ".golangci.yml" - ".github/workflows/sandboxd.yml" workflow_dispatch: diff --git a/.github/workflows/shell.yml b/.github/workflows/shell.yml index 1b57d767..76999ce5 100644 --- a/.github/workflows/shell.yml +++ b/.github/workflows/shell.yml @@ -7,10 +7,12 @@ on: branches: [main] paths: - "**.sh" + - "Makefile" - ".github/workflows/shell.yml" pull_request: paths: - "**.sh" + - "Makefile" - ".github/workflows/shell.yml" workflow_dispatch: diff --git a/Makefile b/Makefile index ae7b48e6..d55553c3 100644 --- a/Makefile +++ b/Makefile @@ -22,6 +22,7 @@ $(LOCALBIN): ## Tool versions GOLANGCILINT_VERSION ?= v2.13.2 +GO_VERSION := $(shell sed -n 's/^go //p' sandboxd/go.mod) GOLANGCILINT_ROOT := $(LOCALBIN)/golangci-lint-$(GOLANGCILINT_VERSION) GOLANGCILINT := $(GOLANGCILINT_ROOT)/golangci-lint @@ -32,7 +33,7 @@ GOLANGCILINT := $(GOLANGCILINT_ROOT)/golangci-lint .PHONY: golangci-lint golangci-lint: $(GOLANGCILINT) $(GOLANGCILINT): - GOBIN=$(GOLANGCILINT_ROOT) go install github.com/golangci/golangci-lint/v2/cmd/golangci-lint@$(GOLANGCILINT_VERSION) + GOBIN=$(GOLANGCILINT_ROOT) GOTOOLCHAIN=go$(GO_VERSION) go install github.com/golangci/golangci-lint/v2/cmd/golangci-lint@$(GOLANGCILINT_VERSION) cloc: ## Count lines of code excluding tests (requires cloc) cloc --exclude-dir=target,dist,node_modules --exclude-ext=json \ diff --git a/boot/Dockerfile b/boot/Dockerfile index 4699732c..1720e9c3 100644 --- a/boot/Dockerfile +++ b/boot/Dockerfile @@ -55,7 +55,7 @@ RUN set -e; \ make olddefconfig; \ for s in $arch_asserts DEVTMPFS BLK_DEV_INITRD \ VIRTIO_PCI VIRTIO_BLK VIRTIO_NET \ - VIRTIO_CONSOLE VIRTIO_VSOCKETS \ + VIRTIO_CONSOLE VIRTIO_VSOCKETS UNIX98_PTYS \ EROFS_FS EROFS_FS_ZIP OVERLAY_FS EXT4_FS \ NAMESPACES USER_NS SECCOMP_FILTER \ BRIDGE VETH BRIDGE_NETFILTER NF_NAT NFT_COMPAT \ diff --git a/docs/deploy.md b/docs/deploy.md index 9a32394b..099b0a26 100644 --- a/docs/deploy.md +++ b/docs/deploy.md @@ -91,6 +91,12 @@ sandboxd reads one JSON file (`-config`, default } ``` +As written the `egress` pool's guests reach nothing: on a bridge lane the NIC +is locked default-deny and the proxy door is bound only for a claim that has a +policy. Give the pool an [`egress`](egress.md) block before expecting traffic; +a tenant's claim of it (`acme` above) needs the tenant's own block as well and +gets the intersection of the two. + | field | default | meaning | |---|---|---| | `listen` | `:7777` | control- and data-plane HTTP listener | @@ -120,10 +126,10 @@ sandboxd reads one JSON file (`-config`, default | `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. The pool follows the target down: warm VMs above it are destroyed once the count has stayed above it for a full decay period, so a burst does not leave the pool parked at its high-water mark | | `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, and again in every clone before it joins the warm pool, so those pages are already faulted into the restored VM when the first command runs — 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 guest image's proxy variables wherever nothing routes directly — the none lane and the locked bridge egress lane — with the proxy not yet serving, since a door pre-bound at refill only starts serving at claim); 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) | +| `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 a non-volume claim tries a warm-peer redirect first; a volume claim answers the 429 with no redirect) | | `audit_log` | false | append every relayed frame that opens an RPC — its op + addressing fields, never payloads — to `/audit.jsonl`; continuation `data`, `data_end`, `stdin`, and `stdin_close` frames pass through unrecorded. The file is 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 `method` (`GET`, `CONNECT`, `SOCKS5`, …), `decision` and `secret` (the ref name, never its value) on `egress` records; preview accesses record as op `preview`, one per request. An opening frame whose first line exceeds 4 KiB records as op `oversized` with no addressing fields | | `idle_hibernate_seconds` | 0 (off) | node-wide idle policy for unpooled claims (template/checkpoint claims): a none-lane claim is hibernated once it has had no open data-plane connection (relay, buffered exec, preview) and no egress request in flight for this long; the clock restarts when the last connection closes, so a long command is never cut short. 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 | -| `archive_after_seconds` | 0 (off) | tier below hibernation: a hibernated claim idle this long is checkpointed to the store and its local VM dropped, freeing the node entirely; the next call that reaches the guest restores it transparently (a checkpoint restore's latency) with a fresh server-default 5m lease. Requires `idle_hibernate_seconds > 0` and must exceed it. Node-wide for unpooled keys; per-pool overrides for that pool | +| `archive_after_seconds` | 0 (off) | tier below hibernation: a hibernated claim idle this long is checkpointed to the store and its local VM dropped, freeing the node entirely; the next call that reaches the guest restores it transparently (a checkpoint restore's latency) with a fresh lease of the length the claim was granted (the server default of 5m when it asked for none); nothing renews a lease. Keep `idle_hibernate_seconds` under the leases your claims use, or a woken sandbox that idles again is destroyed by the reap before it can hibernate, its archive already consumed by the wake; one still in use when its lease ends is destroyed the same way. Requires `idle_hibernate_seconds > 0` and must exceed it. Node-wide for unpooled keys; per-pool overrides for that pool | | `archive_delete_after_seconds` | 0 (keep) | purge an archived claim's store checkpoint this long after it was archived, reclaiming storage; the claim is then gone for good. On archive this retention window replaces the live claim deadline; 0 clears the deadline so the archive is kept forever. Same node-wide/per-pool split | | `mesh` | unset | join a cluster ([Clusters](cluster.md)); unset = single node | | `pools[]` | — | warm pools, keyed by `(template, net, size)`. `warm` defaults to 4 only when both `warm` and `warm_max` are omitted; an explicit `warm: 0` is kept, so `warm: 0` under a `warm_max` starts the pool empty and grows it on demand, and `warm: 0` alone registers the key with no warm VMs (`PUT /v1/pools` never defaults `warm`); `net` is `none` or `egress`; `size` is a tier, below. Retune online without a restart via [`PUT /v1/pools`](sandboxd-api.md#put-v1pools) — omitted pools drain. This is the **first-boot seed**: once a node takes a `PUT /v1/pools`, the applied set persists to `/pools.json` and overrides this section on every later boot (a startup log notes it); delete `pools.json` to return to config-owned pools. Egress stays config-owned either way. See [state ownership](cluster.md#state-ownership) | @@ -299,7 +305,7 @@ here validates on load: "max_claims": 200, "audit_log": true, - "idle_hibernate_seconds": 300, + "idle_hibernate_seconds": 120, "archive_after_seconds": 3600, "archive_delete_after_seconds": 604800, @@ -423,6 +429,7 @@ After=network-online.target [Service] ExecStart=/usr/local/bin/sandboxd -config /etc/sandboxd/config.json Restart=on-failure +KillMode=process Environment=SANDBOXD_LOG_LEVEL=info [Install] @@ -435,7 +442,10 @@ degradations. `journalctl` answers an empty range with `-- No entries --`, so a check that counts lines reads one problem where there are none — count records (`-o json | wc -l`) or test the output for emptiness. -Stopping sandboxd leaves VMs alive; the next start reconciles them. Claimed +Stopping sandboxd leaves VMs alive; the next start reconciles them. The unit's +`KillMode=process` is what guarantees it: systemd signals sandboxd alone, so a +VMM that cocoon could not move into its own scope, and any `cocoon` call still +in flight, outlive the stop instead of dying with the service's cgroup. Claimed sandboxes are reaped when their TTL expires (default 5m, capped at 24h). To empty a node for maintenance, cordon it first: diff --git a/docs/egress.md b/docs/egress.md index 78851cba..19bb1c82 100644 --- a/docs/egress.md +++ b/docs/egress.md @@ -44,8 +44,9 @@ supported*), no authentication (the socket path already carries the sandbox's identity), and a `DOMAINNAME` destination is resolved host-side, so the guest still needs no resolver. Each door serves at most 256 connections per sandbox at a time; later ones sit unserved in the socket backlog until one closes, and -the proxy keeps at most 64 idle upstream connections per sandbox, so a guest -cannot turn its proxy into a host descriptor sink. +the proxy keeps at most 64 idle upstream connections per sandbox — 64 more for +decrypted traffic when the pool intercepts — so a guest cannot turn its proxy +into a host descriptor sink. A SOCKS5 tunnel takes the decision an HTTP `CONNECT` to the same host takes, through the same code: a rule with a nonempty `methods` list that omits @@ -72,7 +73,9 @@ curl --socks5-hostname 127.0.0.1:1080 imaps://imap.example.com/ # allowed: {"s A `net:"egress"` guest owns a real NIC on the host bridge and reaches the same proxy over the same vsock path. To stop it bypassing the proxy, sandboxd locks -the NIC at claim: an nftables netdev table (`sandbox_egress_`) with an +the NIC at claim — until then a warm VM of the pool, and the pool's golden +build, sit on the bridge unlocked, running only the image's own services, never +tenant code: an nftables netdev table (`sandbox_egress_`) with an ingress hook on the guest's tap drops every guest-initiated packet except IPv4 broadcast DHCP, so the guest's only routed egress is the audited vsock proxy. The lock is fail-closed — a claim whose NIC cannot be locked is rejected, not handed @@ -154,7 +157,10 @@ and the secret injection that rides it — is always an explicit act. Root claims have no tenant layer and take the pool's policy whole; a key with no configured pool — a promoted template, or an image cold-booted on demand — has no pool layer, so a tenant's claim of one takes the tenant's policy alone (a -root claim of one stays denied). Secrets are registered separately and +root claim of one stays denied). Which layers a claim has is settled when it +is made and kept for its life: a `PUT /v1/pools` that adds or drops a pool +changes the claims made after it, never a live one (a record from before this +rule resolves against the live pool set). Secrets are registered separately and referenced by name — the value comes from the environment, never the config file. @@ -199,7 +205,10 @@ woken sandbox binds at arm time. request inside an intercepted tunnel. Exact ports only; `0` and repeats are rejected at load. - `secret`: injects the named registered secret's header. A guest-supplied value - for the same header is overwritten. On HTTPS the injection needs `intercept`. + for the same header is overwritten. Inside an HTTPS `CONNECT` tunnel the + injection needs `intercept`; an absolute-form `https://` request on the + forward door needs none — the proxy dials the origin over verified TLS + itself, checks the method, and injects. - `intercept`: terminate a matched HTTPS CONNECT so the request is filtered by method and the secret injected (see below). Only a pool rule may set it. - No policy on a claim ⇒ no egress at all (the proxy is not started). diff --git a/docs/langchain.md b/docs/langchain.md index dcb29525..fa59dd3a 100644 --- a/docs/langchain.md +++ b/docs/langchain.md @@ -20,12 +20,22 @@ schemas, sync-native with `asyncio.to_thread` async bridges): | tool | what it does | |---|---| -| `sandbox_exec` | run a shell command, cut off after 5 minutes with the reply saying so; the budget is one wall clock over the claim, the dial and the command, so a first call that waits on a slow or redirecting cluster still answers inside it; stdout/stderr/exit code; disk state persists across calls | +| `sandbox_exec` | run a shell command, cut off after 5 minutes with the reply saying so (the cut also kills the command in the guest, which can hold the reply up to 5 s more); the budget is one wall clock over the claim, the dial and the command, so a first call that waits on a slow or redirecting cluster still answers inside it; stdout/stderr/exit code; disk state persists across calls | | `sandbox_write_file` | write a text file (atomic on the guest; the parent directory must exist) | | `sandbox_read_file` | read a text file | | `sandbox_list_dir` | list a directory as JSON | -**Branching**: `CocoonToolkit(..., from_checkpoint="ck_...")` claims the +The claim's lease is `ttl_seconds`, one hour by default: nothing renews a +lease, and an agent run outlives the node's 5-minute default. Once the lease +ends the sandbox is gone and every later tool call reports it. + +A failed call — a missing path, a guest error, an expired or unreachable +sandbox — comes back to the model as the tool's error text, not as an +exception out of the agent run. Every tool's first call claims the sandbox +inside the same 5-minute budget. + +**Branching**: `CocoonToolkit(..., from_checkpoint="ck_...")` — the checkpoint +pins the template, lane and size, so `template` and `net` are ignored — claims the sandbox from a [checkpoint](sdk-python.md#checkpoints--branching-and-time-travel)'s captured moment instead of a clean template — agents start from prepared state (dependencies installed, repo cloned) in milliseconds, and every diff --git a/docs/mcp.md b/docs/mcp.md index fbaa39b2..a63593ad 100644 --- a/docs/mcp.md +++ b/docs/mcp.md @@ -23,7 +23,8 @@ sandbox-mcp .` ## Tools -Every tool call is capped at 5 minutes. +Every tool call is capped at 5 minutes; an `exec` that hits the cap is killed +in the guest, which can hold its reply up to 5 s past the cap. | tool | what it does | |---|---| @@ -31,7 +32,7 @@ Every tool call is capped at 5 minutes. | `exec` | run a shell command to completion; returns stdout, stderr and the exit code, each stream keeping its first 1 MiB with `truncated` set past that; a cut-off or dropped run returns the output collected so far next to an `error` field; a hibernated sandbox wakes transparently | | `spawn` | start a command detached and return its pid; output goes to a 256 KiB ring buffer that `logs` replays | | `ps` | list tracked processes (exec, spawn, pty) with state, exit code and start time | -| `logs` | replay up to 256 KiB of a tracked process's newest whole stdout/stderr chunks (+ exit code once ended) | +| `logs` | replay up to 256 KiB of a tracked process's newest whole stdout/stderr chunks (+ exit code once ended); an exited process is forgotten after 5 minutes | | `kill` | signal a tracked process (0 = SIGKILL); an exited process is a no-op success | | `write_file` / `read_file` / `list_dir` | atomic whole-file write (parent must exist); whole-file text read of a regular file up to 1 MiB (a larger file, a directory or a device is an error, invalid UTF-8 replaced, missing path is an error); one-level listing of `{name, kind, size}` entries | | `fork` | clone into N children (1 to the node's `max_fork_count`, default 16) carrying exact memory + disk state, all-or-nothing; the parent keeps running and each child lives one hour | @@ -44,7 +45,10 @@ Every tool call is capped at 5 minutes. | `node_info` | warm pools, live claims, drain state, capacity and mesh peers | Sandbox handles (and their tokens) are held by the server process for the -session. Checkpoints outlive sessions: `branch_checkpoint` accepts any known +session, and released with it: when the MCP client disconnects or the server +is signalled, it destroys every sandbox that session claimed, whatever +`ttl_seconds` asked for. `checkpoint` first if the state must survive. +Checkpoints outlive sessions: `branch_checkpoint` accepts any known id without a listing round-trip. If the connected node does not hold it, the claim follows a live owner probe and redirect, or heals the checkpoint locally when peer healing is enabled. `delete_checkpoint` is different: it acts on the diff --git a/docs/openai-adapter.md b/docs/openai-adapter.md index 8522b00b..8c4e27f3 100644 --- a/docs/openai-adapter.md +++ b/docs/openai-adapter.md @@ -32,7 +32,7 @@ pair over the sync [Python SDK](sdk-python.md), bridged with | SDK surface | cocoon | |---|---| | `create` | `Client.new` — one claimed sandbox per session | -| session `exec` | `Sandbox.run` (stdout/stderr/exit); a per-call `timeout` is `run`'s wall clock and surfaces as `TimeoutError` | +| session `exec` | `Sandbox.run` (stdout/stderr/exit); a per-call `timeout` is `run`'s wall clock: it kills the command in the guest and surfaces as `TimeoutError` | | `read` / `write` | `Sandbox.read_file` / `write_file`; a missing path on `read` → `FileNotFoundError`; a per-call `user` raises `NotImplementedError` (fs verbs carry no user on the wire) | | `persist_workspace` / `hydrate_workspace` | `Sandbox.pull` / `push` (tar) | | exposed port | `Sandbox.proxy_port` | @@ -41,7 +41,8 @@ pair over the sync [Python SDK](sdk-python.md), bridged with `CocoonSandboxClientOptions` carries the node address, api token, template ref, network lane (empty for the node's default, or `none`/`egress`), and -TTL. The session state is +`ttl_seconds`, which defaults to one hour: nothing renews a lease, and a +session outlives the node's 5-minute default. The session state is JSON-serializable, so a run can be resumed against the same sandbox after a process restart. Requires Python 3.10+ (the Agents SDK floor); the underlying `cocoonsandbox` stays 3.9+. diff --git a/docs/sandboxd-api.md b/docs/sandboxd-api.md index eb6f6dce..28cc2b9f 100644 --- a/docs/sandboxd-api.md +++ b/docs/sandboxd-api.md @@ -40,9 +40,10 @@ Auth: `Authorization: Bearer ` (when configured). - `claim_ref` is an optional opaque caller reference echoed by the scoped sandbox index; the aggregated apiserver uses `/` - `no_redirect` is set by the SDK when retrying at a redirect target -- `require_promoted` is an internal redirect field. When a redirect response - sets it, copy it into the `no_redirect` retry so the target cannot cold-boot - a promoted template name if its gossip view is stale +- `require_promoted` is an internal redirect field of volume claims, the only + path that sets or reads it. When a redirect response sets it, copy it into + the `no_redirect` retry so the target cannot cold-boot a promoted template + name if its gossip view is stale - `volumes` is an ordered list of at most eight unique catalog names. `mount` defaults to `/volumes/`; a custom value must be absolute and clean, outside the guest OS tree, unique, and non-nesting within the request. @@ -232,7 +233,7 @@ otherwise only a side effect of the next agent access, so this is the explicit form for warming a sandbox ahead of use. Idempotent on one already running. A hibernated sandbox retains its existing deadline. An archived sandbox instead uses `archive_delete_after_seconds` as its retention deadline -while stored, and waking it starts a fresh server-default 5m lease. 204 on +while stored, and waking it starts a fresh lease of the length the claim was granted (the server default when it asked for none). 204 on success, 404 unknown id or wrong token. ## POST /v1/sandboxes/{id}/fork @@ -318,7 +319,12 @@ protocol: a node answering a `no_redirect` delete speaks only for itself. ## PUT /v1/pools Auth: root only (tenant tokens get 403). Replaces the node's desired warm -targets online — no restart, live claims untouched: +targets online — no restart, and no live claim's VM is touched. On a node with egress policies, +whether a key is pooled decides which [policy layers](egress.md) a claim of it +gets; that is settled when the claim is made, so adding or dropping a pool here +changes only claims made afterwards — a key that gains a pool with no `egress` +block leaves new tenant claims without egress, a key that loses such a pool +gives new tenant claims the tenant's policy alone: ```json {"pools": [{"template": "base:24.04", "net": "none", "size": "small", @@ -351,11 +357,17 @@ fresh info payload. ## POST /v1/sandboxes/{id}/preview -Auth: like fork — node `api_token` in the header, the sandbox's own token -in the body. Mints a signed URL serving a +Auth: node `api_token` in the header, the sandbox's own token in the body. +Unlike fork, promote and checkpoint — where the root token with an empty body +token acts by id as the operator — preview has no operator path: the body +token is required. Mints a signed URL serving a guest HTTP port from a browser: body `{"token": "...", "port": 8080, "ttl_seconds": 0}` → `{"url": "http:///p//"}`. -The URL's life is clamped to the claim's remaining lease. 501 when the node +The URL's life is clamped to the claim's remaining lease; an archived claim +kept forever (`archive_delete_after_seconds: 0`) has no lease, so there the +requested `ttl_seconds` stands unclamped and `ttl_seconds: 0` mints a one-hour +URL; releasing the sandbox ends it early either way. +501 when the node has no `preview_listen`. The signed token embeds the sandbox id, port, and owner `advertise_addr`, so any node's preview listener can serve it (forwarding to the owner's main listener) and a released sandbox's URL simply stops diff --git a/docs/sdk-python.md b/docs/sdk-python.md index cb2641cd..ec92e265 100644 --- a/docs/sdk-python.md +++ b/docs/sdk-python.md @@ -11,9 +11,9 @@ with client.new("ghcr.io/cocoonstack/sandbox/rt:24.04") as sb: `pip install cocoonstack-sandbox` — stdlib-only, no dependencies, and synchronous by design (agent frameworks that need async wrap calls in `asyncio.to_thread`, exactly like the [OpenAI adapter](openai-adapter.md) -does). It matches the [Go SDK](sdk.md) guest and data-plane surface, with two +does). It matches the [Go SDK](sdk.md) guest and data-plane surface, with three exceptions: operator pool retuning (`SetPools`/`SetPoolsCluster`) remains -Go-only, and file and tar payloads are whole `bytes` values rather than +Go-only, `Template.new` takes no `claim_ref`, and file and tar payloads are whole `bytes` values rather than streams (`read_file`/`pull` return them, `write_file`/`push` take them; the wire is chunked, the caller's copy is not). Wire fidelity is pinned by the shared protocol fixture corpus that the Rust guest, Go, and @@ -75,8 +75,10 @@ client = Client("10.0.0.5:7777", api_token="...", timeout=120.0) tenant token (resource-creating verbs only; operator surfaces answer it 403). On a cluster every node shares the root token and the same tenants set. `timeout` bounds every control-plane request and the data-plane dial -and upgrade; a guest stream then lives until the guest ends it, as in the -Go SDK. A caller with a wall clock of its own passes `deadline` to a claim +and upgrade; a guest stream then lives until the guest ends it. Unlike the +Go SDK, where a canceled context closes any call, only `run`/`exec` take a +`timeout` here: every other data-plane call blocks until the guest answers +or the connection drops. A caller with a wall clock of its own passes `deadline` to a claim (below) so the redirect walk cannot outlive it. **Clusters need nothing extra**: dial any node. On a warm miss the entry @@ -189,7 +191,8 @@ gets one connection per call as before. If that deployment also enables `archive_after_seconds`, archiving replaces the original claim deadline with the archive-retention deadline (or no deadline when archives are kept forever). Waking an archive starts a fresh -server-default 5m lease. The existing handle's `sb.deadline` remains the value +lease of the length the claim was granted (the server default when it asked +for none). The existing handle's `sb.deadline` remains the value returned when that handle was created; call `client.sandboxes()` to read the current server deadline after an archive/wake transition. @@ -333,7 +336,19 @@ its end the connection is cut, which makes silkd kill the command, and `exec` returns stdout and raises `ExitError` on a non-zero exit — carrying `code`, `stderr`, and the `stdout` produced before it failed. `run` streams raw bytes through the callbacks (chunk boundaries may split multi-byte sequences) and returns the exit code. `user` de-escalates -inside the guest; `session=` routes the command into a persistent session. +inside the guest; `session=` routes the command into a persistent session, +where only `argv` applies — the session owns the cwd, environment and user, +and stderr arrives merged into stdout. A command's environment is `PATH`, +`TERM` and the lane's proxy variables plus `env`, not the image's `ENV`. The +guest kills a foreground command whose connection dropped only once it next +writes output, so a `run(timeout=)` cut does not leave that to the drop: once +the guest has reported the command's pid, the cut sends that pid a `kill` over +a connection of its own (best effort, bounded to 5 s) before `SandboxTimeout` +is raised. A client that only drops the connection leaves a silent command +running in the guest; `ps` then `kill` stops it. `spawn` outlives the +connection by design. A command run through a session is the session's: it +reports no pid, so no cut kills it and it is not in `ps`; it runs to +completion, and the session's `close()` is what ends the session. ## Background processes @@ -348,6 +363,8 @@ sb.kill(pid) # default SIGKILL `spawn` starts the command detached with a bounded output ring; `logs` replays it, `attach` follows live output until exit (replay and live stream hand off atomically). Killing an already-exited process is a no-op success. +An exited process leaves the table after 5 minutes; from then on its pid +raises `SilkdError` `not_found`. ## Sessions @@ -408,6 +425,10 @@ results = sb.replace(["/work/main.py"], r"foo", "bar") ``` Patterns are regular expressions evaluated in the guest — no shell quoting. +`replace` is atomic per file, not per list: it raises at the first missing, +unreadable or non-UTF-8 path and the files before it stay rewritten, so pass it +paths a `find` just returned. A `find` fails as a whole when one match's frame +would pass the 8 MiB cap. ## Watching @@ -459,14 +480,15 @@ pty.close() ```python sb = client.new("rt:24.04", claim_ref="ns/workload") -client.sandboxes() # id, key, deadline, claim_ref — never tokens +client.sandboxes() # one summary dict per live claim — never tokens client.drain() # cordon: refuse new claims, run leases out client.uncordon() client.attach(owner_addr, id, token) # bind a known handle, no lookup round-trip ``` `sandboxes()` is scoped to the calling token, so a tenant sees only its own -claims. `drain()` leaves live claims alone — poll `info()` until `claimed` is +claims; the fields are those of [`GET /v1/sandboxes`](sandboxd-api.md#get-v1sandboxes). +`drain()` leaves live claims alone — poll `info()` until `claimed` is zero. ## Errors diff --git a/docs/sdk.md b/docs/sdk.md index 17fb3cf7..1b5e5242 100644 --- a/docs/sdk.md +++ b/docs/sdk.md @@ -287,7 +287,8 @@ egress request) is not swept; the idle clock restarts when that connection ends. If that deployment also enables `archive_after_seconds`, archiving replaces the original claim deadline with the archive-retention deadline (or no deadline when archives are kept forever). Waking an archive starts a fresh -server-default 5m lease. The existing handle's `Sandbox.Deadline` is the value +lease of the length the claim was granted (the server default when it asked +for none). The existing handle's `Sandbox.Deadline` is the value returned when that handle was created; call `Client.Sandboxes` to read the current server deadline after an archive/wake transition. @@ -446,7 +447,20 @@ code, err := sb.Run(ctx, sandbox.Cmd{ `Cmd` fields: `Argv` (required), `Cwd`, `Env`, `User` (de-escalation inside the guest), `Session` (run inside a persistent session, below), `Stdin` (nil closes the child's stdin immediately; do not share one blocking reader -across Runs), `Stdout`/`Stderr` (nil discards). +across Runs), `Stdout`/`Stderr` (nil discards). With `Session` set only +`Argv` applies: the session owns the cwd, environment and user, its commands +read `/dev/null`, and stderr arrives merged into stdout. A command's +environment is `PATH`, `TERM` and the lane's proxy variables plus `Env` — not +the image's `ENV`; `HOME` and `USER` are set only with `User`. The guest kills +a foreground command whose connection dropped only once it next writes output, +so `Run` does not leave that to the drop: when its ctx ends after the guest +reported the command's pid, `Run` sends that pid a `kill` over a connection of +its own (best effort, bounded to 5 s) before returning the ctx error. A client +that only drops the connection leaves a silent command (`sleep`, a quiet build) +running in the guest; find its pid with `Ps` and `Kill` it. Use `Spawn` for +work that must outlive the connection. A command run inside a `Session` is the +session's: it reports no pid, so no cancel kills it and it does not appear in +`Ps`; it runs to completion, and `Session.Close` is what ends the session. Non-zero exits surface as `*sandbox.ExitError{Code, Stderr}` from `Exec` (alongside partial stdout); `Run` returns the code directly. @@ -467,7 +481,8 @@ the process has ended (`exited=false` while running); `Attach` follows live output until exit — the replay and the live stream hand off atomically, so no chunk is lost or doubled between them. Killing an already-exited process is a no-op success (its OS pid may be recycled; silkd never signals a -reaped child). +reaped child). An exited process leaves the table after 5 minutes; from then +on its pid answers `not_found` to every verb. ## Sessions @@ -485,7 +500,9 @@ err = sess.Close(ctx) ids, err := sb.Sessions(ctx) // live session ids ``` -Idle sessions are reaped guest-side after 30 minutes. +Idle sessions are reaped guest-side after 30 minutes. `Close` on a session +that is already gone — and `Lsp.Stop` on a server that already exited — answers +silkd's `not_found`; ignore it in a deferred cleanup. ## Files @@ -532,6 +549,10 @@ results, err := sb.Replace(ctx, []string{"/work/main.go"}, `foo`, "bar") ``` Patterns are regular expressions evaluated in the guest — no shell quoting. +`Replace` is atomic per file, not per list: it fails at the first missing, +unreadable or non-UTF-8 path and the files before it stay rewritten, so pass it +paths a `Find` just returned. A `Find` fails as a whole when one match's frame +would pass the 8 MiB cap. ## Watching @@ -591,7 +612,7 @@ apiserver claims under: ```go sb, _ := client.New(ctx, "rt:24.04", sandbox.WithClaimRef("ns/workload")) -list, _ := client.Sandboxes(ctx) // id, key, deadline, claim_ref — never tokens +list, _ := client.Sandboxes(ctx) // one SandboxSummary per live claim — never tokens info, _ := client.Drain(ctx) // cordon: refuse new claims, run leases out info, _ = client.Uncordon(ctx) info, _ = client.SetPools(ctx, pools) // retune warm targets without a restart @@ -600,7 +621,8 @@ sb = client.Attach(ownerAddr, id, token) // bind a known handle, no lookup roun ``` `Sandboxes` is scoped to the calling token, so a tenant sees only its own -claims. `Drain` leaves live claims alone — poll `Info` until `Claimed` is zero. +claims; the fields are those of [`GET /v1/sandboxes`](sandboxd-api.md#get-v1sandboxes). +`Drain` leaves live claims alone — poll `Info` until `Claimed` is zero. ## Error handling diff --git a/docs/security.md b/docs/security.md index b9ecd315..ff5fda0b 100644 --- a/docs/security.md +++ b/docs/security.md @@ -77,8 +77,9 @@ the root `api_token` (operator surfaces, full access), tenant tokens (resource-creating verbs, everything stamped and quota'd per tenant), and per-sandbox tokens (that sandbox only — holding a handle amplifies to nothing node-level). Two capability tokens ride on top: preview URLs are -HMAC-signed, expire with the claim's lease, and die with the sandbox (no -revocation list to leak); a checkpoint id is the unguessable capability to +HMAC-signed, expire with the claim's lease (an archived claim kept forever +has none, so its URL runs for the TTL it was minted with), and die with the +sandbox (no revocation list to leak); a checkpoint id is the unguessable capability to branch it. On a cluster, deleting a checkpoint does not revoke that capability fleet-wide the instant it runs: the delete is best-effort — broadcast to every peer the node currently sees — so a peer that is offline diff --git a/docs/silkd.md b/docs/silkd.md index a8a7d1f5..145fbdec 100644 --- a/docs/silkd.md +++ b/docs/silkd.md @@ -25,9 +25,14 @@ on a daemon that keeps the connection. Binary payloads ride base64 in `data` fields. Frames are capped at 8 MiB; requests carry `"v": 1` and unknown fields are ignored (the forward-compatibility story). -Sessions and processes are server-side state addressed by id — a dropped -connection loses nothing (`attach` resumes). The connection-bound verbs are -`fs_watch`, `pty_open`, `lsp_request`, and `port_forward`. +Sessions and detached processes are server-side state addressed by id — a +dropped connection loses nothing (`attach` resumes). A foreground `exec` is +bound to its connection, but silkd learns of a drop only when a write to the +client fails: a child that is producing output is killed and its pid freed, a +silent one runs to completion — which is why the SDKs' `Run`/`run` answer a +canceled ctx or a timeout with a `kill` of the pid `started` reported. The other +connection-bound verbs are `fs_watch`, `pty_open`, `lsp_request`, and +`port_forward`. The authoritative wire contract is the shared fixture corpus in `protocol/wire/fixtures/v1`, round-tripped by the Rust, Go, and Python test @@ -37,12 +42,12 @@ suites — a frame only one side can parse fails CI. | group | ops | response flow | |---|---|---| -| exec | `exec {argv, cwd?, env?, user?, detach?, session?}` | `started{pid}` → `stdout/stderr{data}`… → `exit{code}`; the client may stream `stdin{data}` / `stdin_close`. `detach` returns after `started`; the process keeps a bounded output ring for later `logs`/`attach`; not combinable with `session` | +| exec | `exec {argv, cwd?, env?, user?, detach?, session?}` | `started{pid}` → `stdout/stderr{data}`… → `exit{code}`; the client may stream `stdin{data}` / `stdin_close`. `detach` returns after `started`; the process keeps a bounded output ring for later `logs`/`attach` and leaves the table 5 minutes after it exits, after which its pid answers `not_found`; not combinable with `session`. With `session` only `argv` is used — `cwd`, `env`, `user` and `stdin` frames are ignored (the session owns them, and its commands read `/dev/null`) — no `started` is sent, and stderr arrives merged into `stdout`. A child's environment is cleared to `PATH`, `TERM` and the lane's proxy variables plus `env`; `HOME` and `USER` are set only with `user` | | procs | `ps` / `kill {pid, signal?}` / `attach {pid}` / `logs {pid}` | `ps` answers one `procs{procs}` frame, `kill` a bare `done`; `logs` and `attach` stream `stdout/stderr{data}`… then `exit{code}` (`logs` closes with `done`, and so does `attach` when the process is gone before its exit). Handles are guest pids; any connection can list, signal, replay, or re-attach live | | sessions | `session_create {id?, cwd?, env?}` / `session_list` / `session_rm {id}` | `session_created{id}` / `sessions{sessions}` / `done`. A session is a real persistent bash; `exec` with `session` runs inside it. Idle sessions are reaped after 30 minutes | | fs | `fs_write {path, mode?}` (+`data`/`data_end` frames) / `fs_read` / `fs_list` / `fs_stat` / `fs_mkdir {parents?}` / `fs_rm {recursive?}` / `fs_rename {from, to}` | `fs_read` streams `data{data}`… → `done` for a regular file and answers `bad_request` for a directory, device or FIFO, `fs_list` streams 4096-entry `entries{entries}` batches → `done`, `fs_stat` answers one `stat{info}`, and the mutating verbs terminate with `done`. Streaming runs both directions; write commits atomically via temp+rename and inherits an overwritten file's mode | | tree | `fs_push {dest}` (+tar as `data` frames) / `fs_pull {path}` | whole trees as tar streams through the guest tar: `fs_pull` streams `data{data}`… → `done`, `fs_push` terminates with `done` | -| search | `fs_find {path, pattern, glob?}` → `match{file, line, content}`… → `done` / `fs_replace {files, pattern, replacement}` → `replaced{file, replacements}`… → `done` | regex as data, no shell quoting; `glob` is anchored `*`/`?` wildcards over file names; find skips binary and >8 MiB files, and replace skips the same 8 MiB bound with a zero count | +| search | `fs_find {path, pattern, glob?}` → `match{file, line, content}`… → `done` / `fs_replace {files, pattern, replacement}` → `replaced{file, replacements}`… → `done` | regex as data, no shell quoting; `glob` is anchored `*`/`?` wildcards over file names; find skips binary and >8 MiB files, and replace skips the same 8 MiB bound with a zero count. One match whose frame would pass the 8 MiB cap ends the find with an `error`. Replace is atomic per file, not per list: it stops with an `error` at the first missing, unreadable or non-UTF-8 file, and the files before it stay rewritten | | watch | `fs_watch {path, recursive?}` | `ready` once armed (events after it are guaranteed captured), then `event{kind, path}` until the client disconnects; watcher or delivery-queue overflow errors arrive as a terminal `error` instead of silently losing events | | pty | `pty_open {cols, rows, cwd?, env?, user?}` / `pty_resize {pid, cols, rows}` | `pty_open` answers `started{pid}`, then a shell under a pseudo-terminal: output as `stdout` frames, input as `stdin` frames, `exit{code}` terminal; `pty_resize` answers `done`. PTYs register in the proc table like any exec | | git | `git_clone {url, path, branch?, depth?, auth?}` / `git_status {path}` / `git_add {path, files}` / `git_commit {path, message, author}` / `git_push {path, auth?}` / `git_pull {path, auth?}` / `git_branch {path, action, name?}` | `git_status_result` (porcelain v2), `git_commit_result{hash}`, `git_branches{current, branches}` for `git_branch {action: "list"}`; every other verb terminates with `done`. A status past about 1 MiB of entries carries `truncated: true` with the head of the list, so one frame never exceeds the cap. `auth` is injected as an in-memory header, never written to guest disk | @@ -85,10 +90,14 @@ own routed network never gets them. ## Limits - 8 MiB frame cap (a malformed peer cannot OOM the daemon) -- foreground exec output is backpressured: a slow client paces the child - rather than losing output; detached observers ride a best-effort broadcast +- foreground exec output is backpressured while the child runs: a slow client + paces it rather than losing output. Once the child exits, the client has 2 + seconds to drain what is still queued (up to 8 MiB) before `exit` is sent + and the rest is dropped. Detached observers ride a best-effort broadcast - no fsync durability on writes (guest page cache semantics) - `fs_push` is network-failure atomic: the stream extracts into a staging dir inside `dest` and merges (local renames) only after the tar terminates - cleanly, so a truncated stream or dropped connection leaves `dest` - untouched; the residual crash window is the merge itself, microseconds + cleanly, so a truncated stream or dropped connection leaves `dest`'s + contents untouched; `dest` itself and any missing parent are created before + the stream starts and stay behind, empty, on failure. The residual crash + window is the merge itself, microseconds diff --git a/e2e/cmd/meshsmoke/main.go b/e2e/cmd/meshsmoke/main.go index 2807ee62..be00c5af 100644 --- a/e2e/cmd/meshsmoke/main.go +++ b/e2e/cmd/meshsmoke/main.go @@ -7,6 +7,7 @@ package main import ( "context" + "errors" "flag" "fmt" "os" @@ -32,6 +33,9 @@ func main() { } func run(addr, peer, token, template string) error { + if peer == "" { + return errors.New("peer is required") + } ctx, cancel := context.WithTimeout(context.Background(), 180*time.Second) defer cancel() @@ -40,7 +44,7 @@ func run(addr, peer, token, template string) error { return fmt.Errorf("via %s: %w", addr, err) } defer func() { _ = sb.Close() }() - if peer != "" && sb.Owner() != peer { + if sb.Owner() != peer { return fmt.Errorf("claim owner %s, want redirect to %s", sb.Owner(), peer) } fmt.Printf(" redirect: claim entered at %s, owned by %s\n", addr, sb.Owner()) @@ -76,7 +80,7 @@ func run(addr, peer, token, template string) error { if got := strings.TrimSpace(out); got != "mesh-marker" { return fmt.Errorf("marker %q, want mesh-marker (clone not from the promoted golden)", got) } - if peer != "" && clone.Owner() != peer { + if clone.Owner() != peer { return fmt.Errorf("clone owner %s, want %s (template owner)", clone.Owner(), peer) } fmt.Printf(" name-based claim: clone on %s carries the marker\n", clone.Owner()) diff --git a/mcp/server_test.go b/mcp/server_test.go index 0a17a606..6a399231 100644 --- a/mcp/server_test.go +++ b/mcp/server_test.go @@ -111,6 +111,20 @@ func TestReadFileStopsAtTheCap(t *testing.T) { } } +func TestLogsKeepsAtMostTheCap(t *testing.T) { + chunk := `{"type":"stdout","data":"` + base64.StdEncoding.EncodeToString(bytes.Repeat([]byte("x"), 256<<10)) + `"}` + "\n" + srv := newTestServer(t, agentRoute(t, func(string) (string, bool) { + return strings.Repeat(chunk, 8) + `{"type":"done"}` + "\n", false + })) + replies := serveLines(t, srv, + `{"jsonrpc":"2.0","id":1,"method":"tools/call","params":{"name":"create_sandbox","arguments":{}}}`, + `{"jsonrpc":"2.0","id":2,"method":"tools/call","params":{"name":"logs","arguments":{"sandbox_id":"sb_1","pid":7}}}`, + ) + if n := strings.Count(toolText(t, replies[2]), "x"); n != execOutputCap { + t.Errorf("logs kept %d bytes of a 2 MiB replay, want the %d cap", n, execOutputCap) + } +} + func TestTrackBoxAfterCloseReleasesTheClaim(t *testing.T) { released := make(chan string, 1) node := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { diff --git a/mcp/tools.go b/mcp/tools.go index fcc0eabc..c2abb9c0 100644 --- a/mcp/tools.go +++ b/mcp/tools.go @@ -252,7 +252,7 @@ func toolLogs(ctx context.Context, s *server, raw json.RawMessage) (string, erro if err != nil { return "", err } - var stdout, stderr strings.Builder + var stdout, stderr cappedOutput code, exited, err := sb.Logs(ctx, args.PID, &stdout, &stderr) if err != nil { return "", err diff --git a/os-image/android/15/Dockerfile b/os-image/android/15/Dockerfile index 05d07785..b958ed3b 100644 --- a/os-image/android/15/Dockerfile +++ b/os-image/android/15/Dockerfile @@ -9,7 +9,7 @@ ARG ANDROID_BASE_IMAGE=ghcr.io/cocoonstack/cocoon/android:15.0 # silkd carrier image (built from silkd/Dockerfile); CI passes the # version-pinned tag. -ARG SILKD_IMAGE=ghcr.io/cocoonstack/sandbox/silkd:0.1.0 +ARG SILKD_IMAGE=ghcr.io/cocoonstack/sandbox/silkd:0.1.12 FROM ${SILKD_IMAGE} AS silkd diff --git a/packaging/sandboxd.service b/packaging/sandboxd.service index 20c56166..d4f8b5c5 100644 --- a/packaging/sandboxd.service +++ b/packaging/sandboxd.service @@ -6,6 +6,7 @@ After=network-online.target [Service] ExecStart=/usr/local/bin/sandboxd -config /etc/sandboxd/config.json Restart=on-failure +KillMode=process Environment=SANDBOXD_LOG_LEVEL=info [Install] diff --git a/sandboxd/config/config.go b/sandboxd/config/config.go index 6d3b65a1..27d587b2 100644 --- a/sandboxd/config/config.go +++ b/sandboxd/config/config.go @@ -423,7 +423,7 @@ func (c *Config) validateClientAdvertise() error { (u.Path != "" && u.Path != "/") || strings.ContainsAny(c.ClientAdvertise, "?#") { return fmt.Errorf("client_advertise must be an http or https origin") } - if ip, _ := netip.ParseAddr(u.Hostname()); ip.IsUnspecified() { + if !routableHost(u.Hostname()) { return fmt.Errorf("client_advertise must name a routable host") } if port := u.Port(); port != "" { @@ -450,7 +450,7 @@ func (c *Config) validateMesh() error { if err != nil { return fmt.Errorf("advertise_addr: %w", err) } - if ip, _ := netip.ParseAddr(host); host == "" || ip.IsUnspecified() { + if !routableHost(host) { return fmt.Errorf("advertise_addr %q is gossiped to peers and must name a routable host", c.AdvertiseAddr) } return nil @@ -618,10 +618,10 @@ func namesHost(addr string) bool { addr = "http://" + addr } u, err := url.Parse(addr) - return err == nil && u.Hostname() != "" && !isUnspecifiedHost(u.Hostname()) + return err == nil && routableHost(u.Hostname()) } -func isUnspecifiedHost(host string) bool { +func routableHost(host string) bool { ip, _ := netip.ParseAddr(host) - return ip.IsUnspecified() + return host != "" && !ip.IsUnspecified() } diff --git a/sandboxd/engine/portconn.go b/sandboxd/engine/portconn.go index 02fdaefc..660e82a8 100644 --- a/sandboxd/engine/portconn.go +++ b/sandboxd/engine/portconn.go @@ -13,7 +13,7 @@ import ( ) // portReadBuf fits silkd's data frames in one buffered read. -const portReadBuf = 64 << 10 +const portReadBuf = wire.BulkChunk*4/3 + 64 var portDataHead = []byte(`{"type":"data","data":"`) diff --git a/sandboxd/engine/portconn_test.go b/sandboxd/engine/portconn_test.go index 759488de..0d8cda57 100644 --- a/sandboxd/engine/portconn_test.go +++ b/sandboxd/engine/portconn_test.go @@ -10,6 +10,8 @@ import ( "os" "path/filepath" "testing" + + "github.com/cocoonstack/sandbox/protocol/wire" ) func TestGuestPortConnFramesBothWays(t *testing.T) { @@ -100,3 +102,13 @@ func TestGuestPortConnRoundTripsFixtures(t *testing.T) { t.Errorf("write frame %q, want fixture %q", line[:n], want) } } + +func TestPortReadBufHoldsOneBulkFrame(t *testing.T) { + frame, err := wire.EncodeResponse(wire.DataResp{Data: make([]byte, wire.BulkChunk)}) + if err != nil { + t.Fatal(err) + } + if n := len(frame) + 1; n > portReadBuf { + t.Errorf("a full bulk data frame is %d bytes with its delimiter, portReadBuf %d cannot take it in one read", n, portReadBuf) + } +} diff --git a/sandboxd/go.mod b/sandboxd/go.mod index b7cd0038..6dfd0785 100644 --- a/sandboxd/go.mod +++ b/sandboxd/go.mod @@ -12,6 +12,7 @@ require ( github.com/google/nftables v0.3.0 github.com/hashicorp/memberlist v0.6.0 github.com/projecteru2/core v0.1.5 + github.com/rs/zerolog v1.35.1 golang.org/x/net v0.58.0 golang.org/x/sync v0.22.0 golang.org/x/sys v0.48.0 @@ -57,7 +58,6 @@ require ( github.com/miekg/dns v1.1.72 // indirect github.com/pkg/errors v0.9.1 // indirect github.com/rogpeppe/go-internal v1.16.0 // indirect - github.com/rs/zerolog v1.35.1 // indirect github.com/sean-/seed v0.0.0-20170313163322-e2103e2c3529 // indirect github.com/vishvananda/netns v0.0.5 // indirect golang.org/x/mod v0.40.0 // indirect diff --git a/sandboxd/pool/archive.go b/sandboxd/pool/archive.go index 4eb4d89e..85673208 100644 --- a/sandboxd/pool/archive.go +++ b/sandboxd/pool/archive.go @@ -218,7 +218,7 @@ func (m *Manager) commitWake(ctx context.Context, sb *types.Sandbox, vmName, soc } ck, deadline := sb.ArchiveCk, sb.Deadline sb.VMName, sb.VsockSocket, sb.ArchiveCk = vmName, sock, "" - sb.Deadline = time.Now().Add(clampTTL(0)) // a woken sandbox is a fresh lease + sb.Deadline = time.Now().Add(clampTTL(types.Seconds(sb.LeaseSeconds))) // a woken sandbox is a fresh lease js := m.store.set(sb) m.mu.Unlock() if err := m.store.commit(js); err != nil { diff --git a/sandboxd/pool/archive_test.go b/sandboxd/pool/archive_test.go index d610382d..cc8768a2 100644 --- a/sandboxd/pool/archive_test.go +++ b/sandboxd/pool/archive_test.go @@ -6,6 +6,7 @@ import ( "os" "path/filepath" "slices" + "strings" "testing" "testing/synctest" "time" @@ -333,10 +334,13 @@ func TestArchiveRetentionPurge(t *testing.T) { m.reapOnce(t.Context()) waitFor(t, func() bool { return archivedCount(m) == 0 }) - waitFor(t, func() bool { return m.Counters().ArchiveDeletes > 0 }) + waitFor(t, func() bool { return m.Counters().Reaps > 0 }) if ckExists(t, m, ck) { t.Error("purge did not delete the archived checkpoint") } + if got := usageEventsOf(t, m, sb.ID); !strings.HasSuffix(got, "archive,archive_delete,reap") { + t.Errorf("events %s, want the purge to close the claim with a reap", got) + } m.mu.Lock() live := m.tenantLive["acme"] @@ -347,6 +351,25 @@ func TestArchiveRetentionPurge(t *testing.T) { }) } +func TestArchiveWakeRegrantsTheClaimsLease(t *testing.T) { + eng := newFakeEngine() + m := newTestManager(t, eng, archivePool(3600)) + sb, err := m.ClaimProvision(t.Context(), testKey, 2*time.Hour, "acme", "", nil) + if err != nil { + t.Fatalf("claim: %v", err) + } + mustArchive(t, m, sb) + if _, _, err := m.WakeAgentSocket(t.Context(), sb.ID, sb.Token); err != nil { + t.Fatalf("wake: %v", err) + } + m.mu.Lock() + lease := time.Until(sb.Deadline) + m.mu.Unlock() + if lease < time.Hour || lease > 2*time.Hour { + t.Errorf("lease after the wake = %v, want the 2h the claim asked for", lease) + } +} + func TestReleaseArchivedDeletesCheckpoint(t *testing.T) { eng := newFakeEngine() m := newTestManager(t, eng, archivePool(3600)) diff --git a/sandboxd/pool/checkpoint_test.go b/sandboxd/pool/checkpoint_test.go index 810fd5f7..15042051 100644 --- a/sandboxd/pool/checkpoint_test.go +++ b/sandboxd/pool/checkpoint_test.go @@ -8,6 +8,7 @@ import ( "path/filepath" "strings" "testing" + "testing/synctest" "time" "github.com/cocoonstack/sandbox/sandboxd/config" @@ -151,6 +152,37 @@ func TestRunRefillsWhileTheRetentionSweepBlocks(t *testing.T) { } } +func TestRunRefillsWhileTheGenerationSweepBlocks(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + eng := newFakeEngine() + m := newTestManager(t, eng, config.PoolSpec{PoolKey: testKey, Warm: 1}) + m.pools[testKey].goldenDir = "/goldens/x" + stall := &stallingSweepStore{Store: m.ckpts, entered: make(chan struct{}, 1), release: make(chan struct{})} + m.ckpts = stall + + ctx, cancel := context.WithCancel(t.Context()) + done := make(chan struct{}) + go func() { + m.Run(ctx) + close(done) + }() + defer func() { + close(stall.release) + cancel() + <-done + }() + time.Sleep(time.Hour + time.Second) + <-stall.entered + + mustClaim(t, m, testKey) + waitFor(t, func() bool { + m.mu.Lock() + defer m.mu.Unlock() + return len(m.pools[testKey].warm) > 0 + }) + }) +} + func TestCheckpointTenantIsolation(t *testing.T) { eng := newFakeEngine() m := newTestManager(t, eng) @@ -264,3 +296,18 @@ func (s *stallingMetasStore) Metas(ctx context.Context) ([][]byte, error) { <-s.release return s.Store.Metas(ctx) } + +type stallingSweepStore struct { + store.Store + entered chan struct{} + release chan struct{} +} + +func (s *stallingSweepStore) SweepGenerations() error { + select { + case s.entered <- struct{}{}: + default: + } + <-s.release + return s.Store.SweepGenerations() +} diff --git a/sandboxd/pool/claim.go b/sandboxd/pool/claim.go index f593243e..389c528c 100644 --- a/sandboxd/pool/claim.go +++ b/sandboxd/pool/claim.go @@ -290,6 +290,10 @@ func (m *Manager) finalizeBatch(ctx context.Context, sbs []*types.Sandbox, ttl t return quotaErr } for _, sb := range sbs { + sb.Layer = types.LayerUnpooled + if _, pooled := m.activePool(sb.Key); pooled { + sb.Layer = types.LayerPooled + } m.claimed[sb.ID] = sb m.tenantDelta(sb.Tenant, 1) } @@ -450,6 +454,8 @@ func (m *Manager) reapOnce(ctx context.Context) { m.disarmEgress(v.id, true) m.purgeArchiveCk(ctx, v.id, v.ck, v.tenant) m.untrack(m.pendingCks, v.ck) + m.counters.reaps.Add(1) + m.recordUsage(ctx, usageEvent{Event: "reap", ID: v.id}) logger.Infof(ctx, "purged archived sandbox %s", v.id) case reapArchive: logSweepResult(ctx, logger, m.archive(ctx, v.sb), "archived expired sandbox "+v.id, "archive expired sandbox "+v.id) @@ -475,14 +481,21 @@ func (m *Manager) purgeArchiveCk(ctx context.Context, id, ck, tenant string) { m.recordUsage(ctx, usageEvent{Event: "archive_delete", ID: id, Reference: ck, Tenant: tenant}) } -// recommit re-persists in the background until success or a newer durable write; detached. +// recommit keeps claims.json catching up with the store: one retrier at a time writes the current state until it is durable and nothing newer is pending. func (m *Manager) recommit(ctx context.Context, snap claimSnapshot) { + if !m.recommitting.CompareAndSwap(false, true) { + return + } go func() { backoff := recommitBackoff for { err := m.store.commit(snap) if err == nil { - return + m.recommitting.Store(false) + if snap = m.store.mark(); !m.store.pending(snap) || !m.recommitting.CompareAndSwap(false, true) { + return + } + continue } log.WithFunc("pool.recommit").Error(ctx, err, "persist claims") time.Sleep(backoff) @@ -558,6 +571,7 @@ func stampIdentity(sb *types.Sandbox, ttl time.Duration) { sb.ID = "sb_" + randHex(8) sb.Token = randHex(16) sb.Deadline = time.Now().Add(ttl) + sb.LeaseSeconds = int(ttl / time.Second) } func clampTTL(ttl time.Duration) time.Duration { diff --git a/sandboxd/pool/claims.go b/sandboxd/pool/claims.go index f68931c7..1e672e36 100644 --- a/sandboxd/pool/claims.go +++ b/sandboxd/pool/claims.go @@ -22,20 +22,22 @@ type claimSnapshot struct { // claimDTO is the persisted projection of a Sandbox, copied so commit marshals off m.mu. type claimDTO struct { - ID string `json:"id"` - VMName string `json:"vm_name"` - Key types.PoolKey `json:"key"` - Token string `json:"token,omitempty"` - Deadline time.Time `json:"deadline,omitzero"` - Tenant string `json:"tenant,omitempty"` - ClaimRef string `json:"claim_ref,omitempty"` - Volumes []types.Volume `json:"volumes,omitempty"` - VsockSocket string `json:"vsock_socket,omitempty"` - TAP string `json:"tap,omitempty"` - HibernateSnap string `json:"hibernate_snap,omitempty"` - PendingSnap string `json:"pending_snap,omitempty"` - ArchiveCk string `json:"archive_ck,omitempty"` - FromCheckpoint string `json:"from_checkpoint,omitempty"` + ID string `json:"id"` + VMName string `json:"vm_name"` + Key types.PoolKey `json:"key"` + Token string `json:"token,omitempty"` + Deadline time.Time `json:"deadline,omitzero"` + LeaseSeconds int `json:"lease_seconds,omitzero"` + Layer types.PolicyLayer `json:"policy_layer,omitempty"` + Tenant string `json:"tenant,omitempty"` + ClaimRef string `json:"claim_ref,omitempty"` + Volumes []types.Volume `json:"volumes,omitempty"` + VsockSocket string `json:"vsock_socket,omitempty"` + TAP string `json:"tap,omitempty"` + HibernateSnap string `json:"hibernate_snap,omitempty"` + PendingSnap string `json:"pending_snap,omitempty"` + ArchiveCk string `json:"archive_ck,omitempty"` + FromCheckpoint string `json:"from_checkpoint,omitempty"` } // claimStore persists claimed sandboxes across daemon restarts; warm VMs are not persisted. @@ -98,6 +100,12 @@ func (s *claimStore) mark() claimSnapshot { return claimSnapshot{seq: s.seq} } +func (s *claimStore) pending(snap claimSnapshot) bool { + s.mu.Lock() + defer s.mu.Unlock() + return snap.seq > s.written +} + // reset rebuilds the whole projection; the startup path, before contention exists. func (s *claimStore) reset(claims map[string]*types.Sandbox) claimSnapshot { s.mu.Lock() @@ -152,7 +160,8 @@ func (s *claimStore) synced() bool { func dtoOf(sb *types.Sandbox) claimDTO { return claimDTO{ ID: sb.ID, VMName: sb.VMName, Key: sb.Key, Token: sb.Token, - Deadline: sb.Deadline, Tenant: sb.Tenant, ClaimRef: sb.ClaimRef, + Deadline: sb.Deadline, LeaseSeconds: sb.LeaseSeconds, Layer: sb.Layer, + Tenant: sb.Tenant, ClaimRef: sb.ClaimRef, Volumes: slices.Clone(sb.Volumes), VsockSocket: sb.VsockSocket, TAP: sb.TAP, HibernateSnap: sb.HibernateSnap, PendingSnap: sb.PendingSnap, ArchiveCk: sb.ArchiveCk, FromCheckpoint: sb.FromCheckpoint, diff --git a/sandboxd/pool/claims_test.go b/sandboxd/pool/claims_test.go index 72f3d744..b3564467 100644 --- a/sandboxd/pool/claims_test.go +++ b/sandboxd/pool/claims_test.go @@ -1,17 +1,79 @@ package pool import ( + "io" "os" "path/filepath" "reflect" + "runtime" "slices" "sync" + "syscall" "testing" "time" "github.com/cocoonstack/sandbox/sandboxd/types" ) +func TestRecommitKeepsOneRetrierWhileWritesFail(t *testing.T) { + m := newTestManager(t, newFakeEngine()) + sb := mustClaim(t, m, testKey) + dir := filepath.Join(t.TempDir(), "absent") + m.store.path = filepath.Join(dir, "claims.json") + + before := runtime.NumGoroutine() + for range 50 { + m.mu.Lock() + snap := m.store.set(sb) + m.mu.Unlock() + m.recommit(t.Context(), snap) + } + if grown := runtime.NumGoroutine() - before; grown > 2 { + t.Fatalf("%d goroutines for 50 failed persists, want one retrier", grown) + } + + if err := os.MkdirAll(dir, 0o750); err != nil { + t.Fatalf("mkdir: %v", err) + } + waitFor(t, func() bool { + _, err := os.Stat(m.store.path) + return err == nil && !m.recommitting.Load() + }) +} + +func TestRecommitDrainsAChangeItWasTooBusyToSee(t *testing.T) { + m := newTestManager(t, newFakeEngine()) + sb := mustClaim(t, m, testKey) + m.store.path = filepath.Join(t.TempDir(), "claims.json") + fifo := m.store.path + ".tmp" + if err := syscall.Mkfifo(fifo, 0o600); err != nil { + t.Fatalf("mkfifo: %v", err) + } + change := func() claimSnapshot { + m.mu.Lock() + defer m.mu.Unlock() + return m.store.set(sb) + } + + m.recommit(t.Context(), change()) + waitFor(t, func() bool { + if m.store.writeMu.TryLock() { + m.store.writeMu.Unlock() + return false + } + return true + }) + m.recommit(t.Context(), change()) + + r, err := os.Open(fifo) + if err != nil { + t.Fatalf("open fifo: %v", err) + } + _, _ = io.Copy(io.Discard, r) + _ = r.Close() + waitFor(t, func() bool { return !m.recommitting.Load() && !m.store.pending(m.store.mark()) }) +} + func TestStoreRoundTrip(t *testing.T) { s := newClaimStore(t.TempDir()) claims := map[string]*types.Sandbox{ diff --git a/sandboxd/pool/egress.go b/sandboxd/pool/egress.go index 356a44e8..fa06ed4e 100644 --- a/sandboxd/pool/egress.go +++ b/sandboxd/pool/egress.go @@ -311,7 +311,11 @@ func (m *Manager) effectivePolicy(sb *types.Sandbox) (egress.Evaluator, bool) { poolPol := m.poolEgress[sb.Key] tenantPol := m.tenantEgress[sb.Tenant] if poolPol == nil { - if _, pooled := m.activePool(sb.Key); pooled || sb.Tenant == "" || tenantPol == nil { + pooled := sb.Layer == types.LayerPooled + if sb.Layer == "" { + _, pooled = m.activePool(sb.Key) + } + if pooled || sb.Tenant == "" || tenantPol == nil { return nil, false } return *tenantPol, true diff --git a/sandboxd/pool/egress_test.go b/sandboxd/pool/egress_test.go index f05a02dc..022c6610 100644 --- a/sandboxd/pool/egress_test.go +++ b/sandboxd/pool/egress_test.go @@ -211,6 +211,36 @@ func TestEgressLaneWakeFailsClosed(t *testing.T) { } } +func TestEffectivePolicyKeepsTheLayerPinnedAtClaim(t *testing.T) { + m := newTestManager(t, newFakeEngine()) + m.tenantEgress = map[string]*egress.Policy{"acme": {Allow: []egress.Rule{{Host: "b.test"}}}} + m.poolEgress = map[types.PoolKey]*egress.Policy{} + + m.pools = map[types.PoolKey]*pool{testKey: newPool(testKey)} + if _, ok := m.effectivePolicy(&types.Sandbox{Key: testKey, Tenant: "acme", Layer: types.LayerUnpooled}); !ok { + t.Error("a claim made on an unpooled key lost its tenant policy when the key gained a pool") + } + m.pools = map[types.PoolKey]*pool{} + if _, ok := m.effectivePolicy(&types.Sandbox{Key: testKey, Tenant: "acme", Layer: types.LayerPooled}); ok { + t.Error("a claim made on a policy-less pool gained the tenant policy when the pool was dropped") + } +} + +func TestClaimRecordsItsPolicyLayer(t *testing.T) { + m := newTestManager(t, newFakeEngine(), config.PoolSpec{PoolKey: testKey, Warm: 1}) + if sb := mustClaim(t, m, testKey); sb.Layer != types.LayerPooled { + t.Errorf("pooled claim layer %q, want %q", sb.Layer, types.LayerPooled) + } + unpooled := types.PoolKey{Template: "promoted-name", Net: types.NetNone, Size: types.SizeSmall} + sb, err := m.ClaimProvision(t.Context(), unpooled, time.Hour, "acme", "", nil) + if err != nil { + t.Fatalf("claim: %v", err) + } + if sb.Layer != types.LayerUnpooled { + t.Errorf("unpooled claim layer %q, want %q", sb.Layer, types.LayerUnpooled) + } +} + func TestEffectivePolicyComposition(t *testing.T) { m := newTestManager(t, newFakeEngine()) both := &egress.Policy{Allow: []egress.Rule{{Host: "a.test"}, {Host: "b.test"}}} diff --git a/sandboxd/pool/pool.go b/sandboxd/pool/pool.go index eb5acee0..7e9b79bf 100644 --- a/sandboxd/pool/pool.go +++ b/sandboxd/pool/pool.go @@ -302,6 +302,8 @@ type Manager struct { tpls store.Store ckptTTL time.Duration ckptSweeping atomic.Bool + genSweeping atomic.Bool + recommitting atomic.Bool // tplSet caches each template id against its owning tenant ("" = operator). tplMu sync.Mutex @@ -404,12 +406,14 @@ func NewManager(ctx context.Context, cfg *config.Config, eng Engine, secrets *eg } m.ckptTTL = time.Duration(cfg.CheckpointTTLHours) * time.Hour m.tplSet = map[string]string{} - if metas, listErr := m.tpls.Metas(ctx); listErr == nil { - for _, raw := range metas { - var rec templateRecord - if json.Unmarshal(raw, &rec) == nil && rec.ID != "" { - m.tplSet[rec.ID] = rec.Tenant - } + metas, listErr := m.tpls.Metas(ctx) + if listErr != nil { + log.WithFunc("pool.NewManager").Warnf(ctx, "list templates skipped: %v", listErr) + } + for _, raw := range metas { + var rec templateRecord + if json.Unmarshal(raw, &rec) == nil && rec.ID != "" { + m.tplSet[rec.ID] = rec.Tenant } } usage, err := newJournal(filepath.Join(cfg.DataDir, "usage.jsonl")) @@ -501,7 +505,7 @@ func (m *Manager) Run(ctx context.Context) { m.shrinkOnce(ctx) go m.retryArchiveDeletes(ctx) case <-storeSweep.C: - m.sweepStoreGenerations(ctx) + go m.sweepStoreGenerations(ctx) go m.sweepExpiredCheckpoints(ctx) } } @@ -592,6 +596,10 @@ func (m *Manager) WithPeerDelete(fn PeerDeleteFunc) { } func (m *Manager) sweepStoreGenerations(ctx context.Context) { + if !m.genSweeping.CompareAndSwap(false, true) { + return + } + defer m.genSweeping.Store(false) logger := log.WithFunc("pool.sweepStoreGenerations") if err := m.ckpts.SweepGenerations(); err != nil { logger.Error(ctx, err, "sweep checkpoint generations") diff --git a/sandboxd/pool/reconcile.go b/sandboxd/pool/reconcile.go index 8d830012..3ee41aed 100644 --- a/sandboxd/pool/reconcile.go +++ b/sandboxd/pool/reconcile.go @@ -36,6 +36,7 @@ func (m *Manager) Reconcile(ctx context.Context) error { owned := map[string]bool{} referenced := map[string]bool{} + var adopted []*types.Sandbox m.mu.Lock() for id, sb := range claims { rec, ok := live[sb.VMName] @@ -55,6 +56,7 @@ func (m *Manager) Reconcile(ctx context.Context) error { case ok && sb.PendingSnap != "" && (snapsErr != nil || slices.Contains(snaps, sb.PendingSnap)): // the journaled intent names the wake image, so adopt it when the snapshot is there sb.HibernateSnap, sb.PendingSnap = sb.PendingSnap, "" + adopted = append(adopted, sb) default: continue } @@ -69,6 +71,9 @@ func (m *Manager) Reconcile(ctx context.Context) error { m.adoptGolden(p) } m.mu.Unlock() + for _, sb := range adopted { + m.recordHibernate(ctx, sb) + } // a crash mid-export leaves a golden *.tmp or a fork-* staging dir nothing in this life reuses tmps, _ := filepath.Glob(filepath.Join(m.goldensDir(), "*.tmp")) forks, _ := filepath.Glob(filepath.Join(m.dataDir, "fork-*")) diff --git a/sandboxd/pool/reconcile_test.go b/sandboxd/pool/reconcile_test.go index 799cc3e1..66f66a61 100644 --- a/sandboxd/pool/reconcile_test.go +++ b/sandboxd/pool/reconcile_test.go @@ -6,12 +6,39 @@ import ( "os" "path/filepath" "slices" + "strings" "testing" + "github.com/cocoonstack/sandbox/sandboxd/config" "github.com/cocoonstack/sandbox/sandboxd/engine" "github.com/cocoonstack/sandbox/sandboxd/types" ) +func TestReconcileJournalsAnAdoptedHibernate(t *testing.T) { + eng := newFakeEngine() + dir := t.TempDir() + m := newTestManagerAt(t, eng, dir, config.PoolSpec{PoolKey: testKey, Warm: 1}) + sb := mustClaim(t, m, testKey) + snap := hibernatePrefix + strings.TrimPrefix(sb.VMName, vmPrefix) + "-abc" + if err := eng.Hibernate(t.Context(), sb.VMName, snap); err != nil { + t.Fatalf("engine hibernate: %v", err) + } + if err := m.store.commit(m.setPendingSnap(sb, snap)); err != nil { + t.Fatalf("journal intent: %v", err) + } + + restarted := newTestManagerAt(t, eng, dir, config.PoolSpec{PoolKey: testKey, Warm: 1}) + if err := restarted.Reconcile(t.Context()); err != nil { + t.Fatalf("reconcile: %v", err) + } + if _, _, err := restarted.WakeAgentSocket(t.Context(), sb.ID, sb.Token); err != nil { + t.Fatalf("wake: %v", err) + } + if got, want := usageEventsOf(t, restarted, sb.ID), "claim,hibernate,wake"; got != want { + t.Errorf("events %s, want %s", got, want) + } +} + func TestReconcileStaleCreateSweep(t *testing.T) { tests := []staleCreateCase{ {"collected frees the record", engine.StaleCreateCollected, nil, false, false, false}, diff --git a/sandboxd/pool/refill.go b/sandboxd/pool/refill.go index 2f77ab76..b25f9ae7 100644 --- a/sandboxd/pool/refill.go +++ b/sandboxd/pool/refill.go @@ -469,11 +469,11 @@ func (m *Manager) findVM(ctx context.Context, name string) (types.VMRecord, bool return vms[i], true, nil } -// runBounded fans f over n items on the refill semaphore, sharing the node-wide budget. func (m *Manager) destroyAll(ctx context.Context, names []string) *sync.WaitGroup { return m.runBounded(ctx, len(names), func(ctx context.Context, i int) { m.destroy(ctx, names[i]) }) } +// runBounded fans f over n items on the refill semaphore, sharing the node-wide budget. func (m *Manager) runBounded(ctx context.Context, n int, f func(context.Context, int)) *sync.WaitGroup { var wg sync.WaitGroup for i := range n { diff --git a/sandboxd/pool/telemetry_test.go b/sandboxd/pool/telemetry_test.go index 5095580c..e512d2fe 100644 --- a/sandboxd/pool/telemetry_test.go +++ b/sandboxd/pool/telemetry_test.go @@ -27,23 +27,8 @@ func TestUsageJournalRecordsLifecycle(t *testing.T) { t.Fatalf("Release: %v", err) } - raw, err := os.ReadFile(filepath.Join(m.dataDir, "usage.jsonl")) - if err != nil { - t.Fatalf("read journal: %v", err) - } - var events []string - for line := range strings.SplitSeq(strings.TrimSpace(string(raw)), "\n") { - var ev usageEvent - if err := json.Unmarshal([]byte(line), &ev); err != nil { - t.Fatalf("bad journal line %q: %v", line, err) - } - if ev.ID == sb.ID { - events = append(events, ev.Event) - } - } - want := []string{"claim", "hibernate", "wake", "release"} - if strings.Join(events, ",") != strings.Join(want, ",") { - t.Errorf("events %v, want %v", events, want) + if got, want := usageEventsOf(t, m, sb.ID), "claim,hibernate,wake,release"; got != want { + t.Errorf("events %s, want %s", got, want) } c := m.Counters() @@ -216,3 +201,22 @@ func TestPreviewDialWritesAuditEvent(t *testing.T) { t.Errorf("audit event %+v", ev) } } + +func usageEventsOf(t *testing.T, m *Manager, id string) string { + t.Helper() + raw, err := os.ReadFile(filepath.Join(m.dataDir, "usage.jsonl")) + if err != nil { + t.Fatalf("read journal: %v", err) + } + var events []string + for line := range strings.SplitSeq(strings.TrimSpace(string(raw)), "\n") { + var ev usageEvent + if err := json.Unmarshal([]byte(line), &ev); err != nil { + t.Fatalf("bad journal line %q: %v", line, err) + } + if ev.ID == id { + events = append(events, ev.Event) + } + } + return strings.Join(events, ",") +} diff --git a/sandboxd/pool/volume_attach_test.go b/sandboxd/pool/volume_attach_test.go index acf48feb..ee5deb60 100644 --- a/sandboxd/pool/volume_attach_test.go +++ b/sandboxd/pool/volume_attach_test.go @@ -4,7 +4,6 @@ import ( "errors" "slices" "strings" - "sync" "testing" "github.com/cocoonstack/sandbox/sandboxd/config" @@ -73,9 +72,7 @@ func TestAttachOnlyClaimAttachesEveryVolumeConcurrently(t *testing.T) { {Name: "dataset", AttachOnly: true}, {Name: "cache", Mode: types.VolumeModeRW, AttachOnly: true}, } - var attaches sync.WaitGroup - attaches.Add(len(requested)) - eng.attachRendezvous = &attaches + rendezvousAttaches(eng, len(requested)) sb, err := m.ClaimProvision(t.Context(), testKey, 0, "", "", requested) if err != nil { @@ -104,9 +101,7 @@ func TestClaimProvisionAttachFailureKeepsHoldsUntilRemoval(t *testing.T) { eng := newFakeEngine() eng.diskAttachErrFor = "broken" eng.removeStall = make(chan struct{}) - var attaches sync.WaitGroup - attaches.Add(2) - eng.attachRendezvous = &attaches + attaches := rendezvousAttaches(eng, 2) m := newVolumeManager(t, eng, []config.VolumeSpec{ {Name: "scratch", Path: scratch, Writable: true}, {Name: "broken", Path: broken}, @@ -170,9 +165,7 @@ func TestClaimWarmAttachFailureKeepsHoldsUntilRemoval(t *testing.T) { eng.diskAttachErrFor = "broken" eng.removeErrFor = "sbx-warm" eng.vms["sbx-warm"] = "/vsock/warm" - var attaches sync.WaitGroup - attaches.Add(2) - eng.attachRendezvous = &attaches + rendezvousAttaches(eng, 2) m := newVolumePoolManager(t, eng, t.TempDir(), []config.VolumeSpec{ {Name: "scratch", Path: scratch, Writable: true}, {Name: "broken", Path: broken}, @@ -216,9 +209,7 @@ func TestFinalizeQuotaFailureKeepsHoldsUntilRemoval(t *testing.T) { eng := newFakeEngine() m := newVolumeManager(t, eng, []config.VolumeSpec{{Name: "scratch", Path: scratch, Writable: true}}) m.maxClaims = 1 - var attaches sync.WaitGroup - attaches.Add(2) - eng.attachRendezvous = &attaches + attaches := rendezvousAttaches(eng, 2) writable := []types.Volume{{Name: "scratch", Mode: types.VolumeModeRW}} claimErr := make(chan error, 1) @@ -291,9 +282,7 @@ func TestFinalizeTenantQuotaFailureQuiescesAndUncountsTenant(t *testing.T) { eng := newFakeEngine() m := newVolumeManager(t, eng, []config.VolumeSpec{{Name: "scratch", Path: scratch, Writable: true}}) m.tenantMax = map[string]int{"acme": 1} - var attaches sync.WaitGroup - attaches.Add(2) - eng.attachRendezvous = &attaches + attaches := rendezvousAttaches(eng, 2) claimErr := make(chan error, 1) go func() { diff --git a/sandboxd/pool/volume_test.go b/sandboxd/pool/volume_test.go index 9a9e7740..8484e4d1 100644 --- a/sandboxd/pool/volume_test.go +++ b/sandboxd/pool/volume_test.go @@ -97,9 +97,7 @@ func TestClaimProvisionBringsMixedVolumesUpConcurrently(t *testing.T) { {Name: "dataset", Mount: "/volumes/dataset"}, {Name: "cache", Mount: "/cache", Mode: types.VolumeModeRW}, } - var attaches sync.WaitGroup - attaches.Add(len(requested)) - eng.attachRendezvous = &attaches + rendezvousAttaches(eng, len(requested)) sb, err := m.ClaimProvision(t.Context(), testKey, 0, "", "", requested) if err != nil { @@ -586,6 +584,13 @@ func newVolumeManager(t *testing.T, eng *fakeEngine, volumes []config.VolumeSpec return newVolumeManagerAt(t, eng, t.TempDir(), volumes) } +func rendezvousAttaches(eng *fakeEngine, n int) *sync.WaitGroup { + var attaches sync.WaitGroup + attaches.Add(n) + eng.attachRendezvous = &attaches + return &attaches +} + func newVolumePoolManager(t *testing.T, eng *fakeEngine, dataDir string, volumes []config.VolumeSpec) *Manager { t.Helper() m, err := NewManager(t.Context(), &config.Config{ diff --git a/sandboxd/pool/watermark.go b/sandboxd/pool/watermark.go index 71fd8088..50411a19 100644 --- a/sandboxd/pool/watermark.go +++ b/sandboxd/pool/watermark.go @@ -50,6 +50,12 @@ func (p *pool) effectiveTarget(now time.Time) int { rate *= math.Exp(-now.Sub(p.lastArrival).Seconds() / rateDecayTau.Seconds()) } dynamic := int(math.Ceil(rate * p.lead.Seconds() * leadSafety)) + if p.lead == 0 && rate > 0 { + dynamic = 1 + } + if dynamic == 1 && rate*rateDecayTau.Seconds() < 1 { + dynamic = 0 + } return max(p.floor, min(dynamic, p.warmMax)) } diff --git a/sandboxd/pool/watermark_test.go b/sandboxd/pool/watermark_test.go index 7375923f..5808c45f 100644 --- a/sandboxd/pool/watermark_test.go +++ b/sandboxd/pool/watermark_test.go @@ -76,6 +76,53 @@ func TestEffectiveTargetOffWithoutWarmMax(t *testing.T) { } } +func TestEffectiveTargetGrowsAnEmptyPoolAndReturnsItToEmpty(t *testing.T) { + now := time.Now() + p := &pool{key: types.PoolKey{}, warmMax: 8} + for i := range 60 { + p.noteArrival(now.Add(time.Duration(i) * 100 * time.Millisecond)) + } + busy := now.Add(6 * time.Second) + if got := p.effectiveTarget(busy); got != 1 { + t.Fatalf("target %d under demand with no measured lead, want 1 so a refill can measure it", got) + } + p.noteLead(500 * time.Millisecond) + if got := p.effectiveTarget(busy); got != 8 { + t.Errorf("target %d at 10/s with a 500 ms lead, want warmMax 8", got) + } + if got := p.effectiveTarget(busy.Add(10 * time.Minute)); got != 0 { + t.Errorf("target %d after ten silent minutes, want the empty floor", got) + } +} + +func TestEffectiveTargetKeepsASlowPoolAheadOfSparseDemand(t *testing.T) { + now := time.Now() + p := &pool{key: types.PoolKey{}, floor: 1, warmMax: 3, lead: 2 * time.Minute, rate: 0.01, lastArrival: now} + if got := p.effectiveTarget(now); got != 3 { + t.Errorf("target %d at 0.01/s with a two minute lead, want 3: the empty-floor cutoff must not reach a sized pool", got) + } +} + +func TestRefillGrowsAnEmptyPoolOnDemand(t *testing.T) { + eng := newFakeEngine() + m := newTestManager(t, eng, config.PoolSpec{PoolKey: testKey, WarmMax: 4}) + p := m.pools[testKey] + p.goldenDir = "/goldens/x" + start := time.Now().Add(-3 * time.Second) + m.mu.Lock() + for i := range 30 { + p.noteArrival(start.Add(time.Duration(i) * 100 * time.Millisecond)) + } + m.mu.Unlock() + + m.refillOnce(t.Context()) + waitFor(t, func() bool { + m.mu.Lock() + defer m.mu.Unlock() + return len(p.warm) > 0 && p.lead > 0 + }) +} + func TestNoteLeadConverges(t *testing.T) { p := &pool{} p.noteLead(time.Second) diff --git a/sandboxd/server/exec.go b/sandboxd/server/exec.go index 4af9280d..e5b976de 100644 --- a/sandboxd/server/exec.go +++ b/sandboxd/server/exec.go @@ -13,6 +13,7 @@ import ( "github.com/projecteru2/core/log" "github.com/cocoonstack/sandbox/protocol/wire" + "github.com/cocoonstack/sandbox/sandboxd/types" ) const ( @@ -63,15 +64,15 @@ func (s *Server) handleExec(w http.ResponseWriter, r *http.Request) { defer done() if req.TimeoutSeconds > 0 { var cancel context.CancelFunc - ctx, cancel = context.WithTimeout(ctx, time.Duration(req.TimeoutSeconds)*time.Second) + ctx, cancel = context.WithTimeout(ctx, types.Seconds(req.TimeoutSeconds)) defer cancel() } defer closeOnCancel(ctx, guest)() 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) } + frame = append(frame, '\n') stdinClose, _ := wire.EncodeRequest(wire.StdinClose{}) frame = append(frame, stdinClose...) frame = append(frame, '\n') diff --git a/sandboxd/server/exec_test.go b/sandboxd/server/exec_test.go index bece9766..e54ae03f 100644 --- a/sandboxd/server/exec_test.go +++ b/sandboxd/server/exec_test.go @@ -58,6 +58,45 @@ func TestExecReturnsBufferedOutput(t *testing.T) { } } +func TestExecAuditsTheFrameWithoutItsDelimiter(t *testing.T) { + ts, srv := newRelayServer(t, func(conn net.Conn) { + defer conn.Close() + r := bufio.NewReader(conn) + for range 2 { + if _, err := r.ReadBytes('\n'); err != nil { + return + } + } + _, _ = io.WriteString(conn, `{"type":"started","pid":7}`+"\n"+`{"type":"exit","code":0}`+"\n") + }) + audited := make(chan []byte, 1) + srv.mgr.(*fakeManager).audited = func(_ string, line []byte) { audited <- bytes.Clone(line) } + if status, body := postExec(t, ts, `{"argv":["true"]}`); status != http.StatusOK { + t.Fatalf("status %d, want 200: %s", status, body) + } + want, _ := wire.EncodeRequest(wire.Exec{Argv: []string{"true"}}) + if got := <-audited; !bytes.Equal(got, want) { + t.Errorf("audited %q, want the bare frame %q", got, want) + } +} + +func TestExecRunsUnderAnOverflowingTimeout(t *testing.T) { + ts, _ := newRelayServer(t, func(conn net.Conn) { + defer conn.Close() + r := bufio.NewReader(conn) + for range 2 { + if _, err := r.ReadBytes('\n'); err != nil { + return + } + } + _, _ = io.WriteString(conn, `{"type":"started","pid":7}`+"\n"+`{"type":"exit","code":0}`+"\n") + }) + status, body := postExec(t, ts, `{"argv":["true"],"timeout_seconds":10000000000}`) + if status != http.StatusOK { + t.Fatalf("status %d, want 200: a huge timeout_seconds must not wrap into an expired deadline: %s", status, body) + } +} + func TestExecMapsSilkdErrors(t *testing.T) { for _, tt := range []struct { kind string diff --git a/sandboxd/store/dir/dir.go b/sandboxd/store/dir/dir.go index b7883e45..87366ca7 100644 --- a/sandboxd/store/dir/dir.go +++ b/sandboxd/store/dir/dir.go @@ -15,7 +15,6 @@ import ( "path/filepath" "regexp" "runtime" - "slices" "strings" "time" @@ -26,7 +25,6 @@ import ( ) const ( - oldSuffix = ".old" // pre-generation crash artifacts, reclaimed at Delete digestPrefix = "digest-" digestWorkerLimit = 8 digestReadBufferSize = 128 << 10 @@ -87,15 +85,10 @@ func (d *Store) Fetch(ctx context.Context, id string) (string, []byte, string, e return "", nil, "", err } dir := filepath.Join(d.root, id, store.ExportGen(meta)) - if _, statErr := os.Stat(dir); errors.Is(statErr, fs.ErrNotExist) { - dir = filepath.Join(d.root, id, store.ExportDir) - switch _, legacyErr := os.Stat(dir); { - case errors.Is(legacyErr, fs.ErrNotExist): - return "", nil, "", store.ErrNotFound - case legacyErr != nil: - return "", nil, "", legacyErr - } - } else if statErr != nil { + switch _, statErr := os.Stat(dir); { + case errors.Is(statErr, fs.ErrNotExist): + return "", nil, "", store.ErrNotFound + case statErr != nil: return "", nil, "", statErr } digest, err := os.ReadFile(filepath.Join(d.root, id, digestName(meta))) //nolint:gosec // id pinned by the instance idRe @@ -139,7 +132,7 @@ func (d *Store) Delete(_ context.Context, id string) error { final := filepath.Join(d.root, id) entries, err := os.ReadDir(final) if errors.Is(err, fs.ErrNotExist) { - return os.RemoveAll(final + oldSuffix) + return nil } if err != nil { return err @@ -152,7 +145,7 @@ func (d *Store) Delete(_ context.Context, id string) error { return err } } - return errors.Join(os.RemoveAll(final), os.RemoveAll(final+oldSuffix)) + return os.RemoveAll(final) } func (d *Store) SweepStaging() error { @@ -277,15 +270,10 @@ func (d *Store) sweepGenerations(id string) (err error) { } current := store.ExportGen(meta) currentDigest := digestName(meta) - hasCurrent := slices.ContainsFunc(entries, func(e fs.DirEntry) bool { return e.Name() == current }) for _, e := range entries { - name := e.Name() - if name == store.MetaFile || name == current || name == currentDigest { + if name := e.Name(); name == store.MetaFile || name == current || name == currentDigest { continue } - if name == store.ExportDir && !hasCurrent { - continue // the legacy flat layout still backs the current meta - } if removeErr := removeAgedEntry(final, e); removeErr != nil { return removeErr } @@ -303,9 +291,6 @@ func touchCurrentGeneration(final string, now time.Time) error { } current := filepath.Join(final, store.ExportGen(meta)) exportErr := os.Chtimes(current, now, now) //nolint:gosec // current is a hash-derived name under our root - if errors.Is(exportErr, fs.ErrNotExist) { - exportErr = os.Chtimes(filepath.Join(final, store.ExportDir), now, now) - } if errors.Is(exportErr, fs.ErrNotExist) { return nil } diff --git a/sandboxd/store/dir/dir_test.go b/sandboxd/store/dir/dir_test.go index d119e3db..328e0342 100644 --- a/sandboxd/store/dir/dir_test.go +++ b/sandboxd/store/dir/dir_test.go @@ -333,33 +333,6 @@ func TestPublishRetriesAfterExpiredInstall(t *testing.T) { } } -func TestFetchLegacyFlatLayout(t *testing.T) { - const id = "ck_00000000000000aa" - root := t.TempDir() - st, err := New(root, store.CheckpointIDRe) - if err != nil { - t.Fatalf("New: %v", err) - } - seedRecord(t, filepath.Join(root, id), "legacy") - - dir, meta, digest, err := st.Fetch(t.Context(), id) - if err != nil { - t.Fatalf("fetch legacy: %v", err) - } - if string(meta) != metaJSON("legacy") || digest != "" || dir != filepath.Join(root, id, store.ExportDir) { - t.Fatalf("legacy fetch = %q %q %q, want the flat export dir without digest", dir, meta, digest) - } - - mustPublish(t, st, id, "modern") - dir, meta, digest, err = st.Fetch(t.Context(), id) - if err != nil { - t.Fatalf("fetch after re-publish: %v", err) - } - if string(meta) != metaJSON("modern") || digest != "" || dir == filepath.Join(root, id, store.ExportDir) { - t.Fatalf("re-published fetch = %q %q %q, want a generation dir without digest", dir, meta, digest) - } -} - func TestPublishSweepsGenerationsBySupersessionAge(t *testing.T) { const id = "ck_00000000000000aa" root := t.TempDir() @@ -441,37 +414,6 @@ func TestSweepSparesPublishingGeneration(t *testing.T) { } } -func TestSweepSparesLegacyFallback(t *testing.T) { - const id = "ck_00000000000000aa" - root := t.TempDir() - st, err := New(root, store.CheckpointIDRe) - if err != nil { - t.Fatalf("New: %v", err) - } - seedRecord(t, filepath.Join(root, id), "legacy") - backdate(t, filepath.Join(root, id, store.MetaFile)) - - if err := st.SweepStaging(); err != nil { - t.Fatalf("sweep: %v", err) - } - if _, _, _, err := st.Fetch(t.Context(), id); err != nil { - t.Fatalf("legacy record swept while still current: %v", err) - } - - backdate(t, filepath.Join(root, id, store.ExportDir)) - mustPublish(t, st, id, "modern") - if _, err := os.Stat(filepath.Join(root, id, store.ExportDir)); err != nil { - t.Fatalf("legacy export reclaimed inside the grace: %v", err) - } - backdate(t, filepath.Join(root, id, store.ExportDir)) - if err := st.SweepStaging(); err != nil { - t.Fatalf("sweep after re-publish: %v", err) - } - if _, err := os.Stat(filepath.Join(root, id, store.ExportDir)); !os.IsNotExist(err) { - t.Fatalf("superseded legacy export survived the graced sweep: %v", err) - } -} - func TestSweepReclaimsOnlyAgedUncommittedGenerations(t *testing.T) { const ( orphanID = "ck_00000000000000aa" @@ -591,30 +533,6 @@ func TestSweepGenerationsToleratesConcurrentDelete(t *testing.T) { } } -func TestDeleteRemovesLegacyResidue(t *testing.T) { - const id = "ck_00000000000000aa" - root := t.TempDir() - st, err := New(root, store.CheckpointIDRe) - if err != nil { - t.Fatalf("New: %v", err) - } - seedRecord(t, filepath.Join(root, id), id) - seedRecord(t, filepath.Join(root, id+oldSuffix), "stale") - - if err := st.Delete(t.Context(), id); err != nil { - t.Fatalf("Delete: %v", err) - } - if _, err := st.ReadMeta(t.Context(), id); !errors.Is(err, store.ErrNotFound) { - t.Errorf("ReadMeta after Delete: %v, want store.ErrNotFound", err) - } - if _, err := os.Stat(filepath.Join(root, id+oldSuffix)); !os.IsNotExist(err) { - t.Errorf("legacy .old residue survived Delete: %v", err) - } - if err := st.Delete(t.Context(), id); err != nil { - t.Errorf("Delete of a missing record: %v, want nil", err) - } -} - func TestMetasSurfacesReadError(t *testing.T) { const id = "ck_00000000000000aa" root := t.TempDir() diff --git a/sandboxd/store/peer/probe_test.go b/sandboxd/store/peer/probe_test.go index f76fe3ac..9462c936 100644 --- a/sandboxd/store/peer/probe_test.go +++ b/sandboxd/store/peer/probe_test.go @@ -10,14 +10,8 @@ import ( ) func TestOwnersReturnsOnly200(t *testing.T) { - ok := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { - w.WriteHeader(http.StatusOK) - })) - defer ok.Close() - miss := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { - w.WriteHeader(http.StatusNotFound) - })) - defer miss.Close() + ok := statusServer(t, http.StatusOK) + miss := statusServer(t, http.StatusNotFound) p := &HTTPProber{Peers: func() []string { return []string{ok.URL, miss.URL} }} owners := p.Owners(t.Context(), testID) @@ -27,10 +21,7 @@ func TestOwnersReturnsOnly200(t *testing.T) { } func TestOwnersAllMissIsEmpty(t *testing.T) { - miss := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { - w.WriteHeader(http.StatusNotFound) - })) - defer miss.Close() + miss := statusServer(t, http.StatusNotFound) p := &HTTPProber{Peers: func() []string { return []string{miss.URL} }} if owners := p.Owners(t.Context(), testID); len(owners) != 0 { @@ -39,16 +30,8 @@ func TestOwnersAllMissIsEmpty(t *testing.T) { } func TestOwnersHungPeerExcludedWithoutBlockingOthers(t *testing.T) { - block := make(chan struct{}) - hung := httptest.NewServer(http.HandlerFunc(func(_ http.ResponseWriter, _ *http.Request) { - <-block - })) - defer hung.Close() - defer close(block) - ok := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { - w.WriteHeader(http.StatusOK) - })) - defer ok.Close() + hung := hungServer(t) + ok := statusServer(t, http.StatusOK) start := time.Now() p := &HTTPProber{Peers: func() []string { return []string{hung.URL, ok.URL} }} @@ -62,16 +45,8 @@ func TestOwnersHungPeerExcludedWithoutBlockingOthers(t *testing.T) { } func TestOwnersReturnsPromptlyWithOneOwner(t *testing.T) { - block := make(chan struct{}) - hung := httptest.NewServer(http.HandlerFunc(func(_ http.ResponseWriter, _ *http.Request) { - <-block - })) - defer hung.Close() - defer close(block) - ok := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { - w.WriteHeader(http.StatusOK) - })) - defer ok.Close() + hung := hungServer(t) + ok := statusServer(t, http.StatusOK) p := &HTTPProber{Peers: func() []string { return []string{hung.URL, ok.URL} }} start := time.Now() @@ -138,11 +113,7 @@ func TestOwnersDedupsAddrs(t *testing.T) { func TestOwnersCapsAtThree(t *testing.T) { var addrs []string for range 4 { - srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { - w.WriteHeader(http.StatusOK) - })) - defer srv.Close() - addrs = append(addrs, srv.URL) + addrs = append(addrs, statusServer(t, http.StatusOK).URL) } p := &HTTPProber{Peers: func() []string { return addrs }} @@ -154,19 +125,9 @@ func TestOwnersCapsAtThree(t *testing.T) { func TestOwnersCancelsStragglersOnceCapReached(t *testing.T) { var addrs []string for range maxRedirectOwners { - srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { - w.WriteHeader(http.StatusOK) - })) - defer srv.Close() - addrs = append(addrs, srv.URL) + addrs = append(addrs, statusServer(t, http.StatusOK).URL) } - block := make(chan struct{}) - hung := httptest.NewServer(http.HandlerFunc(func(_ http.ResponseWriter, _ *http.Request) { - <-block - })) - defer hung.Close() - defer close(block) - addrs = append(addrs, hung.URL) + addrs = append(addrs, hungServer(t).URL) start := time.Now() p := &HTTPProber{Peers: func() []string { return addrs }} @@ -183,11 +144,7 @@ func TestHealOwnersReturnsMoreThanRedirectCap(t *testing.T) { const want = maxRedirectOwners + 2 var addrs []string for range want { - srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { - w.WriteHeader(http.StatusOK) - })) - defer srv.Close() - addrs = append(addrs, srv.URL) + addrs = append(addrs, statusServer(t, http.StatusOK).URL) } p := &HTTPProber{Peers: func() []string { return addrs }} @@ -197,10 +154,7 @@ func TestHealOwnersReturnsMoreThanRedirectCap(t *testing.T) { } func TestHealOwnersIsExhaustiveDespiteAFastOwner(t *testing.T) { - fast := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { - w.WriteHeader(http.StatusOK) - })) - defer fast.Close() + fast := statusServer(t, http.StatusOK) slow := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { time.Sleep(2 * probeGrace) w.WriteHeader(http.StatusOK) @@ -352,3 +306,18 @@ func TestVerifyProbeMACRejectsOutsideWindow(t *testing.T) { t.Error("VerifyProbeMAC accepted a bucket two steps away, want rejected") } } + +func statusServer(t *testing.T, code int) *httptest.Server { + t.Helper() + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { w.WriteHeader(code) })) + t.Cleanup(srv.Close) + return srv +} + +func hungServer(t *testing.T) *httptest.Server { + t.Helper() + block := make(chan struct{}) + srv := httptest.NewServer(http.HandlerFunc(func(http.ResponseWriter, *http.Request) { <-block })) + t.Cleanup(func() { close(block); srv.Close() }) + return srv +} diff --git a/sandboxd/store/s3/s3.go b/sandboxd/store/s3/s3.go index edaeb713..44fa2f25 100644 --- a/sandboxd/store/s3/s3.go +++ b/sandboxd/store/s3/s3.go @@ -15,6 +15,7 @@ import ( "regexp" "slices" "strings" + "time" "github.com/aws/aws-sdk-go-v2/aws" awsconfig "github.com/aws/aws-sdk-go-v2/config" @@ -36,6 +37,7 @@ const ( metaReadConcurrency = 8 publishConcurrency = 4 // files in parallel; each already multiparts internally fetchConcurrency = 4 + fetchBudget = 30 * time.Minute deleteBatch = 1000 // DeleteObjects caps one request at 1000 keys ) @@ -125,11 +127,18 @@ func (s *Store) Fetch(ctx context.Context, id string) (string, []byte, string, e if _, statErr := os.Stat(export); statErr == nil { return export, meta, digest, nil } - _, err, _ = s.fetches.Do(gen, func() (any, error) { - return nil, s.populate(ctx, id, meta, gen) + flight := s.fetches.DoChan(gen, func() (any, error) { + fctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), fetchBudget) + defer cancel() + return nil, s.populate(fctx, id, meta, gen) }) - if err != nil { - return "", nil, "", err + select { + case res := <-flight: + if res.Err != nil { + return "", nil, "", res.Err + } + case <-ctx.Done(): + return "", nil, "", ctx.Err() } return export, meta, digest, nil } @@ -324,14 +333,7 @@ func (s *Store) populate(ctx context.Context, id string, meta []byte, gen string return err } if len(keys) == 0 { - // records published before per-generation prefixes. - exportPrefix = s.key(id, store.ExportDir) + "/" - if keys, err = s.list(ctx, exportPrefix); err != nil { - return err - } - } - if len(keys) == 0 { - return fmt.Errorf("record %s has no export", id) + return store.ErrNotFound } g, gctx := errgroup.WithContext(ctx) g.SetLimit(fetchConcurrency) diff --git a/sandboxd/store/s3/s3_test.go b/sandboxd/store/s3/s3_test.go index e5d17a4a..58202742 100644 --- a/sandboxd/store/s3/s3_test.go +++ b/sandboxd/store/s3/s3_test.go @@ -6,6 +6,7 @@ import ( "context" "crypto/sha256" "encoding/xml" + "errors" "fmt" "io" "maps" @@ -18,6 +19,7 @@ import ( "strings" "sync" "testing" + "time" "github.com/aws/aws-sdk-go-v2/feature/s3/transfermanager" @@ -120,20 +122,54 @@ func TestDeleteRetainsMetaUntilExportsAreGone(t *testing.T) { } } -func TestFetchLegacyExportLayout(t *testing.T) { - const id = "ck_00000000000000aa" +func TestFetchReportsMetaWithoutExportAsNotFound(t *testing.T) { + const id = "ck_00000000000000cc" fake := &fakeS3{objects: map[string][]byte{ - "ck/" + id + "/" + store.MetaFile: []byte(`{"id":"` + id + `"}`), - "ck/" + id + "/" + store.ExportDir + "/disk.img": []byte("legacy-bytes"), + "ck/" + id + "/" + store.MetaFile: []byte(`{"id":"` + id + `"}`), }} st := newTestStore(t, fake) - dir, _, _, err := st.Fetch(t.Context(), id) - if err != nil { - t.Fatalf("Fetch: %v", err) + if _, _, _, err := st.Fetch(t.Context(), id); !errors.Is(err, store.ErrNotFound) { + t.Fatalf("Fetch = %v, want store.ErrNotFound", err) + } +} + +func TestFetchOutlivesTheCallerThatStartedIt(t *testing.T) { + const id = "ck_00000000000000dd" + metaKey := "ck/" + id + "/" + store.MetaFile + fake := &fakeS3{objects: map[string][]byte{}, exportGate: make(chan struct{}), exportEntered: make(chan struct{}, 1)} + st := newTestStore(t, fake) + publishRecord(t, st, id, []byte(`{"id":"`+id+`"}`), "payload") + fake.resetRequests() + + first, hangUp := context.WithCancel(t.Context()) + started := make(chan error, 1) + go func() { + _, _, _, err := st.Fetch(first, id) + started <- err + }() + <-fake.exportEntered + joined := make(chan error, 1) + go func() { + _, _, _, err := st.Fetch(t.Context(), id) + joined <- err + }() + for fake.requestCount(http.MethodGet, metaKey) < 2 { + time.Sleep(time.Millisecond) + } + time.Sleep(50 * time.Millisecond) + hangUp() + select { + case err := <-started: + if !errors.Is(err, context.Canceled) { + t.Fatalf("first caller's Fetch = %v, want its own cancellation", err) + } + case <-time.After(5 * time.Second): + close(fake.exportGate) + t.Fatal("first caller's Fetch still blocked on the stalled download after it hung up") } - 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) + close(fake.exportGate) + if err := <-joined; err != nil { + t.Fatalf("second caller's Fetch = %v after the first caller hung up, want the shared download", err) } } @@ -481,12 +517,21 @@ type fakeS3 struct { deleteBatches [][]string failList bool failDelete bool + exportGate chan struct{} + exportEntered chan struct{} } func (f *fakeS3) ServeHTTP(w http.ResponseWriter, r *http.Request) { + key := strings.TrimPrefix(r.URL.Path, "/testbucket/") + if f.exportGate != nil && r.Method == http.MethodGet && strings.Contains(key, "/"+store.ExportDir) { + select { + case f.exportEntered <- struct{}{}: + default: + } + <-f.exportGate + } f.mu.Lock() defer f.mu.Unlock() - key := strings.TrimPrefix(r.URL.Path, "/testbucket/") if f.requests == nil { f.requests = map[fakeRequest]int{} } diff --git a/sandboxd/store/store.go b/sandboxd/store/store.go index 72a2a095..d25867b8 100644 --- a/sandboxd/store/store.go +++ b/sandboxd/store/store.go @@ -17,7 +17,7 @@ import ( ) const ( - // ExportDir is the snapshot export under a record: /export. + // ExportDir is the snapshot export inside a staging dir; publish renames it to its ExportGen. ExportDir = "export" // MetaFile is the record's metadata, written last so a lister never sees a half-published record. MetaFile = "meta.json" diff --git a/sandboxd/types/api.go b/sandboxd/types/api.go index 2b3ca854..58d72545 100644 --- a/sandboxd/types/api.go +++ b/sandboxd/types/api.go @@ -1,6 +1,7 @@ package types import ( + "math" "time" ) @@ -11,7 +12,7 @@ type TTLField struct { // TTL converts the wire seconds to a duration. func (f TTLField) TTL() time.Duration { - return time.Duration(f.TTLSeconds) * time.Second + return Seconds(f.TTLSeconds) } // ClaimRequest is the wire body of POST /v1/claim. @@ -127,3 +128,8 @@ type PreviewRequest struct { type PreviewResponse struct { URL string `json:"url"` } + +// Seconds converts wire seconds to a duration: a negative count is zero and a huge one saturates instead of wrapping. +func Seconds(n int) time.Duration { + return time.Duration(min(max(int64(n), 0), math.MaxInt64/int64(time.Second))) * time.Second +} diff --git a/sandboxd/types/types.go b/sandboxd/types/types.go index d70334aa..696c3a5c 100644 --- a/sandboxd/types/types.go +++ b/sandboxd/types/types.go @@ -26,6 +26,9 @@ const ( SizeXLarge Size = "xlarge" Size2XLarge Size = "2xlarge" + LayerPooled PolicyLayer = "pooled" + LayerUnpooled PolicyLayer = "unpooled" + RestoreCopy RestoreMode = "copy" RestoreOnDemand RestoreMode = "ondemand" RestoreMmap RestoreMode = "mmap" @@ -64,6 +67,9 @@ var ( // NetShape selects whether the Cloud Hypervisor guest has a NIC. type NetShape string +// PolicyLayer pins, at claim time, whether the key had a pool layer in its egress policy. +type PolicyLayer string + // RestoreMode selects cocoon's clone memory-restore strategy; empty is cocoon's default. type RestoreMode string @@ -142,6 +148,10 @@ type Sandbox struct { Token string `json:"token,omitempty"` Deadline time.Time `json:"deadline,omitzero"` + // LeaseSeconds is the lease the claim was granted; a wake from the archive grants it again. + LeaseSeconds int `json:"lease_seconds,omitzero"` + // Layer is empty on a record older than the field, which resolves against the live pool set. + Layer PolicyLayer `json:"policy_layer,omitempty"` // Tenant names the owning tenant; empty means the operator claimed it. Tenant string `json:"tenant,omitempty"` diff --git a/sandboxd/types/types_test.go b/sandboxd/types/types_test.go index 0234958f..d4a541ed 100644 --- a/sandboxd/types/types_test.go +++ b/sandboxd/types/types_test.go @@ -1,6 +1,29 @@ package types -import "testing" +import ( + "math" + "testing" + "time" +) + +func TestSecondsSaturatesInsteadOfWrapping(t *testing.T) { + for _, n := range []int{9223372037, 10_000_000_000, math.MaxInt64} { + if got := Seconds(n); got <= 24*time.Hour { + t.Errorf("Seconds(%d) = %v, want a duration past any cap, never a wrapped one", n, got) + } + } + if got := (TTLField{TTLSeconds: 9223372037}).TTL(); got <= 24*time.Hour { + t.Errorf("TTL() = %v for a huge ttl_seconds, want it to reach the 24h clamp", got) + } + for _, n := range []int{-1, -10_000_000_000, math.MinInt64} { + if got := Seconds(n); got != 0 { + t.Errorf("Seconds(%d) = %v, want 0 (the server default), never a wrapped positive", n, got) + } + } + if got := Seconds(90); got != 90*time.Second { + t.Errorf("Seconds(90) = %v", got) + } +} func TestValidateSizes(t *testing.T) { for _, tt := range []struct { diff --git a/scripts/egress-e2e.sh b/scripts/egress-e2e.sh index efab59d8..1a26b33c 100755 --- a/scripts/egress-e2e.sh +++ b/scripts/egress-e2e.sh @@ -81,9 +81,10 @@ echo "== start sandboxd ($NET-lane pool + egress policy, reach=$REACH)" DAEMON_PID=$! for _ in $(seq 1 40); do curl -sf "http://$ADDR/healthz" >/dev/null && break; sleep 0.5; done # Wait for the warm pool so the claim is a warm hit (proxy armed at claim). -for _ in $(seq 1 120); do +for i in $(seq 1 120); do curl -sf -H "Authorization: Bearer $TOKEN" "http://$ADDR/v1/info" 2>/dev/null | jq -e '.pools[0].warm >= 1' >/dev/null 2>&1 && break + [[ $i == 120 ]] && { echo "pool never became warm"; exit 1; } sleep 1 done diff --git a/scripts/intercept-e2e.sh b/scripts/intercept-e2e.sh index e0f64ef3..01b6bbdd 100755 --- a/scripts/intercept-e2e.sh +++ b/scripts/intercept-e2e.sh @@ -73,9 +73,10 @@ echo "== start sandboxd (intercept pool; golden build bakes the cluster root via "$DATA/sandboxd" -config "$DATA/config.json" >>"$DATA/daemon.log" 2>&1 & DAEMON_PID=$! for _ in $(seq 1 40); do curl -sf "http://$ADDR/healthz" >/dev/null 2>&1 && break; sleep 0.5; done -for _ in $(seq 1 300); do +for i in $(seq 1 300); do curl -sf -H "Authorization: Bearer $TOKEN" "http://$ADDR/v1/info" 2>/dev/null | jq -e '.pools[0].warm >= 1' >/dev/null 2>&1 && break + [[ $i == 300 ]] && { echo "pool never became warm"; exit 1; } sleep 1 done diff --git a/scripts/socks-e2e.sh b/scripts/socks-e2e.sh index b9b9ab14..90762044 100755 --- a/scripts/socks-e2e.sh +++ b/scripts/socks-e2e.sh @@ -70,9 +70,10 @@ echo "== start sandboxd (small pool opted into SOCKS5, medium pool without a pol "$DATA/sandboxd" -config "$DATA/config.json" >>"$DATA/daemon.log" 2>&1 & DAEMON_PID=$! for _ in $(seq 1 40); do curl -sf "http://$ADDR/healthz" >/dev/null && break; sleep 0.5; done -for _ in $(seq 1 180); do +for i in $(seq 1 180); do curl -sf -H "Authorization: Bearer $TOKEN" "http://$ADDR/v1/info" 2>/dev/null | jq -e '(.pools | length) == 3 and all(.pools[]; .warm >= 1)' >/dev/null 2>&1 && break + [[ $i == 180 ]] && { echo "pools never became warm"; exit 1; } sleep 1 done diff --git a/sdk/go/client.go b/sdk/go/client.go index 64177090..67303c22 100644 --- a/sdk/go/client.go +++ b/sdk/go/client.go @@ -215,13 +215,15 @@ func (c *Client) DeleteTemplate(ctx context.Context, template string, opts ...Op return err } u.Set(noRedirectQueryParam, "1") - if tryErr := tryEach(redirect, func(addr string) error { - _, retryErr := c.deleteTemplates(ctx, addr, u) - return retryErr - }, retryMiss); tryErr != nil { - return fmt.Errorf("delete template at owner: %w", tryErr) + for _, addr := range redirect { + if _, err = c.deleteTemplates(ctx, addr, u); err == nil { + return nil + } + if he, ok := errors.AsType[*APIError](err); ok && he.Status != http.StatusNotFound { + break + } } - return nil + return fmt.Errorf("delete template at owner: %w", err) } func (c *Client) ownerAt(ctx context.Context, addr, id, token string) (string, error) { @@ -333,25 +335,6 @@ func doNoContent(ctx context.Context, c *Client, method, addr, path string, body return nil } -func tryEach(candidates []string, call func(addr string) error, retry func(error) bool) error { - var lastErr error - for _, addr := range candidates { - lastErr = call(addr) - if lastErr == nil { - return nil - } - if !retry(lastErr) { - return lastErr - } - } - return lastErr -} - -func retryMiss(err error) bool { - he, ok := errors.AsType[*APIError](err) - return !ok || he.Status == http.StatusNotFound -} - func retryTransient(err error) bool { he, ok := errors.AsType[*APIError](err) if !ok { diff --git a/sdk/go/keepalive_test.go b/sdk/go/keepalive_test.go index 78436354..c574988d 100644 --- a/sdk/go/keepalive_test.go +++ b/sdk/go/keepalive_test.go @@ -39,6 +39,23 @@ func TestCallsShareOneConnection(t *testing.T) { } } +func TestRefusedStreamOpenKeepsItsConnection(t *testing.T) { + var upgrades atomic.Int32 + fake := silkdtest.NewFake(t.TempDir()) + sb := testSandbox(t, newAgentServer(t, func(c net.Conn) { + upgrades.Add(1) + fake.ServeConn(c) + })) + for range 3 { + if _, err := sb.Watch(t.Context(), "/missing", false); err == nil { + t.Fatal("watch on a missing path succeeded") + } + } + if got := upgrades.Load(); got != 1 { + t.Errorf("upgrades = %d, want 1: an error frame ends the RPC, not the connection", got) + } +} + func TestOldDaemonDialsPerCall(t *testing.T) { var upgrades atomic.Int32 sb := testSandbox(t, newAgentServer(t, func(c net.Conn) { diff --git a/sdk/go/port.go b/sdk/go/port.go index 5af3042d..00375db5 100644 --- a/sdk/go/port.go +++ b/sdk/go/port.go @@ -138,8 +138,7 @@ func (s *Sandbox) openStream(ctx context.Context, req wire.Request) (*PortConn, return nil, err } if _, err = expect[wire.Ready](ctx, conn); err != nil { - l.close() - return nil, err + return nil, l.done(err) } pr, pw := io.Pipe() p := &PortConn{conn: conn, stop: l.close, out: pr} diff --git a/sdk/go/proc.go b/sdk/go/proc.go index b9ca12f9..30b28ff6 100644 --- a/sdk/go/proc.go +++ b/sdk/go/proc.go @@ -68,7 +68,7 @@ func (s *Sandbox) drainProc(ctx context.Context, req wire.Request, stdout, stder return 0, false, err } defer l.close() - code, exited, err := pumpStdio(ctx, conn, stdout, stderr) + _, code, exited, err := pumpStdio(ctx, conn, stdout, stderr) // logs closes with done after the exit frame if _, replay := req.(*wire.Logs); replay && exited && err == nil { err = terminalErr(ctx, conn) diff --git a/sdk/go/pty.go b/sdk/go/pty.go index b04db459..117ab49c 100644 --- a/sdk/go/pty.go +++ b/sdk/go/pty.go @@ -104,8 +104,7 @@ func (s *Sandbox) OpenPty(ctx context.Context, opts PtyOpts) (*Pty, error) { } started, err := expect[wire.Started](ctx, conn) if err != nil { - l.close() - return nil, err + return nil, l.done(err) } pr, pw := io.Pipe() diff --git a/sdk/go/sandbox.go b/sdk/go/sandbox.go index 495c8932..b3f72d93 100644 --- a/sdk/go/sandbox.go +++ b/sdk/go/sandbox.go @@ -16,6 +16,7 @@ import ( const ( releaseTimeout = 30 * time.Second + killWait = 5 * time.Second stdinChunk = 32 * 1024 ) @@ -100,7 +101,7 @@ func (s *Sandbox) Exec(ctx context.Context, argv ...string) (string, error) { } // Run executes cmd in the sandbox, streaming stdio over one relayed silkd -// connection, and returns the exit code. +// connection, and returns the exit code; a ctx that ends first kills the command. func (s *Sandbox) Run(ctx context.Context, cmd Cmd) (int, error) { if len(cmd.Argv) == 0 { return 0, fmt.Errorf("empty argv") @@ -123,7 +124,7 @@ func (s *Sandbox) Run(ctx context.Context, cmd Cmd) (int, error) { }() } - code, exited, err := pumpStdio(ctx, conn, cmd.Stdout, cmd.Stderr) + pid, code, exited, err := pumpStdio(ctx, conn, cmd.Stdout, cmd.Stderr) // a pump still parked in Read would write into the next RPC, so its connection is not kept select { case <-stdinDone: @@ -131,6 +132,11 @@ func (s *Sandbox) Run(ctx context.Context, cmd Cmd) (int, error) { default: } if err != nil { + if ctx.Err() != nil && pid != 0 { + kctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), killWait) + _ = s.Kill(kctx, pid, 0) + cancel() + } return 0, err } if !exited { diff --git a/sdk/go/sandbox_test.go b/sdk/go/sandbox_test.go index 1defa563..6aa5d8c6 100644 --- a/sdk/go/sandbox_test.go +++ b/sdk/go/sandbox_test.go @@ -1,6 +1,8 @@ package sandbox import ( + "bufio" + "bytes" "context" "errors" "io" @@ -77,6 +79,49 @@ func TestRunContextCancel(t *testing.T) { } } +func TestRunCancelKillsTheCommandItStarted(t *testing.T) { + killed := make(chan uint32, 1) + sb := testSandbox(t, newAgentServer(t, func(conn net.Conn) { + defer conn.Close() + r := bufio.NewReader(conn) + for { + line, err := r.ReadBytes('\n') + if err != nil { + return + } + req, err := wire.DecodeRequest(bytes.TrimSpace(line)) + if err != nil { + return + } + switch req := req.(type) { + case *wire.Info: + _, _ = io.WriteString(conn, `{"type":"info","version":"test","proto":2}`+"\n") + case *wire.Exec: + _, _ = io.WriteString(conn, `{"type":"started","pid":7}`+"\n") + _, _ = io.Copy(io.Discard, r) + return + case *wire.Kill: + killed <- req.PID + _, _ = io.WriteString(conn, `{"type":"done"}`+"\n") + } + } + })) + + ctx, cancel := context.WithTimeout(t.Context(), 100*time.Millisecond) + defer cancel() + if _, err := sb.Run(ctx, Cmd{Argv: []string{"sleep", "300"}}); !errors.Is(err, context.DeadlineExceeded) { + t.Fatalf("got %v, want deadline exceeded", err) + } + select { + case pid := <-killed: + if pid != 7 { + t.Errorf("killed pid %d, want 7", pid) + } + default: + t.Fatal("no kill reached the guest before Run returned") + } +} + func TestUpgradeKeepsCoalescedBytes(t *testing.T) { mux := http.NewServeMux() mux.HandleFunc("GET /v1/sandboxes/{id}/agent", func(w http.ResponseWriter, r *http.Request) { diff --git a/sdk/go/utils.go b/sdk/go/utils.go index e7b16a68..8c11cb33 100644 --- a/sdk/go/utils.go +++ b/sdk/go/utils.go @@ -71,33 +71,34 @@ func (s *Sandbox) downloadRPC(ctx context.Context, req wire.Request, sink func([ return l.done(drainData(ctx, conn, sink)) } -// pumpStdio copies stdout/stderr frames to the writers until the terminal frame: exit carries the code, done means the stream ended without one. -func pumpStdio(ctx context.Context, conn *silkd.Conn, stdout, stderr io.Writer) (code int32, exited bool, err error) { +// pumpStdio copies stdout/stderr frames to the writers until the terminal frame: exit carries the code, done means the stream ended without one; pid is the one started reported, zero before it. +func pumpStdio(ctx context.Context, conn *silkd.Conn, stdout, stderr io.Writer) (pid uint32, code int32, exited bool, err error) { stdout = cmp.Or(stdout, io.Discard) stderr = cmp.Or(stderr, io.Discard) for { resp, err := recv(ctx, conn) if err != nil { - return 0, false, err + return pid, 0, false, err } switch resp := resp.(type) { case *wire.Started: + pid = resp.PID case *wire.Stdout: if _, err := stdout.Write(resp.Data); err != nil { - return 0, false, err + return pid, 0, false, err } case *wire.Stderr: if _, err := stderr.Write(resp.Data); err != nil { - return 0, false, err + return pid, 0, false, err } case *wire.Exit: - return resp.Code, true, nil + return pid, resp.Code, true, nil case *wire.Done: - return 0, false, nil + return pid, 0, false, nil case *wire.ErrorResp: - return 0, false, resp + return pid, 0, false, resp default: - return 0, false, unexpected(resp) + return pid, 0, false, unexpected(resp) } } } diff --git a/sdk/go/watch.go b/sdk/go/watch.go index 8d5e1b50..33be6a1f 100644 --- a/sdk/go/watch.go +++ b/sdk/go/watch.go @@ -87,8 +87,7 @@ func (s *Sandbox) Watch(ctx context.Context, path string, recursive bool) (*Watc return nil, err } if _, err = expect[wire.Ready](ctx, conn); err != nil { - l.close() - return nil, err + return nil, l.done(err) } w := &Watcher{ events: make(chan wire.Event, 16), diff --git a/sdk/langchain/cocoonsandbox_langchain/toolkit.py b/sdk/langchain/cocoonsandbox_langchain/toolkit.py index 048eb918..e3ff3282 100644 --- a/sdk/langchain/cocoonsandbox_langchain/toolkit.py +++ b/sdk/langchain/cocoonsandbox_langchain/toolkit.py @@ -11,7 +11,7 @@ import time from collections.abc import Callable -from cocoonsandbox import Client, Sandbox, SilkdError +from cocoonsandbox import Client, Sandbox, SandboxError from langchain_core.tools import StructuredTool, ToolException from pydantic import BaseModel, Field @@ -42,7 +42,7 @@ def __init__( api_token: str = "", template: str = "rt:24.04", net: str = "", - ttl_seconds: int = 0, + ttl_seconds: int = 3600, from_checkpoint: str = "", ) -> None: self._client = Client(addr, api_token=api_token, timeout=CALL_TIMEOUT) @@ -108,12 +108,12 @@ def close(self) -> None: sb.close() def sandbox(self, deadline: float | None = None) -> Sandbox: - """The claimed sandbox, claiming (or branching) on first use within the caller's deadline.""" + """The claimed sandbox, claiming (or branching) on first use by deadline, else within CALL_TIMEOUT.""" with self._lock: if self._closed: raise RuntimeError("toolkit is closed") if self._sb is None: - self._sb = self._claim(deadline) + self._sb = self._claim(time.monotonic() + CALL_TIMEOUT if deadline is None else deadline) return self._sb def _claim(self, deadline: float | None = None) -> Sandbox: @@ -122,11 +122,17 @@ def _claim(self, deadline: float | None = None) -> Sandbox: return self._client.new(self._template, net=self._net, ttl_seconds=self._ttl, deadline=deadline) def _tool(self, name: str, description: str, schema: type[BaseModel], func: Callable[..., str]) -> StructuredTool: + def run(**kwargs: object) -> str: + try: + return func(**kwargs) + except SandboxError as exc: + raise ToolException(str(exc)) from exc + async def arun(**kwargs): - return await asyncio.to_thread(func, **kwargs) + return await asyncio.to_thread(run, **kwargs) return StructuredTool.from_function( - func=func, coroutine=arun, name=name, description=description, args_schema=schema, handle_tool_error=True + func=run, coroutine=arun, name=name, description=description, args_schema=schema, handle_tool_error=True ) def _exec(self, command: str, cwd: str = "") -> str: @@ -158,12 +164,7 @@ def _write_file(self, path: str, content: str) -> str: return f"wrote {path}" def _read_file(self, path: str) -> str: - try: - return self.sandbox().read_file(path).decode(errors="replace") - except SilkdError as exc: - if exc.kind != "not_found": - raise - raise ToolException(f"{path}: no such file") from None + return self.sandbox().read_file(path).decode(errors="replace") def _list_dir(self, path: str) -> str: return json.dumps(self.sandbox().list_dir(path)) diff --git a/sdk/langchain/tests/test_toolkit.py b/sdk/langchain/tests/test_toolkit.py index 59cdcb85..761137d0 100644 --- a/sdk/langchain/tests/test_toolkit.py +++ b/sdk/langchain/tests/test_toolkit.py @@ -2,10 +2,12 @@ shaping, lazy claim, and double-close safety — no real node.""" import asyncio +import socket +import threading import time import pytest -from cocoonsandbox import SilkdError +from cocoonsandbox import Sandbox, SilkdError from cocoonsandbox_langchain import CocoonToolkit from cocoonsandbox_langchain.toolkit import CALL_TIMEOUT @@ -43,7 +45,15 @@ def test_file_tools_round_trip(monkeypatch): assert "a.txt" in list_dir.invoke({"path": "/w"}) -def test_exec_claims_inside_the_call_budget(monkeypatch): +@pytest.mark.parametrize( + ("name", "args"), + [ + ("sandbox_exec", {"command": "echo hi"}), + ("sandbox_write_file", {"path": "/w/a.txt", "content": "body"}), + ("sandbox_list_dir", {"path": "/w"}), + ], +) +def test_first_tool_call_claims_inside_the_call_budget(monkeypatch, name, args): kit = CocoonToolkit("127.0.0.1:1") seen = [] @@ -52,7 +62,7 @@ def claim(deadline=None): return FakeSandbox() monkeypatch.setattr(kit, "_claim", claim) - kit.get_tools()[0].invoke({"command": "echo hi"}) + next(t for t in kit.get_tools() if t.name == name).invoke(args) assert len(seen) == 1 and seen[0] is not None assert 0 < seen[0] - time.monotonic() <= CALL_TIMEOUT, seen[0] @@ -67,6 +77,30 @@ def claim(deadline=None): assert kit.get_tools()[0].invoke({"command": "echo hi"}) == f"cut off after {CALL_TIMEOUT}s" +def test_a_guest_error_comes_back_as_a_tool_error(monkeypatch): + kit, fake = hooked(monkeypatch) + + def refused(path, data): + raise SilkdError("not_found", "no such directory") + + fake.write_file = refused + tool = next(t for t in kit.get_tools() if t.name == "sandbox_write_file") + assert "no such directory" in tool.invoke({"path": "/nope/a.txt", "content": "body"}) + + +def test_default_lease_outlives_an_agent_run(monkeypatch): + kit = CocoonToolkit("127.0.0.1:1") + seen = {} + + def new(template, **kwargs): + seen.update(kwargs) + return FakeSandbox() + + monkeypatch.setattr(kit._client, "new", new) + kit.get_tools()[0].invoke({"command": "echo hi"}) + assert seen["ttl_seconds"] == 3600, seen + + def test_close_releases_once(monkeypatch): kit, fake = hooked(monkeypatch) kit.get_tools()[0].invoke({"command": "x"}) @@ -82,6 +116,31 @@ def test_use_after_close_raises(monkeypatch): kit.sandbox() +def test_read_file_reports_a_missing_path_as_a_tool_error(monkeypatch): + kit, fake = hooked(monkeypatch) + + def missing(path): + raise SilkdError("not_found", "no such file") + + fake.read_file = missing + tool = next(t for t in kit.get_tools() if t.name == "sandbox_read_file") + assert "no such file" in tool.invoke({"path": "/nope"}) + + +def test_a_stalled_agent_upgrade_comes_back_as_a_tool_error(): + server = socket.create_server(("127.0.0.1", 0)) + addr = f"127.0.0.1:{server.getsockname()[1]}" + threading.Thread(target=serve_silence, args=(server,), daemon=True).start() + kit = CocoonToolkit(addr) + kit._client.timeout = 0.2 + kit._sb = Sandbox(kit._client, "sb_1", "tok", addr) + tool = next(t for t in kit.get_tools() if t.name == "sandbox_read_file") + try: + assert "timed out" in tool.invoke({"path": "/etc/hostname"}) + finally: + server.close() + + class FakeSandbox: def __init__(self): self.closed = 0 @@ -119,12 +178,8 @@ def hooked(monkeypatch): return kit, fake -def test_read_file_reports_a_missing_path_as_a_tool_error(monkeypatch): - kit, fake = hooked(monkeypatch) - - def missing(path): - raise SilkdError("not_found", "no such file") - - fake.read_file = missing - tool = next(t for t in kit.get_tools() if t.name == "sandbox_read_file") - assert "no such file" in tool.invoke({"path": "/nope"}) +def serve_silence(server: socket.socket) -> None: + conn, _ = server.accept() + conn.recv(4096) + time.sleep(1) + conn.close() diff --git a/sdk/openai/cocoonsandbox_openai/adapter.py b/sdk/openai/cocoonsandbox_openai/adapter.py index 21834ede..b5efff73 100644 --- a/sdk/openai/cocoonsandbox_openai/adapter.py +++ b/sdk/openai/cocoonsandbox_openai/adapter.py @@ -31,7 +31,7 @@ class CocoonSandboxClientOptions(BaseSandboxClientOptions): api_token: str = "" template: str = "rt:24.04" net: str = "" - ttl_seconds: int = 0 + ttl_seconds: int = 3600 class CocoonSandboxSessionState(SandboxSessionState): @@ -53,6 +53,7 @@ class CocoonSandboxSession(BaseSandboxSession): def __init__(self, *, state: CocoonSandboxSessionState) -> None: self.state = state self._proxies: list[socket.socket] = [] + self._sb: Sandbox | None = None @classmethod def from_state(cls, state: CocoonSandboxSessionState) -> CocoonSandboxSession: @@ -120,9 +121,11 @@ async def _shutdown_backend(self) -> None: self._proxies.clear() def _sandbox(self) -> Sandbox: - s = self.state - client = Client(s.addr, api_token=s.api_token) - return Sandbox(client=client, id=s.sandbox_id, token=s.sandbox_token, owner=s.owner or s.addr) + if self._sb is None: + s = self.state + client = Client(s.addr, api_token=s.api_token) + self._sb = Sandbox(client=client, id=s.sandbox_id, token=s.sandbox_token, owner=s.owner or s.addr) + return self._sb def _abs(self, path: Path | str) -> Path: path = Path(path) diff --git a/sdk/openai/tests/test_adapter.py b/sdk/openai/tests/test_adapter.py index 146738ed..87ff58a4 100644 --- a/sdk/openai/tests/test_adapter.py +++ b/sdk/openai/tests/test_adapter.py @@ -14,12 +14,15 @@ from cocoonsandbox import ProtocolError, SilkdError from cocoonsandbox_openai import CocoonSandboxClient, CocoonSandboxClientOptions, CocoonSandboxSessionState +CLAIMS: list = [] + class FakeNode(BaseHTTPRequestHandler): def do_POST(self): length = int(self.headers.get("Content-Length") or 0) - _ = self.rfile.read(length) + body = self.rfile.read(length) if self.path == "/v1/claim": + self.server.claims.append(json.loads(body)) self._reply(200, {"id": "sb_1", "token": "tok", "owner_addr": self.headers["Host"]}) elif self.path.endswith("/release"): self._reply(204, None) @@ -44,6 +47,8 @@ def log_message(self, *args): @pytest.fixture def node(): server = HTTPServer(("127.0.0.1", 0), FakeNode) + server.claims = CLAIMS + CLAIMS.clear() threading.Thread(target=server.serve_forever, daemon=True).start() yield f"127.0.0.1:{server.server_port}" server.shutdown() @@ -66,6 +71,27 @@ async def go(): asyncio.run(go()) +def test_default_lease_outlives_an_agent_run(node): + async def go(): + client = CocoonSandboxClient() + session = await client.create(options=CocoonSandboxClientOptions(addr=node)) + await client.delete(session) + + asyncio.run(go()) + assert CLAIMS[0]["ttl_seconds"] == 3600, CLAIMS + + +def test_operations_share_one_sandbox_handle(node): + async def go(): + client = CocoonSandboxClient() + session = await client.create(options=CocoonSandboxClientOptions(addr=node, template="rt:24.04")) + inner = session._inner + assert inner._sandbox() is inner._sandbox() + await client.delete(session) + + asyncio.run(go()) + + def test_exec_maps_stdio_and_exit(node, monkeypatch): class FakeSandbox: def __init__(self, **kw): diff --git a/sdk/python/cocoonsandbox/conn.py b/sdk/python/cocoonsandbox/conn.py index d3a0fb21..a8018a2d 100644 --- a/sdk/python/cocoonsandbox/conn.py +++ b/sdk/python/cocoonsandbox/conn.py @@ -12,7 +12,7 @@ from collections.abc import Iterator from typing import Any, BinaryIO, Protocol, TypeVar -from .errors import APIError, ProtocolError, SandboxTimeout, SilkdError +from .errors import APIError, ProtocolError, SandboxError, SandboxTimeout, SilkdError from .frames import MAX_FRAME, decode_response, encode_request KEEP_ALIVE_CONNS = 8 @@ -40,7 +40,10 @@ def __init__(self, sock: socket.socket, reader: BinaryIO) -> None: self._reader = reader def send(self, op: str, **fields: object) -> None: - self._sock.sendall(encode_request(op, **fields)) + try: + self._sock.sendall(encode_request(op, **fields)) + except OSError as exc: + raise ProtocolError(f"write failed: {exc}") from exc def settimeout(self, timeout: float | None) -> None: self._sock.settimeout(timeout) @@ -165,7 +168,7 @@ def dial_agent( try: sock = socket.create_connection((host, port), timeout=remaining_timeout(timeout, deadline)) except OSError as exc: - raise ProtocolError(f"dial {addr}: {exc}") from exc + raise _handshake_error(f"dial {addr}", exc) from exc sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1) reader = None try: @@ -176,7 +179,7 @@ def dial_agent( try: sock.do_handshake() except OSError as exc: - raise ProtocolError(f"tls handshake {addr}: {exc}") from exc + raise _handshake_error(f"tls handshake {addr}", exc) from exc request = ( f"GET /v1/sandboxes/{sandbox_id}/agent HTTP/1.1\r\n" f"Host: {endpoint.netloc}\r\n" @@ -213,10 +216,12 @@ def dial_agent( remaining_timeout(timeout, deadline) sock.settimeout(None) return Conn(sock, reader) - except Exception: + except Exception as exc: if reader is not None: reader.close() sock.close() + if isinstance(exc, OSError) and not isinstance(exc, SandboxError): + raise _handshake_error(f"agent upgrade {addr}", exc) from exc raise @@ -228,3 +233,9 @@ def remaining_timeout(timeout: float, deadline: float | None, what: str = "agent if remaining <= 0: raise SandboxTimeout(f"{what} timed out") return min(timeout, remaining) + + +def _handshake_error(what: str, exc: OSError) -> SandboxError: + if isinstance(exc, (TimeoutError, socket.timeout)): + return SandboxTimeout(f"{what} timed out") + return ProtocolError(f"{what}: {exc}") diff --git a/sdk/python/cocoonsandbox/sandbox.py b/sdk/python/cocoonsandbox/sandbox.py index 3e9c8e19..29ea2434 100644 --- a/sdk/python/cocoonsandbox/sandbox.py +++ b/sdk/python/cocoonsandbox/sandbox.py @@ -6,6 +6,7 @@ import socket import threading import time +from collections import deque from collections.abc import Callable, Iterator from typing import TYPE_CHECKING, Any, cast @@ -18,6 +19,9 @@ if TYPE_CHECKING: from .client import Client +_ACCEPT_POLL_SECONDS = 1.0 +_KILL_WAIT_SECONDS = 5.0 + class Sandbox: """One claimed microVM.""" @@ -93,11 +97,12 @@ def run( on_stderr: Callable[[bytes], object] | None = None, timeout: float | None = None, ) -> int: - """Streams raw output bytes and returns the exit code; timeout bounds the entire call.""" + """Streams raw output bytes and returns the exit code; timeout bounds the call and kills what it cuts.""" if timeout is not None and timeout <= 0: raise ValueError("timeout must be positive") deadline = None if timeout is None else time.monotonic() + timeout expired = threading.Event() + started: deque[int] = deque(maxlen=1) try: conn = self._connect(deadline) except (ProtocolError, TimeoutError): @@ -105,7 +110,7 @@ def run( raise SandboxTimeout(f"command did not finish within {timeout}s") from None raise with self._lease(conn): - watchdog = _arm_watchdog(conn, deadline, expired) + watchdog = None if deadline is None else _arm_watchdog(conn, deadline, expired) pump = threading.Thread(target=_feed_stdin, args=(conn, stdin), daemon=True) if stdin else None try: conn.send( @@ -120,11 +125,13 @@ def run( if pump is not None: pump.start() else: - with contextlib.suppress(OSError): + with contextlib.suppress(ProtocolError): conn.send("stdin_close") - code = _pump_stdio(conn, on_stdout, on_stderr) + code = _pump_stdio(conn, on_stdout, on_stderr, started.append) except (ProtocolError, OSError): if expired.is_set(): + if started: + self._kill_after_cut(started[0]) raise SandboxTimeout(f"command did not finish within {timeout}s") from None raise except SilkdError: @@ -419,16 +426,24 @@ def _open_stream(self, op: str, expect: str = "ready", **fields: object) -> tupl try: conn.send(op, **fields) frame = _expect(conn, expect) + except SilkdError: + self._park(conn) + raise except BaseException: conn.close() raise return conn, frame def _proxy_accept_loop(self, listener: socket.socket, port: int) -> None: - with contextlib.suppress(OSError): - while True: + listener.settimeout(_ACCEPT_POLL_SECONDS) + while True: + try: local, _ = listener.accept() - threading.Thread(target=self._proxy_conn, args=(local, port), daemon=True).start() + except socket.timeout: + continue + except OSError: + return + threading.Thread(target=self._proxy_conn, args=(local, port), daemon=True).start() def _proxy_conn(self, local: socket.socket, port: int) -> None: try: @@ -450,7 +465,7 @@ def pump_out() -> None: pump = threading.Thread(target=pump_out, daemon=True) pump.start() try: - with contextlib.suppress(OSError): + with contextlib.suppress(SandboxError, OSError): while True: chunk = local.recv(FS_CHUNK) if not chunk: @@ -487,6 +502,16 @@ def _drain_proc( _expect(conn, "done") return code + def _kill_after_cut(self, pid: int) -> None: + deadline = time.monotonic() + _KILL_WAIT_SECONDS + with contextlib.suppress(SandboxError, OSError), self._lease(self._connect(deadline)) as conn: + watchdog = _arm_watchdog(conn, deadline, threading.Event()) + try: + conn.send("kill", pid=pid) + _expect(conn, "done") + finally: + watchdog.cancel() + class Session(_Closeable): """A persistent shell inside the sandbox, addressed by id.""" @@ -612,10 +637,7 @@ def _send_chunks(conn: Conn, data: bytes, op: str = "data", chunk: int = FS_CHUN conn.send(op, data=view[off : off + chunk]) -def _arm_watchdog(conn: Conn, deadline: float | None, expired: threading.Event) -> threading.Timer | None: - if deadline is None: - return None - +def _arm_watchdog(conn: Conn, deadline: float, expired: threading.Event) -> threading.Timer: def cut() -> None: expired.set() conn.abort() @@ -634,16 +656,21 @@ def _feed_stdin(conn: Conn, stdin: bytes) -> None: def _pump_stdio( - conn: Conn, on_stdout: Callable[[bytes], object] | None, on_stderr: Callable[[bytes], object] | None + conn: Conn, + on_stdout: Callable[[bytes], object] | None, + on_stderr: Callable[[bytes], object] | None, + on_started: Callable[[int], object] | None = None, ) -> int | None: for frame in conn.recv_until("exit", "done"): t = frame["type"] - if t == "stdout" and on_stdout: - on_stdout(frame["data"]) + if t == "started" and on_started: + on_started(cast(int, _need(frame, "pid"))) + elif t == "stdout" and on_stdout: + on_stdout(_need(frame, "data")) elif t == "stderr" and on_stderr: - on_stderr(frame["data"]) + on_stderr(_need(frame, "data")) elif t == "exit": - return cast(int, frame["code"]) + return cast(int, _need(frame, "code")) return None @@ -672,5 +699,5 @@ def _drain_data(conn: Conn) -> bytes: chunks = [] for frame in conn.recv_until("done"): if frame["type"] == "data": - chunks.append(frame["data"]) + chunks.append(_need(frame, "data")) return b"".join(chunks) diff --git a/sdk/python/cocoonsandbox/template.py b/sdk/python/cocoonsandbox/template.py index 7573ce29..a7261c54 100644 --- a/sdk/python/cocoonsandbox/template.py +++ b/sdk/python/cocoonsandbox/template.py @@ -14,7 +14,7 @@ class Template: """A promoted template on its owner node.""" - def __init__(self, client: Client, addr: str, name: str, net: str, size: str, content_digest: str = ""): + def __init__(self, client: Client, addr: str, name: str, net: str, size: str, content_digest: str = "") -> None: self._client = client self._addr = addr self.name = name diff --git a/sdk/python/tests/test_endpoint.py b/sdk/python/tests/test_endpoint.py index 1a20ffad..89869730 100644 --- a/sdk/python/tests/test_endpoint.py +++ b/sdk/python/tests/test_endpoint.py @@ -3,7 +3,7 @@ import pytest -from cocoonsandbox import Client, ProtocolError +from cocoonsandbox import Client, SandboxTimeout from cocoonsandbox.conn import dial_agent from cocoonsandbox.endpoint import _endpoint_url @@ -58,7 +58,7 @@ def stall() -> None: thread = threading.Thread(target=stall, daemon=True) thread.start() - with pytest.raises(ProtocolError): + with pytest.raises(SandboxTimeout): dial_agent(f"https://127.0.0.1:{server.getsockname()[1]}", "sb_1", "token", 0.05) assert done.wait(5), "timed-out handshake kept its socket open" thread.join() diff --git a/sdk/python/tests/test_hardening.py b/sdk/python/tests/test_hardening.py index 7d280b4b..c2a97b2f 100644 --- a/sdk/python/tests/test_hardening.py +++ b/sdk/python/tests/test_hardening.py @@ -26,6 +26,17 @@ def test_dial_agent_wraps_refused_connection(dead_addr): dial_agent(dead_addr, "sb_1", "tok", 0.5) +def test_send_on_a_dead_socket_raises_a_typed_error(): + ours, theirs = socket.socketpair() + theirs.close() + conn = Conn(ours, ours.makefile("rb")) + try: + with pytest.raises(ProtocolError): + conn.send("info") + finally: + conn.close() + + def test_watcher_propagates_silkd_error(): client_sock, guest_sock = socket.socketpair() guest_sock.sendall(json.dumps({"type": "error", "kind": "not_found", "message": "gone"}).encode() + b"\n") diff --git a/sdk/python/tests/test_keepalive.py b/sdk/python/tests/test_keepalive.py index 8e57fd4c..37e53e6f 100644 --- a/sdk/python/tests/test_keepalive.py +++ b/sdk/python/tests/test_keepalive.py @@ -10,6 +10,7 @@ import pytest from conftest import accept_upgrade, sandbox_at, wait_until +from cocoonsandbox import SilkdError from cocoonsandbox.frames import KEEP_ALIVE_PROTO INPUT_OPS = ("stdin", "stdin_close", "data", "data_end") @@ -28,6 +29,19 @@ def test_calls_share_one_connection(): assert agent.upgrades == 1 +def test_a_refused_stream_open_keeps_its_connection(): + agent = FakeAgent() + sb = sandbox_at(agent.addr) + try: + for _ in range(3): + with pytest.raises(SilkdError): + sb.watch("/missing") + finally: + sb._pool.drain() + agent.stop() + assert agent.upgrades == 1 + + def test_old_daemon_dials_per_call(): agent = FakeAgent(proto=1) sb = sandbox_at(agent.addr) diff --git a/sdk/python/tests/test_proc.py b/sdk/python/tests/test_proc.py index 440d99b9..ad5f7eaa 100644 --- a/sdk/python/tests/test_proc.py +++ b/sdk/python/tests/test_proc.py @@ -3,8 +3,10 @@ import threading +import pytest from conftest import sandbox_at +from cocoonsandbox import ProtocolError from cocoonsandbox.frames import FS_CHUNK @@ -43,6 +45,13 @@ def test_attach_returns_exit_code(monkeypatch): assert out == [b"late"] +@pytest.mark.parametrize("frame", [{"type": "stdout"}, {"type": "stderr"}, {"type": "exit"}]) +def test_attach_reports_a_frame_without_its_field_as_a_protocol_error(monkeypatch, frame): + sb, _ = fake_sandbox(monkeypatch, [frame]) + with pytest.raises(ProtocolError): + sb.attach(41, on_stdout=bytes, on_stderr=bytes) + + def test_run_pumps_stdin_while_reading_output(monkeypatch): sb = sandbox_at("127.0.0.1:1", keep_alive=0) blocking = BlockingStdinConn([{"type": "exit", "code": 0}], buffer_frames=1) diff --git a/sdk/python/tests/test_proxy.py b/sdk/python/tests/test_proxy.py index 06f7f25b..03ee24cf 100644 --- a/sdk/python/tests/test_proxy.py +++ b/sdk/python/tests/test_proxy.py @@ -3,6 +3,7 @@ can't see it, only running the accept loop does.""" import socket +import threading import time from conftest import sandbox_at @@ -86,3 +87,13 @@ def test_proxy_port_ends_the_local_connection_when_the_guest_port_is_closed(monk c.close() finally: listener.close() + + +def test_proxy_port_retires_its_accept_thread_when_the_listener_closes(): + sb = sandbox_at("127.0.0.1:1", keep_alive=0) + known = set(threading.enumerate()) + listener = sb.proxy_port("127.0.0.1:0", 80) + (accept_loop,) = set(threading.enumerate()) - known + listener.close() + accept_loop.join(5) + assert not accept_loop.is_alive() diff --git a/sdk/python/tests/test_stream.py b/sdk/python/tests/test_stream.py index fb9e966c..042448bf 100644 --- a/sdk/python/tests/test_stream.py +++ b/sdk/python/tests/test_stream.py @@ -1,6 +1,7 @@ """The client timeout bounds the dial and the upgrade, not the frames that follow: a guest stream outlives it.""" +import contextlib import json import socket import threading @@ -9,7 +10,9 @@ import pytest from conftest import accept_upgrade, sandbox_at -from cocoonsandbox import Sandbox +from cocoonsandbox import ProtocolError, Sandbox, SandboxTimeout +from cocoonsandbox import conn as conn_module +from cocoonsandbox import sandbox as sandbox_module TIMEOUT = 0.2 @@ -49,15 +52,34 @@ def serve_port_forward(server: socket.socket, quiet: float, ops: list[str]) -> N conn.close() -def serve_started_then_hang(server: socket.socket, quiet: float) -> None: +def serve_started_then_hang(server: socket.socket, quiet: float, kills: list[dict]) -> None: conn, _ = server.accept() reader = accept_upgrade(conn) reader.readline() conn.sendall(b'{"type":"started","pid":7}\n') + killer, _ = server.accept() + kills.append(json.loads(accept_upgrade(killer).readline())) + killer.sendall(b'{"type":"done"}\n') + killer.close() time.sleep(quiet) conn.close() +def serve_started_then_trickle_the_kill_reply(server: socket.socket, kills: list[dict]) -> None: + conn, _ = server.accept() + reader = accept_upgrade(conn) + reader.readline() + conn.sendall(b'{"type":"started","pid":7}\n') + killer, _ = server.accept() + kills.append(json.loads(accept_upgrade(killer).readline())) + with contextlib.suppress(OSError): + for byte in b'{"type":"done"}\n': + time.sleep(0.75 * TIMEOUT) + killer.sendall(bytes([byte])) + killer.close() + conn.close() + + def serve_silence(server: socket.socket) -> None: conn, _ = server.accept() conn.recv(4096) @@ -80,10 +102,11 @@ def test_port_stream_outlives_the_client_timeout(): assert ops == ["port_forward"] -def test_run_timeout_cuts_a_silent_command(): +def test_run_timeout_cuts_a_silent_command_and_kills_it(): server = socket.create_server(("127.0.0.1", 0)) addr = f"127.0.0.1:{server.getsockname()[1]}" - threading.Thread(target=serve_started_then_hang, args=(server, 5 * TIMEOUT), daemon=True).start() + kills: list[dict] = [] + threading.Thread(target=serve_started_then_hang, args=(server, 5 * TIMEOUT, kills), daemon=True).start() sb = legacy_sandbox(addr) started = time.monotonic() try: @@ -92,6 +115,24 @@ def test_run_timeout_cuts_a_silent_command(): finally: server.close() assert time.monotonic() - started < 3 * TIMEOUT + assert kills == [{"v": 1, "op": "kill", "pid": 7}], kills + + +def test_run_timeout_bounds_the_kill_it_sends(monkeypatch): + monkeypatch.setattr(sandbox_module, "_KILL_WAIT_SECONDS", TIMEOUT) + server = socket.create_server(("127.0.0.1", 0)) + addr = f"127.0.0.1:{server.getsockname()[1]}" + kills: list[dict] = [] + threading.Thread(target=serve_started_then_trickle_the_kill_reply, args=(server, kills), daemon=True).start() + sb = legacy_sandbox(addr) + started = time.monotonic() + try: + with pytest.raises(TimeoutError): + sb.run(["sleep", "9"], timeout=TIMEOUT) + finally: + server.close() + assert time.monotonic() - started < 4 * TIMEOUT + assert kills == [{"v": 1, "op": "kill", "pid": 7}], kills def test_run_timeout_cuts_a_blocked_exec_send(monkeypatch): @@ -116,8 +157,13 @@ def test_dial_is_still_bounded_by_the_client_timeout(): sb = legacy_sandbox(addr) started = time.monotonic() try: - with pytest.raises(OSError): + with pytest.raises(SandboxTimeout): sb.dial_port(5000) finally: server.close() assert time.monotonic() - started < 3 * TIMEOUT + + +@pytest.mark.parametrize(("exc", "want"), [(socket.timeout(), SandboxTimeout), (ConnectionResetError(), ProtocolError)]) +def test_handshake_failures_are_typed(exc, want): + assert isinstance(conn_module._handshake_error("agent upgrade", exc), want)