diff --git a/docs/cluster.md b/docs/cluster.md index aab2b16ad..b288f83c0 100644 --- a/docs/cluster.md +++ b/docs/cluster.md @@ -79,7 +79,7 @@ digest. Nodes gossip only their currently available catalog names: host paths and access lists never leave the node. After config load the set appears on the next gossip tick; later image distribution or removal is detected the same way. The node epoch bumps only when one of the gossiped sets — warm counts, -template hashes, volume names — actually changes. +template hashes, volume names — or the config digest actually changes. A writable name (`writable: true`) still needs its catalog entry — name, access list, and the `writable` flag — declared identically on every node diff --git a/docs/deploy.md b/docs/deploy.md index 1c2cdf136..4d4cb182d 100644 --- a/docs/deploy.md +++ b/docs/deploy.md @@ -1028,7 +1028,7 @@ Effects on what already runs: point at a different variable, but only one that was already set when sandboxd started. -A successful reload writes an `op:"config_reload"` audit record listing what +A reload that changes a setting writes an `op:"config_reload"` audit record listing what changed. Each node reloads its own file; on a cluster, distribute the file and reload every node. diff --git a/docs/egress.md b/docs/egress.md index 38ff153b5..5717bdd82 100644 --- a/docs/egress.md +++ b/docs/egress.md @@ -267,7 +267,8 @@ recorded as a decision of its own before its inner requests. The HTTP and CONNECT doors deny a host longer than 253 bytes (the DNS limit); the audit record clips it. A sandbox records at most 10 `deny` and `rate_limited` decisions per second after a burst of 100, dropping -the rest, so a guest cannot flood the journals. +the rest, so a guest cannot flood the journals. `rate_limited` comes from +[`egress_conn_rate`](deploy.md#configuration). ## Claim env and secrets diff --git a/docs/langchain.md b/docs/langchain.md index 56aeb3c24..2066b3f26 100644 --- a/docs/langchain.md +++ b/docs/langchain.md @@ -23,7 +23,7 @@ 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 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_exec` | run a shell command, with an optional `cwd`, 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 | diff --git a/docs/mcp.md b/docs/mcp.md index d3c09a1ae..ed7d0c70b 100644 --- a/docs/mcp.md +++ b/docs/mcp.md @@ -29,18 +29,18 @@ in the guest, which can hold its reply up to 5 s past the cap. | tool | what it does | |---|---| | `create_sandbox` | claim a microVM and return its id plus deadline; optional `template`, `net` (`none` default, or `egress`), `size` (`small` default, `medium`, `large`, `xlarge`, `2xlarge`) and `ttl_seconds` (0 means one hour); warm claims take milliseconds, nothing renews the deadline | -| `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 | +| `exec` | run a shell command to completion, with an optional `cwd`; 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, with an optional `cwd`, 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 the newest 256 KiB of a tracked process's stdout/stderr (+ 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 | +| `kill` | send `signal` to 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 | -| `checkpoint` | capture full state without stopping; returns a `checkpoint_id` that can be branched repeatedly | +| `checkpoint` | capture full state without stopping, with an optional `name`; returns a `checkpoint_id` that can be branched repeatedly | | `branch_checkpoint` | claim a fresh sandbox from a checkpoint's captured moment; it lives one hour | | `list_checkpoints` / `delete_checkpoint` | checkpoint lifecycle | | `hibernate` | snapshot + stop, freeing memory while keeping id, files and processes; the next call that reaches the guest wakes it | -| `promote` | publish the sandbox as a named template on its node; re-promoting replaces it | +| `promote` | publish the sandbox as the template `template_name` on its node; re-promoting replaces it | | `release` | destroy the sandbox and its files; the session forgets the id, so a second release is rejected as unknown | | `node_info` | warm pools, promoted templates, live claims, drain state, capacity and mesh peers; needs the operator token | diff --git a/docs/sandboxd-api.md b/docs/sandboxd-api.md index 4f6826a1f..286b58252 100644 --- a/docs/sandboxd-api.md +++ b/docs/sandboxd-api.md @@ -2,7 +2,8 @@ All bodies are JSON. Three token kinds: -- **root** — the node-level `api_token`: full access to every endpoint. +- **root** — the node-level `api_token`: every endpoint except the agent relay, exec, + owner and the preview mint, which need the sandbox's own token. - **tenant** — a token the [tenant API](#tenants-v1tenants) added: accepted by the resource-creating verbs (claim, fork, promote, checkpoint create/claim, preview mint) and the tenant-scoped listings/deletes below; everything a @@ -15,7 +16,7 @@ All bodies are JSON. Three token kinds: [`meta_store`](deploy.md#shared-tenant-database), a tenant token the node has not cached answers `503` while the database is unreachable; retry. - **sandbox** — every claimed sandbox carries its own bearer token guarding - the sandbox-scoped endpoints, regardless of the other two. + the sandbox-scoped endpoints; each endpoint says whether root also acts by id. Endpoints below that say "node API token" accept root or tenant unless marked root-only. Errors are `{"error": "message"}` with the status codes @@ -530,7 +531,8 @@ Auth: the sandbox's own token, or the root token by id (operator). Restores a hibernated (or archived) sandbox and leaves it running — waking is 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 +running. A hibernated sandbox retains its existing deadline, unless it is past its lease +and archives at expiry, which starts a fresh lease. An archived sandbox instead uses `archive_delete_after_seconds` as its retention deadline while stored, and waking it starts a fresh lease of the length the claim was granted (the server default when it asked for none). A [failed](deploy.md#dead-vmms-and-host-restarts) sandbox is cold-booted from @@ -866,8 +868,8 @@ resolving — no revocation list. A browser request to the minted URL answers 50 Auth: node API token; body `{"token": "", "name": "..."}` (name optional); the root `api_token` with an empty body `token` acts by id as the operator. Captures the sandbox's full state without stopping it and -answers `200 {"checkpoint": {id, name, sandbox_id, key, tenant?, -created_at}}` — `tenant` records the calling tenant, absent for root. +answers `200 {"checkpoint": {id, name?, sandbox_id, key, tenant?, created_at, +policy_source?, no_egress?, guest_env?}}` — `tenant` records the calling tenant, absent for root. 400 bad body or name, 401 bad api token, 404 unknown id or wrong sandbox token, 409 egress-lane sandbox, one with volumes attached (see [egress](egress.md)), or an archived one (`POST /wake` or an exec wakes it @@ -893,7 +895,7 @@ and answers with a redirect shaped like a warm-miss the claim. See the [placement lifecycle](cluster.md#checkpoints-on-a-cluster). 404 for an unknown checkpoint (locally, and after redirect and heal both -miss), 409 for an egress-lane checkpoint (see [egress](egress.md)), 429 node +miss), 429 node or calling tenant at `max_claims`, the node at `max_running`, the node draining, the disk under `disk_reserve`, or memory over `memory_limit`, 503 when the node's concurrent-heal cap is already full — retryable, and the response carries a @@ -927,7 +929,7 @@ part of the public API; an SDK caller has no reason to call it directly. Auth: node API token. Lists this node's checkpoints, newest first. A tenant sees only its own records; root sees everything. A checkpoint backing an archived claim is that claim's wake image, not a listable record. Answers -`200 {"checkpoints": [{id, name, sandbox_id, key, tenant?, created_at}]}`, the +`200 {"checkpoints": [{id, name?, sandbox_id, key, tenant?, created_at, policy_source?, no_egress?, guest_env?}]}`, the record shape `POST /v1/sandboxes/{id}/checkpoint` returns. ## DELETE /v1/checkpoints/{id} @@ -942,7 +944,7 @@ healed replica does not outlive it — eventual cleanup, not a fleet-wide revocation.** A peer offline during the broadcast keeps its copy until the checkpoint TTL ages it out, so an id-holder can still branch it for that window; [placement lifecycle](cluster.md#delete-is-eventual-not-a-fleet-wide-revocation) -has the bound and why heal requires a nonzero, fleet-matching TTL. A shared +has the bound and why heal requires a nonzero, fleet-matching TTL. The s3 checkpoint store skips the broadcast: every node already resolves every record directly, so there is no replica to chase. `?no_forward=1` marks a delete already arriving from another node's own broadcast, so it is not @@ -1185,8 +1187,8 @@ them). 426 without the upgrade header, 404 unknown sandbox, 409 a failed sandbox claim of a replicating pool that is already open when the lease turns overdue is closed, whether or not it has a replica yet. -A hibernated or archived sandbox wakes on the first frame that opens an RPC, -unless the sandbox's pool sets [`reader`](deploy.md#reading-sleeping-claims-without-waking-them). +A hibernated or archived sandbox wakes when the upgrade request arrives, before the `101`, +unless its last sleep is reader-safe, which the pool's [`reader`](deploy.md#reading-sleeping-claims-without-waking-them) setting provides. With it, the node answers the `101` before it touches the sandbox, reads the first frame, and serves `fs_stat`, `fs_list`, `fs_read` and `fs_pull` from a reader VM, a warm VM of the sandbox's pool that mounts a copy of the sandbox's disk read-only: the sandbox stays diff --git a/docs/sdk-python.md b/docs/sdk-python.md index 4bb3cc38d..44d96ce49 100644 --- a/docs/sdk-python.md +++ b/docs/sdk-python.md @@ -187,7 +187,7 @@ Responses include `volumes_want`, `volumes`, `states`, and `converging`. name the three states. A pending state with `converging=False` needs an explicit retry. `VolumeError`, exported from `cocoonsandbox` and `cocoonsandbox.errors`, retains a complete failure view in `.response`. Complete replies update both -public sets; incomplete errors preserve the last view. Serialize calls on one +public sets (`sb.volumes_want`, `sb.volumes`); incomplete errors preserve the last view. Serialize calls on one handle. See the [replacement tutorial](live-volumes.md) and [API recovery contract](sandboxd-api.md#partial-failure-and-recovery). @@ -387,7 +387,7 @@ split multi-byte sequences) and returns the exit code. `user` de-escalates 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 +`TERM`, the lane's proxy variables and the claim env, 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 diff --git a/docs/sdk.md b/docs/sdk.md index e46db3b5c..3cccb7f96 100644 --- a/docs/sdk.md +++ b/docs/sdk.md @@ -278,7 +278,7 @@ and `.Converging`. `VolumePhase` has `VolumeAttached`, `VolumePending`, and `VolumeDetaching` constants. Python exposes `VolumePhase` and the corresponding `VOLUME_ATTACHED`, `VOLUME_PENDING`, and `VOLUME_DETACHING` constants. Each state can include an `error`. Methods use the client API token on the owning node -and update both public sets from each complete response, including a partial failure. Serialize changes to one handle as with +and update both public sets (`Sandbox.VolumesWant`, `Sandbox.Volumes`) from each complete response, including a partial failure. Serialize changes to one handle as with `Renew`. Returned views do not share the handle's slices or dictionaries. Go returns `*sandbox.VolumeError` with a `Response` alongside the returned view. @@ -329,7 +329,8 @@ err = sb.SetEnv(ctx, nil) // replaces the whole env; nil clears it `WithEnv` sets the claim's own env under the [claim rules](sandboxd-api.md#post-v1claim). An entry with a nil `Guest` is delivered into the guest; `Guest: new(false)` keeps it host-side, where only the node's egress reads it (see -[secrets](egress.md#claim-env-and-secrets)). `Env`, `SetEnv` and `PatchEnv` +[secrets](egress.md#claim-env-and-secrets)), and `Inject: &sandbox.EnvInject{Hosts: …}` +has egress send it to those hosts. `Env`, `SetEnv` and `PatchEnv` call [the env verbs](sandboxd-api.md#getputpatch-v1sandboxesidenv) on the claim's owner with the client's API token, not the sandbox token. `PatchEnv` sets each entry, removes each nil one and keeps the rest as stored, @@ -547,7 +548,7 @@ the guest), `Session` (run inside a persistent session, below), `Stdin` 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 +environment is `PATH`, `TERM`, the lane's proxy variables and the claim env, 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 @@ -716,10 +717,13 @@ info, _ = client.SetPools(ctx, pools) // retune warm targets without a resta res, _ := client.SetPoolsCluster(ctx, pools) // per-node results; retry the failures sb = client.Attach(ownerAddr, id, token) // bind a known handle, no lookup round-trip -err = client.PutTenant(ctx, sandbox.TenantSpec{Name: "u:42", Token: tok, MaxClaims: 10}) +err = client.PutTenant(ctx, sandbox.TenantSpec{Name: "u:42", Token: tok, MaxClaims: 10, EgressClass: "desk"}) +err = client.DeleteTenant(ctx, "u:42") res, _ = client.PutTenantCluster(ctx, sandbox.TenantSpec{Name: "u:42", MaxClaims: 20}) // token kept res, _ = client.DeleteTenantCluster(ctx, "u:42") page, _ := client.Tenants(ctx, "", 0) // page.Tenants: names, caps, live claims, removed tenants; never tokens; page.Next pages +page, _ = client.SetTenants(ctx, specs) // replace the whole set +res, _ = client.SetTenantsCluster(ctx, specs) ``` `Sandboxes` is scoped to the calling token, so a tenant sees only its own diff --git a/docs/security.md b/docs/security.md index 07e6f5666..8387d635c 100644 --- a/docs/security.md +++ b/docs/security.md @@ -162,7 +162,7 @@ Facts to plan around, stated so the boundary is honest: `meta_store.usage` the usage journal also lands in the [shared usage table](deploy.md#shared-usage-table); the audit journal stays local. -- **Rate limiting is capacity-based only** (`max_claims`, `max_running`, +- **API rate limiting is capacity-based only** (`max_claims`, `max_running`, `disk_reserve`, `memory_limit` and per-tenant caps answer 429). The API assumes callers inside the trust boundary; front it with your own limiter if semi-trusted automation can reach it. - **A dead node's running sandboxes die with it** unless their pool diff --git a/e2e/tls_client.py b/e2e/tls_client.py index 0fb3e5c4c..81fd34dcf 100644 --- a/e2e/tls_client.py +++ b/e2e/tls_client.py @@ -1,5 +1,6 @@ -"""Exercise both SDK planes through the TLS cluster fixture.""" +"""Exercise both SDK planes through the TLS cluster fixture. Usage: tls_client.py ENTRY OWNER CA""" +import argparse import socket import ssl import sys @@ -8,8 +9,12 @@ def main() -> None: - entry, owner, ca = sys.argv[1:] - context = ssl.create_default_context(cafile=ca) + parser = argparse.ArgumentParser(description=__doc__) + for name in ("entry", "owner", "ca"): + parser.add_argument(name) + args = parser.parse_args() + entry, owner = args.entry, args.owner + context = ssl.create_default_context(cafile=args.ca) client = Client(entry, api_token="node-token", ssl_context=context) with client.new("rt:24.04") as sb: assert sb.owner == owner, sb.owner @@ -49,4 +54,4 @@ def main() -> None: if __name__ == "__main__": - main() + sys.exit(main()) diff --git a/mcp/server.go b/mcp/server.go index 089db43f5..d30b2b5f9 100644 --- a/mcp/server.go +++ b/mcp/server.go @@ -56,11 +56,7 @@ type server struct { } func newServer(addr, token, template string) (*server, error) { - var opts []sandbox.ClientOption - if token != "" { - opts = append(opts, sandbox.WithAPIToken(token)) - } - client, err := sandbox.Connect(addr, opts...) + client, err := sandbox.Connect(addr, sandbox.WithAPIToken(token)) if err != nil { return nil, err } diff --git a/mcp/tools.go b/mcp/tools.go index 209682d7b..93d20954b 100644 --- a/mcp/tools.go +++ b/mcp/tools.go @@ -111,14 +111,7 @@ func toolCreateSandbox(ctx context.Context, s *server, raw jsontext.Value) (stri } // An agent session outlives the node's 5-minute default, and nothing renews a lease ttl := cmp.Or(time.Duration(args.TTLSeconds)*time.Second, defaultToolTTL) - opts := []sandbox.Option{sandbox.WithTimeout(ttl)} - if args.Net != "" { - opts = append(opts, sandbox.WithNetwork(sandbox.NetShape(args.Net))) - } - if args.Size != "" { - opts = append(opts, sandbox.WithSize(sandbox.Size(args.Size))) - } - sb, err := s.client.New(ctx, cmp.Or(args.Template, s.template), opts...) + sb, err := s.client.New(ctx, cmp.Or(args.Template, s.template), sandbox.WithTimeout(ttl), sandbox.WithNetwork(sandbox.NetShape(args.Net)), sandbox.WithSize(sandbox.Size(args.Size))) if err != nil { return "", err } @@ -472,10 +465,7 @@ func unmarshalInput(raw []byte, v any) error { } func jsonText(v any) string { - out, err := json.Marshal(v, jsontext.AllowInvalidUTF8(true), json.Deterministic(true)) - if err != nil { - return fmt.Sprintf("%+v", v) - } + out, _ := json.Marshal(v, jsontext.AllowInvalidUTF8(true), json.Deterministic(true)) return string(out) } diff --git a/protocol/wire/frame.go b/protocol/wire/frame.go index f06d8bd2e..73f3c741e 100644 --- a/protocol/wire/frame.go +++ b/protocol/wire/frame.go @@ -1,9 +1,4 @@ -// Package wire is the Go binding of the silkd wire protocol, shared by the -// SDK and sandboxd: newline-delimited JSON frames, RPCs back to back on one -// connection, requests tagged by "op", responses by "type", binary payloads -// base64 in data fields. The authoritative contract is the shared corpus in -// protocol/wire/fixtures/v1 — silkd's Rust tests and this package's tests -// round-trip the same files. +// Package wire is the Go binding of the silkd wire protocol; the corpus in protocol/wire/fixtures/v1 is the authoritative contract. package wire import ( @@ -18,8 +13,7 @@ import ( ) const ( - // ProtoVersion is stamped into every request as "v"; silkd ignores - // unknown fields, which is the forward-compatibility story. + // ProtoVersion is stamped into every request as "v"; silkd ignores unknown fields. ProtoVersion = 1 // KeepAliveProto is the InfoResp.Proto from which silkd serves RPCs back to back on one connection. KeepAliveProto = 2 @@ -128,11 +122,11 @@ var ( } // bulkResponses decode their canonical shape without json.Unmarshal; any other shape takes responseDecoders. - bulkResponses = bulkDecoders( - func(d []byte) Response { return &Stdout{Data: d} }, - func(d []byte) Response { return &Stderr{Data: d} }, - func(d []byte) Response { return &DataResp{Data: d} }, - ) + bulkResponses = map[string]bulkResponse{ + Stdout{}.RespType(): {[]byte(`{"type":"stdout","data":"`), func(d []byte) Response { return &Stdout{Data: d} }}, + Stderr{}.RespType(): {[]byte(`{"type":"stderr","data":"`), func(d []byte) Response { return &Stderr{Data: d} }}, + DataResp{}.RespType(): {[]byte(`{"type":"data","data":"`), func(d []byte) Response { return &DataResp{Data: d} }}, + } dataRespHead = bulkResponses["data"].head ) @@ -145,11 +139,9 @@ type Response interface{ RespType() string } type respDecoder func([]byte) (Response, error) -type bulkMaker func([]byte) Response - type bulkResponse struct { head []byte - mk bulkMaker + mk func([]byte) Response } // reqPtr and respPtr turn a mistyped decoder table entry into a compile error. @@ -240,8 +232,7 @@ type StdinClose struct{} func (StdinClose) Op() string { return "stdin_close" } -// FsWrite streams Data frames into a file, atomically; nil Mode inherits or -// defaults. +// FsWrite streams Data frames into a file, atomically; nil Mode inherits or defaults. type FsWrite struct { Path string `json:"path"` Mode *uint32 `json:"mode,omitzero"` @@ -308,8 +299,7 @@ type FsPull struct { func (FsPull) Op() string { return "fs_pull" } -// FsFind streams Match frames for lines under Path matching Pattern; Glob -// narrows the walk to file names matching it (`*` and `?` wildcards). +// FsFind streams Match frames for lines under Path matching Pattern; Glob narrows the walk to file names matching it (`*` and `?` wildcards). type FsFind struct { Path string `json:"path"` Pattern string `json:"pattern"` @@ -318,8 +308,7 @@ type FsFind struct { func (FsFind) Op() string { return "fs_find" } -// FsReplace rewrites Pattern to Replacement in each file, streaming one -// Replaced frame per file then Done. +// FsReplace rewrites Pattern to Replacement in each file, streaming one Replaced frame per file then Done. type FsReplace struct { Files []string `json:"files"` Pattern string `json:"pattern"` @@ -336,8 +325,7 @@ type FsWatch struct { func (FsWatch) Op() string { return "fs_watch" } -// PtyOpen runs the guest shell under a pseudo-terminal: Started, then Stdout -// frames out and Stdin frames in, until the shell exits (Exit). +// PtyOpen runs the guest shell under a pseudo-terminal: Started, then Stdout frames out and Stdin frames in, until the shell exits (Exit). type PtyOpen struct { Cols uint16 `json:"cols"` Rows uint16 `json:"rows"` @@ -427,8 +415,7 @@ type GitBranch struct { func (GitBranch) Op() string { return "git_branch" } -// LspStart spawns the flavor image's language server for Language, -// rooted at Root. +// LspStart spawns the flavor image's language server for Language, rooted at Root. type LspStart struct { Language string `json:"language"` Root string `json:"root,omitempty"` @@ -483,8 +470,7 @@ type Stderr struct { func (Stderr) RespType() string { return "stderr" } -// Exit is the terminal frame of a foreground exec; -1 means killed or -// unknown. +// Exit is the terminal frame of a foreground exec; -1 means killed or unknown. type Exit struct { Code int32 `json:"code"` } @@ -496,8 +482,7 @@ type Done struct{} func (Done) RespType() string { return "done" } -// Ready acknowledges an armed watch (events after it are guaranteed -// captured) or a connected port_forward. +// Ready acknowledges an armed watch (events after it are guaranteed captured) or a connected port_forward. type Ready struct{} func (Ready) RespType() string { return "ready" } @@ -678,8 +663,7 @@ func DecodeRequest(line []byte) (Request, error) { return dec(line) } -// DecodeResponse parses one frame into its type's concrete Go type. Byte -// fields are freshly allocated per frame, so callers may retain them. +// DecodeResponse parses one frame into its type's concrete Go type; byte fields are freshly allocated per frame, so callers may retain them. func DecodeResponse(line []byte) (Response, error) { return DecodeResponseInto(line, nil) } @@ -734,26 +718,13 @@ func RequestOp(line []byte) string { // IsContinuation reports whether a request line feeds the RPC in flight instead of opening one. func IsContinuation(line []byte) bool { - op, err := frameTag(line, reqTagHead, "op") - if err != nil { - return false - } - switch string(op) { + switch RequestOp(line) { case Data{}.Op(), DataEnd{}.Op(), Stdin{}.Op(), StdinClose{}.Op(): return true } return false } -func bulkDecoders(mks ...bulkMaker) map[string]bulkResponse { - out := make(map[string]bulkResponse, len(mks)) - for _, mk := range mks { - tag := mk(nil).RespType() - out[tag] = bulkResponse{head: []byte(`{"type":"` + tag + `","data":"`), mk: mk} - } - return out -} - // decodeBulk decodes a canonical bulk frame's payload into scratch, or a fresh buffer when scratch is nil; ok is false for any other shape. func decodeBulk(line, head []byte, scratch *[]byte) ([]byte, bool) { after, ok := bytes.CutPrefix(line, head) diff --git a/sandboxd/archives/pg/pg.go b/sandboxd/archives/pg/pg.go index b8c3fa258..04afb9b2b 100644 --- a/sandboxd/archives/pg/pg.go +++ b/sandboxd/archives/pg/pg.go @@ -9,6 +9,7 @@ import ( "time" "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgtype" "github.com/jackc/pgx/v5/pgxpool" "github.com/cocoonstack/sandbox/sandboxd/archives" @@ -59,7 +60,7 @@ func (r *Registry) Put(ctx context.Context, rec archives.Record) (int64, error) deadline = EXCLUDED.deadline, claim = EXCLUDED.claim, updated_at = now() WHERE sandboxd_archives.owner = EXCLUDED.owner RETURNING epoch`, - rec.ID, rec.Tenant, rec.TokenSHA256, rec.Owner, rec.Checkpoint, nullTime(rec.Deadline), []byte(rec.Claim), r.cell) + rec.ID, rec.Tenant, rec.TokenSHA256, rec.Owner, rec.Checkpoint, pgtype.Timestamptz{Time: rec.Deadline, Valid: !rec.Deadline.IsZero()}, []byte(rec.Claim), r.cell) } func (r *Registry) Replace(ctx context.Context, rec archives.Record, expect int64) (int64, error) { @@ -69,7 +70,7 @@ func (r *Registry) Replace(ctx context.Context, rec archives.Record, expect int6 return r.advance(ctx, `UPDATE sandboxd_archives SET tenant = $5, token_sha256 = $6, epoch = nextval('sandboxd_archives_epoch'), checkpoint = $7, deadline = $8, claim = $9, updated_at = now() WHERE cell = $1 AND id = $2 AND epoch = $3 AND owner = $4 AND NOT consumed - RETURNING epoch`, r.cell, rec.ID, expect, rec.Owner, rec.Tenant, rec.TokenSHA256, rec.Checkpoint, nullTime(rec.Deadline), []byte(rec.Claim)) + RETURNING epoch`, r.cell, rec.ID, expect, rec.Owner, rec.Tenant, rec.TokenSHA256, rec.Checkpoint, pgtype.Timestamptz{Time: rec.Deadline, Valid: !rec.Deadline.IsZero()}, []byte(rec.Claim)) } func (r *Registry) Take(ctx context.Context, id string, expect int64, owner string) (int64, error) { @@ -131,7 +132,7 @@ func (r *Registry) Get(ctx context.Context, id string) (archives.Record, bool, e case errors.Is(err, pgx.ErrNoRows): return archives.Record{}, false, nil case err != nil: - return archives.Record{}, false, err + return archives.Record{}, false, unavailable(err) } return rec, true, nil } @@ -202,30 +203,16 @@ func (r *Registry) collect(ctx context.Context, sql string, args ...any) ([]arch func scan(row pgx.Row) (archives.Record, error) { var ( rec archives.Record - deadline *time.Time + deadline pgtype.Timestamptz claim []byte ) - err := row.Scan(&rec.ID, &rec.Tenant, &rec.TokenSHA256, &rec.Owner, &rec.Epoch, &rec.Consumed, &rec.Checkpoint, &deadline, &claim) - switch { - case errors.Is(err, pgx.ErrNoRows): + if err := row.Scan(&rec.ID, &rec.Tenant, &rec.TokenSHA256, &rec.Owner, &rec.Epoch, &rec.Consumed, &rec.Checkpoint, &deadline, &claim); err != nil { return archives.Record{}, err - case err != nil: - return archives.Record{}, unavailable(err) - } - if deadline != nil { - rec.Deadline = *deadline } - rec.Claim = claim + rec.Deadline, rec.Claim = deadline.Time, claim return rec, nil } -func nullTime(t time.Time) *time.Time { - if t.IsZero() { - return nil - } - return &t -} - func unavailable(err error) error { return fmt.Errorf("%w: %w", archives.ErrUnavailable, err) } diff --git a/sandboxd/backends.go b/sandboxd/backends.go index 235c2efb5..a2c1cbc7c 100644 --- a/sandboxd/backends.go +++ b/sandboxd/backends.go @@ -23,7 +23,6 @@ import ( func openBackends(ctx context.Context, cfg *config.Config) (pool.Backends, error) { var b pool.Backends var err error - // Checkpoints and templates are two id-namespaced views (ck_*, tp_*) over one store root if b.Checkpoints, err = newStoreView(ctx, cfg, "checkpoint-staging", store.CheckpointIDRe); err != nil { return pool.Backends{}, err } diff --git a/sandboxd/config/config.go b/sandboxd/config/config.go index e9f495f09..e06e784c6 100644 --- a/sandboxd/config/config.go +++ b/sandboxd/config/config.go @@ -206,7 +206,7 @@ func (mc *MeshConfig) ParsedBind() (string, int, error) { if err != nil { return "", 0, fmt.Errorf("mesh bind port %q: %w", portStr, err) } - if host == "" { + if !RoutableHost(host) { return "", 0, fmt.Errorf("mesh bind needs an explicit host (got %q); a wildcard advertises an unroutable address", mc.Bind) } return host, port, nil diff --git a/sandboxd/config/config_test.go b/sandboxd/config/config_test.go index cf019d271..9670d1dcc 100644 --- a/sandboxd/config/config_test.go +++ b/sandboxd/config/config_test.go @@ -142,6 +142,8 @@ func TestLoadRejectsInvalid(t *testing.T) { {"mesh with unspecified advertise", `{"advertise_addr":"0.0.0.0:7777","pools":[],"mesh":{"bind":"node1:7946"}}`, "routable host"}, {"mesh bind missing port", `{"pools":[],"mesh":{"bind":"node1"}}`, "mesh bind"}, {"mesh bind wildcard host", `{"pools":[],"mesh":{"bind":":7946"}}`, "explicit host"}, + {"mesh bind unspecified ipv4", `{"pools":[],"mesh":{"bind":"0.0.0.0:7946"}}`, "explicit host"}, + {"mesh bind unspecified ipv6", `{"pools":[],"mesh":{"bind":"[::]:7946"}}`, "explicit host"}, {"mesh cluster key not base64", `{"pools":[],"mesh":{"bind":"node1:7946","cluster_key":"not!base64"}}`, "not valid base64"}, {"mesh cluster key wrong length", `{"pools":[],"mesh":{"bind":"node1:7946","cluster_key":"YWJj"}}`, "want 16, 24, or 32"}, {"checkpoint peer heal without mesh", `{"checkpoint_peer_heal":true,"pools":[]}`, "requires an encrypted mesh"}, diff --git a/sandboxd/config/reload.go b/sandboxd/config/reload.go index a1d3787b9..6919c8bdb 100644 --- a/sandboxd/config/reload.go +++ b/sandboxd/config/reload.go @@ -32,10 +32,6 @@ var reloadableFields = map[string]bool{ "archive_delete_after_seconds": true, } -type poolTargets struct { - warm, warmMax, idle, archiveAfter, archiveDelete int -} - // ReloadDiff sorts next's differences from c into reloadable settings it changes and parts their own API owns, which a reload leaves alone; any other difference needs a restart and fails. func (c *Config) ReloadDiff(next *Config) (changed, ignored []string, err error) { cur, nxt := reflect.ValueOf(c).Elem(), reflect.ValueOf(next).Elem() @@ -91,8 +87,8 @@ func diffPools(cur, next []PoolSpec) (changed, ignored, restart []string) { return changed, ignored, restart } -func targetsOf(s PoolSpec) poolTargets { - return poolTargets{s.Warm, s.WarmMax, s.IdleHibernateSeconds, s.ArchiveAfterSeconds, s.ArchiveDeleteAfterSeconds} +func targetsOf(s PoolSpec) [5]int { + return [5]int{s.Warm, s.WarmMax, s.IdleHibernateSeconds, s.ArchiveAfterSeconds, s.ArchiveDeleteAfterSeconds} } func poolLabel(k types.PoolKey) string { diff --git a/sandboxd/egress/ca_issue.go b/sandboxd/egress/ca_issue.go index b0378782a..32f82f22e 100644 --- a/sandboxd/egress/ca_issue.go +++ b/sandboxd/egress/ca_issue.go @@ -22,20 +22,10 @@ func GenerateRoot(commonName string) (certPEM, keyPEM []byte, err error) { if err != nil { return nil, nil, fmt.Errorf("generate root key: %w", err) } - serial, err := serialNumber() + tmpl, err := caTemplate(commonName, rootValidity) if err != nil { return nil, nil, err } - now := time.Now() - tmpl := &x509.Certificate{ - SerialNumber: serial, - Subject: pkix.Name{CommonName: commonName}, - NotBefore: now.Add(-clockSkew), - NotAfter: now.Add(rootValidity), - KeyUsage: x509.KeyUsageCertSign | x509.KeyUsageCRLSign, - BasicConstraintsValid: true, - IsCA: true, - } return marshalCA(tmpl, tmpl, key, key) } @@ -53,22 +43,29 @@ func IssueIntermediate(rootCertPEM, rootKeyPEM []byte, nodeID string) (certPEM, if err != nil { return nil, nil, fmt.Errorf("generate intermediate key: %w", err) } - serial, err := serialNumber() + tmpl, err := caTemplate("sandbox egress intermediate "+nodeID, interValidity) if err != nil { return nil, nil, err } + tmpl.MaxPathLenZero = true + return marshalCA(tmpl, root, rootKey, key) +} + +func caTemplate(commonName string, validity time.Duration) (*x509.Certificate, error) { + serial, err := serialNumber() + if err != nil { + return nil, err + } now := time.Now() - tmpl := &x509.Certificate{ + return &x509.Certificate{ SerialNumber: serial, - Subject: pkix.Name{CommonName: "sandbox egress intermediate " + nodeID}, + Subject: pkix.Name{CommonName: commonName}, NotBefore: now.Add(-clockSkew), - NotAfter: now.Add(interValidity), + NotAfter: now.Add(validity), KeyUsage: x509.KeyUsageCertSign | x509.KeyUsageCRLSign, BasicConstraintsValid: true, IsCA: true, - MaxPathLenZero: true, - } - return marshalCA(tmpl, root, rootKey, key) + }, nil } func marshalCA(tmpl, parent *x509.Certificate, signer, key *ecdsa.PrivateKey) (certPEM, keyPEM []byte, err error) { diff --git a/sandboxd/egress/intercept.go b/sandboxd/egress/intercept.go index 5124fb981..33f1173d3 100644 --- a/sandboxd/egress/intercept.go +++ b/sandboxd/egress/intercept.go @@ -93,8 +93,7 @@ type interceptHandler struct { func (h *interceptHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) { p := h.proxy rule, decision := p.policy.EvalInner(h.host, r.Method, h.port) - ev, route := p.routeEvent(Event{Method: r.Method, Host: h.host, Port: h.port, Decision: decision}) - p.relay(w, r, ev, rule, route, true, func(out *http.Request) { + p.relay(w, r, Event{Method: r.Method, Host: h.host, Port: h.port, Decision: decision}, rule, true, func(out *http.Request) { out.URL.Scheme = "https" out.URL.Host = h.authority // Keep the guest's Host when it names the CONNECT host: SigV4 signing breaks on a rewrite. diff --git a/sandboxd/egress/policy.go b/sandboxd/egress/policy.go index 87624a78f..42e5d54de 100644 --- a/sandboxd/egress/policy.go +++ b/sandboxd/egress/policy.go @@ -200,6 +200,10 @@ type Evaluator interface { Eval(host, method string, port uint16) (Rule, Decision) EvalHost(host string, port uint16) (Rule, Decision) EvalInner(host, method string, port uint16) (Rule, Decision) +} + +type PolicyEvaluator interface { + Evaluator ServesSocks() bool InterceptsHost(pattern string) bool } @@ -209,41 +213,38 @@ type composite struct { } // Compose intersects a pool and a tenant policy; the pool rule wins on a double allow. -func Compose(pool, tenant Policy) Evaluator { +func Compose(pool, tenant Policy) PolicyEvaluator { return composite{pool: pool, tenant: tenant} } func (c composite) Eval(host, method string, port uint16) (Rule, Decision) { rule, pd := c.pool.Eval(host, method, port) - if pd != DecisionAllow { - return Rule{}, DecisionDeny - } - if _, td := c.tenant.Eval(host, method, port); td != DecisionAllow { - return Rule{}, DecisionDeny + if pd == DecisionAllow { + if _, td := c.tenant.Eval(host, method, port); td == DecisionAllow { + return rule, DecisionAllow + } } - return rule, DecisionAllow + return Rule{}, DecisionDeny } func (c composite) EvalHost(host string, port uint16) (Rule, Decision) { rule, pd := c.pool.EvalHost(host, port) - if pd != DecisionAllow { - return Rule{}, DecisionDeny - } - if _, td := c.tenant.EvalHost(host, port); td != DecisionAllow { - return Rule{}, DecisionDeny + if pd == DecisionAllow { + if _, td := c.tenant.EvalHost(host, port); td == DecisionAllow { + return rule, DecisionAllow + } } - return rule, DecisionAllow + return Rule{}, DecisionDeny } func (c composite) EvalInner(host, method string, port uint16) (Rule, Decision) { rule, pd := c.pool.EvalInner(host, method, port) - if pd != DecisionAllow { - return Rule{}, DecisionDeny - } - if _, td := c.tenant.Eval(host, method, port); td != DecisionAllow { - return Rule{}, DecisionDeny + if pd == DecisionAllow { + if _, td := c.tenant.Eval(host, method, port); td == DecisionAllow { + return rule, DecisionAllow + } } - return rule, DecisionAllow + return Rule{}, DecisionDeny } // ServesSocks is the pool's call: the tenant's rules already gate every tunnel through Eval. diff --git a/sandboxd/egress/proxy.go b/sandboxd/egress/proxy.go index 0f6de2976..e9d1e57e7 100644 --- a/sandboxd/egress/proxy.go +++ b/sandboxd/egress/proxy.go @@ -269,7 +269,7 @@ func (p *Proxy) serveConnect(w http.ResponseWriter, r *http.Request) { return } decision, intercept := p.tunnelDecision(host, port, true) - ev, route := p.routeEvent(Event{Method: r.Method, Host: host, Port: port, Decision: decision}) + ev := Event{Method: r.Method, Host: host, Port: port, Decision: decision} if intercept { p.record(ev) p.serveIntercept(w, r, host, port) @@ -280,7 +280,7 @@ func (p *Proxy) serveConnect(w http.ResponseWriter, r *http.Request) { denied(w, host) return } - upstream, err := p.router.Dial(r.Context(), route, "tcp", net.JoinHostPort(host, strconv.Itoa(int(port)))) + upstream, err := p.router.Dial(r.Context(), p.router.Route(), "tcp", net.JoinHostPort(host, strconv.Itoa(int(port)))) ev.attach(upstream, err) p.record(ev) if err != nil { @@ -347,17 +347,17 @@ func (p *Proxy) serveForward(w http.ResponseWriter, r *http.Request) { return } rule, decision := p.policy.Eval(host, r.Method, port) - ev, route := p.routeEvent(Event{Method: r.Method, Host: host, Port: port, Decision: decision}) - p.relay(w, r, ev, rule, route, false, nil) + p.relay(w, r, Event{Method: r.Method, Host: host, Port: port, Decision: decision}, rule, false, nil) } -func (p *Proxy) relay(w http.ResponseWriter, r *http.Request, ev Event, rule Rule, route string, mitm bool, prepare func(*http.Request)) { +func (p *Proxy) relay(w http.ResponseWriter, r *http.Request, ev Event, rule Rule, mitm bool, prepare func(*http.Request)) { if ev.Decision == DecisionDeny { p.record(ev) denied(w, ev.Host) return } + route := p.router.Route() ctx := r.Context() var got *net.Conn if p.ends || route != "" { @@ -455,13 +455,6 @@ func (p *Proxy) transport(route string, mitm bool) *http.Transport { return tr } -func (p *Proxy) routeEvent(ev Event) (Event, string) { - if ev.Decision != DecisionAllow { - return ev, "" - } - return ev, p.router.Route() -} - func (p *Proxy) reportTransfer(ev Event, sent, received int64) { if p.transfer != nil { p.transfer(ev, sent, received) diff --git a/sandboxd/egress/socks.go b/sandboxd/egress/socks.go index 9925effe1..46251eb1c 100644 --- a/sandboxd/egress/socks.go +++ b/sandboxd/egress/socks.go @@ -64,7 +64,7 @@ func (p *Proxy) serveSocks(ctx context.Context, conn net.Conn) { if intercept || port == 0 { decision = DecisionDeny } - ev, route := p.routeEvent(Event{Method: methodSOCKS, Host: host, Port: port, Decision: decision}) + ev := Event{Method: methodSOCKS, Host: host, Port: port, Decision: decision} if decision == DecisionDeny { p.record(ev) _ = socksReply(conn, socksDenied) @@ -72,7 +72,7 @@ func (p *Proxy) serveSocks(ctx context.Context, conn net.Conn) { } dialCtx, cancel := context.WithTimeout(ctx, socksTimeout) defer cancel() - upstream, err := p.router.Dial(dialCtx, route, "tcp", net.JoinHostPort(host, strconv.Itoa(int(port)))) + upstream, err := p.router.Dial(dialCtx, p.router.Route(), "tcp", net.JoinHostPort(host, strconv.Itoa(int(port)))) ev.attach(upstream, err) p.record(ev) if err != nil { diff --git a/sandboxd/engine/cloneargs_test.go b/sandboxd/engine/cloneargs_test.go index 14868be7c..7ea2a8078 100644 --- a/sandboxd/engine/cloneargs_test.go +++ b/sandboxd/engine/cloneargs_test.go @@ -25,8 +25,8 @@ func TestCloneArgsRestoreMode(t *testing.T) { t.Run(tt.name, func(t *testing.T) { e := New("cocoon", []string{"br0"}, nil, false, false, tt.mode) for _, args := range [][]string{ - e.cloneArgs("/goldens/g1", "sbx-1", tt.key), - e.cloneSnapArgs("ck_1", "sbx-1", tt.key), + e.cloneArgs("sbx-1", tt.key, "--from-dir", "/goldens/g1"), + e.cloneArgs("sbx-1", tt.key, "ck_1"), e.restoreCmdArgs("sbx-1", "sbx-hib-1"), } { i := slices.Index(args, "--restore-mode") @@ -50,8 +50,8 @@ func TestLifecycleArgsApplyDirectIOPolicy(t *testing.T) { cold := e.runColdArgs("sbx-1", key, "") for _, args := range [][]string{ cold, - e.cloneArgs("/goldens/g1", "sbx-1", key), - e.cloneSnapArgs("ck_1", "sbx-1", key), + e.cloneArgs("sbx-1", key, "--from-dir", "/goldens/g1"), + e.cloneArgs("sbx-1", key, "ck_1"), } { if !slices.Contains(args, want) { t.Errorf("args %v missing %q", args, want) @@ -77,7 +77,7 @@ func TestEgressVMsSpreadOverEveryConfiguredShard(t *testing.T) { t.Run(name, func(t *testing.T) { counts := map[string]int{} for i := range 4000 { - args := tt.e.cloneArgs("/goldens/g1", "sbx-pool1-"+strconv.Itoa(i), key) + args := tt.e.cloneArgs("sbx-pool1-"+strconv.Itoa(i), key, "--from-dir", "/goldens/g1") j := slices.Index(args, tt.flag) if j < 0 { t.Fatalf("args %v carry no %s", args, tt.flag) @@ -126,7 +126,7 @@ func TestNoBalloonReachesColdBootsOnly(t *testing.T) { if got := slices.Contains(e.runColdArgs("sbx-1", key, ""), "--no-balloon"); got != noBalloon { t.Errorf("cold args carry --no-balloon = %v, want %v", got, noBalloon) } - if slices.Contains(e.cloneArgs("/goldens/g1", "sbx-1", key), "--no-balloon") { + if slices.Contains(e.cloneArgs("sbx-1", key, "--from-dir", "/goldens/g1"), "--no-balloon") { t.Error("clone args carry --no-balloon; clones inherit it from the golden") } }) @@ -145,7 +145,7 @@ func TestColdArgsSizeTheDiskOnlyWhenAsked(t *testing.T) { t.Errorf("sized cold args %v, want --storage %s verbatim", args, storage) } } - if slices.Contains(e.cloneArgs("/goldens/g1", "sbx-1", key), "--storage") { + if slices.Contains(e.cloneArgs("sbx-1", key, "--from-dir", "/goldens/g1"), "--storage") { t.Error("clone args carry --storage; clones inherit the golden's disk") } } diff --git a/sandboxd/engine/engine.go b/sandboxd/engine/engine.go index 15ff99f76..206338949 100644 --- a/sandboxd/engine/engine.go +++ b/sandboxd/engine/engine.go @@ -127,12 +127,12 @@ func (e *Engine) VersionWarning(ctx context.Context) (version, warning string) { // Clone restores a VM from an exported golden directory. func (e *Engine) Clone(ctx context.Context, fromDir, name string, key types.PoolKey) (types.VMRecord, error) { - return e.runRecord(ctx, dumpTimeout, dumpTermGrace, e.cloneArgs(fromDir, name, key)...) + return e.runRecord(ctx, dumpTimeout, dumpTermGrace, e.cloneArgs(name, key, "--from-dir", fromDir)...) } // CloneSnap clones from a local-store snapshot by name — fork's fast path. func (e *Engine) CloneSnap(ctx context.Context, snap, name string, key types.PoolKey) (types.VMRecord, error) { - return e.runRecord(ctx, dumpTimeout, dumpTermGrace, e.cloneSnapArgs(snap, name, key)...) + return e.runRecord(ctx, dumpTimeout, dumpTermGrace, e.cloneArgs(name, key, snap)...) } // RunCold boots a VM from the template image with a storage-sized disk (empty = cocoon's default), returning its lifecycle record. @@ -317,7 +317,7 @@ func (e *Engine) DialGuestPort(ctx context.Context, vsockSocket string, port uin _ = conn.Close() return nil, fmt.Errorf("port_forward %d: %s", port, respFail(resp)) } - return newGuestPortConn(conn, r), nil + return &guestPortConn{Conn: conn, r: r}, nil } func (e *Engine) dialVsock(ctx context.Context, vsockSocket string, port int) (net.Conn, error) { @@ -366,15 +366,10 @@ func (e *Engine) probe(ctx context.Context, vsockSocket string, port int, timeou } } -func (e *Engine) cloneArgs(fromDir, name string, key types.PoolKey) []string { +func (e *Engine) cloneArgs(name string, key types.PoolKey, source ...string) []string { + args := append([]string{"vm", "clone"}, source...) // --pull: a checkpoint/template export carries only COW + memory, not the base image. - args := []string{"vm", "clone", "--from-dir", fromDir, argName, name, "--pull", argOutput, formatJSON, e.directIOArg()} - args = append(args, e.restoreArgs()...) - return append(args, e.netArgs(name, key, false)...) -} - -func (e *Engine) cloneSnapArgs(snap, name string, key types.PoolKey) []string { - args := []string{"vm", "clone", snap, argName, name, "--pull", argOutput, formatJSON, e.directIOArg()} + args = append(args, argName, name, "--pull", argOutput, formatJSON, e.directIOArg()) args = append(args, e.restoreArgs()...) return append(args, e.netArgs(name, key, false)...) } diff --git a/sandboxd/engine/env.go b/sandboxd/engine/env.go index 3fdf1c84c..00d38bb7c 100644 --- a/sandboxd/engine/env.go +++ b/sandboxd/engine/env.go @@ -7,7 +7,5 @@ const guestEnvPath = "/run/silkd.env" // WriteGuestEnv replaces the guest's env file, root-only. func (e *Engine) WriteGuestEnv(ctx context.Context, vsockSocket string, doc []byte) error { - ctx, cancel := context.WithTimeout(ctx, cmdTimeout) - defer cancel() return e.silkdWriteFile(ctx, vsockSocket, guestEnvPath, 0o600, doc) } diff --git a/sandboxd/engine/imds.go b/sandboxd/engine/imds.go index bd858d2ac..ba92639ee 100644 --- a/sandboxd/engine/imds.go +++ b/sandboxd/engine/imds.go @@ -12,8 +12,6 @@ var ErrNoInstanceMetadata = errors.New("image serves no instance metadata") // WriteInstanceMetadata replaces the instance-metadata document silkd serves on the guest's 169.254.169.254. func (e *Engine) WriteInstanceMetadata(ctx context.Context, vsockSocket string, doc []byte) error { - ctx, cancel := context.WithTimeout(ctx, cmdTimeout) - defer cancel() err := e.silkdWriteFile(ctx, vsockSocket, instanceMetadataPath, 0o600, doc) if isNotFound(err) { return ErrNoInstanceMetadata diff --git a/sandboxd/engine/lane.go b/sandboxd/engine/lane.go index 8b99a356a..f581b993e 100644 --- a/sandboxd/engine/lane.go +++ b/sandboxd/engine/lane.go @@ -10,7 +10,5 @@ const laneFile = "/etc/silkd-lane" // MarkLane writes the verdict into the guest, where silkd and the image's units read it; the nft lock is invisible from inside. func (e *Engine) MarkLane(ctx context.Context, vsockSocket string, lane types.NetRoute) error { - ctx, cancel := context.WithTimeout(ctx, cmdTimeout) - defer cancel() return e.silkdWriteFile(ctx, vsockSocket, laneFile, 0o644, []byte(lane+"\n")) } diff --git a/sandboxd/engine/portconn.go b/sandboxd/engine/portconn.go index 6ac222a5e..23c47a3f2 100644 --- a/sandboxd/engine/portconn.go +++ b/sandboxd/engine/portconn.go @@ -24,10 +24,6 @@ type guestPortConn struct { pending []byte } -func newGuestPortConn(conn net.Conn, r *bufio.Reader) *guestPortConn { - return &guestPortConn{Conn: conn, r: r} -} - func (g *guestPortConn) Read(p []byte) (int, error) { for len(g.pending) == 0 { line, err := g.readFrame() diff --git a/sandboxd/engine/portconn_test.go b/sandboxd/engine/portconn_test.go index 6d4936e37..5801621ed 100644 --- a/sandboxd/engine/portconn_test.go +++ b/sandboxd/engine/portconn_test.go @@ -123,7 +123,7 @@ func TestGuestPortConnRoundTripsFixtures(t *testing.T) { client2, silk2 := net.Pipe() t.Cleanup(func() { _ = client2.Close(); _ = silk2.Close() }) - gc2 := newGuestPortConn(client2, bufio.NewReaderSize(client2, portReadBuf)) + gc2 := &guestPortConn{Conn: client2, r: bufio.NewReaderSize(client2, portReadBuf)} respError := fixture("resp_error.json") go func() { _, _ = silk2.Write(respError) }() if _, err := gc2.Read(buf); err == nil || errors.Is(err, io.EOF) { @@ -132,7 +132,7 @@ func TestGuestPortConnRoundTripsFixtures(t *testing.T) { client3, silk3 := net.Pipe() t.Cleanup(func() { _ = client3.Close(); _ = silk3.Close() }) - gc3 := newGuestPortConn(client3, bufio.NewReaderSize(client3, portReadBuf)) + gc3 := &guestPortConn{Conn: client3, r: bufio.NewReaderSize(client3, portReadBuf)} go func() { _, _ = gc3.Write([]byte("hello")) }() line := make([]byte, 256) n, _ = silk3.Read(line) @@ -196,7 +196,7 @@ func newTestGuestPortConn(t *testing.T) (net.Conn, *guestPortConn) { t.Helper() client, silk := net.Pipe() t.Cleanup(func() { _ = client.Close(); _ = silk.Close() }) - return silk, newGuestPortConn(client, bufio.NewReaderSize(client, portForwardMax)) + return silk, &guestPortConn{Conn: client, r: bufio.NewReaderSize(client, portForwardMax)} } func writeDataFrames(t *testing.T, silk net.Conn, chunks ...[]byte) { diff --git a/sandboxd/engine/silkd.go b/sandboxd/engine/silkd.go index 0e5abd729..67534515e 100644 --- a/sandboxd/engine/silkd.go +++ b/sandboxd/engine/silkd.go @@ -31,7 +31,6 @@ func (e *guestExitError) Error() string { return fmt.Sprintf("exit code %d: %s", e.code, e.output) } -// silkdSession is a dialed silkd conn bound to a ctx, with wire-typed request/reply helpers. type silkdSession struct { conn net.Conn sc *bufio.Scanner @@ -104,42 +103,31 @@ func (e *Engine) silkdStream(ctx context.Context, vsockSocket string, req wire.R } } -func (e *Engine) silkdList(ctx context.Context, vsockSocket, path string) ([]wire.DirEntry, error) { - var entries []wire.DirEntry - err := e.silkdStream(ctx, vsockSocket, wire.FsList{Path: path}, func(frame wire.Response) error { - resp, ok := frame.(*wire.Entries) +func (e *Engine) silkdCollect[F wire.Response, E any](ctx context.Context, vsockSocket string, req wire.Request, limit int, over error, items func(F) []E) ([]E, error) { + var out []E + err := e.silkdStream(ctx, vsockSocket, req, func(frame wire.Response) error { + resp, ok := frame.(F) if !ok { return fmt.Errorf("unexpected silkd frame %q", frame.RespType()) } - entries = append(entries, resp.Entries...) - if len(entries) > silkdListMax { - return fmt.Errorf("silkd list of %s exceeds %d entries", path, silkdListMax) + out = append(out, items(resp)...) + if len(out) > limit { + return over } return nil }) if err != nil { return nil, err } - return entries, nil + return out, nil +} + +func (e *Engine) silkdList(ctx context.Context, vsockSocket, path string) ([]wire.DirEntry, error) { + return e.silkdCollect(ctx, vsockSocket, wire.FsList{Path: path}, silkdListMax, fmt.Errorf("silkd list of %s exceeds %d entries", path, silkdListMax), func(r *wire.Entries) []wire.DirEntry { return r.Entries }) } func (e *Engine) silkdReadFile(ctx context.Context, vsockSocket, path string, limit int) ([]byte, error) { - var data []byte - err := e.silkdStream(ctx, vsockSocket, wire.FsRead{Path: path}, func(frame wire.Response) error { - resp, ok := frame.(*wire.DataResp) - if !ok { - return fmt.Errorf("unexpected silkd frame %q", frame.RespType()) - } - data = append(data, resp.Data...) - if len(data) > limit { - return fmt.Errorf("silkd read of %s exceeds %d bytes", path, limit) - } - return nil - }) - if err != nil { - return nil, err - } - return data, nil + return e.silkdCollect(ctx, vsockSocket, wire.FsRead{Path: path}, limit, fmt.Errorf("silkd read of %s exceeds %d bytes", path, limit), func(r *wire.DataResp) []byte { return r.Data }) } // silkdOnce sends a request answered by a single T frame and no Done, so it cannot ride silkdStream. @@ -166,6 +154,8 @@ func (e *Engine) silkdOnce[T wire.Response](ctx context.Context, vsockSocket str } func (e *Engine) silkdWriteFile(ctx context.Context, vsockSocket, path string, mode uint32, data []byte) error { + ctx, cancel := context.WithTimeout(ctx, cmdTimeout) + defer cancel() s, err := e.dialSilkdSession(ctx, vsockSocket) if err != nil { return err diff --git a/sandboxd/main.go b/sandboxd/main.go index 19c3dad85..bb34632bb 100644 --- a/sandboxd/main.go +++ b/sandboxd/main.go @@ -117,14 +117,11 @@ func main() { defer func() { _ = msh.Shutdown() }() placer = msh mgr.SetTemplateNotifier(func() { - msh.UpdateSelf(ctx, mgr.WarmCounts(), mgr.TemplateHashes(), mgr.VolumeNames()) + msh.UpdateSelf(mgr.WarmCounts(), mgr.TemplateHashes(), mgr.VolumeNames()) }) - republish = func() { msh.UpdateDigest(ctx, mgr.ClusterDigest()) } + republish = func() { msh.UpdateDigest(mgr.ClusterDigest()) } mgr.OnTenants(republish) - clusterKey, err := cfg.Mesh.DecodedKey() - if err != nil { - logger.Fatalf(ctx, err, "decode mesh cluster key") - } + clusterKey, _ := cfg.Mesh.DecodedKey() if len(clusterKey) > 0 { probeKey = peer.DeriveProbeKey(clusterKey) } @@ -216,25 +213,19 @@ func servePreview(ctx context.Context, srv *http.Server) { func startMesh(ctx context.Context, cfg *config.Config, mgr *pool.Manager) (*mesh.Mesh, error) { mc := cfg.Mesh mlCfg := memberlist.DefaultLANConfig() - host, port, err := mc.ParsedBind() - if err != nil { - return nil, err - } + host, port, _ := mc.ParsedBind() mlCfg.BindAddr, mlCfg.BindPort = host, port mlCfg.AdvertiseAddr, mlCfg.AdvertisePort = host, port mlCfg.PushPullInterval = gossipInterval mlCfg.LogOutput = io.Discard // memberlist's own logs bypass core/log - key, err := mc.DecodedKey() - if err != nil { - return nil, err - } + key, _ := mc.DecodedKey() self := mesh.NodeState{NodeID: mc.ID(), Addr: cfg.AdvertiseAddr, ClientAddr: cfg.ClientAdvertise, Digest: mgr.ClusterDigest()} - msh, err := mesh.New(ctx, mlCfg, self, key, cfg.DataDir) + msh, err := mesh.New(ctx, mlCfg, self, key) if err != nil { return nil, err } - msh.UpdateSelf(ctx, mgr.WarmCounts(), mgr.TemplateHashes(), mgr.VolumeNames()) + msh.UpdateSelf(mgr.WarmCounts(), mgr.TemplateHashes(), mgr.VolumeNames()) if err := msh.Join(mc.Join); err != nil { _ = msh.Shutdown() return nil, err @@ -249,7 +240,7 @@ func gossipNodeState(ctx context.Context, msh *mesh.Mesh, mgr *pool.Manager) { case <-ctx.Done(): return case <-tick: - msh.UpdateSelf(ctx, mgr.WarmCounts(), mgr.TemplateHashes(), mgr.VolumeNames()) + msh.UpdateSelf(mgr.WarmCounts(), mgr.TemplateHashes(), mgr.VolumeNames()) } } } diff --git a/sandboxd/mesh/client_endpoint_test.go b/sandboxd/mesh/client_endpoint_test.go index c01745b42..71e1d29c1 100644 --- a/sandboxd/mesh/client_endpoint_test.go +++ b/sandboxd/mesh/client_endpoint_test.go @@ -6,7 +6,7 @@ import ( ) func TestClientAddressesDoNotReplaceInternalAddresses(t *testing.T) { - m := newBoundMesh(t, t.TempDir()) + m := newBoundMesh(t) if m.Members()[0].ClientAddr != "https://self.example" { t.Fatal("self client address was not published") } diff --git a/sandboxd/mesh/epoch.go b/sandboxd/mesh/epoch.go deleted file mode 100644 index 656cad1e4..000000000 --- a/sandboxd/mesh/epoch.go +++ /dev/null @@ -1,26 +0,0 @@ -package mesh - -import ( - "math" - "os" - "strconv" - "strings" - - "github.com/cocoonstack/sandbox/sandboxd/utils" -) - -func loadEpoch(path string) uint64 { - raw, err := os.ReadFile(path) //nolint:gosec // node-local data-dir path - if err != nil { - return 0 - } - n, err := strconv.ParseUint(strings.TrimSpace(string(raw)), 10, 64) - if err != nil || n > math.MaxInt64 { - return 0 - } - return n -} - -func storeEpoch(path string, epoch uint64) error { - return utils.WriteFileSync(path, []byte(strconv.FormatUint(epoch, 10)), 0o600) -} diff --git a/sandboxd/mesh/mesh.go b/sandboxd/mesh/mesh.go index d38e01f35..4a23002e5 100644 --- a/sandboxd/mesh/mesh.go +++ b/sandboxd/mesh/mesh.go @@ -8,7 +8,6 @@ import ( "fmt" "maps" "math/rand/v2" - "path/filepath" "slices" "sync" "time" @@ -17,11 +16,7 @@ import ( "github.com/projecteru2/core/log" ) -const ( - leaveTimeout = time.Second - - epochLease = 1024 -) +const leaveTimeout = time.Second // NodeState is one node's gossiped placement view; the higher Epoch wins a merge. type NodeState struct { @@ -39,12 +34,8 @@ type nodeMatch func(NodeState) bool // Mesh is the node's view of the cluster and its own gossiped state. type Mesh struct { - ml *memberlist.Memberlist - epochPath string - ctx context.Context - - updateMu sync.Mutex - leased uint64 + ml *memberlist.Memberlist + ctx context.Context mu sync.Mutex self NodeState @@ -53,20 +44,14 @@ type Mesh struct { } // New starts a mesh member listening per cfg. -func New(ctx context.Context, cfg *memberlist.Config, self NodeState, secretKey []byte, dataDir string) (*Mesh, error) { - epochPath := filepath.Join(dataDir, "mesh-epoch") - self.Epoch = max(uint64(time.Now().UnixNano()), loadEpoch(epochPath)+1) +func New(ctx context.Context, cfg *memberlist.Config, self NodeState, secretKey []byte) (*Mesh, error) { + self.Epoch = uint64(time.Now().UnixNano()) self.Pools = map[string]int{} m := &Mesh{ - ctx: ctx, - epochPath: epochPath, - leased: self.Epoch + epochLease, - self: self, - view: map[string]NodeState{self.NodeID: self}, - live: map[string]struct{}{}, - } - if err := storeEpoch(m.epochPath, m.leased); err != nil { - return nil, fmt.Errorf("persist mesh epoch: %w", err) + ctx: ctx, + self: self, + view: map[string]NodeState{self.NodeID: self}, + live: map[string]struct{}{}, } cfg.Name = self.NodeID @@ -85,9 +70,6 @@ func New(ctx context.Context, cfg *memberlist.Config, self NodeState, secretKey // Join contacts seed members; an empty list leaves a mesh of one. func (m *Mesh) Join(seeds []string) error { - if len(seeds) == 0 { - return nil - } if _, err := m.ml.Join(seeds); err != nil { return fmt.Errorf("join mesh: %w", err) } @@ -95,8 +77,8 @@ func (m *Mesh) Join(seeds []string) error { } // UpdateSelf republishes this node's counts, templates, and volumes; both slices must be sorted. -func (m *Mesh) UpdateSelf(ctx context.Context, pools map[string]int, templates, volumes []string) { - m.publish(ctx, func(self *NodeState) bool { +func (m *Mesh) UpdateSelf(pools map[string]int, templates, volumes []string) { + m.publish(func(self *NodeState) bool { if maps.Equal(self.Pools, pools) && slices.Equal(self.Templates, templates) && slices.Equal(self.Volumes, volumes) { return false } @@ -106,8 +88,8 @@ func (m *Mesh) UpdateSelf(ctx context.Context, pools map[string]int, templates, } // UpdateDigest republishes this node's config digest after a runtime change to the cluster-invariant config. -func (m *Mesh) UpdateDigest(ctx context.Context, digest string) { - m.publish(ctx, func(self *NodeState) bool { +func (m *Mesh) UpdateDigest(digest string) { + m.publish(func(self *NodeState) bool { if self.Digest == digest { return false } @@ -265,31 +247,15 @@ func (m *Mesh) owners(match nodeMatch) []string { return owners[:min(len(owners), 2)] } -// publish applies change under a fresh epoch; the epoch lease is persisted first so a restart never reuses one. -func (m *Mesh) publish(ctx context.Context, change func(*NodeState) bool) { - m.updateMu.Lock() - defer m.updateMu.Unlock() +// publish applies change under a fresh epoch. +func (m *Mesh) publish(change func(*NodeState) bool) { m.mu.Lock() - probe := m.self - if !change(&probe) { - m.mu.Unlock() + defer m.mu.Unlock() + if !change(&m.self) { return } - epoch := m.self.Epoch + 1 - m.mu.Unlock() - if epoch > m.leased { - leased := epoch + epochLease - if err := storeEpoch(m.epochPath, leased); err != nil { - log.WithFunc("mesh.publish").Warnf(ctx, "persist epoch: %v", err) - return - } - m.leased = leased - } - m.mu.Lock() - change(&m.self) - m.self.Epoch = epoch + m.self.Epoch++ m.view[m.self.NodeID] = m.self - m.mu.Unlock() } func (m *Mesh) admit(nodeID string) { diff --git a/sandboxd/mesh/mesh_test.go b/sandboxd/mesh/mesh_test.go index 2d093d140..e3b3f7b32 100644 --- a/sandboxd/mesh/mesh_test.go +++ b/sandboxd/mesh/mesh_test.go @@ -5,7 +5,6 @@ import ( "io" "log" "maps" - "path/filepath" "slices" "testing" "time" @@ -32,7 +31,7 @@ func TestMergeKeepsHigherEpoch(t *testing.T) { func TestMergeNeverOverwritesSelf(t *testing.T) { m := newTestMesh(t, "a") - m.UpdateSelf(t.Context(), map[string]int{"k": 3}, nil, nil) + m.UpdateSelf(map[string]int{"k": 3}, nil, nil) mergeStates(t, m, []NodeState{{NodeID: "a", Addr: "evil:9999", Epoch: 999, Pools: map[string]int{"k": 0}}}) @@ -45,7 +44,7 @@ func TestMergeNeverOverwritesSelf(t *testing.T) { func TestCandidatesExcludeSelfAndEmpty(t *testing.T) { m := newTestMesh(t, "a") - m.UpdateSelf(t.Context(), map[string]int{"k": 5}, nil, nil) + m.UpdateSelf(map[string]int{"k": 5}, nil, nil) mergeStates(t, m, []NodeState{ {NodeID: "b", Addr: "b:7777", Epoch: 1, Pools: map[string]int{"k": 2}}, {NodeID: "c", Addr: "c:7777", Epoch: 1, Pools: map[string]int{"k": 0}}, @@ -63,7 +62,7 @@ func TestCandidatesExcludeSelfAndEmpty(t *testing.T) { func TestTemplateOwnersExcludeSelfAndUnknown(t *testing.T) { m := newTestMesh(t, "a") - m.UpdateSelf(t.Context(), nil, []string{"tpl"}, nil) + m.UpdateSelf(nil, []string{"tpl"}, nil) mergeStates(t, m, []NodeState{ {NodeID: "b", Addr: "b:7777", Epoch: 1, Templates: []string{"tpl", "other"}}, {NodeID: "c", Addr: "c:7777", Epoch: 1, Templates: []string{"other"}}, @@ -88,7 +87,7 @@ func TestForgetPrunesDeadNode(t *testing.T) { t.Errorf("candidates after forget %v, want nil (b pruned)", got) } - m.UpdateSelf(t.Context(), map[string]int{"k": 1}, nil, nil) + m.UpdateSelf(map[string]int{"k": 1}, nil, nil) m.forget("a") if len(m.Members()) != 1 { t.Error("forget removed self") @@ -121,7 +120,7 @@ func TestForgottenNodeStaysGoneUntilItRejoins(t *testing.T) { func TestPoolPeersIncludeAColdPeerOfThePool(t *testing.T) { m := newTestMesh(t, "a") - m.UpdateSelf(t.Context(), map[string]int{"pool": 0}, nil, nil) + m.UpdateSelf(map[string]int{"pool": 0}, nil, nil) mergeStates(t, m, []NodeState{ {NodeID: "b", Addr: "b:7777", Epoch: 1, Pools: map[string]int{"pool": 0}}, {NodeID: "c", Addr: "c:7777", Epoch: 1, Pools: map[string]int{"other": 3}}, @@ -176,7 +175,7 @@ func TestTwoNodeClusterGossipsPools(t *testing.T) { if err := b.mesh.Join([]string{a.addr}); err != nil { t.Fatalf("join: %v", err) } - a.mesh.UpdateSelf(t.Context(), map[string]int{"kk": 4}, []string{"tpl-hash"}, []string{"dataset"}) + a.mesh.UpdateSelf(map[string]int{"kk": 4}, []string{"tpl-hash"}, []string{"dataset"}) deadline := time.Now().Add(10 * time.Second) for time.Now().Before(deadline) { @@ -199,7 +198,7 @@ func TestARuntimeDigestChangeReachesThePeer(t *testing.T) { if err := b.mesh.Join([]string{a.addr}); err != nil { t.Fatalf("join: %v", err) } - a.mesh.UpdateDigest(t.Context(), "rotated") + a.mesh.UpdateDigest("rotated") deadline := time.Now().Add(10 * time.Second) for time.Now().Before(deadline) { if b.mesh.ConfigMismatches() == 1 { @@ -212,7 +211,7 @@ func TestARuntimeDigestChangeReachesThePeer(t *testing.T) { func TestVolumeOwnersRequireEveryNameAndExcludeSelf(t *testing.T) { m := newTestMesh(t, "a") - m.UpdateSelf(t.Context(), nil, nil, []string{"dataset", "weights"}) + m.UpdateSelf(nil, nil, []string{"dataset", "weights"}) mergeStates(t, m, []NodeState{ {NodeID: "b", Addr: "b:7777", Epoch: 1, Volumes: []string{"dataset", "weights"}}, {NodeID: "c", Addr: "c:7777", Epoch: 1, Volumes: []string{"dataset"}}, @@ -264,7 +263,7 @@ func TestTemplateVolumeOwnersUseTrueIntersection(t *testing.T) { func TestVolumeHoldersCountSelfAndPeers(t *testing.T) { m := newTestMesh(t, "a") - m.UpdateSelf(t.Context(), nil, nil, []string{"dataset"}) + m.UpdateSelf(nil, nil, []string{"dataset"}) mergeStates(t, m, []NodeState{ {NodeID: "b", Addr: "b:7777", Epoch: 1, Volumes: []string{"dataset", "weights"}}, {NodeID: "c", Addr: "c:7777", Epoch: 1, Volumes: []string{"weights"}}, @@ -287,10 +286,9 @@ func mergeStates(t *testing.T, m *Mesh, states []NodeState) { func newTestMesh(t *testing.T, id string) *Mesh { t.Helper() return &Mesh{ - epochPath: filepath.Join(t.TempDir(), "mesh-epoch"), - self: NodeState{NodeID: id, Addr: id + ":7777", Pools: map[string]int{}, Digest: id + "-digest"}, - view: map[string]NodeState{id: {NodeID: id, Addr: id + ":7777"}}, - live: map[string]struct{}{}, + self: NodeState{NodeID: id, Addr: id + ":7777", Pools: map[string]int{}, Digest: id + "-digest"}, + view: map[string]NodeState{id: {NodeID: id, Addr: id + ":7777"}}, + live: map[string]struct{}{}, } } @@ -309,7 +307,7 @@ func startNode(t *testing.T, host string, port int, id string) *node { cfg.AdvertiseAddr = host cfg.PushPullInterval = 200 * time.Millisecond cfg.Logger = discardLogger() - m, err := New(t.Context(), cfg, NodeState{NodeID: id, Addr: id + ":7777", Digest: "shared"}, nil, t.TempDir()) + m, err := New(t.Context(), cfg, NodeState{NodeID: id, Addr: id + ":7777", Digest: "shared"}, nil) if err != nil { t.Fatalf("new mesh %s: %v", id, err) } diff --git a/sandboxd/mesh/state_test.go b/sandboxd/mesh/state_test.go index 74e00e9b5..0ac508a3d 100644 --- a/sandboxd/mesh/state_test.go +++ b/sandboxd/mesh/state_test.go @@ -1,109 +1,21 @@ package mesh import ( - "os" - "path/filepath" "sync" "testing" "github.com/hashicorp/memberlist" ) -func TestEpochRoundTrip(t *testing.T) { - p := filepath.Join(t.TempDir(), "e") - if loadEpoch(p) != 0 { - t.Error("missing epoch file must read 0") - } - if err := storeEpoch(p, 42); err != nil { - t.Fatalf("store: %v", err) - } - if got := loadEpoch(p); got != 42 { - t.Errorf("loadEpoch = %d, want 42", got) - } -} - -func TestLoadEpochRejectsImplausibleValue(t *testing.T) { - p := filepath.Join(t.TempDir(), "e") - - if err := os.WriteFile(p, []byte("18446744073709551615"), 0o600); err != nil { - t.Fatalf("write: %v", err) - } - if got := loadEpoch(p); got != 0 { - t.Errorf("loadEpoch of a MaxUint64 file = %d, want 0 (treated as corrupt)", got) - } -} - -func TestEpochSeededAbovePersistedFloor(t *testing.T) { - dir := t.TempDir() - huge := uint64(1) << 62 - if err := storeEpoch(filepath.Join(dir, "mesh-epoch"), huge); err != nil { - t.Fatalf("store: %v", err) - } - m := newBoundMesh(t, dir) - - if m.self.Epoch != huge+1 { - t.Errorf("epoch seeded %d, want persisted floor + 1 = %d (a backwards clock must land above peers' last-seen epoch)", m.self.Epoch, huge+1) - } -} - -func TestUpdateSelfPersistFailKeepsOldState(t *testing.T) { - dir := t.TempDir() - m := newBoundMesh(t, dir) - before := m.self.Epoch - - m.epochPath = filepath.Join(dir, "gone", "mesh-epoch") - m.leased = before - m.UpdateSelf(t.Context(), map[string]int{"k": 1}, nil, nil) - if m.self.Epoch != before { - t.Errorf("self epoch advanced to %d on a failed persist, want %d held", m.self.Epoch, before) - } - if len(m.self.Pools) != 0 { - t.Errorf("self pools published %v despite a failed persist", m.self.Pools) - } -} - -func TestUpdateSelfPersistsEpoch(t *testing.T) { - dir := t.TempDir() - m := newBoundMesh(t, dir) - before := m.self.Epoch - m.UpdateSelf(t.Context(), map[string]int{"k": 1}, nil, nil) - if got := loadEpoch(filepath.Join(dir, "mesh-epoch")); got <= before { - t.Errorf("persisted epoch %d did not advance past the seed %d", got, before) - } -} - -func TestUpdateSelfLeasesEpochFloor(t *testing.T) { - dir := t.TempDir() - m := newBoundMesh(t, dir) - path := filepath.Join(dir, "mesh-epoch") - seeded := loadEpoch(path) - if seeded <= m.self.Epoch { - t.Fatalf("seeded floor %d, want above the live epoch %d", seeded, m.self.Epoch) - } - for i := range 8 { - m.UpdateSelf(t.Context(), map[string]int{"k": i + 1}, nil, nil) - } - if got := loadEpoch(path); got != seeded { - t.Errorf("floor rewritten to %d during %d leased ticks, want %d held", got, 8, seeded) - } - m.mu.Lock() - m.self.Epoch = seeded - m.mu.Unlock() - m.UpdateSelf(t.Context(), map[string]int{"k": 99}, nil, nil) - if got := loadEpoch(path); got <= m.self.Epoch { - t.Errorf("floor %d not above the published epoch %d after crossing the lease", got, m.self.Epoch) - } -} - func TestUpdateSelfConcurrentDropsNothing(t *testing.T) { - m := newBoundMesh(t, t.TempDir()) + m := newBoundMesh(t) base := m.self.Epoch for i := range 100 { a := map[string]int{"a": i + 1} b := map[string]int{"b": i + 1} var wg sync.WaitGroup - wg.Go(func() { m.UpdateSelf(t.Context(), a, nil, nil) }) - wg.Go(func() { m.UpdateSelf(t.Context(), b, nil, nil) }) + wg.Go(func() { m.UpdateSelf(a, nil, nil) }) + wg.Go(func() { m.UpdateSelf(b, nil, nil) }) wg.Wait() want := base + uint64(2*(i+1)) @@ -114,17 +26,17 @@ func TestUpdateSelfConcurrentDropsNothing(t *testing.T) { } func TestUpdateSelfBumpsOnlyWhenVolumesChange(t *testing.T) { - m := newBoundMesh(t, t.TempDir()) + m := newBoundMesh(t) base := m.self.Epoch - m.UpdateSelf(t.Context(), nil, nil, []string{"dataset"}) + m.UpdateSelf(nil, nil, []string{"dataset"}) if m.self.Epoch != base+1 { t.Fatalf("first volume publish epoch=%d, want %d", m.self.Epoch, base+1) } - m.UpdateSelf(t.Context(), nil, nil, []string{"dataset"}) + m.UpdateSelf(nil, nil, []string{"dataset"}) if m.self.Epoch != base+1 { t.Errorf("unchanged volume republished at epoch=%d", m.self.Epoch) } - m.UpdateSelf(t.Context(), nil, nil, []string{"dataset", "weights"}) + m.UpdateSelf(nil, nil, []string{"dataset", "weights"}) if m.self.Epoch != base+2 { t.Errorf("changed volume epoch=%d, want %d", m.self.Epoch, base+2) } @@ -140,24 +52,24 @@ func TestConfigDigestMismatch(t *testing.T) { } func TestUpdateDigestRepublishesUnderAFreshEpoch(t *testing.T) { - m := newBoundMesh(t, t.TempDir()) + m := newBoundMesh(t) base := m.self.Epoch - m.UpdateDigest(t.Context(), "d1") + m.UpdateDigest("d1") if m.self.Epoch != base { t.Errorf("an unchanged digest republished at epoch=%d", m.self.Epoch) } - m.UpdateDigest(t.Context(), "d2") + m.UpdateDigest("d2") if m.self.Epoch != base+1 || m.view["self"].Digest != "d2" { t.Errorf("epoch=%d view digest %q, want epoch %d and d2", m.self.Epoch, m.view["self"].Digest, base+1) } } -func newBoundMesh(t *testing.T, dataDir string) *Mesh { +func newBoundMesh(t *testing.T) *Mesh { t.Helper() cfg := memberlist.DefaultLocalConfig() cfg.BindPort = 0 cfg.Logger = discardLogger() - m, err := New(t.Context(), cfg, NodeState{NodeID: "self", Addr: "self:7777", ClientAddr: "https://self.example", Digest: "d1"}, nil, dataDir) + m, err := New(t.Context(), cfg, NodeState{NodeID: "self", Addr: "self:7777", ClientAddr: "https://self.example", Digest: "d1"}, nil) if err != nil { t.Fatalf("new mesh: %v", err) } diff --git a/sandboxd/outbound/claim.go b/sandboxd/outbound/claim.go index e520730f9..237389ae9 100644 --- a/sandboxd/outbound/claim.go +++ b/sandboxd/outbound/claim.go @@ -60,14 +60,6 @@ func (c *claim) EvalInner(host, method string, port uint16) (egress.Rule, egress return c.current().EvalInner(host, method, port) } -func (c *claim) ServesSocks() bool { - return c.current().ServesSocks() -} - -func (c *claim) InterceptsHost(pattern string) bool { - return c.current().InterceptsHost(pattern) -} - func (c *claim) Header(name string) (header, value string, ok bool) { secrets := c.h.o.View().secrets if header, value, ok = secrets.Header(name); !ok { diff --git a/sandboxd/outbound/view.go b/sandboxd/outbound/view.go index 0bc0fca0c..440470338 100644 --- a/sandboxd/outbound/view.go +++ b/sandboxd/outbound/view.go @@ -85,7 +85,7 @@ func (v *View) HasClass(name string) bool { } // Resolve is sb's effective policy, pool ∩ tenant: root has no tenant layer, an unpooled key no pool one, a NoEgress claim none at all; pooled runs only when the key has no pool policy. -func (v *View) Resolve(sb *types.Sandbox, pooled PooledFunc) (egress.Evaluator, bool) { +func (v *View) Resolve(sb *types.Sandbox, pooled PooledFunc) (egress.PolicyEvaluator, bool) { if sb.NoEgress { return nil, false } diff --git a/sandboxd/pool/checkpoint.go b/sandboxd/pool/checkpoint.go index d781978a5..3491f0e7b 100644 --- a/sandboxd/pool/checkpoint.go +++ b/sandboxd/pool/checkpoint.go @@ -213,9 +213,6 @@ func (m *Manager) claimLoaded(ctx context.Context, ckpt types.Checkpoint, o Clai if ckpt.Archive { return nil, ErrUnknownCheckpoint // a wake image, not a branchable checkpoint } - if !ckpt.Key.Capturable() { - return nil, ErrNoEgressFork - } dir, _, _, err := m.ckpts.Fetch(ctx, ckpt.ID) if errors.Is(err, store.ErrNotFound) { return nil, ErrUnknownCheckpoint // deleted between the pre-check and the lock diff --git a/sandboxd/pool/claim.go b/sandboxd/pool/claim.go index 03a229674..0739ef9eb 100644 --- a/sandboxd/pool/claim.go +++ b/sandboxd/pool/claim.go @@ -208,7 +208,7 @@ func (m *Manager) AgentSocket(id, token string) (string, error) { if !ok { return "", ErrUnknownSandbox } - if sb.Replicated() && m.isFenced() { + if sb.Replicated && m.isFenced() { return "", ErrLeaseLost } // No activity stamp: a control-plane poll must not keep an idle sandbox awake @@ -462,7 +462,9 @@ func (m *Manager) tenantDelta(tenant string, delta int) { func (m *Manager) addClaimLocked(sb *types.Sandbox) { if m.archives != nil { - sb.SetReplicated(m.view.Load().poolReplicate[sb.PolicyKey()] > 0 || sb.HasReplica()) + if replicated := m.view.Load().poolReplicate[sb.PolicyKey()] > 0 || sb.HasReplica(); sb.Replicated != replicated { + sb.Replicated = replicated + } } m.claimed[sb.ID] = sb m.tenantDelta(sb.Tenant, 1) diff --git a/sandboxd/pool/claims.go b/sandboxd/pool/claims.go index c4951e723..e998d7c0a 100644 --- a/sandboxd/pool/claims.go +++ b/sandboxd/pool/claims.go @@ -118,11 +118,6 @@ func (s *claimStore) load() (map[string]*types.Sandbox, error) { s.mu.Lock() defer s.mu.Unlock() for _, sb := range claims { - for _, volumes := range [][]types.Volume{sb.Volumes, sb.VolumesWant} { - for i := range volumes { - volumes[i].AttachOnly = volumes[i].Mount == "" - } - } if sb.VolumesWant == nil { sb.VolumesWant = sb.Volumes } diff --git a/sandboxd/pool/failover_test.go b/sandboxd/pool/failover_test.go index c436de56b..84e286b1d 100644 --- a/sandboxd/pool/failover_test.go +++ b/sandboxd/pool/failover_test.go @@ -1407,12 +1407,12 @@ func TestFailoverArchiveExportsNothingForARowAnotherNodeHolds(t *testing.T) { if _, err := c.reg.Put(t.Context(), archives.Record{ID: sb.ID, Owner: "b", Checkpoint: "ck_00000000000000ee", Claim: []byte(`{}`)}); err != nil { t.Fatalf("put: %v", err) } - published := &publishCountingStore{Store: a.ckpts} + published := &failingPublishStore{Store: a.ckpts} a.ckpts = published if err := a.archive(t.Context(), sb); !errors.Is(err, archives.ErrLost) { t.Fatalf("archive of a claim another node holds: %v, want ErrLost", err) } - if n := published.n.Load(); n != 0 { + if n := published.publishes.Load(); n != 0 { t.Errorf("archive published %d exports, want none", n) } } @@ -2057,16 +2057,6 @@ func (r *hookRegistry) Consume(ctx context.Context, id string, expect int64) (in return epoch, err } -type publishCountingStore struct { - store.Store - n atomic.Int32 -} - -func (s *publishCountingStore) Publish(ctx context.Context, staging, id, lineage string) error { - s.n.Add(1) - return s.Store.Publish(ctx, staging, id, lineage) -} - type unreadableStore struct { store.Store } diff --git a/sandboxd/pool/lease.go b/sandboxd/pool/lease.go index 791da7dae..41b8f3d16 100644 --- a/sandboxd/pool/lease.go +++ b/sandboxd/pool/lease.go @@ -81,6 +81,9 @@ func (m *Manager) trip(ctx context.Context) { func (m *Manager) unfence(ctx context.Context, start time.Time) { logger := log.WithFunc("pool.unfence") fenced := m.fence.Load() + if !fenced.fenced { + return + } m.mu.Lock() held := map[string]*types.Sandbox{} for id, sb := range m.claimed { @@ -204,7 +207,7 @@ func (m *Manager) confirmOwnership(ctx context.Context, sb *types.Sandbox, rec a // fenceWatch gates sb at resolution and returns the channel that closes when its streams must end; nil for an unreplicated claim. func (m *Manager) fenceWatch(sb *types.Sandbox) (<-chan struct{}, error) { - if !sb.Replicated() { + if !sb.Replicated { return nil, nil } st := m.fence.Load() diff --git a/sandboxd/pool/lease_test.go b/sandboxd/pool/lease_test.go index 49b27fe5c..ff0c0b640 100644 --- a/sandboxd/pool/lease_test.go +++ b/sandboxd/pool/lease_test.go @@ -41,6 +41,15 @@ func TestLeaseRenewLoopKeepsTheNodeLeasedUntilItStops(t *testing.T) { }) } +func TestAnUnfenceOnALiftedNodeKeepsTheOpenFenceChannel(t *testing.T) { + m := newTestManager(t, newFakeEngine()) + lifted := m.fence.Load() + m.unfence(t.Context(), time.Now()) + if m.fence.Load() != lifted { + t.Error("an unfence replaced a lifted state, so open streams would miss the next trip") + } +} + func TestLeaseOverdueRenewalFencesReplicatedClaimsOnly(t *testing.T) { synctest.Test(t, func(t *testing.T) { c := newReplicaCluster(t) diff --git a/sandboxd/pool/operator.go b/sandboxd/pool/operator.go index f28c34e38..bf9c2b425 100644 --- a/sandboxd/pool/operator.go +++ b/sandboxd/pool/operator.go @@ -63,14 +63,11 @@ func (m *Manager) SetInstanceMetadata(ctx context.Context, id string, doc []byte return m.eng.WriteInstanceMetadata(ctx, sb.VsockSocket, doc) } -// resolve authorizes id under cred; an unclaimed slot must never match an empty token. +// resolve authorizes id under cred. func (m *Manager) resolve(id string, cred Cred) (*types.Sandbox, bool) { if cred.Operator { return m.byID(id) } - if cred.Token == "" { - return nil, false - } return m.claim(id, cred.Token) } diff --git a/sandboxd/pool/reader_test.go b/sandboxd/pool/reader_test.go index 553f15606..ddd03a47d 100644 --- a/sandboxd/pool/reader_test.go +++ b/sandboxd/pool/reader_test.go @@ -62,7 +62,7 @@ func TestReaderObeysTheNodeLeaseFence(t *testing.T) { eng := newFakeEngine() m := newReaderManager(t, eng) replicated, plain := mustClaim(t, m, testKey), mustClaim(t, m, testKey) - replicated.SetReplicated(true) + replicated.Replicated = true hibernateClaim(t, m, replicated) hibernateClaim(t, m, plain) seedReaderVMs(m, testKey, "sbx-reader-1", "sbx-reader-2") @@ -90,7 +90,7 @@ func TestReaderConnectionEndsWhenTheFenceTrips(t *testing.T) { m := newReaderManager(t, eng) m.liftFence() replicated, plain := mustClaim(t, m, testKey), mustClaim(t, m, testKey) - replicated.SetReplicated(true) + replicated.Replicated = true hibernateClaim(t, m, replicated) hibernateClaim(t, m, plain) seedReaderVMs(m, testKey, "sbx-reader-1", "sbx-reader-2") diff --git a/sandboxd/pool/reap.go b/sandboxd/pool/reap.go index da667282b..5b0b0303e 100644 --- a/sandboxd/pool/reap.go +++ b/sandboxd/pool/reap.go @@ -85,9 +85,6 @@ func (m *Manager) reapOnce(ctx context.Context) { } return bad }) - if len(expired) == 0 { - return - } m.mu.Lock() keep := expired[:0] var purgeCks, purged []string diff --git a/sandboxd/pool/replica_test.go b/sandboxd/pool/replica_test.go index 54b0b9c21..0a688516f 100644 --- a/sandboxd/pool/replica_test.go +++ b/sandboxd/pool/replica_test.go @@ -312,7 +312,7 @@ func TestReplicaFollowsThePolicyPoolOfAPromotedTemplate(t *testing.T) { sb.Key, sb.PolicySource = tplKey("promoted:v1"), testKey a.addClaimLocked(sb) a.mu.Unlock() - if !sb.Replicated() { + if !sb.Replicated { t.Error("a claim of a template promoted out of a replicating pool is not marked replicated") } @@ -421,8 +421,8 @@ func TestReplicaLeftoverJournalWithFailoverOffStillServesStreams(t *testing.T) { t.Fatalf("reconcile: %v", err) } held, ok := restarted.byID(sb.ID) - if !ok || held.ReplicaCk != "ck_leftover" || held.Replicated() { - t.Fatalf("claim held %v replica %q replicated %v, want the leftover kept and not replicated", ok, held.ReplicaCk, held.Replicated()) + if !ok || held.ReplicaCk != "ck_leftover" || held.Replicated { + t.Fatalf("claim held %v replica %q replicated %v, want the leftover kept and not replicated", ok, held.ReplicaCk, held.Replicated) } _, fenced, done, err := restarted.WakeAgentSocket(t.Context(), sb.ID, sb.Token) if err != nil || fenced != nil { diff --git a/sandboxd/pool/volume_mutation_test.go b/sandboxd/pool/volume_mutation_test.go index 0830ce33b..1dc2cc6de 100644 --- a/sandboxd/pool/volume_mutation_test.go +++ b/sandboxd/pool/volume_mutation_test.go @@ -1,7 +1,6 @@ package pool import ( - "bytes" "context" "errors" "fmt" @@ -60,13 +59,6 @@ func TestLiveVolumeFourModesAndMixedRoundtrip(t *testing.T) { t.Fatalf("explicit false inherited attach-only: %v", err) } assertLiveJournal(t, m, sb) - raw, err := os.ReadFile(m.store.path) - if err != nil { - t.Fatalf("ReadFile: %v", err) - } - if err = os.WriteFile(m.store.path, bytes.ReplaceAll(raw, []byte(`,"attach_only":true`), nil), 0o600); err != nil { - t.Fatalf("WriteFile: %v", err) - } restarted := newVolumeManagerAt(t, eng, m.dataDir, catalog) if err = restarted.Reconcile(t.Context()); err != nil { t.Fatalf("Reconcile: %v", err) @@ -527,7 +519,6 @@ func TestDesiredVolumesJournalMigration(t *testing.T) { want []types.Volume }{ {"legacy", "", []types.Volume{{Name: "data", Mount: "/data"}}}, - {"legacy attach-only desired", `,"volumes_want":[{"name":"raw","mode":"rw"}]`, []types.Volume{{Name: "raw", Mode: types.VolumeModeRW, AttachOnly: true}}}, {"explicit empty", `,"volumes_want":[]`, nil}, {"desired", `,"volumes_want":[{"name":"other","mount":"/other"}]`, []types.Volume{{Name: "other", Mount: "/other"}}}, } { diff --git a/sandboxd/poolset/pg/pg.go b/sandboxd/poolset/pg/pg.go index 1d7e90ced..cb75f27fe 100644 --- a/sandboxd/poolset/pg/pg.go +++ b/sandboxd/poolset/pg/pg.go @@ -54,7 +54,7 @@ func (s *Source) Load(ctx context.Context) (*poolset.Set, int64, error) { case errors.Is(err, pgx.ErrNoRows): return nil, 0, nil case err != nil: - return nil, 0, unavailable(err) + return nil, 0, fmt.Errorf("%w: %w", poolset.ErrUnavailable, err) } if err = json.Unmarshal(raw, &set.Pools); err != nil { return nil, 0, fmt.Errorf("decode cell %q pool set: %w", s.cell, err) @@ -76,7 +76,7 @@ func (s *Source) Store(ctx context.Context, set poolset.Set) (int64, error) { version = sandboxd_pool_sets.version + 1, updated_at = now() RETURNING version) SELECT version, pg_notify($4, $1) FROM up`, s.cell, set.ConfigSeed, raw, channel).Scan(&version, nil); err != nil { - return 0, unavailable(err) + return 0, fmt.Errorf("%w: %w", poolset.ErrUnavailable, err) } return version, nil } @@ -93,7 +93,3 @@ func (s *Source) Close() error { s.pool.Close() return nil } - -func unavailable(err error) error { - return fmt.Errorf("%w: %w", poolset.ErrUnavailable, err) -} diff --git a/sandboxd/server/metrics.go b/sandboxd/server/metrics.go index fa3135c24..1a75b0d75 100644 --- a/sandboxd/server/metrics.go +++ b/sandboxd/server/metrics.go @@ -10,7 +10,6 @@ import ( "github.com/cocoonstack/sandbox/sandboxd/pool" ) -// handleMetrics renders Prometheus text format by hand. func (s *Server) handleMetrics(w http.ResponseWriter, _ *http.Request) { pools, g := s.mgr.Info() c := s.mgr.Counters() diff --git a/sandboxd/server/reader.go b/sandboxd/server/reader.go index 13fd9c69a..64ba9e581 100644 --- a/sandboxd/server/reader.go +++ b/sandboxd/server/reader.go @@ -144,7 +144,7 @@ func (r *router) dialReader() error { func (r *router) handover(head []byte) (net.Conn, func(), bool) { r.closeReader() - guest, done, err := r.s.openGuest(r.ctx, r.id, r.token) + guest, done, err := r.s.openGuest(r.ctx, r.id, r.token, r.s.mgr.WakeAgentSocket) if err != nil { r.fail(err) return nil, nil, false @@ -277,8 +277,8 @@ func (s *Server) serveSleeping(ctx context.Context, w http.ResponseWriter, id, t copyBoth(client, s.audited(ctx, id, io.MultiReader(bytes.NewReader(rt.replay), rt.br)), guest, silkdEnds, s.noSplice) } -func (s *Server) openGuest(ctx context.Context, id, token string) (net.Conn, func(), error) { - sock, fenced, done, err := s.mgr.WakeAgentSocket(ctx, id, token) +func (s *Server) openGuest(ctx context.Context, id, token string, wake agentWaker) (net.Conn, func(), error) { + sock, fenced, done, err := wake(ctx, id, token) if err != nil { return nil, nil, err } diff --git a/sandboxd/server/relay.go b/sandboxd/server/relay.go index 8bd0a9275..468e03274 100644 --- a/sandboxd/server/relay.go +++ b/sandboxd/server/relay.go @@ -73,38 +73,30 @@ func (s *Server) handleAgent(w http.ResponseWriter, r *http.Request) { if !ok { return } - s.relay(r.Context(), id, client, clientBuf, guest) + s.splice(client, s.audited(r.Context(), id, clientReader(client, clientBuf)), guest, switchingProtocols, silkdEnds) } // wakeGuest dials the sandbox's silkd, waking a hibernated VM first unless wake answers ErrSleeping; on failure it has already answered the request. func (s *Server) wakeGuest(ctx context.Context, w http.ResponseWriter, id, token string, wake agentWaker) (guest net.Conn, release func(), sleeping, ok bool) { - sock, fenced, done, err := wake(ctx, id, token) + guest, done, err := s.openGuest(ctx, id, token, wake) switch { + case err == nil: + return guest, done, false, true case errors.Is(err, pool.ErrSleeping): return nil, nil, true, false - case writePoolErr(w, err): - return nil, nil, false, false - case err != nil: - log.WithFunc("server.wakeGuest").Errorf(ctx, err, "agent socket for %s", id) - writeErr(w, http.StatusInternalServerError, "sandbox lookup failed") - return nil, nil, false, false - } - guest, err = s.dialer.DialSilkd(ctx, sock) - if err != nil { - done() + case errors.Is(err, errUnreachable): if errors.Is(err, context.Canceled) && errors.Is(ctx.Err(), context.Canceled) { log.WithFunc("server.wakeGuest").Debugf(ctx, "client canceled the dial of silkd for %s: %v", id, err) } else { log.WithFunc("server.wakeGuest").Errorf(ctx, err, "dial silkd for %s", id) } writeErr(w, http.StatusBadGateway, "guest agent unreachable") - return nil, nil, false, false + case writePoolErr(w, err): + default: + log.WithFunc("server.wakeGuest").Errorf(ctx, err, "agent socket for %s", id) + writeErr(w, http.StatusInternalServerError, "sandbox lookup failed") } - return guest, fenceGuest(guest, fenced, done), false, true -} - -func (s *Server) relay(ctx context.Context, id string, client net.Conn, clientBuf *bufio.Reader, guest net.Conn) { - s.splice(client, s.audited(ctx, id, clientReader(client, clientBuf)), guest, switchingProtocols, silkdEnds) + return nil, nil, false, false } func (s *Server) audited(ctx context.Context, id string, clientR io.Reader) io.Reader { @@ -208,9 +200,6 @@ func (t *auditTee) endLine() { } func fenceGuest(guest net.Conn, fenced <-chan struct{}, done func()) func() { - if fenced == nil { - return done - } stop := pool.CloseOnFence(fenced, guest) return func() { stop(); done() } } diff --git a/sandboxd/server/server_http.go b/sandboxd/server/server_http.go index ac2aeb3cc..94e3944fc 100644 --- a/sandboxd/server/server_http.go +++ b/sandboxd/server/server_http.go @@ -120,10 +120,7 @@ func decodeBody[T any](w http.ResponseWriter, r *http.Request) (T, bool) { func decodeOptionalBody[T any](w http.ResponseWriter, r *http.Request) (T, bool) { var v T err := json.UnmarshalDecode(jsontext.NewDecoder(http.MaxBytesReader(w, r.Body, maxBodyBytes)), &v, json.MatchCaseInsensitiveNames(true)) - if errors.Is(err, io.EOF) { - return v, true - } - if err != nil { + if err != nil && !errors.Is(err, io.EOF) { writeErr(w, http.StatusBadRequest, "invalid request body") return v, false } @@ -153,9 +150,6 @@ func validRequest(w http.ResponseWriter, errs ...error) bool { } func metadataFilter(pairs []string) (types.Metadata, error) { - if len(pairs) == 0 { - return nil, nil - } filter := make(types.Metadata, len(pairs)) for _, pair := range pairs { k, v, ok := strings.Cut(pair, "=") diff --git a/sandboxd/store/digest.go b/sandboxd/store/digest.go index 0bf7c5bb1..200d513f9 100644 --- a/sandboxd/store/digest.go +++ b/sandboxd/store/digest.go @@ -26,10 +26,7 @@ type DigestFile struct { // ChunkCount returns how many DigestChunkSize chunks size spans. func ChunkCount(size int64) int64 { - if size == 0 { - return 0 - } - return (size-1)/DigestChunkSize + 1 + return (size + DigestChunkSize - 1) / DigestChunkSize } // AssembleDigest returns the canonical v2 digest of indexed file chunks. diff --git a/sandboxd/store/dir/dir.go b/sandboxd/store/dir/dir.go index 1250f7296..49e14c32e 100644 --- a/sandboxd/store/dir/dir.go +++ b/sandboxd/store/dir/dir.go @@ -183,7 +183,6 @@ func (d *Store) sweep() error { for _, e := range entries { var sweepErr error switch { - // ReadDir + suffix, not Glob: the root path may hold glob metacharacters. case strings.HasSuffix(e.Name(), ".tmp"): sweepErr = removeAgedEntry(d.root, e) case e.IsDir() && d.idRe.MatchString(e.Name()): @@ -278,21 +277,15 @@ func (d *Store) publish(_ context.Context, staging, id, digest string) error { } // sweepGenerations reclaims only entries whose install or supersession time is outside the grace. -func (d *Store) sweepGenerations(id string) (err error) { +func (d *Store) sweepGenerations(id string) error { final := filepath.Join(d.root, id) var keep []string - metaFile, err := os.Open(filepath.Join(final, store.MetaFile)) //nolint:gosec // id pinned by the instance idRe + meta, err := os.ReadFile(filepath.Join(final, store.MetaFile)) //nolint:gosec // id pinned by the instance idRe switch { case errors.Is(err, fs.ErrNotExist): case err != nil: return err default: - defer func() { err = errors.Join(err, metaFile.Close()) }() - // The open file pins one inode across a concurrent meta rename. - meta, readErr := io.ReadAll(metaFile) - if readErr != nil { - return readErr - } keep = []string{store.MetaFile, store.LabelsFile, store.ExportGen(meta), digestName(meta)} } entries, err := os.ReadDir(final) @@ -383,9 +376,6 @@ func hashExport(ctx context.Context, root string) (string, error) { if err != nil { return "", err } - if err := ctx.Err(); err != nil { - return "", err - } if err := hashChunks(ctx, sources, min(runtime.GOMAXPROCS(0), digestWorkerLimit), hashChunk); err != nil { return "", err } @@ -431,23 +421,17 @@ func hashChunk(source *digestSource, chunk int, buffer []byte) error { if err != nil { return fmt.Errorf("open export entry %s: %w", source.digest.Path, err) } + defer func() { _ = f.Close() }() offset := int64(chunk) * store.DigestChunkSize size := min(store.DigestChunkSize, source.digest.Size-offset) if sum, hole := store.HoleSum(f, offset, size); hole { source.digest.Chunks[chunk] = sum - if err := f.Close(); err != nil { - return fmt.Errorf("close export entry %s: %w", source.digest.Path, err) - } return nil } hash := sha256.New() - n, copyErr := io.CopyBuffer(hash, io.NewSectionReader(f, offset, size), buffer) - closeErr := f.Close() - if copyErr != nil { - return fmt.Errorf("hash export entry %s chunk %d: %w", source.digest.Path, chunk, copyErr) - } - if closeErr != nil { - return fmt.Errorf("close export entry %s: %w", source.digest.Path, closeErr) + n, err := io.CopyBuffer(hash, io.NewSectionReader(f, offset, size), buffer) + if err != nil { + return fmt.Errorf("hash export entry %s chunk %d: %w", source.digest.Path, chunk, err) } if n != size { return fmt.Errorf("export entry %s changed size while hashing", source.digest.Path) diff --git a/sandboxd/store/peer/broadcast.go b/sandboxd/store/peer/broadcast.go index 7b5978aee..142283a10 100644 --- a/sandboxd/store/peer/broadcast.go +++ b/sandboxd/store/peer/broadcast.go @@ -23,9 +23,6 @@ type Broadcaster struct { // Delete sends DELETE ?no_forward=1 to every peer in parallel; failures are logged, not returned. func (b *Broadcaster) Delete(ctx context.Context, id string) { addrs := slices.Compact(slices.Sorted(slices.Values(b.Peers()))) - if len(addrs) == 0 { - return - } logger := log.WithFunc("peer.Broadcaster.Delete") var wg sync.WaitGroup for _, addr := range addrs { diff --git a/sandboxd/store/peer/peer.go b/sandboxd/store/peer/peer.go index 8ac240855..5f0613096 100644 --- a/sandboxd/store/peer/peer.go +++ b/sandboxd/store/peer/peer.go @@ -31,9 +31,6 @@ func NewHealer(owners Owners, puller Puller) *Healer { // Pull fetches id into staging from the first owner whose transfer validates. func (h *Healer) Pull(ctx context.Context, id, staging string, validate Validate) error { - if h.owners == nil || h.puller == nil { - return store.ErrNotFound - } addrs := h.owners(id) if len(addrs) == 0 { return store.ErrNotFound @@ -56,7 +53,7 @@ func (h *Healer) pullFrom(ctx context.Context, id, staging string, addrs []strin err := h.puller.Pull(attemptCtx, addr, id, staging) cancel() if err != nil { - if !errors.Is(err, ErrNotFound) { + if !errors.Is(err, store.ErrNotFound) { errs = append(errs, fmt.Errorf("peer %s: %w", addr, err)) } continue diff --git a/sandboxd/store/peer/peer_test.go b/sandboxd/store/peer/peer_test.go index 61bea09e8..41486c06b 100644 --- a/sandboxd/store/peer/peer_test.go +++ b/sandboxd/store/peer/peer_test.go @@ -17,13 +17,6 @@ import ( const testID = "ck_00000000000000aa" -func TestInertWithoutOwnersOrPuller(t *testing.T) { - h := NewHealer(nil, nil) - if err := h.Pull(t.Context(), testID, t.TempDir(), nil); !errors.Is(err, store.ErrNotFound) { - t.Fatalf("Pull error = %v, want store.ErrNotFound", err) - } -} - func TestPullWritesStaging(t *testing.T) { puller := &fakePuller{records: map[string]map[string]string{"peer-a:7777": record()}} h := healerOf(puller, "peer-a:7777") @@ -176,7 +169,7 @@ func (p *fakePuller) Pull(_ context.Context, addr, _, dst string) error { } files, ok := p.records[addr] if !ok { - return ErrNotFound + return store.ErrNotFound } for name, content := range files { full := filepath.Join(dst, name) diff --git a/sandboxd/store/peer/probe.go b/sandboxd/store/peer/probe.go index c77efa29a..89f2bd4ae 100644 --- a/sandboxd/store/peer/probe.go +++ b/sandboxd/store/peer/probe.go @@ -46,25 +46,21 @@ type HTTPProber struct { ProbeKey []byte redirectFlight singleflight.Group - healFlight singleflight.Group cacheMu sync.Mutex cache map[string]redirectCacheEntry - // epoch is bumped by every Forget, so a fan-out started before a delete cannot cache its result. - epoch uint64 // cacheTTL overrides redirectCacheTTL; a test seam, 5 seconds is slow to wait out. cacheTTL time.Duration } func (p *HTTPProber) Owners(ctx context.Context, id string) []string { - owners, start, hit := p.cacheLookup(id) - if hit { + if owners, hit := p.cacheLookup(id); hit { return owners } v, _, _ := p.redirectFlight.Do(id, func() (any, error) { owners := p.fanOut(ctx, id, maxRedirectOwners, probeGrace) - p.cachePut(id, owners, start) + p.cachePut(id, owners) return owners, nil }) return v.([]string) @@ -72,17 +68,13 @@ func (p *HTTPProber) Owners(ctx context.Context, id string) []string { // HealOwners is Owners' wide twin: a heal wants the largest source list, uncached. func (p *HTTPProber) HealOwners(ctx context.Context, id string) []string { - v, _, _ := p.healFlight.Do(id, func() (any, error) { - // No grace: a heal must not stop early on a fast owner and hide a slower valid source. - return p.fanOut(ctx, id, maxHealOwners, 0), nil - }) - return v.([]string) + // No grace: a heal must not stop early on a fast owner and hide a slower valid source. + return p.fanOut(ctx, id, maxHealOwners, 0) } func (p *HTTPProber) Forget(id string) { p.cacheMu.Lock() defer p.cacheMu.Unlock() - p.epoch++ delete(p.cache, id) } @@ -130,27 +122,22 @@ func (p *HTTPProber) fanOut(ctx context.Context, id string, maxOwners int, grace } } -// cacheLookup returns id's live owners and the current epoch the caller carries into the fan-out. -func (p *HTTPProber) cacheLookup(id string) (owners []string, epoch uint64, hit bool) { +func (p *HTTPProber) cacheLookup(id string) ([]string, bool) { p.cacheMu.Lock() defer p.cacheMu.Unlock() entry, ok := p.cache[id] if !ok || time.Now().After(entry.expires) { - return nil, p.epoch, false + return nil, false } - return entry.owners, p.epoch, true + return entry.owners, true } -// cachePut records a positive result unless a Forget bumped the epoch since the fan-out began. -func (p *HTTPProber) cachePut(id string, owners []string, start uint64) { +func (p *HTTPProber) cachePut(id string, owners []string) { if len(owners) == 0 { return } p.cacheMu.Lock() defer p.cacheMu.Unlock() - if p.epoch != start { - return - } if p.cache == nil { p.cache = make(map[string]redirectCacheEntry) } diff --git a/sandboxd/store/peer/probe_test.go b/sandboxd/store/peer/probe_test.go index 9462c9362..588e8ac76 100644 --- a/sandboxd/store/peer/probe_test.go +++ b/sandboxd/store/peer/probe_test.go @@ -59,39 +59,6 @@ func TestOwnersReturnsPromptlyWithOneOwner(t *testing.T) { } } -func TestForgetDuringFlightPreventsStaleCache(t *testing.T) { - release := make(chan struct{}) - inFlight := make(chan struct{}) - probed := sync.OnceFunc(func() { close(inFlight) }) - var hits atomic.Int32 - owner := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { - hits.Add(1) - probed() - <-release - w.WriteHeader(http.StatusOK) - })) - defer owner.Close() - - p := &HTTPProber{Peers: func() []string { return []string{owner.URL} }} - var owners []string - done := make(chan struct{}) - go func() { owners = p.Owners(t.Context(), testID); close(done) }() - - <-inFlight - p.Forget(testID) - close(release) - <-done - - if len(owners) != 1 { - t.Fatalf("owners = %v, want the one that answered", owners) - } - - p.Owners(t.Context(), testID) - if got := hits.Load(); got != 2 { - t.Errorf("owner probed %d times, want 2: a Forget mid-flight must not leave a stale positive cached", got) - } -} - func TestOwnersDedupsAddrs(t *testing.T) { var hits atomic.Int32 ok := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { diff --git a/sandboxd/store/peer/transport.go b/sandboxd/store/peer/transport.go index 35a812ec7..c1e30bc9d 100644 --- a/sandboxd/store/peer/transport.go +++ b/sandboxd/store/peer/transport.go @@ -16,15 +16,11 @@ import ( "strconv" "strings" "sync" - "time" "github.com/cocoonstack/sandbox/sandboxd/store" ) const ( - // Generous for a guest memory image, but a wedged peer must fail so the next owner is tried. - pullTimeout = 30 * time.Minute - maxRecordBytes = 1 << 40 // 1 TiB exportPrefix = store.ExportDir + "/" @@ -37,9 +33,6 @@ const ( paxChunk = "SANDBOX.chunk" ) -// ErrNotFound reports that a peer does not hold the requested record. -var ErrNotFound = errors.New("peer does not hold record") - // Puller fetches a record from a peer into a local directory. type Puller interface { Pull(ctx context.Context, addr, id, dst string) error @@ -47,15 +40,10 @@ type Puller interface { // HTTPPuller pulls records over sandboxd's control-plane HTTP port. type HTTPPuller struct { - // A nil Client uses a default with no overall timeout; the deadline comes from the context. - Client *http.Client - Token string + Token string } func (p *HTTPPuller) Pull(ctx context.Context, addr, id, dst string) error { - ctx, cancel := context.WithTimeout(ctx, pullTimeout) - defer cancel() - req, err := http.NewRequestWithContext(ctx, http.MethodGet, checkpointURL(addr, id, "/blob"), nil) if err != nil { return err @@ -64,8 +52,7 @@ func (p *HTTPPuller) Pull(ctx context.Context, addr, id, dst string) error { req.Header.Set("Authorization", "Bearer "+p.Token) } - client := cmp.Or(p.Client, http.DefaultClient) - resp, err := client.Do(req) + resp, err := http.DefaultClient.Do(req) if err != nil { return fmt.Errorf("pull %s from %s: %w", id, addr, err) } @@ -74,7 +61,7 @@ func (p *HTTPPuller) Pull(ctx context.Context, addr, id, dst string) error { switch resp.StatusCode { case http.StatusOK: case http.StatusNotFound: - return ErrNotFound + return store.ErrNotFound default: return fmt.Errorf("pull %s from %s: unexpected status %d", id, addr, resp.StatusCode) } @@ -227,19 +214,10 @@ func peerURL(addr, path string) string { // safeJoin resolves name under root, rejecting absolute paths and names that escape root. func safeJoin(root, name string) (string, error) { - if name == "" || filepath.IsAbs(name) || strings.Contains(name, `\`) { + if !filepath.IsLocal(name) || strings.Contains(name, `\`) { return "", fmt.Errorf("untar: refusing entry %q", name) } - // clean as-is, never "/"+name: a leading slash absorbs a leading ".." instead of rejecting it. - clean := filepath.Clean(name) - if clean == ".." || strings.HasPrefix(clean, ".."+string(os.PathSeparator)) { - return "", fmt.Errorf("untar: entry %q escapes destination", name) - } - target := filepath.Join(root, clean) - if target != root && !strings.HasPrefix(target, root+string(os.PathSeparator)) { - return "", fmt.Errorf("untar: entry %q escapes destination", name) - } - return target, nil + return filepath.Join(root, name), nil } // tarInto emits only regular files and directories (a symlink would steer the reader outside dst); a file is a size entry plus its non-zero chunks. @@ -249,10 +227,7 @@ func tarInto(src, prefix string, tw *tar.Writer) error { if err != nil { return err } - rel, err := filepath.Rel(src, path) - if err != nil { - return err - } + rel, _ := filepath.Rel(src, path) if rel == "." { return nil } diff --git a/sandboxd/store/peer/transport_test.go b/sandboxd/store/peer/transport_test.go index e7537e65e..56a6e19ea 100644 --- a/sandboxd/store/peer/transport_test.go +++ b/sandboxd/store/peer/transport_test.go @@ -161,10 +161,10 @@ func TestHTTPPullerNotFound(t *testing.T) { })) defer srv.Close() - p := &HTTPPuller{Client: srv.Client()} + p := &HTTPPuller{} err := p.Pull(t.Context(), srv.URL, "ck_0000000000000001", t.TempDir()) - if !errors.Is(err, ErrNotFound) { - t.Fatalf("Pull error = %v, want ErrNotFound so the next owner is tried", err) + if !errors.Is(err, store.ErrNotFound) { + t.Fatalf("Pull error = %v, want store.ErrNotFound so the next owner is tried", err) } } @@ -181,7 +181,7 @@ func TestHTTPPullerStopsReadingAfterAFailedUntar(t *testing.T) { })) t.Cleanup(srv.Close) start := time.Now() - err := (&HTTPPuller{Client: srv.Client()}).Pull(t.Context(), srv.URL, "ck_0000000000000001", t.TempDir()) + err := (&HTTPPuller{}).Pull(t.Context(), srv.URL, "ck_0000000000000001", t.TempDir()) if err == nil || time.Since(start) > 5*time.Second { t.Fatalf("pull of a non-tar body: %v after %s, want a prompt untar error", err, time.Since(start)) } @@ -201,7 +201,7 @@ func TestHTTPPullerPresentsToken(t *testing.T) { })) defer srv.Close() - p := &HTTPPuller{Client: srv.Client(), Token: "fleet-token"} + p := &HTTPPuller{Token: "fleet-token"} dst := t.TempDir() if err := p.Pull(t.Context(), srv.URL, "ck_0000000000000001", dst); err != nil { t.Fatalf("Pull: %v", err) diff --git a/sandboxd/store/s3/s3.go b/sandboxd/store/s3/s3.go index 9ac202cde..71e398ae9 100644 --- a/sandboxd/store/s3/s3.go +++ b/sandboxd/store/s3/s3.go @@ -46,7 +46,6 @@ const ( publishConcurrency = 4 // files scanned in parallel fetchConcurrency = 8 // chunk GETs in flight per fetch transferBudget = 30 * time.Minute - deleteBatch = 1000 // DeleteObjects caps one request at 1000 keys ) var chunkBufs = sync.Pool{New: func() any { @@ -57,6 +56,7 @@ var chunkBufs = sync.Pool{New: func() any { type publishFile struct { path string rel string + size int64 } type manifest struct { @@ -89,9 +89,6 @@ type Store struct { // New builds the backend; ctx bounds the credential-chain resolution. func New(ctx context.Context, cfg store.S3Config, stagingRoot string, idRe *regexp.Regexp) (*Store, error) { - if cfg.Bucket == "" { - return nil, errors.New("s3 checkpoint store needs a bucket") - } // A prefix without its trailing slash hides every record from the delimiter listing. if cfg.Prefix != "" && !strings.HasSuffix(cfg.Prefix, "/") { cfg.Prefix += "/" @@ -170,11 +167,7 @@ func (s *Store) Fetch(ctx context.Context, id string) (string, []byte, string, e func (s *Store) FetchFile(ctx context.Context, id, rel, dst string) error { ctx, cancel := context.WithTimeout(ctx, transferBudget) defer cancel() - meta, _, err := s.readMeta(ctx, id) - if err != nil { - return err - } - man, err := s.readManifest(ctx, id, store.ExportGen(meta)) + man, err := s.currentManifest(ctx, id) if err != nil { return err } @@ -248,8 +241,8 @@ func (s *Store) Delete(ctx context.Context, id string) error { return err } drop := []string{s.key(id, store.LabelsFile)} - for _, key := range objectKeys(objs) { - man, refreshErr := s.refreshManifest(ctx, key) + for _, obj := range objs { + man, refreshErr := s.refreshManifest(ctx, *obj.Key) if refreshErr != nil { return refreshErr } @@ -610,25 +603,23 @@ func (s *Store) putObject(ctx context.Context, key string, body []byte, metadata } func (s *Store) deleteKeys(ctx context.Context, keys []string) error { - for chunk := range slices.Chunk(keys, deleteBatch) { - objs := make([]s3types.ObjectIdentifier, len(chunk)) - for i := range chunk { - objs[i] = s3types.ObjectIdentifier{Key: &chunk[i]} - } - out, err := s.client.DeleteObjects(ctx, &awss3.DeleteObjectsInput{ - Bucket: &s.bucket, - Delete: &s3types.Delete{Objects: objs, Quiet: new(true)}, - }) - if err != nil { - return fmt.Errorf("delete %d objects: %w", len(chunk), err) - } - for _, e := range out.Errors { - // Strict backends error on deleting an absent key where AWS succeeds silently. - if code := aws.ToString(e.Code); code == "NoSuchKey" || code == "NoSuchVersion" { - continue - } - return fmt.Errorf("delete %s: %s", aws.ToString(e.Key), aws.ToString(e.Message)) + objs := make([]s3types.ObjectIdentifier, len(keys)) + for i := range keys { + objs[i] = s3types.ObjectIdentifier{Key: &keys[i]} + } + out, err := s.client.DeleteObjects(ctx, &awss3.DeleteObjectsInput{ + Bucket: &s.bucket, + Delete: &s3types.Delete{Objects: objs, Quiet: new(true)}, + }) + if err != nil { + return fmt.Errorf("delete %d objects: %w", len(keys), err) + } + for _, e := range out.Errors { + // Strict backends error on deleting an absent key where AWS succeeds silently. + if code := aws.ToString(e.Code); code == "NoSuchKey" || code == "NoSuchVersion" { + continue } + return fmt.Errorf("delete %s: %s", aws.ToString(e.Key), aws.ToString(e.Message)) } return nil } @@ -663,15 +654,8 @@ func (u *chunkUploader) scan(ctx context.Context, file publishFile) (store.Diges return store.DigestFile{}, manifestFile{}, err } defer func() { _ = f.Close() }() - info, err := f.Stat() - if err != nil { - return store.DigestFile{}, manifestFile{}, err - } - if err = store.RequireRegular(file.rel, info.Mode()); err != nil { - return store.DigestFile{}, manifestFile{}, err - } - entry := manifestFile{Path: file.rel, Size: info.Size()} - digest, err := store.ScanFile(f, file.rel, info.Size(), getChunkBuf, func(c store.Chunk) error { + entry := manifestFile{Path: file.rel, Size: file.size} + digest, err := store.ScanFile(f, file.rel, file.size, getChunkBuf, func(c store.Chunk) error { entry.Chunks = append(entry.Chunks, manifestChunk{Index: c.Index, SHA256: c.SHA256}) return u.upload(ctx, c) }) @@ -705,21 +689,13 @@ func (u *chunkUploader) claim(sha string) bool { func publishFiles(staging string) ([]publishFile, error) { var files []publishFile - err := store.WalkExport(filepath.Join(staging, store.ExportDir), func(rel, abs string, _ fs.FileInfo) error { - files = append(files, publishFile{path: abs, rel: rel}) + err := store.WalkExport(filepath.Join(staging, store.ExportDir), func(rel, abs string, info fs.FileInfo) error { + files = append(files, publishFile{path: abs, rel: rel, size: info.Size()}) return nil }) return files, err } -func objectKeys(objs []s3types.Object) []string { - keys := make([]string, len(objs)) - for i, obj := range objs { - keys[i] = *obj.Key - } - return keys -} - // pruneIdleGenerations drops cached generations of one record that no Fetch used within the grace; a hit refreshes the mtime. func pruneIdleGenerations(dir, keep string) { entries, err := os.ReadDir(dir) diff --git a/sandboxd/store/store.go b/sandboxd/store/store.go index 00f53f428..a81261883 100644 --- a/sandboxd/store/store.go +++ b/sandboxd/store/store.go @@ -75,7 +75,6 @@ type Store interface { Metas(ctx context.Context) ([]Record, error) // SetLabels replaces a record's labels; nil removes them. The caller holds the record lock. SetLabels(ctx context.Context, id string, labels []byte) error - // Delete removes a record. Delete(ctx context.Context, id string) error // SweepStaging removes abandoned staging left by a crash mid-publish. SweepStaging() error @@ -98,14 +97,6 @@ func ExportGenHash(meta []byte) string { return hex.EncodeToString(sum[:8]) } -// RequireRegular rejects an export entry that is not a regular file. -func RequireRegular(rel string, mode fs.FileMode) error { - if !mode.IsRegular() { - return fmt.Errorf("export entry %s is not a regular file (%s)", rel, mode.Type()) - } - return nil -} - // WalkExport calls fn for every file under a staged export root with its slash-separated path relative to root; any non-regular entry fails the walk. func WalkExport(root string, fn func(rel, path string, info fs.FileInfo) error) error { return filepath.WalkDir(root, func(path string, entry fs.DirEntry, walkErr error) error { @@ -121,17 +112,14 @@ func WalkExport(root string, fn func(rel, path string, info fs.FileInfo) error) if entry.IsDir() { return nil } - rel, err := filepath.Rel(root, path) - if err != nil { - return err - } + rel, _ := filepath.Rel(root, path) rel = filepath.ToSlash(rel) info, err := entry.Info() if err != nil { return err } - if err := RequireRegular(rel, info.Mode()); err != nil { - return err + if !info.Mode().IsRegular() { + return fmt.Errorf("export entry %s is not a regular file (%s)", rel, info.Mode().Type()) } return fn(rel, path, info) }) diff --git a/sandboxd/store/storetest/storetest.go b/sandboxd/store/storetest/storetest.go index 121e13110..7e700f52c 100644 --- a/sandboxd/store/storetest/storetest.go +++ b/sandboxd/store/storetest/storetest.go @@ -172,15 +172,8 @@ func runDigestContract(t *testing.T, st store.Store) { t.Helper() ctx := t.Context() const id = "ck_00000000000000cc" - staging, err := st.Stage(id) - if err != nil { - t.Fatalf("Stage digested: %v", err) - } - writeExport(t, staging, "z.bin", "z") + staging := StageRecord(t, st, id, MetaJSON(id), "z.bin", "z") writeExport(t, staging, "nested/a.bin", "a") - if err = os.WriteFile(filepath.Join(staging, store.MetaFile), []byte(MetaJSON(id)), 0o600); err != nil { - t.Fatalf("write digested meta: %v", err) - } digest, commit, err := st.PrepareDigested(ctx, staging, id, "") if err != nil { t.Fatalf("PrepareDigested: %v", err) @@ -207,14 +200,7 @@ func runDigestContract(t *testing.T, st store.Store) { t.Errorf("Metas %+v, %v; want the one record with digest %q", recs, err, want) } rejectNonRegularReplacement(t, st, id, want) - staging, err = st.Stage(id) - if err != nil { - t.Fatalf("Stage replacement: %v", err) - } - writeExport(t, staging, "z.bin", "zz") - if err = os.WriteFile(filepath.Join(staging, store.MetaFile), []byte(`{"id":"`+id+`","gen":3}`), 0o600); err != nil { - t.Fatalf("write replacement meta: %v", err) - } + staging = StageRecord(t, st, id, `{"id":"`+id+`","gen":3}`, "z.bin", "zz") next, commit, err := st.PrepareDigested(ctx, staging, id, "") if err != nil { t.Fatalf("PrepareDigested replacement: %v", err) diff --git a/sandboxd/tenants/file/file.go b/sandboxd/tenants/file/file.go index ca26a2d6e..12d3ef1ff 100644 --- a/sandboxd/tenants/file/file.go +++ b/sandboxd/tenants/file/file.go @@ -103,39 +103,31 @@ func (s *Source) List(_ context.Context, after string, limit int) ([]tenants.Ten } func (s *Source) Put(_ context.Context, r tenants.TenantRecord) (tenants.Change, error) { - return s.apply(func(cur *set) ([]tenants.TenantRecord, error) { + return s.apply(func(cur *set) (map[string]tenants.TenantRecord, error) { kept, err := tenants.KeepTokens(cur.byName, []tenants.TenantRecord{r}) if err != nil { return nil, err } next := maps.Clone(cur.byName) maps.Copy(next, kept) - return slices.Collect(maps.Values(next)), nil + return next, nil }) } func (s *Source) Delete(_ context.Context, name string) (tenants.Change, error) { - return s.apply(func(cur *set) ([]tenants.TenantRecord, error) { + return s.apply(func(cur *set) (map[string]tenants.TenantRecord, error) { if _, ok := cur.byName[name]; !ok { return nil, tenants.ErrUnknown } next := maps.Clone(cur.byName) delete(next, name) - return slices.Collect(maps.Values(next)), nil + return next, nil }) } func (s *Source) Replace(_ context.Context, records []tenants.TenantRecord) (tenants.Change, error) { - return s.apply(func(cur *set) ([]tenants.TenantRecord, error) { - next, err := tenants.KeepTokens(cur.byName, records) - if err != nil { - return nil, err - } - out := make([]tenants.TenantRecord, len(records)) - for i, r := range records { - out[i] = next[r.Name] - } - return out, nil + return s.apply(func(cur *set) (map[string]tenants.TenantRecord, error) { + return tenants.KeepTokens(cur.byName, records) }) } @@ -161,7 +153,7 @@ func (s *Source) Close() error { return nil } -func (s *Source) apply(next func(cur *set) ([]tenants.TenantRecord, error)) (tenants.Change, error) { +func (s *Source) apply(next func(cur *set) (map[string]tenants.TenantRecord, error)) (tenants.Change, error) { s.mu.Lock() defer s.mu.Unlock() cur := s.set.Load() @@ -169,7 +161,7 @@ func (s *Source) apply(next func(cur *set) ([]tenants.TenantRecord, error)) (ten if err != nil { return tenants.Change{}, err } - updated, err := newSet(records, s.rootSum) + updated, err := newSet(slices.Collect(maps.Values(records)), s.rootSum) if err != nil { return tenants.Change{}, err } diff --git a/sandboxd/tenants/pg/pg.go b/sandboxd/tenants/pg/pg.go index e132d49d6..69243b1a7 100644 --- a/sandboxd/tenants/pg/pg.go +++ b/sandboxd/tenants/pg/pg.go @@ -56,10 +56,8 @@ type Source struct { byName map[string]entry misses map[[sha256.Size]byte]time.Time inflight map[string]struct{} - // seq counts evictions; a read caches only if neither its name nor the whole cache was evicted after it started. - seq uint64 - evicted map[string]uint64 - floor uint64 + // seq counts evictions; a read caches only if none landed after it started. + seq uint64 stop context.CancelFunc done chan struct{} @@ -79,7 +77,6 @@ func Open(ctx context.Context, dsn string) (*Source, error) { byName: map[string]entry{}, misses: map[[sha256.Size]byte]time.Time{}, inflight: map[string]struct{}{}, - evicted: map[string]uint64{}, stop: stop, done: make(chan struct{}), } @@ -121,7 +118,7 @@ func (s *Source) Resolve(ctx context.Context, sum [sha256.Size]byte) (*tenants.T case err != nil: return nil, unavailable(err) } - if s.fresh(r.Name, start) { + if s.seq == start { s.cache(&r, sum) } return &r, nil @@ -338,15 +335,10 @@ func (s *Source) write(ctx context.Context, key string, f func(tx pgx.Tx) error) return nil } -// fresh reports whether a read started at start may cache name; callers hold s.mu. -func (s *Source) fresh(name string, start uint64) bool { - return start >= s.floor && s.evicted[name] <= start -} - // settle caches a by-name read that no eviction overtook; callers hold s.mu. func (s *Source) settle(name string, start uint64, r tenants.TenantRecord, ok bool, err error) { switch { - case err != nil || !s.fresh(name, start): + case err != nil || s.seq != start: case ok: sum, _ := hex.DecodeString(r.TokenSHA256) s.cache(&r, [sha256.Size]byte(sum)) @@ -368,17 +360,14 @@ func (s *Source) evict(key string) { s.mu.Lock() defer s.mu.Unlock() s.seq++ - if key == flushAll || len(s.evicted) >= maxCached { + if key == flushAll { s.clearCache() - clear(s.evicted) - s.floor = s.seq return } if e := s.byName[key]; e.rec != nil { delete(s.byToken, e.sum) } delete(s.byName, key) - s.evicted[key] = s.seq clear(s.misses) } diff --git a/sandboxd/tenants/pg/pg_test.go b/sandboxd/tenants/pg/pg_test.go index 4c65c1b76..817e67b60 100644 --- a/sandboxd/tenants/pg/pg_test.go +++ b/sandboxd/tenants/pg/pg_test.go @@ -152,26 +152,6 @@ func TestARoleWithoutCreateServesAPreCreatedTable(t *testing.T) { } } -func TestAnUnrelatedEvictionKeepsALookupCacheable(t *testing.T) { - s := &Source{evicted: map[string]uint64{}} - start := s.seq - s.seq++ - s.evicted["other"] = s.seq - if !s.fresh("u-1", start) { - t.Error("another tenant's eviction discarded a read of u-1") - } - s.seq++ - s.evicted["u-1"] = s.seq - if s.fresh("u-1", start) { - t.Error("a read that raced u-1's own eviction stays cacheable") - } - s.seq++ - s.floor = s.seq - if s.fresh("u-2", start) { - t.Error("a read that raced a full flush stays cacheable") - } -} - func openIn(t *testing.T, dsn string) *Source { t.Helper() s, err := Open(t.Context(), dsn) diff --git a/sandboxd/types/env.go b/sandboxd/types/env.go index bbc7e8c50..894874c1d 100644 --- a/sandboxd/types/env.go +++ b/sandboxd/types/env.go @@ -166,15 +166,6 @@ func (e Env) Validate() error { return nil } -// Hidden returns the value of a host-only entry. -func (e Env) Hidden(name string) (string, bool) { - v, ok := e[name] - if !ok || v.InGuest() { - return "", false - } - return v.Value, true -} - // HasInject reports whether any entry injects into egress requests. func (e Env) HasInject() bool { for _, v := range e { diff --git a/sandboxd/types/env_test.go b/sandboxd/types/env_test.go index 89a561943..6a85dd423 100644 --- a/sandboxd/types/env_test.go +++ b/sandboxd/types/env_test.go @@ -196,10 +196,12 @@ func TestSandboxPublishesItsEnvToTheProxy(t *testing.T) { func TestEnvSplitsGuestAndHostOnlyEntries(t *testing.T) { env := Env{"G": {Value: "g"}, "H": {Value: "h", Guest: new(false)}, "T": {Value: "t", Guest: new(true)}} - if v, ok := env.Hidden("H"); !ok || v != "h" { - t.Errorf("Hidden(H) = %q %v, want h", v, ok) + sb := &Sandbox{} + sb.SetEnv(env) + if v, ok := sb.HiddenEnv("H"); !ok || v != "h" { + t.Errorf("HiddenEnv(H) = %q %v, want h", v, ok) } - if _, ok := env.Hidden("G"); ok { + if _, ok := sb.HiddenEnv("G"); ok { t.Error("a guest entry resolved as host-only") } if !env.HasGuest() || (Env{"H": env["H"]}).HasGuest() { diff --git a/sandboxd/types/types.go b/sandboxd/types/types.go index 576a7101d..12bac3784 100644 --- a/sandboxd/types/types.go +++ b/sandboxd/types/types.go @@ -65,7 +65,7 @@ var ( } ) -// NetShape selects whether the Cloud Hypervisor guest has a NIC. +// NetShape selects whether the guest has a NIC. type NetShape string // PolicyLayer pins, at claim time, whether the key had a pool layer in its egress policy. @@ -295,7 +295,6 @@ type Sandbox struct { // ReplicaPending is the checkpoint a replication round is about to register; the row may already name it. ReplicaPending string `json:"replica_pending,omitempty"` - // FromCheckpoint names the checkpoint this sandbox branched from; empty otherwise. FromCheckpoint string `json:"from_checkpoint,omitempty"` // TemplateDigest identifies the promoted-template export this sandbox was cloned from. TemplateDigest string `json:"template_digest,omitempty"` @@ -315,11 +314,11 @@ type Sandbox struct { // Waking marks a wake in flight, which counts as running; guarded by the manager mutex. Waking bool `json:"-"` + // Replicated marks a claim of a replicating pool; it is set before the claim is published, written only on a change, and read without a lock. + Replicated bool `json:"-"` // lastActivity is unix-nanos of the last data-plane connection, stamped lock-free. lastActivity atomic.Int64 - // replicated marks a claim of a replicating pool; it is set before the claim is published and read without a lock. - replicated bool // open counts the data-plane connections held right now. open atomic.Int32 // live is the env as of the last SetEnv, read by the egress proxy without the manager mutex. @@ -338,16 +337,6 @@ func (s *Sandbox) TouchAt(t time.Time) { s.lastActivity.Store(t.UnixNano()) } // LastSeen returns the last data-plane activity time. func (s *Sandbox) LastSeen() time.Time { return time.Unix(0, s.lastActivity.Load()) } -// SetReplicated marks the claim as one of a replicating pool; it writes only a change, so a re-add of the same claim never races a reader. -func (s *Sandbox) SetReplicated(v bool) { - if s.replicated != v { - s.replicated = v - } -} - -// Replicated reports whether the claim's pool replicates it. -func (s *Sandbox) Replicated() bool { return s.replicated } - // HasReplica reports whether the claim's registry row may name a replica. func (s *Sandbox) HasReplica() bool { return s.ReplicaCk != "" || s.ReplicaPending != "" } @@ -382,7 +371,9 @@ func (s *Sandbox) SetEnv(e Env) { // HiddenEnv returns a host-only entry's value as last published. func (s *Sandbox) HiddenEnv(name string) (string, bool) { if view := s.live.Load(); view != nil { - return view.env.Hidden(name) + if v, ok := view.env[name]; ok && !v.InGuest() { + return v.Value, true + } } return "", false } diff --git a/sandboxd/types/volume.go b/sandboxd/types/volume.go index b87a955aa..f55fd0b1f 100644 --- a/sandboxd/types/volume.go +++ b/sandboxd/types/volume.go @@ -61,12 +61,9 @@ func ValidDirectIO(mode string) bool { // VolumeNames projects the entries' names in order; nil for none. func VolumeNames(volumes []Volume) []string { - if len(volumes) == 0 { - return nil - } - names := make([]string, len(volumes)) - for i, volume := range volumes { - names[i] = volume.Name + var names []string + for _, volume := range volumes { + names = append(names, volume.Name) } return names } diff --git a/sandboxd/utils/utils.go b/sandboxd/utils/utils.go index a75d88122..8d4180f7a 100644 --- a/sandboxd/utils/utils.go +++ b/sandboxd/utils/utils.go @@ -74,18 +74,16 @@ func WriteFileSync(path string, data []byte, perm os.FileMode) error { // ReplaceFileSync writes data to the open temp file f, fsyncs it, renames it over path and fsyncs the directory; f is closed and removed on failure. func ReplaceFileSync(f *os.File, path string, data []byte) (err error) { tmp := f.Name() - renamed := false defer func() { - if !renamed { + if err != nil { + _ = f.Close() _ = os.Remove(tmp) } }() if _, err = f.Write(data); err != nil { - _ = f.Close() return fmt.Errorf("write %s: %w", tmp, err) } if err = f.Sync(); err != nil { - _ = f.Close() return fmt.Errorf("sync %s: %w", tmp, err) } if err = f.Close(); err != nil { @@ -94,7 +92,6 @@ func ReplaceFileSync(f *os.File, path string, data []byte) (err error) { if err = os.Rename(tmp, path); err != nil { return fmt.Errorf("rename %s: %w", path, err) } - renamed = true return SyncDir(filepath.Dir(path)) } diff --git a/sdk/langchain/tests/test_toolkit.py b/sdk/langchain/tests/test_toolkit.py index 12b8b8766..c8945045e 100644 --- a/sdk/langchain/tests/test_toolkit.py +++ b/sdk/langchain/tests/test_toolkit.py @@ -11,6 +11,39 @@ from cocoonsandbox_langchain.toolkit import CALL_TIMEOUT +class FakeSandbox: + def __init__(self): + self.closed = 0 + self.close_error = None + self.files = {} + + def run(self, argv, cwd="", on_stdout=None, on_stderr=None, timeout=None, **_): + assert argv[:2] == ["sh", "-c"], argv + assert timeout is not None and 0 < timeout <= 300, timeout + if argv[2] == "boom": + on_stderr(b"kaboom\n") + return 3 + if argv[2] == "hang": + on_stdout(b"partial\n") + raise TimeoutError("cut") + on_stdout(f"ran: {argv[2]}\n".encode()) + return 0 + + def write_file(self, path, data): + self.files[path] = data + + def read_file(self, path): + return self.files[path] + + def list_dir(self, path): + return [{"name": "a.txt", "kind": "file", "size": 3}] + + def close(self): + self.closed += 1 + if self.close_error is not None: + raise self.close_error + + def test_tools_shape(monkeypatch): kit, _ = hooked(monkeypatch) tools = kit.get_tools() @@ -152,39 +185,6 @@ def test_a_stalled_agent_upgrade_comes_back_as_a_tool_error(): server.close() -class FakeSandbox: - def __init__(self): - self.closed = 0 - self.close_error = None - self.files = {} - - def run(self, argv, cwd="", on_stdout=None, on_stderr=None, timeout=None, **_): - assert argv[:2] == ["sh", "-c"], argv - assert timeout is not None and 0 < timeout <= 300, timeout - if argv[2] == "boom": - on_stderr(b"kaboom\n") - return 3 - if argv[2] == "hang": - on_stdout(b"partial\n") - raise TimeoutError("cut") - on_stdout(f"ran: {argv[2]}\n".encode()) - return 0 - - def write_file(self, path, data): - self.files[path] = data - - def read_file(self, path): - return self.files[path] - - def list_dir(self, path): - return [{"name": "a.txt", "kind": "file", "size": 3}] - - def close(self): - self.closed += 1 - if self.close_error is not None: - raise self.close_error - - def hooked(monkeypatch): kit = CocoonToolkit("127.0.0.1:1") fake = FakeSandbox() diff --git a/sdk/openai/cocoonsandbox_openai/__init__.py b/sdk/openai/cocoonsandbox_openai/__init__.py index a39bf6199..73876512a 100644 --- a/sdk/openai/cocoonsandbox_openai/__init__.py +++ b/sdk/openai/cocoonsandbox_openai/__init__.py @@ -1,11 +1,4 @@ -"""OpenAI Agents SDK sandbox provider for cocoon microVMs. - -from cocoonsandbox_openai import CocoonSandboxClient, CocoonSandboxClientOptions - -client = CocoonSandboxClient() -session = await client.create(options=CocoonSandboxClientOptions( - addr="10.0.0.5:7777", api_token="...", template="ghcr.io/cocoonstack/sandbox/rt:24.04")) -""" +"""OpenAI Agents SDK sandbox provider for cocoon microVMs.""" from .adapter import ( CocoonSandboxClient, diff --git a/sdk/python/cocoonsandbox/__init__.py b/sdk/python/cocoonsandbox/__init__.py index 449f37de8..bcf30fac8 100644 --- a/sdk/python/cocoonsandbox/__init__.py +++ b/sdk/python/cocoonsandbox/__init__.py @@ -1,13 +1,4 @@ -"""Python SDK for the cocoon sandbox control plane: claim a microVM from a -sandboxd node, run commands over the relayed silkd protocol, branch it with -checkpoints, release it. stdlib-only. - - from cocoonsandbox import Client - - client = Client("node:7777", api_token="...") - with client.new("ghcr.io/cocoonstack/sandbox/rt:24.04") as sb: - print(sb.exec("echo", "hello")) -""" +"""Python SDK for the cocoon sandbox control plane; stdlib-only.""" from .checkpoint import Checkpoint from .client import Client diff --git a/sdk/python/cocoonsandbox/client.py b/sdk/python/cocoonsandbox/client.py index cdf8587ec..c3737de96 100644 --- a/sdk/python/cocoonsandbox/client.py +++ b/sdk/python/cocoonsandbox/client.py @@ -15,7 +15,7 @@ from .checkpoint import Checkpoint from .conn import Conn, dial_agent, is_quiet, remaining_timeout from .endpoint import _endpoint_url -from .errors import APIError, SandboxTimeout, require_field +from .errors import APIError, SandboxTimeout, _error_message, require_field from .sandbox import Sandbox from .template import _claim_body, _template_query from .volumes import Volume @@ -259,7 +259,6 @@ def _roundtrip( return resp.status, raw def _take(self, key: tuple[str, str], timeout: float) -> http.client.HTTPConnection: - """Reuses a parked connection the node has not closed within keep_alive, else dials.""" with self._idle_lock: idle = self._idle.get(key, []) while idle: @@ -280,13 +279,6 @@ def _tls(self) -> ssl.SSLContext: return self._ssl_context -def _error_message(raw: bytes) -> str: - try: - return cast(str, json.loads(raw)["error"]) - except (ValueError, KeyError, TypeError): - return raw.decode(errors="replace").strip() - - def _retry_miss(exc: APIError) -> bool: return exc.status in (404, 0) diff --git a/sdk/python/cocoonsandbox/conn.py b/sdk/python/cocoonsandbox/conn.py index 2dc5c21d2..b37cc17e4 100644 --- a/sdk/python/cocoonsandbox/conn.py +++ b/sdk/python/cocoonsandbox/conn.py @@ -12,7 +12,7 @@ from collections.abc import Callable, Iterator from typing import Any, BinaryIO, Protocol, TypeVar, cast -from .errors import APIError, ProtocolError, SandboxError, SandboxTimeout, SilkdError +from .errors import APIError, ProtocolError, SandboxError, SandboxTimeout, SilkdError, _error_message from .frames import MAX_FRAME, decode_response, encode_request KEEP_ALIVE_CONNS = 8 @@ -24,8 +24,6 @@ class _Closeable(Protocol): - """Context-manager mixin for handles whose exit is just close().""" - def __enter__(self: _CloseableT) -> _CloseableT: return self @@ -294,5 +292,5 @@ def _read_upgrade(reader: BinaryIO, arm: Callable[[], None]) -> None: raise ProtocolError("invalid content-length in upgrade reply") from exc if code != 101: arm() - body = reader.read(min(body_len, MAX_FRAME)).decode(errors="replace") if body_len else "" - raise APIError("agent upgrade", code, body.strip() or status.strip()) + body = reader.read(min(body_len, MAX_FRAME)) if body_len else b"" + raise APIError("agent upgrade", code, _error_message(body) or status.strip()) diff --git a/sdk/python/cocoonsandbox/errors.py b/sdk/python/cocoonsandbox/errors.py index 7daf97da4..52b26e073 100644 --- a/sdk/python/cocoonsandbox/errors.py +++ b/sdk/python/cocoonsandbox/errors.py @@ -2,7 +2,8 @@ from __future__ import annotations -from typing import TYPE_CHECKING, Any +import json +from typing import TYPE_CHECKING, Any, cast if TYPE_CHECKING: from .volumes import SandboxVolumesResponse @@ -63,3 +64,10 @@ def require_field(reply: dict[str, Any], key: str, verb: str) -> Any: if key not in reply: raise APIError(verb, 0, f"reply without {key}") return reply[key] + + +def _error_message(raw: bytes) -> str: + try: + return cast(str, json.loads(raw)["error"]) + except (ValueError, KeyError, TypeError): + return raw.decode(errors="replace").strip() diff --git a/sdk/python/tests/test_proc.py b/sdk/python/tests/test_proc.py index e1b6327d0..b71692232 100644 --- a/sdk/python/tests/test_proc.py +++ b/sdk/python/tests/test_proc.py @@ -9,6 +9,25 @@ from cocoonsandbox.frames import FS_CHUNK +class BlockingStdinConn(FakeConn): + """A guest that stops draining stdin until its output is read, so send blocks past the buffer.""" + + def __init__(self, frames, buffer_frames): + super().__init__(frames) + self._room = threading.Semaphore(buffer_frames) + self._read = False + + def send(self, op, **fields): + if op == "stdin" and not self._read and not self._room.acquire(blocking=False): + assert self._room.acquire(timeout=5), "stdin send deadlocked" + super().send(op, **fields) + + def recv(self): + self._read = True + self._room.release() + return super().recv() + + def test_spawn_returns_pid(monkeypatch): sb, conn = fake_sandbox(monkeypatch, [{"type": "started", "pid": 41}]) assert sb.spawn("sleep", "30") == 41 @@ -75,25 +94,6 @@ def test_a_text_stdin_is_refused_before_dialing(monkeypatch, call): assert dials == [] -class BlockingStdinConn(FakeConn): - """A guest that stops draining stdin until its output is read, so send blocks past the buffer.""" - - def __init__(self, frames, buffer_frames): - super().__init__(frames) - self._room = threading.Semaphore(buffer_frames) - self._read = False - - def send(self, op, **fields): - if op == "stdin" and not self._read and not self._room.acquire(blocking=False): - assert self._room.acquire(timeout=5), "stdin send deadlocked" - super().send(op, **fields) - - def recv(self): - self._read = True - self._room.release() - return super().recv() - - def fake_sandbox(monkeypatch, frames): sb = sandbox_at("127.0.0.1:1", keep_alive=0) conn = FakeConn(frames) diff --git a/sdk/python/tests/test_upgrade.py b/sdk/python/tests/test_upgrade.py index 65bc8e654..6d033fa7b 100644 --- a/sdk/python/tests/test_upgrade.py +++ b/sdk/python/tests/test_upgrade.py @@ -74,6 +74,19 @@ def test_rejected_upgrade_is_an_api_error(): assert agent.early == ["fs_mkdir"], agent.early +def test_rejected_upgrade_reports_the_servers_error_text(): + body = b'{"error":"unknown sandbox"}' + head = b"HTTP/1.1 404 Not Found\r\nContent-Type: application/json\r\nContent-Length: %d\r\n\r\n" % len(body) + agent = EarlyFrameAgent(head + body) + sb = sandbox_at(agent.addr, timeout=5, keep_alive=0) + try: + with pytest.raises(APIError) as refused: + sb.mkdir("/w") + finally: + agent.stop() + assert (refused.value.status, refused.value.message) == (404, "unknown sandbox"), refused.value + + def test_run_timeout_cuts_an_unanswered_upgrade(black_hole): sb = sandbox_at(black_hole, timeout=30, keep_alive=0) started = time.monotonic() diff --git a/sdk/python/tests/test_wire_binding.py b/sdk/python/tests/test_wire_binding.py index 226317c2c..a9525541e 100644 --- a/sdk/python/tests/test_wire_binding.py +++ b/sdk/python/tests/test_wire_binding.py @@ -80,6 +80,27 @@ UNSENT = {"req_session_create": {"id"}} +class BranchActionConn(FakeConn): + """Records the action each git_branch verb puts on the wire.""" + + def __init__(self, sent): + super().__init__([]) + self.sent = sent + self.action = "" + + def send(self, op, **fields): + self.action = fields.get("action") + self.sent.append(self.action) + + def recv(self): + if self.action == "list": + return {"type": "git_branches", "branches": [], "current": ""} + return {"type": "done"} + + def recv_until(self, *terminal): + yield self.recv() + + @pytest.mark.parametrize("stem,replies,invoke", CASES, ids=lambda c: c if isinstance(c, str) else "") def test_call_site_matches_fixture(monkeypatch, stem, replies, invoke): fixture = json.loads((FIXTURES / f"{stem}.json").read_text()) @@ -122,27 +143,6 @@ def test_git_branch_actions_come_from_the_corpus(monkeypatch): assert set(sent) == set(enums["git_branch_action"]) -class BranchActionConn(FakeConn): - """Records the action each git_branch verb puts on the wire.""" - - def __init__(self, sent): - super().__init__([]) - self.sent = sent - self.action = "" - - def send(self, op, **fields): - self.action = fields.get("action") - self.sent.append(self.action) - - def recv(self): - if self.action == "list": - return {"type": "git_branches", "branches": [], "current": ""} - return {"type": "done"} - - def recv_until(self, *terminal): - yield self.recv() - - def fake_sandbox(monkeypatch, replies): """A Sandbox whose _dial yields a real Conn over a socketpair to a scripted guest thread.""" sent = [] diff --git a/silkd/src/pty.rs b/silkd/src/pty.rs index 9991c1c13..163282308 100644 --- a/silkd/src/pty.rs +++ b/silkd/src/pty.rs @@ -123,9 +123,9 @@ async fn pump( ) -> i32 { // stdin rides its own task so a slow write cannot stall output or reaping. let (disc_tx, mut disc_rx) = oneshot::channel::<()>(); - tokio::spawn(pump_stdin(Arc::clone(master), client, disc_tx)); + let stdin = tokio::spawn(pump_stdin(Arc::clone(master), client, disc_tx)); - let mut buf = [0u8; READ_CHUNK]; + let mut buf = vec![0u8; READ_CHUNK]; let mut frame = Vec::new(); let mut eof = false; let mut disc = false; @@ -134,6 +134,7 @@ async fn pump( status = child.wait() => { let code = status.map_or(-1, sysutil::exit_code); drain(master, proc, out, &mut buf, &mut frame).await; + stdin.abort(); return code; } // kill the shell so child.wait publishes its real signalled code. @@ -153,6 +154,7 @@ async fn pump( proc.emit_bytes(false, &buf[..n]); if proto::write_chunk_frame(out, &mut frame, "stdout", &buf[..n]).await.is_err() { let _ = child.start_kill(); + stdin.abort(); return sysutil::wait_code(child).await; } } diff --git a/silkd/src/session.rs b/silkd/src/session.rs index 04251d33a..9f2e1b3b3 100644 --- a/silkd/src/session.rs +++ b/silkd/src/session.rs @@ -66,7 +66,6 @@ impl Table { stdout, _child: child, cmd_buf: String::new(), - buf: vec![0u8; READ_CHUNK], acc: Vec::new(), frame: Vec::new(), }), @@ -183,7 +182,6 @@ struct Io { stdout: ChildStdout, _child: Child, cmd_buf: String, - buf: Vec, acc: Vec, frame: Vec, } @@ -211,20 +209,19 @@ impl Io { let keep = mb.len().saturating_sub(1); self.acc.clear(); loop { - let n = self.stdout.read(&mut self.buf).await?; + self.acc.reserve(READ_CHUNK); + let n = self.stdout.read_buf(&mut self.acc).await?; if n == 0 { return Err(io::Error::new(io::ErrorKind::UnexpectedEof, "shell exited")); } - self.acc.extend_from_slice(&self.buf[..n]); if let Some(pos) = memchr::memmem::find(&self.acc, mb) { emit(&mut out, &mut self.frame, &self.acc[..pos]).await; self.acc.drain(..pos + mb.len()); while memchr::memchr(b'\n', &self.acc).is_none() && self.acc.len() < EXIT_TAIL_MAX { - let m = self.stdout.read(&mut self.buf).await?; - if m == 0 { + self.acc.reserve(READ_CHUNK); + if self.stdout.read_buf(&mut self.acc).await? == 0 { break; } - self.acc.extend_from_slice(&self.buf[..m]); } // parse only the exit-code line: a background writer can land bytes after the newline. let end = memchr::memchr(b'\n', &self.acc).unwrap_or(self.acc.len()); diff --git a/silkd/tests/fs_e2e.rs b/silkd/tests/fs_e2e.rs index 6fde8be13..d0391871d 100644 --- a/silkd/tests/fs_e2e.rs +++ b/silkd/tests/fs_e2e.rs @@ -1,6 +1,5 @@ //! Filesystem verb E2E over the in-memory duplex, against real temp paths. -#![allow(clippy::unwrap_used, clippy::expect_used)] mod common; use std::os::unix::fs::{MetadataExt, PermissionsExt}; diff --git a/silkd/tests/keepalive_e2e.rs b/silkd/tests/keepalive_e2e.rs index a9fb69890..209bea571 100644 --- a/silkd/tests/keepalive_e2e.rs +++ b/silkd/tests/keepalive_e2e.rs @@ -1,6 +1,5 @@ //! RPCs back to back on one connection, and the frames that cross an RPC boundary. -#![allow(clippy::unwrap_used, clippy::expect_used)] mod common; use std::sync::Arc; diff --git a/silkd/tests/pty_e2e.rs b/silkd/tests/pty_e2e.rs index 71e9da61a..bba55342d 100644 --- a/silkd/tests/pty_e2e.rs +++ b/silkd/tests/pty_e2e.rs @@ -120,6 +120,37 @@ async fn resize_unknown_pid_is_not_found() { assert_error(&frames[0], "not_found"); } +#[tokio::test] +async fn stdin_nobody_reads_after_the_shell_exits_does_not_wedge_the_next_request() { + let state = Arc::new(State::new()); + let (mut cw, mut lines, handle) = connect(&state); + send(&mut cw, json!({"op":"pty_open","cols":80,"rows":24})).await; + read_until(&mut lines, |v| v["type"] == "started").await; + send( + &mut cw, + json!({"op":"stdin","data":b64(b"sleep 20 &\nexit\n")}), + ) + .await; + read_until(&mut lines, |v| v["type"] == "exit").await; + + let chunk = b64(&b"unread line\n".repeat(2048)); + let answered = tokio::time::timeout(Duration::from_secs(8), async { + for _ in 0..64 { + send(&mut cw, json!({"op":"stdin","data":chunk})).await; + } + send(&mut cw, json!({"op":"ps"})).await; + read_until(&mut lines, |v| v.get("procs").is_some()).await; + }) + .await; + assert!( + answered.is_ok(), + "the next request is wedged behind stdin nobody reads" + ); + + cw.shutdown().await.unwrap(); + handle.await.unwrap().unwrap(); +} + #[tokio::test] async fn pty_disconnect_tears_down_without_spin() { let state = Arc::new(State::new()); diff --git a/silkd/tests/watch_e2e.rs b/silkd/tests/watch_e2e.rs index b1266a3d5..9bd5d5585 100644 --- a/silkd/tests/watch_e2e.rs +++ b/silkd/tests/watch_e2e.rs @@ -1,6 +1,5 @@ //! fs.watch E2E: a connection-bound event stream ended by client disconnect. -#![allow(clippy::unwrap_used, clippy::expect_used)] mod common; use std::sync::Arc;