diff --git a/.github/workflows/sandboxd.yml b/.github/workflows/sandboxd.yml index 78063b1e4..e84e19521 100644 --- a/.github/workflows/sandboxd.yml +++ b/.github/workflows/sandboxd.yml @@ -15,6 +15,7 @@ on: - "protocol/**" - "Makefile" - ".golangci.yml" + - ".goreleaser.yml" - ".github/workflows/sandboxd.yml" pull_request: paths: @@ -26,6 +27,7 @@ on: - "protocol/**" - "Makefile" - ".golangci.yml" + - ".goreleaser.yml" - ".github/workflows/sandboxd.yml" workflow_dispatch: @@ -57,9 +59,11 @@ jobs: go-version-file: sandboxd/go.mod cache-dependency-path: "**/go.sum" - - name: Build sandboxd - working-directory: sandboxd - run: go build ./... + - name: Check the release config + uses: goreleaser/goreleaser-action@v7.3.0 + with: + version: v2.18.2 + args: check - name: Lint run: make go-lint diff --git a/Makefile b/Makefile index b0c3de6f5..6a6ea142b 100644 --- a/Makefile +++ b/Makefile @@ -15,6 +15,8 @@ EXTRACT_IMAGE ?= $(BOOT_IMAGE) GO_MODULES := protocol/wire sandboxd sdk/go e2e mcp GO_OSES := linux darwin +.DEFAULT_GOAL := help + ## Location to install dependencies to LOCALBIN ?= $(shell pwd)/bin $(LOCALBIN): diff --git a/docs/advanced-network.md b/docs/advanced-network.md index afa5df1d8..9668f8b34 100644 --- a/docs/advanced-network.md +++ b/docs/advanced-network.md @@ -69,8 +69,7 @@ answers. | 10 | 2 | flags | 0 | bit 0 `ON`: packet mode is on for this box | Flag bits 1 to 15 are reserved and sent as 0; a peer ignores a bit it does -not know. A later revision uses one to open several streams per guest, one -per `tun0` queue. +not know. A host that answers without `ON` closes the stream after the hello; the guest removes `tun0`. A host that refuses a guest answers without `ON` rather than @@ -129,7 +128,7 @@ guest -> tun0 -> silkd --vsock 2051--> sandboxd --VXLAN--> exit (vxlan0, forward disk write on the claim path; each release is appended to `/tun-freed.log` so its reuse window outlives a restart. A node with a `mesh` must have `meta_store.cell` and allocates it in the cell's - [table](deploy.md#shared-tun-address-table), one round trip per claim, and stays off (proxy mode only) while the + [table](deploy.md#shared-tun-address-table), one round trip per candidate address, and stays off (proxy mode only) while the meta store is unreachable. A freed address is reused after `tun.reuse_after_seconds`, so the subnet's size over that window bounds the sustained claim rate: a `/8` with the 5-day default carries about 38 new tun @@ -139,8 +138,8 @@ guest -> tun0 -> silkd --vsock 2051--> sandboxd --VXLAN--> exit (vxlan0, forward whatever the guest wrote; a GSO frame is cut into wire-sized packets with full checksums; each gets an Ethernet header (claim MAC to the exit's MAC) and a VXLAN header and leaves for the exit picked by the claim's profile - label (`default` when it has none). The outer source port follows a hash of - the inner 5-tuple. + label (`default` when it has none). The outer source port is one of `senders` + sockets, picked by a hash of the protocol, the ports and the inner destination. - **Inbound**: a VXLAN packet from an exit's VTEP whose inner destination is a claim on this node goes back to `100.64.0.2` and onto that claim's stream; anything else is dropped. Inbound passes only for a flow the claim opened, @@ -152,7 +151,7 @@ guest -> tun0 -> silkd --vsock 2051--> sandboxd --VXLAN--> exit (vxlan0, forward and port-unreachable reach the guest's socket. - **ARP**: sandboxd answers ARP from the exits for its own claims only, announces a claim with a gratuitous ARP when it is armed (claim, wake, - move-in), after a reload of `tun.exits` and every 45 seconds, and never + move-in), after every config reload and every 45 seconds, and never forwards a frame the guest built. - **Policy**: a flow table keyed by 5-tuple; the first packet of a flow is judged by `EvalFlow` (bare `host: "*"` rules only) and the internal-address diff --git a/docs/benchmarks.md b/docs/benchmarks.md index d9a1871c8..1170bd94c 100644 --- a/docs/benchmarks.md +++ b/docs/benchmarks.md @@ -74,9 +74,9 @@ On exit it removes every `sbx-*` VM on the cocoon root, a serving sandboxd's claims included, so run it only on a node that serves no claims. The daemon also reclaims every `sbx-` VM at start. -Knobs (environment variables): `WARM`/`WARM_N` (warm-pool depth and burst -size — the burst must stay within the depth, or refill loses the race and -the tail measures clones), `CLONE_N`, `COLD_N`, `RPC_N`, `PULL_MB`, +Knobs (environment variables): `WARM`/`WARM_N` (warm-pool depth and the +number of back-to-back warm-tier claims, which must stay within the depth, or +refill loses the race and the tail measures clones), `CLONE_N`, `COLD_N`, `RPC_N`, `PULL_MB`, `PULL_N`, `BURST_N` (concurrent clone-claim burst; 0 skips the stage), `ADDR`/`TOKEN` (the throwaway daemon's listen address and token). diff --git a/docs/cluster.md b/docs/cluster.md index 027e5d982..09876815c 100644 --- a/docs/cluster.md +++ b/docs/cluster.md @@ -3,7 +3,7 @@ A cluster is a set of sandboxd nodes joined through a [hashicorp/memberlist](https://github.com/hashicorp/memberlist) SWIM mesh. Gossip carries only placement hints — per-pool warm counts, promoted-template -hashes, available volume names, each node's data-plane address, and a digest +hashes, available volume names, each node's data-plane and client addresses, and a digest of the cluster-invariant config. Gossip carries no per-sandbox state, so a stale view costs at most one extra redirect, never correctness. A single node with no seeds is a valid @@ -444,7 +444,7 @@ guest resumes on the source unless the target committed. **Counting.** The target admits the move like a new claim: drain, `disk_reserve`, `memory_limit`, node `max_claims` and `max_running`, the -tenant's `max_claims` and `move_concurrency`, and holds a reservation from admission to its commit. A +tenant's `max_claims`, and the node's `move_concurrency`, and holds a reservation from admission to its commit. A [cordon](sandboxd-api.md#post-v1cordon) refuses only new claims, so a cordoned target admits the move. The reservation counts in every cap but is no claim: listings, reaps and gossip never see it. At the commit the reservation becomes the claim under one lock, @@ -596,7 +596,9 @@ checkpoint store and keeps each claim's registry row on the latest snapshot: ports, preview, fork, checkpoint and promote answer `503 lease lost; retry` (a read through a [reader](deploy.md#reading-sleeping-claims-without-waking-them) after the `101` gets an `error` frame instead), and `/owner` answers 404 so - a `Lookup` moves on to the other nodes. A claim is replicated when its policy + a `Lookup` moves on to the other nodes. Its proxy, SOCKS5 and intercept doors deny those claims and the trip + ends their open tunnels, and a new claim of a replicating pool meets an + [admission refusal](sandboxd-api.md#admission-refusals), which redirects it to a peer. A claim is replicated when its policy pool has `replicate_seconds`, from the moment it is claimed, adopted or loaded, whether or not a round has run, so the fence covers it before its first replica; the setting is restart-only so that decision cannot go stale. diff --git a/docs/deploy.md b/docs/deploy.md index 7a01127d7..90e40da2a 100644 --- a/docs/deploy.md +++ b/docs/deploy.md @@ -104,7 +104,7 @@ the intersection of the two. | `cocoon_bin` | `cocoon` | cocoon CLI binary | | `restore_mode` | unset | clone and wake-restore memory mode: `copy`, `ondemand`, or `mmap`; use `mmap` for dense pools | | `no_direct_io` | false | use buffered writable disks for Cloud Hypervisor cold boots and clones; recommended for dense ephemeral pools to avoid direct-I/O CoW journal contention | -| `sync_claims` | false | fsync `claims.json` and its directory on every commit (claim, release, renew, hibernate, wake, fork, archive, reap), about 0.3 ms each on NVMe; without it a host power loss can leave an empty journal, which fails the next start's reconcile until the file is removed, after which hibernate snapshots are reaped as orphans and archive checkpoints stay stranded. Off by default; turn it on when running claims must come back from a power loss (see `vmm_restart`) | +| `sync_claims` | false | fsync `claims.json` and its directory on every commit (claim, release, renew, hibernate, wake, fork, archive, reap), about 0.3 ms each on NVMe; without it a host power loss can leave an empty journal, which fails the next start's reconcile until the file is removed, after which hibernate snapshots are reaped as orphans and archive checkpoints stay stranded. Off by default; turn it on when running claims must come back from a power loss (see `vmm_restart`). It does not cover `usage.jsonl`, whose last lines a power loss can still drop | | `cocoond_socket` | unset | `cocoon daemon`'s read-only API socket (by default `/cocoond.sock`); set, the node detects VMM exits from its event stream and polls only as a 60 s backstop. See [dead VMMs](#dead-vmms-and-host-restarts) | | `vmm_restart` | `cold` | what the node does with a running claim whose VMM exited outside a transition (a VMM crash, or a host restart that took it): `cold` cold-boots the claim's own VM from its disk under the same sandbox id and token (guest memory, processes, and the instance-metadata document are lost), `none` keeps the claim listed as `failed` for the caller to release or wake. Egress-lane and volume claims, a VM record gone from cocoon, and a VMM that exits more than 3 times in 10 minutes are marked `failed` under either value. See [dead VMMs](#dead-vmms-and-host-restarts) | | `no_balloon` | false | boot pool and template VMs without the virtio-balloon (cocoon otherwise returns 25% of guest memory to the host); clones inherit it from the golden. A guest that thrashes before deflate-on-OOM fires — a 16G build tier running a large typecheck — needs its whole memory | @@ -429,7 +429,7 @@ generation sweep. | `/meta.json` | The live record, uploaded last as the commit marker. Absent once the record is deleted. Its current generation is `export-` plus the first 16 hex characters of the SHA-256 of the object body. | | `/labels.json` | Optional labels, removed with the record. | | `manifests//.json` | One JSON manifest per published generation: `lineage` and, per file, `path`, `size` and the present `chunks` (`index`, `sha256`). `` is `export-<16 hex>`. | -| `chunks//` | A 16 MiB chunk. A lineage is the dedup scope: it defaults to the record id, and every checkpoint and archive of one claim shares the claim's sandbox id. A chunk is shared by every record of its lineage that names the same digest, and only by those. | +| `chunks//` | A chunk of up to 16 MiB; a file's last chunk is shorter. A lineage is the dedup scope: it defaults to the record id, and every checkpoint and archive of one claim shares the claim's sandbox id. A chunk is shared by every record of its lineage that names the same digest, and only by those. | | `lineages//` | Zero-byte marker: record `` published into another record's lineage ``, written after the manifest and before `meta.json`; a record in its own lineage writes none. A publish into a shared lineage reads only the records these list, so a crash between manifest and marker only costs a missed dedup. It only drives dedup, never chunk liveness. | **Offline reclamation only.** Stop every sandboxd node using the prefix and @@ -1202,7 +1202,7 @@ some of them on a node: | `portsmoke`, `guestserver` | the guest-port relay; `guestserver` is its guest listener | `port-e2e.sh` | | `ringsmoke` | the output ring cap and exec as a user | | | `androidsmoke`, `browsersmoke` | the android and browser flavors | | -| `pullbench`, `pushbench`, `rpcbench`, `qaab` | performance measurements | | +| `pullbench`, `pushbench`, `rpcbench` | performance measurements | | `scripts/bench.sh` is the published [benchmark](benchmarks.md#reproduce) procedure and `scripts/boot-bench.sh` times the boot phases. diff --git a/docs/egress.md b/docs/egress.md index 481c01ca6..539361867 100644 --- a/docs/egress.md +++ b/docs/egress.md @@ -158,7 +158,7 @@ proxy over the same vsock path. To stop it bypassing the proxy, sandboxd locks the NIC at claim — until then a warm VM of the pool, and the pool's golden build, sit on the bridge unlocked, running only the image's own services, never tenant code: an nftables netdev table (`sandbox_egress_`) with an -ingress hook on the guest's tap drops every guest-initiated packet, so the guest's +ingress hook on the guest's tap drops every packet the guest sends, so the guest's only egress is the audited vsock proxy. A locked guest therefore cannot renew a DHCP lease and loses its NIC address when the lease ends, which nothing on this lane needs. The diff --git a/docs/sdk.md b/docs/sdk.md index a03027bac..d06cf7e10 100644 --- a/docs/sdk.md +++ b/docs/sdk.md @@ -219,7 +219,7 @@ without `Stdin`, `WriteFile`, `OpenPty`, `Watch`, `DialPort`, `Lsp.Request`, `Logs`, `Attach`, and the file, git, process and session calls. `Run` with `Stdin` and `Push` stream the caller's reader, so they fail once and the next call reaches the new owner. An -error from the caller's own writer is returned as it is and never retried, and a +error from the caller's own writer is returned wrapped (match it with `errors.Is`) and never retried, and a released sandbox still answers 404. Volume sandboxes and egress-lane sandboxes cannot hibernate, fork, checkpoint, or promote, nor take `WithArchiveOnExpire`; each answers 409, as does a promote diff --git a/docs/security.md b/docs/security.md index 9841f08c8..66d999d4f 100644 --- a/docs/security.md +++ b/docs/security.md @@ -34,7 +34,7 @@ exposing any part of a deployment beyond a single trusted host. `host: "*"` rules and the internal-address guard allow, under per-claim rate, flow and quota limits. On the egress lane: the same vsock paths plus a NIC whose every - guest-initiated packet is dropped by an nftables lock in the host + outgoing packet, replies and ARP included, is dropped by an nftables lock in the host root netns ([egress](egress.md)); the lock is fail-closed and applied before the claim is handed out. On both lanes the proxy refuses loopback, private, link-local (cloud metadata), CGN, the diff --git a/docs/silkd.md b/docs/silkd.md index 6478feffd..48487b0dc 100644 --- a/docs/silkd.md +++ b/docs/silkd.md @@ -28,7 +28,7 @@ RPC still runs ends that RPC's input and is served after it, except in the upload of `fs_write` or `fs_push`: there a request frame before `data_end` fails the upload with `bad_request` and is dropped. A client may close after any terminal frame. A frame that does not parse is answered with -`bad_request` and closes the connection. Binary payloads ride base64 in `data` +`bad_request` and, outside an upload, closes the connection. Binary payloads ride base64 in `data` fields. Frames are capped at 8 MiB; requests carry `"v": 1` and unknown fields are ignored (the forward-compatibility story). @@ -42,7 +42,7 @@ connection-bound verbs are `fs_watch`, `pty_open`, `lsp_request`, and `port_forward`. The authoritative wire contract is the shared fixture corpus in -`protocol/wire/fixtures/v1`, round-tripped by the Rust and Go test suites; the +`protocol/wire/fixtures/v1`: the Go suite round-trips every fixture, Rust parses the requests and round-trips the responses, and the Python suite round-trips the request frames and checks that each response decodes to the fixture's `type` tag and `data` — a frame the Rust or Go side cannot parse fails CI. diff --git a/e2e/cmd/androidsmoke/main.go b/e2e/cmd/androidsmoke/main.go index 8efa5af0d..2ac37a084 100644 --- a/e2e/cmd/androidsmoke/main.go +++ b/e2e/cmd/androidsmoke/main.go @@ -80,7 +80,7 @@ func run(addr, token, template string) error { } func initTree(ctx context.Context, sb *sandbox.Sandbox) error { - // The framework (zygote, system_server) starts minutes after init on a nested first boot; silkd answers long before + // The framework starts minutes after init on a nested first boot; silkd answers earlier. deadline := time.Now().Add(5 * time.Minute) for { out, err := sb.Exec(ctx, "/system/bin/sh", "-c", shellPath+"; ps -A") diff --git a/e2e/cmd/browsersmoke/main.go b/e2e/cmd/browsersmoke/main.go index b735ba531..f1c47e029 100644 --- a/e2e/cmd/browsersmoke/main.go +++ b/e2e/cmd/browsersmoke/main.go @@ -104,7 +104,7 @@ func openTarget(ctx context.Context, sb *sandbox.Sandbox) error { } func cdpJSON(ctx context.Context, sb *sandbox.Sandbox, method, path string) (map[string]any, error) { - body, err := harness.HTTPOverPort(ctx, sb, cdpPort, method, path, nil) + body, err := harness.HTTPOverPort(ctx, sb, cdpPort, method, path) if err != nil { return nil, err } diff --git a/e2e/cmd/coldproof/main.go b/e2e/cmd/coldproof/main.go index 3d2651dea..f6334fe93 100644 --- a/e2e/cmd/coldproof/main.go +++ b/e2e/cmd/coldproof/main.go @@ -16,7 +16,7 @@ func main() { var ( addr = flag.String("addr", "127.0.0.1:7777", "sandboxd address") token = flag.String("token", "", "node api token") - template = flag.String("template", "rt:24.04", "template ref (must be unpooled for a true cold boot)") + template = flag.String("template", "rt:24.04", "template ref; only an unpooled one measures a cold boot") n = flag.Int("n", 1, "iterations") ) flag.Parse() diff --git a/e2e/cmd/crossnode/main.go b/e2e/cmd/crossnode/main.go index 92a1bcddc..8bb88527f 100644 --- a/e2e/cmd/crossnode/main.go +++ b/e2e/cmd/crossnode/main.go @@ -3,6 +3,7 @@ package main import ( "context" + "errors" "flag" "fmt" "os" @@ -29,6 +30,9 @@ func main() { } func run(addrA, addrB, token, template string) error { + if addrB == "" { + return errors.New("-b is required") + } ctx, cancel := context.WithTimeout(context.Background(), 15*time.Minute) defer cancel() diff --git a/e2e/cmd/egresssmoke/main.go b/e2e/cmd/egresssmoke/main.go index 3214a84d7..2427a574b 100644 --- a/e2e/cmd/egresssmoke/main.go +++ b/e2e/cmd/egresssmoke/main.go @@ -1,15 +1,5 @@ -// egresssmoke is the guarded-egress acceptance: a sandbox reaches an allowed -// origin through the host egress proxy with a host-injected credential it never -// holds, direct egress is blocked (no NIC on the none lane, an nft lock on the -// egress lane), and a disallowed host is denied. -// -// Injection is checked against a public echo (-echo) reached through the proxy, -// which the pool policy must allow; a host-local origin bound at -reach is the -// negative control the guest must fail to reach directly. Example config: -// -// "secrets":[{"name":"probe","header":"X-Egress-Token","value_env":"EGRESS_PROBE_TOKEN"}], -// "pools":[{"template":"…","net":"none","size":"small","warm":1, -// "egress":{"allow":[{"host":"postman-echo.com","secret":"probe"}]}}] +// egresssmoke checks guarded egress: an allowed origin gets the host-injected +// credential, direct egress is blocked, and a disallowed host is denied. package main import ( @@ -36,19 +26,18 @@ func main() { netShape := flag.String("net", "none", "claim lane: none|egress") reach := flag.String("reach", "127.0.0.1", "host the guest reaches the origin at") nicAddr := flag.String("nicaddr", "", "static CIDR to bring the egress-lane NIC up with (default route via -reach)") - guarded := flag.Bool("guarded", true, "pool has an egress policy; false asserts the unlocked NIC reaches directly (negative control)") echo := flag.String("echo", "postman-echo.com", "public HTTP host echoing request headers at /get, for the injection check") flag.Parse() - if err := run(*addr, *token, *template, *wantToken, *netShape, *reach, *nicAddr, *echo, *guarded); err != nil { + if err := run(*addr, *token, *template, *wantToken, *netShape, *reach, *nicAddr, *echo); err != nil { fmt.Fprintln(os.Stderr, "egresssmoke:", err) os.Exit(1) } fmt.Println("EGRESSSMOKE PASS") } -func run(addr, token, template, wantToken, netShape, reach, nicAddr, echo string, guarded bool) error { - if guarded && wantToken == "" { +func run(addr, token, template, wantToken, netShape, reach, nicAddr, echo string) error { + if wantToken == "" { return errors.New("-secret is required for the injection check") } ctx, cancel := context.WithTimeout(context.Background(), 5*time.Minute) @@ -79,16 +68,9 @@ func run(addr, token, template, wantToken, netShape, reach, nicAddr, echo string } out, _ := sb.Exec(ctx, "sh", "-c", fmt.Sprintf("curl -s -m 3 --noproxy '*' %s || echo BLOCKED", target)) - if guarded && !strings.Contains(out, "BLOCKED") { + if !strings.Contains(out, "BLOCKED") { return fmt.Errorf("guest reached the origin without the proxy: %q", out) } - if !guarded { - if !strings.Contains(out, "REACHED") { - return fmt.Errorf("unlocked NIC did not reach the origin: %q", out) - } - fmt.Println(" unlocked NIC reaches the origin directly (negative control)") - return nil - } fmt.Println(" direct egress blocked") // The SSRF guard refuses private and loopback destinations even when allow-listed, so injection needs a public echo. @@ -107,8 +89,12 @@ func run(addr, token, template, wantToken, netShape, reach, nicAddr, echo string } fmt.Println(" allowed origin reached with no -x: proxy variables present in the exec") - if out, _ := sb.Exec(ctx, "sh", "-c", "env; cat /proc/1/environ 2>/dev/null | tr '\\0' '\\n'"); strings.Contains(out, wantToken) { - return fmt.Errorf("secret value leaked into the guest") + env, err := sb.Exec(ctx, "sh", "-c", "env; cat /proc/1/environ 2>/dev/null | tr '\\0' '\\n'") + if err != nil { + return fmt.Errorf("leak probe: %w", err) + } + if strings.Contains(env, wantToken) { + return errors.New("secret value leaked into the guest") } fmt.Println(" secret absent from guest env") diff --git a/e2e/cmd/guestserver/main.go b/e2e/cmd/guestserver/main.go index efaf6de58..1af4ed8cf 100644 --- a/e2e/cmd/guestserver/main.go +++ b/e2e/cmd/guestserver/main.go @@ -12,7 +12,6 @@ import ( "sync" ) -// halfCloseSink greets, shuts its write side, and keeps reading what the client sends after. type halfCloseSink struct { mu sync.Mutex got string diff --git a/e2e/cmd/meshsmoke/main.go b/e2e/cmd/meshsmoke/main.go index a2d4a08b5..b26bc0a51 100644 --- a/e2e/cmd/meshsmoke/main.go +++ b/e2e/cmd/meshsmoke/main.go @@ -80,10 +80,10 @@ func run(addr, peer, token, template string) error { return fmt.Errorf("name-based delete via %s: %w", addr, err) } if err := client.DeleteTemplate(ctx, "mesh-tpl", sandbox.WithNetwork(sandbox.NetNone), sandbox.WithSize(sandbox.Small)); err == nil { - return fmt.Errorf("second delete succeeded, want not-found") + return errors.New("second delete succeeded, want not-found") } if err := tpl.Delete(ctx); err == nil { - return fmt.Errorf("owner-handle delete after name-based delete succeeded, want not-found") + return errors.New("owner-handle delete after name-based delete succeeded, want not-found") } fmt.Println(" delete: followed gossip to the owner; gone everywhere after") return nil diff --git a/e2e/cmd/portsmoke/main.go b/e2e/cmd/portsmoke/main.go index affe92cc4..e8279e996 100644 --- a/e2e/cmd/portsmoke/main.go +++ b/e2e/cmd/portsmoke/main.go @@ -19,9 +19,8 @@ import ( ) const ( - // guestPort is the port an edge proxy relays to; the relay is port-agnostic either way. - guestPort = 49983 - // halfClosePort greets and shuts its write side, then keeps reading. + // An edge proxy relays to this port; the relay itself is port-agnostic. + guestPort = 49983 halfClosePort = 49984 // deadPort has no listener, so silkd's connect refusal must surface as 502. deadPort = 49985 @@ -90,7 +89,6 @@ func run(addr, token, template, listener string, hold time.Duration) error { fmt.Printf(" ok %-24s %5.1fs\n", step.name, time.Since(t0).Seconds()) } if hold > 0 { - // The listener is live now, for the operator's edge smoke to drive fmt.Printf("SANDBOX %s %s %s %d\n", sb.ID, sb.Token(), sb.Owner(), guestPort) select { case <-time.After(hold): @@ -118,10 +116,7 @@ func startGuestListener(ctx context.Context, sb *sandbox.Sandbox, listener strin return fmt.Errorf("spawn listener: %w", err) } bound := harness.Poll(ctx, listenerReadyWait, 500*time.Millisecond, func() error { - out, err := sb.Exec(ctx, "sh", "-c", fmt.Sprintf("ss -ltn 'sport = :%d' | grep -c LISTEN", guestPort)) - if err == nil && strings.TrimSpace(out) == "0" { - return errors.New("not listening") - } + _, err := sb.Exec(ctx, "sh", "-c", fmt.Sprintf("ss -ltn 'sport = :%d' | grep -c LISTEN", guestPort)) return err }) if bound != nil { diff --git a/e2e/cmd/qaab/main.go b/e2e/cmd/qaab/main.go deleted file mode 100644 index 76a174853..000000000 --- a/e2e/cmd/qaab/main.go +++ /dev/null @@ -1,198 +0,0 @@ -// qaab measures one A/B arm: a concurrent cold-claim burst, then a concurrent in-guest write burst, as one JSON row. -package main - -import ( - "context" - "encoding/json/v2" - "flag" - "fmt" - "os" - "os/signal" - "slices" - "strconv" - "strings" - "sync" - "syscall" - "time" - - sandbox "github.com/cocoonstack/sandbox/sdk/go" -) - -// One dd keeps exactly one virtio queue busy, so the writers are the queue fan-out -const ddScript = `pids=; for i in $(seq 1 %d); do dd if=/dev/zero of=/root/qaab.$i bs=%s count=%d oflag=direct conv=fsync 2>/dev/null & pids="$pids $!"; done; status=0; for pid in $pids; do wait "$pid" || status=1; done; exit "$status"` - -type stats struct { - P50 float64 `json:"p50"` - P90 float64 `json:"p90"` - Max float64 `json:"max"` -} - -type row struct { - Label string `json:"label"` - N int `json:"n"` - IOBlock string `json:"io_bs"` - IOCount int `json:"io_count"` - IOJobs int `json:"io_jobs"` - Claim stats `json:"claim_ms"` - ClaimWallMS float64 `json:"claim_wall_ms"` - IO stats `json:"io_ms"` - IOWallMS float64 `json:"io_wall_ms"` - IOMiBps float64 `json:"io_aggregate_mibps"` - Failures int `json:"failures"` -} - -func main() { - var ( - addr = flag.String("addr", "127.0.0.1:18777", "sandboxd address") - token = flag.String("token", "", "node api token") - template = flag.String("template", "", "template ref; leave it unpooled so every claim cold-boots") - size = flag.String("size", "2xlarge", "size tier") - label = flag.String("label", "arm", "arm label for the output row") - n = flag.Int("n", 24, "concurrent sandboxes") - ioBS = flag.String("io-bs", "1M", "dd block size per write") - ioCount = flag.Int("io-count", 64, "dd block count per writer") - ioJobs = flag.Int("io-jobs", 8, "concurrent in-guest writers per sandbox") - ) - flag.Parse() - if err := run(*addr, *token, *template, *size, *label, *ioBS, *n, *ioCount, *ioJobs); err != nil { - fmt.Fprintln(os.Stderr, "qaab:", err) - os.Exit(1) - } -} - -func run(addr, token, template, size, label, ioBS string, n, ioCount, ioJobs int) error { - blockSize := blockBytes(ioBS) - if blockSize == 0 { - return fmt.Errorf("--io-bs %q must be a plain byte count or carry a k or M suffix", ioBS) - } - ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) - defer stop() - client, err := sandbox.Connect(addr, sandbox.WithAPIToken(token)) - if err != nil { - return err - } - - boxes := make([]*sandbox.Sandbox, n) - claimMS := make([]float64, n) - claimErrs := make([]error, n) - - claimStart := time.Now() - var claims sync.WaitGroup - for i := range n { - claims.Go(func() { - start := time.Now() - sb, claimErr := client.New(ctx, template, - sandbox.WithSize(sandbox.Size(size)), sandbox.WithTimeout(15*time.Minute)) - claimMS[i] = msSince(start) - boxes[i], claimErrs[i] = sb, claimErr - }) - } - claims.Wait() - claimWall := msSince(claimStart) - - ioMS := make([]float64, n) - ioErrs := make([]error, n) - ioStart := time.Now() - var writes sync.WaitGroup - for i, sb := range boxes { - if sb == nil { - continue - } - writes.Go(func() { - start := time.Now() - var stderr strings.Builder - code, runErr := sb.Run(ctx, sandbox.Cmd{ - Argv: []string{"sh", "-c", fmt.Sprintf(ddScript, ioJobs, ioBS, ioCount)}, - Stderr: &stderr, - }) - ioMS[i] = msSince(start) - if runErr == nil && code != 0 { - runErr = fmt.Errorf("writers exit %d: %s", code, strings.TrimSpace(stderr.String())) - } - ioErrs[i] = runErr - }) - } - writes.Wait() - ioWall := msSince(ioStart) - - releaseErrs := make([]error, n) - for i, sb := range boxes { - if sb == nil { - continue - } - if closeErr := sb.Close(); closeErr != nil { - releaseErrs[i] = fmt.Errorf("release sandbox %s: %w", sb.ID, closeErr) - } - } - - claimOK := observed(claimMS, claimErrs) - ioOK := observed(ioMS, ioErrs) - out := row{ - Label: label, N: n, IOBlock: ioBS, IOCount: ioCount, IOJobs: ioJobs, - Claim: summarize(claimOK), ClaimWallMS: claimWall, - IO: summarize(ioOK), IOWallMS: ioWall, - Failures: failed(claimErrs) + failed(ioErrs) + failed(releaseErrs), - } - if ioWall > 0 { - written := float64(len(ioOK)*ioCount*ioJobs) * float64(blockSize) / (1 << 20) - out.IOMiBps = written / (ioWall / 1000) - } - encoded, err := json.Marshal(out) - if err != nil { - return err - } - fmt.Println(string(encoded)) - - for _, e := range slices.Concat(claimErrs, ioErrs, releaseErrs) { - if e != nil { - return e - } - } - return nil -} - -func msSince(t time.Time) float64 { return float64(time.Since(t).Microseconds()) / 1000 } - -func blockBytes(bs string) int { - unit := 1 - switch { - case strings.HasSuffix(bs, "k"), strings.HasSuffix(bs, "K"): - unit, bs = 1<<10, bs[:len(bs)-1] - case strings.HasSuffix(bs, "M"): - unit, bs = 1<<20, bs[:len(bs)-1] - } - n, err := strconv.Atoi(bs) - if err != nil { - return 0 - } - return n * unit -} - -func failed(errs []error) int { - n := 0 - for _, e := range errs { - if e != nil { - n++ - } - } - return n -} - -func observed(values []float64, errs []error) []float64 { - out := make([]float64, 0, len(values)) - for i, v := range values { - if errs[i] == nil && v > 0 { - out = append(out, v) - } - } - slices.Sort(out) - return out -} - -func summarize(sorted []float64) stats { - if len(sorted) == 0 { - return stats{} - } - at := func(p float64) float64 { return sorted[int(float64(len(sorted)-1)*p)] } - return stats{P50: at(0.5), P90: at(0.9), Max: sorted[len(sorted)-1]} -} diff --git a/e2e/cmd/rpcbench/main.go b/e2e/cmd/rpcbench/main.go index ab5b2f3cd..c1dbb92bf 100644 --- a/e2e/cmd/rpcbench/main.go +++ b/e2e/cmd/rpcbench/main.go @@ -1,13 +1,5 @@ -// rpcbench measures the one-connection-per-RPC overhead on a live node: mode -// A dials+upgrades per RPC by hand; mode C is the SDK's own path, one kept -// connection serving RPCs back to back. A and C interleave sample by sample -// and swap which one leads, so node drift cannot land on one arm. Run by -// hand, or by scripts/bench.sh; it claims its own sandbox: -// -// rpcbench -addr -token -template -n 200 -// -// An https:// address drives the hand-dialed mode through the TLS edge the -// SDK already uses, so A pays a handshake per RPC the way a real client does. +// rpcbench compares dial-per-RPC against the SDK's kept connection on a live +// node, interleaved sample by sample with the lead swapped. package main import ( @@ -192,7 +184,7 @@ func dialAgent(ctx context.Context, addr, id, token string, tlsCfg *tls.Config) } scheme = "https" } - req, err := http.NewRequest(http.MethodGet, scheme+"://"+host+"/v1/sandboxes/"+id+"/agent", nil) + req, err := http.NewRequestWithContext(ctx, http.MethodGet, scheme+"://"+host+"/v1/sandboxes/"+id+"/agent", nil) if err != nil { _ = raw.Close() return nil, err diff --git a/e2e/cmd/smoke/main.go b/e2e/cmd/smoke/main.go index 1c45e08f6..0371240f1 100644 --- a/e2e/cmd/smoke/main.go +++ b/e2e/cmd/smoke/main.go @@ -211,7 +211,7 @@ func smokeWatch(ctx context.Context, sb *sandbox.Sandbox) error { return nil } case <-deadline: - return fmt.Errorf("no event for w.txt within 5s") + return errors.New("no event for w.txt within 5s") } } } @@ -475,7 +475,7 @@ func smokeCheckpoint(ctx context.Context, sb *sandbox.Sandbox) error { return fmt.Errorf("delete checkpoint: %w", err) } if _, err = ckpt.New(ctx); err == nil { - return fmt.Errorf("branch from deleted checkpoint succeeded") + return errors.New("branch from deleted checkpoint succeeded") } return nil } @@ -500,7 +500,7 @@ func smokeTree(ctx context.Context, sb *sandbox.Sandbox) error { full := archive("CLOBBERED") half := io.MultiReader(bytes.NewReader(full[:len(full)/2]), iotest.ErrReader(errors.New("cut"))) if err := sb.Push(ctx, "/root/tree", half); err == nil { - return fmt.Errorf("truncated push reported success") + return errors.New("truncated push reported success") } if out, err := sb.Exec(ctx, "cat", "/root/tree/t.txt"); err != nil || strings.TrimSpace(out) != "tree-marker" { return fmt.Errorf("tree changed by failed push: %q, %v", out, err) @@ -522,7 +522,7 @@ func smokeTree(ctx context.Context, sb *sandbox.Sandbox) error { return fmt.Errorf("pull: %w", err) } if !bytes.Contains(pulled.Bytes(), []byte("tree-marker")) { - return fmt.Errorf("pulled tar missing pushed content") + return errors.New("pulled tar missing pushed content") } return nil } @@ -579,8 +579,8 @@ func smokeLsp(ctx context.Context, client *sandbox.Client, base *sandbox.Sandbox defer func() { _ = stream.Close() }() r := bufio.NewReader(stream) - init := `{"jsonrpc":"2.0","id":1,"method":"initialize","params":{"rootUri":"file:///work/lsp","capabilities":{}}}` - if err = lspWrite(stream, init); err != nil { + initReq := `{"jsonrpc":"2.0","id":1,"method":"initialize","params":{"rootUri":"file:///work/lsp","capabilities":{}}}` + if err = lspWrite(stream, initReq); err != nil { return err } if _, err = lspReadResponse(r, 1); err != nil { diff --git a/e2e/internal/harness/harness.go b/e2e/internal/harness/harness.go index e6743012e..bcb5010fc 100644 --- a/e2e/internal/harness/harness.go +++ b/e2e/internal/harness/harness.go @@ -1,4 +1,4 @@ -// Package harness carries the claim prologue shared by the e2e tools. +// Package harness carries helpers shared by the e2e tools. package harness import ( @@ -41,6 +41,8 @@ func (p PortRelay) Dial(ctx context.Context, port uint16) (RelayConn, error) { if err != nil { return nil, err } + stop := context.AfterFunc(ctx, func() { _ = conn.Close() }) + defer stop() req, err := Request(ctx, p.URL(port), p.Token, true) if err != nil { _ = conn.Close() @@ -113,17 +115,13 @@ func Poll(ctx context.Context, budget, every time.Duration, try func() error) er } // HTTPOverPort sends one HTTP/1.1 request over the port relay and returns a 200 body; the Host stays "localhost" because Chrome's DevTools allowlist accepts it but not proxied vhosts. -func HTTPOverPort(ctx context.Context, sb *sandbox.Sandbox, port uint16, method, path string, body []byte) ([]byte, error) { +func HTTPOverPort(ctx context.Context, sb *sandbox.Sandbox, port uint16, method, path string) ([]byte, error) { pc, err := sb.DialPort(ctx, port) if err != nil { return nil, err } defer func() { _ = pc.Close() }() - head := fmt.Sprintf("%s %s HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n", method, path) - if body != nil { - head += fmt.Sprintf("Content-Type: application/json\r\nContent-Length: %d\r\n", len(body)) - } - if _, err = fmt.Fprintf(pc, "%s\r\n%s", head, body); err != nil { + if _, err = fmt.Fprintf(pc, "%s %s HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n", method, path); err != nil { return nil, err } resp, err := http.ReadResponse(bufio.NewReader(pc), nil) diff --git a/e2e/wire_parity_test.go b/e2e/wire_parity_test.go index a0990daf8..06df4219b 100644 --- a/e2e/wire_parity_test.go +++ b/e2e/wire_parity_test.go @@ -16,6 +16,7 @@ import ( "github.com/cocoonstack/sandbox/sandboxd/config" "github.com/cocoonstack/sandbox/sandboxd/pool" "github.com/cocoonstack/sandbox/sandboxd/server" + "github.com/cocoonstack/sandbox/sandboxd/tenants" "github.com/cocoonstack/sandbox/sandboxd/types" ) @@ -62,7 +63,7 @@ var wirePairs = []struct { {"PoolSpec", true, config.PoolSpec{}, []string{"egress", "warmup", "egress_upstream_env", "capture_trim", "storage", "replicate_seconds", "reader", "net", "size", "tun"}}, {"tenantsUpdate", true, server.TenantsRequest{}, nil}, {"tenantUpdate", true, server.TenantRequest{}, nil}, - {"TenantSpec", true, reflect.Zero(reflect.TypeOf(server.TenantsRequest{}.Tenants).Elem()).Interface(), []string{"token"}}, + {"TenantSpec", true, tenants.TenantSpec{}, []string{"token"}}, {"TenantInfo", false, pool.TenantInfo{}, nil}, {"TenantList", false, pool.TenantPage{}, nil}, } diff --git a/mcp/e2e.py b/mcp/e2e.py index b7408e04d..12db4c7cb 100644 --- a/mcp/e2e.py +++ b/mcp/e2e.py @@ -32,6 +32,8 @@ def call(self, method: str, params: dict | None = None) -> dict: self.proc.stdin.write((json.dumps(req) + "\n").encode()) self.proc.stdin.flush() line = self.proc.stdout.readline() + if not line: + raise RuntimeError("sandbox-mcp exited") resp = json.loads(line) assert resp["id"] == self.seq, resp assert "error" not in resp, resp diff --git a/mcp/main.go b/mcp/main.go index 47df6b041..6a42abd4a 100644 --- a/mcp/main.go +++ b/mcp/main.go @@ -21,7 +21,8 @@ func main() { fmt.Fprintln(os.Stderr, "setup log:", err) os.Exit(1) } - ctx, stop := signal.NotifyContext(ctx, os.Interrupt, syscall.SIGTERM) + signal.Ignore(syscall.SIGPIPE) + ctx, stop := signal.NotifyContext(ctx, os.Interrupt, syscall.SIGTERM, syscall.SIGHUP) defer stop() addr := flag.String("addr", cmp.Or(os.Getenv("SANDBOXD_ADDR"), "127.0.0.1:7777"), "sandboxd address") token := flag.String("token", os.Getenv("SANDBOXD_TOKEN"), "node api token") diff --git a/mcp/server.go b/mcp/server.go index ef90058bd..2d8423b71 100644 --- a/mcp/server.go +++ b/mcp/server.go @@ -19,12 +19,17 @@ import ( const ( protocolVersion = "2024-11-05" - execTimeout = 5 * time.Minute + toolTimeout = 5 * time.Minute execOutputCap = 1 << 20 - defaultToolTTL = time.Hour + // Outlives the node's 5-minute default; nothing renews a tool claim. + defaultToolTTL = time.Hour ) -var version = "devel" +var ( + version = "devel" + + replyOpts = json.JoinOptions(jsontext.AllowInvalidUTF8(true), json.Deterministic(true)) +) type rpcRequest struct { ID jsontext.Value `json:"id"` @@ -36,7 +41,7 @@ type rpcResponse struct { JSONRPC string `json:"jsonrpc"` ID jsontext.Value `json:"id"` Result any `json:"result,omitzero"` - Error *rpcErrorBody `json:"error,omitempty"` + Error *rpcErrorBody `json:"error,omitzero"` } type rpcErrorBody struct { @@ -87,7 +92,7 @@ func (s *server) serve(ctx context.Context, r *bufio.Reader, w io.Writer) error continue // notification (e.g. notifications/initialized) } resp := s.dispatch(ctx, &req) - out, err := json.Marshal(resp, jsontext.AllowInvalidUTF8(true), json.Deterministic(true)) + out, err := json.Marshal(resp, replyOpts) if err != nil { return err } @@ -115,7 +120,11 @@ func (s *server) dispatch(ctx context.Context, req *rpcRequest) rpcResponse { if err := unmarshalInput(req.Params, &call); err != nil { return rpcError(req.ID, -32602, "invalid params") } - text, err := s.callTool(ctx, call.Name, call.Arguments) + i := slices.IndexFunc(tools, func(t tool) bool { return t.name == call.Name }) + if i < 0 { + return rpcError(req.ID, -32602, "unknown tool: "+call.Name) + } + text, err := s.callTool(ctx, tools[i], call.Arguments) if err != nil { return result(req.ID, map[string]any{ "content": []map[string]any{{"type": "text", "text": cmp.Or(text, err.Error())}}, @@ -132,14 +141,10 @@ func (s *server) dispatch(ctx context.Context, req *rpcRequest) rpcResponse { } } -func (s *server) callTool(ctx context.Context, name string, args jsontext.Value) (string, error) { - ctx, cancel := context.WithTimeout(ctx, execTimeout) +func (s *server) callTool(ctx context.Context, t tool, args jsontext.Value) (string, error) { + ctx, cancel := context.WithTimeout(ctx, toolTimeout) defer cancel() - i := slices.IndexFunc(tools, func(t tool) bool { return t.name == name }) - if i < 0 { - return "", fmt.Errorf("unknown tool %q", name) - } - return tools[i].handler(ctx, s, args) + return t.handler(ctx, s, args) } func (s *server) box(id string) (*sandbox.Sandbox, error) { diff --git a/mcp/tools.go b/mcp/tools.go index fe39f460e..33a3534a0 100644 --- a/mcp/tools.go +++ b/mcp/tools.go @@ -183,7 +183,6 @@ func toolCreateSandbox(ctx context.Context, s *server, raw jsontext.Value) (stri if args.TTLSeconds < 0 { return "", fmt.Errorf("ttl_seconds must not be negative, got %d", args.TTLSeconds) } - // 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) 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 { @@ -437,7 +436,7 @@ func unmarshalInput(raw []byte, v any) error { } func jsonText(v any) string { - out, _ := json.Marshal(v, jsontext.AllowInvalidUTF8(true), json.Deterministic(true)) + out, _ := json.Marshal(v, replyOpts) return string(out) } diff --git a/sandboxd/archives/pg/nodes.sql b/sandboxd/archives/pg/nodes.sql deleted file mode 100644 index 1f1da61c8..000000000 --- a/sandboxd/archives/pg/nodes.sql +++ /dev/null @@ -1,6 +0,0 @@ -CREATE TABLE IF NOT EXISTS sandboxd_nodes ( - cell text NOT NULL, - node text NOT NULL, - lease_until timestamptz NOT NULL, - PRIMARY KEY (cell, node) -); diff --git a/sandboxd/archives/pg/pg.go b/sandboxd/archives/pg/pg.go index 15b37bcc4..766f890e0 100644 --- a/sandboxd/archives/pg/pg.go +++ b/sandboxd/archives/pg/pg.go @@ -15,8 +15,7 @@ import ( ) const ( - table = "sandboxd_archives" - nodesTable = "sandboxd_nodes" + table = "sandboxd_archives" leaseExpired = `NOT EXISTS (SELECT 1 FROM sandboxd_nodes n WHERE n.cell = sandboxd_archives.cell AND n.node = sandboxd_archives.owner AND n.lease_until > now())` ) @@ -25,9 +24,6 @@ var ( //go:embed schema.sql schema string - //go:embed nodes.sql - nodesSchema string - cas = metastore.CAS{Lost: archives.ErrLost, Unavailable: archives.ErrUnavailable} ) @@ -45,10 +41,6 @@ func Open(ctx context.Context, dsn, cell string) (*Registry, error) { if err != nil { return nil, err } - if err = metastore.EnsureTable(ctx, pool, nodesTable, nodesSchema); err != nil { - pool.Close() - return nil, err - } return &Registry{pool: pool, cell: cell}, nil } diff --git a/sandboxd/archives/pg/pg_test.go b/sandboxd/archives/pg/pg_test.go index d00a1f3c1..9d1020474 100644 --- a/sandboxd/archives/pg/pg_test.go +++ b/sandboxd/archives/pg/pg_test.go @@ -99,18 +99,6 @@ func TestADownDatabaseIsUnavailable(t *testing.T) { } } -func TestOpenAddsTheNodesTableToAnArchivesOnlyDatabase(t *testing.T) { - dsn := metastoretest.PGSchema(t) - first := open(t, dsn, "c1") - if _, err := first.pool.Exec(t.Context(), `DROP TABLE sandboxd_nodes`); err != nil { - t.Fatalf("drop nodes: %v", err) - } - second := open(t, dsn, "c1") - if err := second.Renew(t.Context(), "a", time.Minute); err != nil { - t.Errorf("renew on a database that predates the nodes table: %v", err) - } -} - func TestLeasesAreScopedToTheirCell(t *testing.T) { dsn := metastoretest.PGSchema(t) c1, c2 := open(t, dsn, "c1"), open(t, dsn, "c2") diff --git a/sandboxd/archives/pg/schema.sql b/sandboxd/archives/pg/schema.sql index cb51e31c8..4587691db 100644 --- a/sandboxd/archives/pg/schema.sql +++ b/sandboxd/archives/pg/schema.sql @@ -15,3 +15,9 @@ CREATE TABLE IF NOT EXISTS sandboxd_archives ( ); CREATE INDEX IF NOT EXISTS sandboxd_archives_deadline ON sandboxd_archives (cell, deadline) WHERE deadline IS NOT NULL; CREATE INDEX IF NOT EXISTS sandboxd_archives_consumed ON sandboxd_archives (cell) WHERE consumed; +CREATE TABLE IF NOT EXISTS sandboxd_nodes ( + cell text NOT NULL, + node text NOT NULL, + lease_until timestamptz NOT NULL, + PRIMARY KEY (cell, node) +); diff --git a/sandboxd/config/config.go b/sandboxd/config/config.go index 0a718b744..af5ae49ef 100644 --- a/sandboxd/config/config.go +++ b/sandboxd/config/config.go @@ -514,6 +514,7 @@ func (c *Config) validate() error { {"idle_hibernate_seconds", c.IdleHibernateSeconds}, {"checkpoint_ttl_hours", c.CheckpointTTLHours}, {"max_claims", c.MaxClaims}, + {"max_fork_count", c.MaxForkCount}, {"max_running", c.MaxRunning}, {"move_concurrency", c.MoveConcurrency}, {"refill_concurrency", c.RefillConcurrency}, @@ -529,9 +530,6 @@ func (c *Config) validate() error { if len(c.Networks) > 0 && c.guardsEgressLane() { return fmt.Errorf("guarded egress needs a bridge lane, not a CNI network: the tap lives in the VM netns and cannot be locked") } - if c.MaxForkCount < 1 { - return fmt.Errorf("max_fork_count must be at least 1, got %d", c.MaxForkCount) - } if err := c.RestoreMode.Validate(); err != nil { return fmt.Errorf("restore_mode: %w", err) } @@ -559,7 +557,7 @@ func (c *Config) validate() error { return fmt.Errorf("checkpoint_peer_heal requires checkpoint_ttl_hours > 0: the ttl is what ages out a healed replica a delete broadcast missed") } if c.CheckpointPeerHeal && c.APIToken == "" { - return fmt.Errorf("checkpoint_peer_heal requires api_token: without it resolveScope leaves the raw checkpoint blob GET reachable with no credential") + return fmt.Errorf("checkpoint_peer_heal requires api_token: without it the raw checkpoint blob GET needs no credential") } if c.MoveConcurrency > 0 && !c.Movable() { return fmt.Errorf("move_concurrency requires meta_store with a cell and an encrypted mesh (set mesh.cluster_key)") @@ -573,7 +571,7 @@ func (c *Config) validate() error { if err != nil { return err } - return c.validateEgress(secrets) + return c.validatePools(secrets) } func (c *Config) validateReplication() error { @@ -688,7 +686,7 @@ func (c *Config) validateMesh() error { return nil } -func (c *Config) validateEgress(secrets map[string]struct{}) error { +func (c *Config) validatePools(secrets map[string]struct{}) error { for _, p := range c.Pools { if err := p.Validate(); err != nil { return fmt.Errorf("pool %q: %w", p.Template, err) diff --git a/sandboxd/egress/bench_test.go b/sandboxd/egress/bench_test.go index 0354fe8ae..916da9b4a 100644 --- a/sandboxd/egress/bench_test.go +++ b/sandboxd/egress/bench_test.go @@ -18,7 +18,7 @@ const benchKey = "EAABsbCS1iHgBAKZBZCx9ZAZBe7ZAZCYqZBfZC0ZD" func BenchmarkSignLeaf(b *testing.B) { ca := testCAOnly(b) for b.Loop() { - if _, err := ca.SignLeaf("bench.example.com"); err != nil { + if _, err := ca.signLeaf("bench.example.com"); err != nil { b.Fatalf("sign: %v", err) } } diff --git a/sandboxd/egress/ca.go b/sandboxd/egress/ca.go index 5f88be677..99b9e4cb8 100644 --- a/sandboxd/egress/ca.go +++ b/sandboxd/egress/ca.go @@ -84,8 +84,8 @@ func (c *CA) CertPEM() []byte { return c.rootPEM } // Fingerprint is the SHA-256 of the baked root PEM, so a rotated intermediate keeps it. func (c *CA) Fingerprint() string { return c.rootFP } -// SignLeaf issues a fresh interception leaf for host, chained [leaf, intermediate]. -func (c *CA) SignLeaf(host string) (*tls.Certificate, error) { +// signLeaf issues a fresh interception leaf for host, chained [leaf, intermediate]. +func (c *CA) signLeaf(host string) (*tls.Certificate, error) { serial, err := serialNumber() if err != nil { return nil, err diff --git a/sandboxd/egress/ca_test.go b/sandboxd/egress/ca_test.go index 096a0873a..107c8ac73 100644 --- a/sandboxd/egress/ca_test.go +++ b/sandboxd/egress/ca_test.go @@ -36,7 +36,7 @@ func TestSignLeafChainsToRoot(t *testing.T) { } for _, host := range []string{"api.github.com", "192.0.2.10"} { t.Run(host, func(t *testing.T) { - crt, err := ca.SignLeaf(host) + crt, err := ca.signLeaf(host) if err != nil { t.Fatalf("sign leaf: %v", err) } diff --git a/sandboxd/egress/intercept.go b/sandboxd/egress/intercept.go index 33f1173d3..af79d3fed 100644 --- a/sandboxd/egress/intercept.go +++ b/sandboxd/egress/intercept.go @@ -59,11 +59,11 @@ func (p *Proxy) serveIntercept(w http.ResponseWriter, r *http.Request, host stri srv := &http.Server{ Handler: &interceptHandler{proxy: p, host: host, port: port, authority: net.JoinHostPort(host, strconv.Itoa(int(port)))}, ReadHeaderTimeout: interceptTimeout, + IdleTimeout: idleConnTimeout, } _ = srv.Serve(ln) } -// leafFor returns the cached leaf for host, signing a fresh one on miss or near-expiry. func (p *Proxy) leafFor(host string) (*tls.Certificate, error) { p.leafMu.Lock() defer p.leafMu.Unlock() @@ -71,7 +71,7 @@ func (p *Proxy) leafFor(host string) (*tls.Certificate, error) { if crt, ok := p.leaves[host]; ok && (time.Now().Before(crt.Leaf.NotAfter.Add(-leafRenewBefore)) || !crt.Leaf.NotAfter.Before(p.ca.interCert.NotAfter)) { return crt, nil } - crt, err := p.ca.SignLeaf(host) + crt, err := p.ca.signLeaf(host) if err != nil { return nil, err } diff --git a/sandboxd/egress/policy.go b/sandboxd/egress/policy.go index 3a740eec7..cf80de3da 100644 --- a/sandboxd/egress/policy.go +++ b/sandboxd/egress/policy.go @@ -5,6 +5,7 @@ import ( "encoding/json/jsontext" "fmt" "net/http" + "net/netip" "slices" "strings" ) @@ -119,6 +120,9 @@ func (p Policy) Validate() error { case "*.": return fmt.Errorf("allow[%d]: host %q needs a domain after the wildcard", i, r.Host) } + if _, ipErr := netip.ParseAddr(r.Host); ipErr != nil && r.Host != "*" && strings.ContainsAny(strings.TrimPrefix(r.Host, "*."), "*:/ ") { + return fmt.Errorf("allow[%d]: host %q is not a host name or a leading \"*.\" wildcard", i, r.Host) + } if r.Intercept == InterceptInject && r.Secret != "" { return fmt.Errorf("allow[%d]: an intercept \"inject\" rule splices for claims without a credential, so it must not carry a secret", i) } diff --git a/sandboxd/egress/policy_test.go b/sandboxd/egress/policy_test.go index 1395e4b18..b7ace506d 100644 --- a/sandboxd/egress/policy_test.go +++ b/sandboxd/egress/policy_test.go @@ -188,6 +188,10 @@ func TestPolicyValidate(t *testing.T) { {"good rules", Policy{Allow: []Rule{{Host: "api.github.com"}, {Host: "*.dev"}, {Host: "*"}}}, false}, {"empty host", Policy{Allow: []Rule{{Host: ""}}}, true}, {"bare wildcard", Policy{Allow: []Rule{{Host: "*."}}}, true}, + {"IPv6 literal", Policy{Allow: []Rule{{Host: "2606:4700:4700::1111"}}}, false}, + {"inner wildcard", Policy{Allow: []Rule{{Host: "api.*.com"}}}, true}, + {"wildcard without a dot", Policy{Allow: []Rule{{Host: "*example.com"}}}, true}, + {"host with a port", Policy{Allow: []Rule{{Host: "example.com:443"}}}, true}, {"socks5 with a tunnel rule", Policy{Socks5: true, Allow: []Rule{{Host: "api.github.com"}}}, false}, {"socks5 without a tunnel rule", Policy{Socks5: true, Allow: []Rule{{Host: "api.github.com", Methods: []string{"GET"}}}}, true}, {"ports", Policy{Allow: []Rule{{Host: "api.github.com", Ports: []uint16{443, 993}}}}, false}, diff --git a/sandboxd/egress/proxy.go b/sandboxd/egress/proxy.go index 9e4666cc3..b0027d152 100644 --- a/sandboxd/egress/proxy.go +++ b/sandboxd/egress/proxy.go @@ -173,7 +173,7 @@ func New(policy Evaluator, secrets Secrets, ca *CA, router Router, audit AuditFu conns: map[net.Conn]struct{}{}, } // The stdlib default of 2 idle conns per host re-dials bursty same-host traffic. - pools := &connPools{tr: &http.Transport{DialContext: direct, MaxIdleConns: maxIdleConns, MaxIdleConnsPerHost: 8, IdleConnTimeout: idleConnTimeout}, routed: map[routeKey]*http.Transport{}} + pools := &connPools{tr: &http.Transport{DialContext: direct, MaxIdleConns: maxIdleConns, MaxIdleConnsPerHost: 8, IdleConnTimeout: idleConnTimeout, DisableCompression: true}, routed: map[routeKey]*http.Transport{}} if ca != nil { pools.mitm = pools.tr.Clone() pools.mitm.TLSClientConfig = &tls.Config{MinVersion: tls.VersionTLS12} @@ -205,10 +205,16 @@ func (p *Proxy) ServeHTTP(w http.ResponseWriter, r *http.Request) { // Close ends every hijacked tunnel, which outlives http.Server.Close; idempotent. func (p *Proxy) Close() { + p.connMu.Lock() + p.closed = true + p.connMu.Unlock() + p.DropTunnels() +} + +func (p *Proxy) DropTunnels() { p.connMu.Lock() conns := slices.Collect(maps.Keys(p.conns)) clear(p.conns) - p.closed = true p.connMu.Unlock() for _, conn := range conns { _ = conn.Close() @@ -435,7 +441,6 @@ func (p *Proxy) relay(w http.ResponseWriter, r *http.Request, ev Event, rule Rul } } -// transport returns the pool for route; the direct route keeps the base transports. func (p *Proxy) transport(route string, mitm bool) *http.Transport { pools := p.pools.Load() base := pools.tr @@ -473,6 +478,48 @@ func (p *Proxy) record(ev Event) { p.audit(ev) } +// inject sets the rule's secret, then on an intercepted request the claim credentials the guest asked for, and returns a scrubber for the query credentials it filled. +func (p *Proxy) inject(rule Rule, out *http.Request, host string, mitm bool) (string, *scrubber) { + if p.secrets == nil { + return "", nil + } + var names, set []string + if rule.Secret != "" { + if header, value, ok := p.secrets.Header(rule.Secret); ok && value != "" { + out.Header.Set(header, value) + names, set = append(names, rule.Secret), append(set, header) + } + } + if !mitm { + return strings.Join(names, ","), nil + } + var filled, bodyCreds []Credential + for _, c := range p.secrets.Credentials(strings.ToLower(host)) { + if c.Value == "" { + continue + } + if c.Query != "" { + if q := replaceParam(out.URL.RawQuery, c.Query, c.Placeholder, c.Value); q != out.URL.RawQuery { + out.URL.RawQuery, filled = q, append(filled, c) + } else if c.Body { + bodyCreds = append(bodyCreds, c) + } + continue + } + if (c.Placeholder != "" && out.Header.Get(c.Header) != c.Placeholder) || + slices.ContainsFunc(set, func(s string) bool { return strings.EqualFold(s, c.Header) }) { + continue + } + out.Header.Set(c.Header, c.Value) + names, set = append(names, "claim:"+c.Name), append(set, c.Header) + } + filled = append(filled, substituteBody(out, bodyCreds)...) + for _, c := range filled { + names = append(names, "claim:"+c.Name) + } + return strings.Join(names, ","), newScrubber(filled) +} + // countingBody counts the request body bytes the transport reads, possibly after RoundTrip returns. type countingBody struct { io.ReadCloser @@ -527,48 +574,6 @@ func (sc *scrubber) header(h http.Header) { } } -// inject sets the rule's secret, then on an intercepted request the claim credentials the guest asked for, and returns a scrubber for the query credentials it filled. -func (p *Proxy) inject(rule Rule, out *http.Request, host string, mitm bool) (string, *scrubber) { - if p.secrets == nil { - return "", nil - } - var names, set []string - if rule.Secret != "" { - if header, value, ok := p.secrets.Header(rule.Secret); ok && value != "" { - out.Header.Set(header, value) - names, set = append(names, rule.Secret), append(set, header) - } - } - if !mitm { - return strings.Join(names, ","), nil - } - var filled, bodyCreds []Credential - for _, c := range p.secrets.Credentials(strings.ToLower(host)) { - if c.Value == "" { - continue - } - if c.Query != "" { - if q := replaceParam(out.URL.RawQuery, c.Query, c.Placeholder, c.Value); q != out.URL.RawQuery { - out.URL.RawQuery, filled = q, append(filled, c) - } else if c.Body { - bodyCreds = append(bodyCreds, c) - } - continue - } - if (c.Placeholder != "" && out.Header.Get(c.Header) != c.Placeholder) || - slices.ContainsFunc(set, func(s string) bool { return strings.EqualFold(s, c.Header) }) { - continue - } - out.Header.Set(c.Header, c.Value) - names, set = append(names, "claim:"+c.Name), append(set, c.Header) - } - filled = append(filled, substituteBody(out, bodyCreds)...) - for _, c := range filled { - names = append(names, "claim:"+c.Name) - } - return strings.Join(names, ","), newScrubber(filled) -} - // scrubReader streams a response body through a scrubber, holding back only a tail that may begin a match. type scrubReader struct { src io.Reader @@ -746,7 +751,7 @@ func hostPort(authority string, def uint16) (host string, port uint16, ok bool) return literal, def, closed && parseErr == nil && addr.Is6() && len(literal) <= maxHostLen } n, err := strconv.ParseUint(text, 10, 16) - return host, uint16(n), err == nil && n != 0 && len(host) <= maxHostLen + return host, uint16(n), err == nil && n != 0 && host != "" && len(host) <= maxHostLen } func tcpAddrPort(a net.Addr) netip.AddrPort { diff --git a/sandboxd/engine/portconn.go b/sandboxd/engine/portconn.go index 8f1220bcc..28058096f 100644 --- a/sandboxd/engine/portconn.go +++ b/sandboxd/engine/portconn.go @@ -38,8 +38,9 @@ func (g *guestPortConn) Read(p []byte) (int, error) { continue } var frame struct { - Type string `json:"type"` - Data []byte `json:"data"` + Type string `json:"type"` + Data []byte `json:"data"` + Message string `json:"message"` } if err := json.Unmarshal(line, &frame); err != nil { return 0, fmt.Errorf("port frame: %w", err) @@ -50,7 +51,7 @@ func (g *guestPortConn) Read(p []byte) (int, error) { case "done", "": return 0, io.EOF default: - return 0, fmt.Errorf("port stream ended: %s", frame.Type) + return 0, fmt.Errorf("port stream ended: %s %s", frame.Type, frame.Message) } } n := copy(p, g.pending) diff --git a/sandboxd/engine/silkd.go b/sandboxd/engine/silkd.go index 9b1e7c5ad..12e4b26b1 100644 --- a/sandboxd/engine/silkd.go +++ b/sandboxd/engine/silkd.go @@ -130,7 +130,7 @@ func (e *Engine) silkdReadFile(ctx context.Context, vsockSocket, path string, li 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. +// silkdOnce sends a request answered by exactly one T frame. func (e *Engine) silkdOnce[T wire.Response](ctx context.Context, vsockSocket string, req wire.Request) error { s, err := e.dialSilkdSession(ctx, vsockSocket) if err != nil { diff --git a/sandboxd/main.go b/sandboxd/main.go index 1b68e7c1e..d6440d728 100644 --- a/sandboxd/main.go +++ b/sandboxd/main.go @@ -118,7 +118,7 @@ func main() { if err != nil { logger.Fatalf(ctx, err, "start mesh") } - defer func() { _ = msh.Shutdown() }() + context.AfterFunc(ctx, func() { _ = msh.Shutdown() }) placer = msh mgr.SetTemplateNotifier(func() { msh.UpdateSelf(mgr.WarmCounts(), mgr.TemplateHashes(), mgr.VolumeNames()) diff --git a/sandboxd/metastore/metastore.go b/sandboxd/metastore/metastore.go index b8287fbf6..0d86d1822 100644 --- a/sandboxd/metastore/metastore.go +++ b/sandboxd/metastore/metastore.go @@ -38,7 +38,7 @@ func Open(ctx context.Context, dsn, table, schema string) (*pgxpool.Pool, *pgx.C if err != nil { return nil, nil, fmt.Errorf("open meta store: %w", err) } - if err = EnsureTable(ctx, pool, table, schema); err != nil { + if err = ensureTable(ctx, pool, table, schema); err != nil { pool.Close() return nil, nil, err } @@ -86,8 +86,7 @@ func Listen(ctx context.Context, cfg *pgx.ConnConfig, channel string, synced fun } } -// EnsureTable runs schema only when table is missing. -func EnsureTable(ctx context.Context, pool *pgxpool.Pool, table, schema string) error { +func ensureTable(ctx context.Context, pool *pgxpool.Pool, table, schema string) error { qctx, cancel := context.WithTimeout(ctx, QueryTimeout) defer cancel() if exists, err := tableExists(qctx, pool, table); err != nil || exists { diff --git a/sandboxd/outbound/claim.go b/sandboxd/outbound/claim.go index ab67cd799..9280eca2c 100644 --- a/sandboxd/outbound/claim.go +++ b/sandboxd/outbound/claim.go @@ -142,6 +142,9 @@ func (c *claim) limiter(r config.EgressConnRate) *rate.Limiter { // current denies everything once a reload leaves the claim with no policy. func (c *claim) current() egress.PolicyEvaluator { + if c.sb.Replicated && c.h.o.Fenced != nil && c.h.o.Fenced() { + return egress.Policy{} + } v := c.h.o.View() if r := c.cur.Load(); r.view == v { return r.eval diff --git a/sandboxd/outbound/claim_test.go b/sandboxd/outbound/claim_test.go index 41d728fe7..98eb7fa27 100644 --- a/sandboxd/outbound/claim_test.go +++ b/sandboxd/outbound/claim_test.go @@ -127,6 +127,25 @@ func TestBlockedDestinationNeverReachesTheUpstream(t *testing.T) { } } +func TestAFencedNodeDeniesEgressToItsReplicatedClaimsOnly(t *testing.T) { + var fenced atomic.Bool + h := upstreamHost(t, &config.Config{}, func(o *Options) { o.Fenced = fenced.Load }) + allowAll := egress.Policy{Allow: []egress.Rule{{Host: "*"}}} + replicated := newClaim(h, &types.Sandbox{ID: "r", Replicated: true}, h.o.View(), allowAll) + plain := newClaim(h, &types.Sandbox{ID: "p"}, h.o.View(), allowAll) + fenced.Store(true) + if _, d := replicated.Eval("example.com", http.MethodGet, 443); d == egress.DecisionAllow { + t.Error("a fenced node let a replicated claim out") + } + if _, d := plain.Eval("example.com", http.MethodGet, 443); d != egress.DecisionAllow { + t.Errorf("a fenced node refused an unreplicated claim: %v", d) + } + fenced.Store(false) + if _, d := replicated.Eval("example.com", http.MethodGet, 443); d != egress.DecisionAllow { + t.Errorf("a lifted fence still refuses the replicated claim: %v", d) + } +} + func TestInternallyAllowedDestinationDialsDirect(t *testing.T) { var direct atomic.Int32 h := upstreamHost(t, &config.Config{}, func(o *Options) { diff --git a/sandboxd/outbound/dns.go b/sandboxd/outbound/dns.go index bd693f390..a566735e6 100644 --- a/sandboxd/outbound/dns.go +++ b/sandboxd/outbound/dns.go @@ -114,7 +114,11 @@ func (r *resolver) serveConn(conn net.Conn, datagram bool) { if err != nil { return } - r.inflight <- struct{}{} + select { + case r.inflight <- struct{}{}: + case <-r.ctx.Done(): + return + } buf := udpBufs.Get().(*[dnsMsgMax]byte) answer, ok := r.answer(query, buf[:], datagram) if ok { diff --git a/sandboxd/outbound/outbound.go b/sandboxd/outbound/outbound.go index 715ea02e3..da971f03a 100644 --- a/sandboxd/outbound/outbound.go +++ b/sandboxd/outbound/outbound.go @@ -129,6 +129,9 @@ func (h *Host) ArmProxy(ctx context.Context, sb *types.Sandbox) error { if d != nil { d.close() } + if sb.TunAddr.IsValid() && sb.VsockSocket != "" { + h.tunDown(sb.ID, sb.VsockSocket) + } return nil } id, tenant := sb.ID, sb.Tenant @@ -227,6 +230,20 @@ func (h *Host) Disarm(id string, removed bool) { } } +func (h *Host) DropTunnels(ids []string) { + var proxies []*egress.Proxy + h.mu.Lock() + for _, id := range ids { + if d := h.doors[id]; d != nil && d.proxy != nil { + proxies = append(proxies, d.proxy) + } + } + h.mu.Unlock() + for _, p := range proxies { + p.DropTunnels() + } +} + // DNSMalformed counts guest DNS queries dropped as malformed, with the connection that carried each. func (h *Host) DNSMalformed() uint64 { return h.dnsMalformed.Load() diff --git a/sandboxd/outbound/tun.go b/sandboxd/outbound/tun.go index d14b47131..c55bb708f 100644 --- a/sandboxd/outbound/tun.go +++ b/sandboxd/outbound/tun.go @@ -50,7 +50,13 @@ func (h *Host) RearmTun(ctx context.Context, sb *types.Sandbox) { h.dropTun(old) v := h.o.View() policy, ok := v.Resolve(sb) - if d == nil || !ok { + if d == nil { + return + } + if !ok { + if sb.TunAddr.IsValid() { + h.tunDown(sb.ID, sb.VsockSocket) + } return } h.armTun(ctx, sb, d, v, newClaim(h, sb, v, policy)) @@ -168,7 +174,7 @@ func (h *Host) armTun(ctx context.Context, sb *types.Sandbox, d *doors, v *View, logger.Warnf(ctx, "tun_up %s: %v", id, err) } if _, armed := h.TunUsed(id); !armed { - _ = h.o.Engine.TunDown(vctx, sock) + h.tunDown(id, sock) } }() } diff --git a/sandboxd/pool/archive.go b/sandboxd/pool/archive.go index 8a490af17..60b8c894f 100644 --- a/sandboxd/pool/archive.go +++ b/sandboxd/pool/archive.go @@ -61,7 +61,7 @@ func (m *Manager) archiveOnce(ctx context.Context) { continue // a reap-triggered archive is already exporting this one } after := m.archiveAfterFor(sb.Key) - if after <= 0 || now.Sub(sb.LastSeen()) < after { + if after <= 0 || now.Sub(sb.LastSeen()) < after || now.Before(sb.ArchiveRetryAt) { continue } m.archiving[sb.ID] = struct{}{} @@ -82,10 +82,17 @@ func (m *Manager) archiveOnce(ctx context.Context) { }() } -func (m *Manager) archive(ctx context.Context, sb *types.Sandbox) error { +func (m *Manager) archive(ctx context.Context, sb *types.Sandbox) (err error) { // Must finish even if the sweep ctx is canceled, or the record diverges from the store. ctx = context.WithoutCancel(ctx) - defer m.untrack(m.archiving, sb.ID) + defer func() { + m.mu.Lock() + delete(m.archiving, sb.ID) + if err != nil { + sb.ArchiveRetryAt = time.Now().Add(archiveRetryDelay) + } + m.mu.Unlock() + }() m.mu.Lock() ok := m.claimed[sb.ID] == sb && sb.HibernateSnap != "" && sb.ArchiveCk == "" m.mu.Unlock() @@ -93,7 +100,7 @@ func (m *Manager) archive(ctx context.Context, sb *types.Sandbox) error { return errWokeMeanwhile } sb.Transition.Lock() - err := m.resolveIntent(ctx, sb) + err = m.resolveIntent(ctx, sb) snap := sb.HibernateSnap sb.Transition.Unlock() if err != nil { @@ -114,6 +121,7 @@ func (m *Manager) archive(ctx context.Context, sb *types.Sandbox) error { // A hibernated source is copied from its wake image, so no VM starts. ck, srcSnap, err := m.publishCheckpoint(ctx, sb, ckID, "archive", sb.Tenant, snap, true) if err != nil { + m.deleteOrphanArchiveCk(ctx, ckID) return err } // The claim must still hold the exact snapshot the export captured, or the ck is stale diff --git a/sandboxd/pool/checkpoint_cleanup_test.go b/sandboxd/pool/checkpoint_cleanup_test.go index de2c3e9b7..e7d19811e 100644 --- a/sandboxd/pool/checkpoint_cleanup_test.go +++ b/sandboxd/pool/checkpoint_cleanup_test.go @@ -118,7 +118,7 @@ func TestCheckpointCleanupWaitsForTheClaimWrite(t *testing.T) { } else if err := m.store.commit(m.store.mark()); err != nil { t.Fatalf("commit the journal: %v", err) } - other := mustClaim(t, m, testKey) + other := mustClaim(t, m, plainKey) m.mu.Lock() other.LeaseSeconds++ m.store.set(other) diff --git a/sandboxd/pool/claim.go b/sandboxd/pool/claim.go index 5eeeef992..186d4d9aa 100644 --- a/sandboxd/pool/claim.go +++ b/sandboxd/pool/claim.go @@ -4,6 +4,7 @@ import ( "context" "fmt" "net" + "slices" "sync" "sync/atomic" "time" @@ -341,7 +342,7 @@ func (m *Manager) releaseResolved(ctx context.Context, id string, sb *types.Sand ctx = context.WithoutCancel(ctx) m.releaseTun(ctx, id, tunAddr) if ck != "" { - m.purgeArchiveCk(ctx, id, ck, sb.Tenant) // archived: no local VM + m.purgeArchiveCk(ctx, id, ck, sb.Tenant) m.untrack(m.pendingCks, ck) } td := m.quiesceVolumes(ctx, sb) @@ -360,7 +361,7 @@ func (m *Manager) releaseResolved(ctx context.Context, id string, sb *types.Sand } } m.out.Disarm(id, removed) - m.dropSnap(ctx, snap) + m.dropSnapAfterTransition(ctx, sb, snap) return err } @@ -608,6 +609,9 @@ func (m *Manager) finalizeBatch(ctx context.Context, sbs []*types.Sandbox, ttl t if err == nil { err = m.quotaErr(len(sbs), sbs[0].Tenant) } + if err == nil && m.fencedReplicating(sbs[0].PolicyKey()) { + err = errFencedReplicating + } if err != nil { m.mu.Unlock() for _, sb := range sbs { @@ -649,6 +653,7 @@ func (m *Manager) finalizeBatch(ctx context.Context, sbs []*types.Sandbox, ttl t // rollbackClaim unwinds a claim batch: a NIC that cannot be locked is never handed out. func (m *Manager) rollbackClaim(ctx context.Context, sbs []*types.Sandbox, billed bool) { m.mu.Lock() + sbs = slices.DeleteFunc(slices.Clone(sbs), func(sb *types.Sandbox) bool { return m.claimed[sb.ID] != sb }) ids := make([]string, len(sbs)) holds := make([][]types.Volume, len(sbs)) for i, sb := range sbs { @@ -672,6 +677,10 @@ func (m *Manager) rollbackClaim(ctx context.Context, sbs []*types.Sandbox, bille } } +func (m *Manager) fencedReplicating(key types.PoolKey) bool { + return m.archives != nil && m.view.Load().poolReplicate[key] > 0 && m.isFenced() +} + func (m *Manager) abortVolumeClaim(ctx context.Context, vmName string, reserved *[]types.Volume) { m.removeOrRetry(ctx, vmName, "", volumeTeardown{holds: *reserved}) *reserved = nil @@ -704,6 +713,9 @@ func (m *Manager) admitOptions(ctx context.Context, key types.PoolKey, o ClaimOp if err := m.validate(key); err != nil { return nil, nil, nil, err } + if m.fencedReplicating(key) { + return nil, nil, nil, errFencedReplicating + } if err := archivable(key, len(o.Volumes) > 0, o.OnExpire); err != nil { return nil, nil, nil, err } diff --git a/sandboxd/pool/expiry_test.go b/sandboxd/pool/expiry_test.go index 1e9768781..0dcd3bdab 100644 --- a/sandboxd/pool/expiry_test.go +++ b/sandboxd/pool/expiry_test.go @@ -101,6 +101,13 @@ func TestExpiryArchiveRetriesOnlyTheExportAfterItFails(t *testing.T) { } exports := failing.publishes.Load() failing.fail.Store(false) + m.mu.Lock() + spaced := sb.ArchiveRetryAt.After(time.Now()) + sb.ArchiveRetryAt = time.Time{} + m.mu.Unlock() + if !spaced { + t.Error("a failed archive was retried without spacing") + } m.reapOnce(t.Context()) waitFor(t, func() bool { return archivedCount(m) == 1 }) if m.Counters().Hibernates != 1 || failing.publishes.Load() != exports+1 { @@ -135,6 +142,13 @@ func TestExpiryArchiveRetriesTheHibernateAfterItFails(t *testing.T) { eng.mu.Lock() eng.hibernateErr = nil eng.mu.Unlock() + m.mu.Lock() + spaced := sb.ArchiveRetryAt.After(time.Now()) + sb.ArchiveRetryAt = time.Time{} + m.mu.Unlock() + if !spaced { + t.Error("a failed archive was retried without spacing") + } m.reapOnce(t.Context()) waitFor(t, func() bool { return archivedCount(m) == 1 }) } diff --git a/sandboxd/pool/failover.go b/sandboxd/pool/failover.go index 6715c9e8b..b85c5ea67 100644 --- a/sandboxd/pool/failover.go +++ b/sandboxd/pool/failover.go @@ -193,6 +193,9 @@ func (m *Manager) refusedReplace(ctx context.Context, sb *types.Sandbox) error { switch { case err != nil: return err + case ok && rec.Owner == m.nodeID && rec.Consumed: + m.reputArchive(ctx, sb) + return nil case ok && rec.Owner == m.nodeID: _, _ = m.commitIfLive(ctx, sb, func() { sb.ArchiveEpoch = max(sb.ArchiveEpoch, rec.Epoch) }) return nil diff --git a/sandboxd/pool/fork.go b/sandboxd/pool/fork.go index 9684c97fe..59ebca3ea 100644 --- a/sandboxd/pool/fork.go +++ b/sandboxd/pool/fork.go @@ -5,7 +5,6 @@ import ( "errors" "fmt" "os" - "path/filepath" "time" "github.com/cocoonstack/sandbox/sandboxd/types" @@ -100,11 +99,10 @@ func (m *Manager) forkSource(ctx context.Context, sb *types.Sandbox) (create vmP if err != nil { return nil, nil, false, err } - exportDir := filepath.Join(dir, "export") // cocoon wants the target absent - if err = m.eng.SnapshotExport(ctx, sb.HibernateSnap, exportDir); err != nil { + if err = m.eng.SnapshotExport(ctx, sb.HibernateSnap, dir); err != nil { _ = os.RemoveAll(dir) return nil, nil, false, err } - return func(name string) (types.VMRecord, error) { return m.eng.Clone(ctx, exportDir, name, sb.Key) }, + return func(name string) (types.VMRecord, error) { return m.eng.Clone(ctx, dir, name, sb.Key) }, func() { _ = os.RemoveAll(dir) }, inherited, nil } diff --git a/sandboxd/pool/fork_test.go b/sandboxd/pool/fork_test.go index 802d38343..b3c5d7181 100644 --- a/sandboxd/pool/fork_test.go +++ b/sandboxd/pool/fork_test.go @@ -294,6 +294,10 @@ func TestForkReadsTheParentsOnExpireUnderTheLock(t *testing.T) { parent := mustClaim(t, m, testKey) stop := make(chan struct{}) var wg sync.WaitGroup + t.Cleanup(func() { + close(stop) + wg.Wait() + }) wg.Go(func() { action := types.ExpireArchive for { @@ -318,6 +322,4 @@ func TestForkReadsTheParentsOnExpireUnderTheLock(t *testing.T) { t.Fatalf("Fork: %v", err) } } - close(stop) - wg.Wait() } diff --git a/sandboxd/pool/golden.go b/sandboxd/pool/golden.go index 76a04a595..2c314c8e8 100644 --- a/sandboxd/pool/golden.go +++ b/sandboxd/pool/golden.go @@ -20,7 +20,7 @@ func (m *Manager) buildGolden(ctx context.Context, p *pool) { logger := log.WithFunc("pool.buildGolden") hash := p.hash name := vmPrefix + "gb-" + hash - snap := goldenPrefix + hash + snap := goldenPrefix + hash + "-" + randHex(6) final := filepath.Join(m.goldensDir(), hash) imageID := m.imageIDs(ctx)[p.key.Template] v := m.view.Load() diff --git a/sandboxd/pool/hibernate.go b/sandboxd/pool/hibernate.go index 112ed7a2d..b9577a3f5 100644 --- a/sandboxd/pool/hibernate.go +++ b/sandboxd/pool/hibernate.go @@ -339,7 +339,7 @@ func (m *Manager) dropStale(ctx context.Context, sb *types.Sandbox) { } } -// settlePendingSnap settles a dangling hibernate intent and bills an adopted one, which never was. +// settlePendingSnap settles a dangling hibernate intent and records the hibernate it adopts. func (m *Manager) settlePendingSnap(ctx context.Context, sb *types.Sandbox) (adopted bool, err error) { if sb.PendingSnap == "" { return false, nil @@ -390,5 +390,5 @@ func (m *Manager) recordHibernate(ctx context.Context, sb *types.Sandbox) { func skipIdle(sb *types.Sandbox) bool { return sb.Key.Net == types.NetEgress || hasVolumes(sb) || - sb.HibernateSnap != "" || sb.ArchiveCk != "" || sb.Failed != "" || sb.Busy() + sb.HibernateSnap != "" || sb.ArchiveCk != "" || sb.PendingSnap != "" || sb.SaveSnap != "" || sb.Failed != "" || sb.Busy() } diff --git a/sandboxd/pool/hibernate_test.go b/sandboxd/pool/hibernate_test.go index eeb78dbfc..0e23c9ce1 100644 --- a/sandboxd/pool/hibernate_test.go +++ b/sandboxd/pool/hibernate_test.go @@ -478,6 +478,24 @@ func TestReleaseHibernatedDropsSnapshot(t *testing.T) { } } +func TestReleaseDuringAnExportDropsTheSnapshotOnceTheExportEnds(t *testing.T) { + eng := newFakeEngine() + m := newTestManager(t, eng) + sb := mustClaim(t, m, testKey) + hibernateClaim(t, m, sb) + snap := eng.hibernates[0] + + sb.Transition.Lock() + if err := m.Release(t.Context(), sb.ID, Cred{Token: sb.Token}); err != nil { + t.Fatalf("Release: %v", err) + } + if eng.snapRemoved(snap) { + t.Fatal("the snapshot was dropped while an export still held it") + } + sb.Transition.Unlock() + waitFor(t, func() bool { return eng.snapRemoved(snap) }) +} + func TestReleaseMidHibernateDropsOrphanSnapshot(t *testing.T) { synctest.Test(t, func(t *testing.T) { eng := newFakeEngine() diff --git a/sandboxd/pool/lease.go b/sandboxd/pool/lease.go index 60671b756..86cb097ef 100644 --- a/sandboxd/pool/lease.go +++ b/sandboxd/pool/lease.go @@ -73,10 +73,23 @@ func (m *Manager) trip(ctx context.Context) { } close(old.ch) log.WithFunc("pool.trip").Warnf(ctx, "node lease not renewed for %s; replicated claims answer 503 until it is", leaseWindow) + m.out.DropTunnels(m.replicatedIDs()) return } } +func (m *Manager) replicatedIDs() []string { + m.mu.Lock() + defer m.mu.Unlock() + var ids []string + for id, sb := range m.claimed { + if sb.Replicated { + ids = append(ids, id) + } + } + return ids +} + // unfence confirms each replicated claim against its row, then lifts the fence if that finished inside the window the renewal at start still covers. func (m *Manager) unfence(ctx context.Context, start time.Time) { logger := log.WithFunc("pool.unfence") @@ -118,7 +131,7 @@ func (m *Manager) unfence(ctx context.Context, start time.Time) { continue } if td := m.forgetAdopted(ctx, sb); td != nil { - m.out.DisarmTun(ctx, sb.ID, "") + m.out.Disarm(sb.ID, false) go func() { sb.Transition.Lock() defer sb.Transition.Unlock() diff --git a/sandboxd/pool/lease_test.go b/sandboxd/pool/lease_test.go index ed68205eb..d5bd778d6 100644 --- a/sandboxd/pool/lease_test.go +++ b/sandboxd/pool/lease_test.go @@ -331,6 +331,22 @@ func TestLeaseANodeLiftedAfterAnOutageAdoptsOnlyOnceEveryOwnerCouldRenew(t *test }) } +func TestLeaseFencedNodeRefusesNewClaimsOfReplicatingPoolsOnly(t *testing.T) { + c := newReplicaCluster(t) + a := c.node(t, "a", t.TempDir(), keepAnHour) + a.tripFence() + if _, err := claimAny(t.Context(), a, testKey, ClaimOptions{}); !errors.Is(err, ErrQuota) { + t.Errorf("claim of a replicating pool while fenced: %v, want ErrQuota", err) + } + if _, err := claimAny(t.Context(), a, plainKey, ClaimOptions{}); err != nil { + t.Errorf("claim of an unreplicated pool while fenced: %v", err) + } + a.liftFence() + if _, err := claimAny(t.Context(), a, testKey, ClaimOptions{}); err != nil { + t.Errorf("claim once the fence lifted: %v", err) + } +} + func TestLeaseFencedNodeDoesNotAnswerAsOwnerOfReplicatedClaims(t *testing.T) { c := newReplicaCluster(t) a := c.node(t, "a", t.TempDir(), keepAnHour) diff --git a/sandboxd/pool/move.go b/sandboxd/pool/move.go index bd44e3d76..8737d72d0 100644 --- a/sandboxd/pool/move.go +++ b/sandboxd/pool/move.go @@ -90,7 +90,7 @@ func (m *Manager) moveDir(id string) string { return filepath.Join(m.dataDir, movesDir, id) } -// A target marks a committed move started and aborts a stale pending or final one; only the source deletes an ended one. +// sweepStaleMoves marks a stale committed move started on its target, aborts a stale pending or final one on either end, and deletes an ended one on its source. func (m *Manager) sweepStaleMoves(ctx context.Context, now time.Time) { logger := log.WithFunc("pool.sweepStaleMoves") recs, err := m.moves.Stale(ctx, now.Add(-moveStaleAge)) diff --git a/sandboxd/pool/pool.go b/sandboxd/pool/pool.go index 5c946ec72..384d08c65 100644 --- a/sandboxd/pool/pool.go +++ b/sandboxd/pool/pool.go @@ -43,6 +43,7 @@ const ( livenessInterval = 10 * time.Second imageInterval = time.Minute buildRetryDelay = 30 * time.Second + archiveRetryDelay = time.Minute claimProbeTimeout = 15 * time.Second coldProbeTimeout = 90 * time.Second // A clone answers in ~1s even saturated, so a silent one is replaced, not waited on @@ -132,11 +133,13 @@ var ( errNotSleeping = errors.New("sandbox is no longer hibernated or archived") errWokeMeanwhile = errors.New("woke between sweep and hibernate") errNoEgressTap = errors.New("egress-lane claim has no lockable tap") - errNoFailover = errors.New("archive_failover is off, so no peer can adopt the claim") - errNoArchiveRow = errors.New("archived before archive_failover was on, so no registry row names it") - errArchiveBusy = errors.New("an archive of the claim is in flight") - errNoAdopter = errors.New("no peer adopted the claim") - errMovesOff = fmt.Errorf("%w: live moves need a meta store cell and an encrypted mesh", ErrNotMovable) + + errFencedReplicating = fmt.Errorf("%w: node lease lost", ErrQuota) + errNoFailover = errors.New("archive_failover is off, so no peer can adopt the claim") + errNoArchiveRow = errors.New("archived before archive_failover was on, so no registry row names it") + errArchiveBusy = errors.New("an archive of the claim is in flight") + errNoAdopter = errors.New("no peer adopted the claim") + errMovesOff = fmt.Errorf("%w: live moves need a meta store cell and an encrypted mesh", ErrNotMovable) ) // Engine is the slice of the cocoon driver the manager consumes. diff --git a/sandboxd/pool/reader.go b/sandboxd/pool/reader.go index 4e8a29eed..7207ad573 100644 --- a/sandboxd/pool/reader.go +++ b/sandboxd/pool/reader.go @@ -64,7 +64,9 @@ func (m *Manager) DialReader(ctx context.Context, id, token string) (conn net.Co conn, err = m.eng.DialReader(ctx, r.sock) if err != nil { m.releaseReader(r) - m.dropReader(ctx, r, true) + if ctx.Err() == nil { + m.dropReader(ctx, r, true) + } return nil, "", fmt.Errorf("%w: %w", ErrNoReader, err) } return newHeldConn(conn, fenced, func() { m.releaseReader(r) }), r.source, nil diff --git a/sandboxd/pool/reap.go b/sandboxd/pool/reap.go index 9ce89de44..2628e96dc 100644 --- a/sandboxd/pool/reap.go +++ b/sandboxd/pool/reap.go @@ -52,7 +52,7 @@ func (m *Manager) reapOnce(ctx context.Context) { purges = append(purges, sb.ArchiveCk) case m.archivesAtDeadline(sb): // Archive instead of destroy, kept in m.claimed for archive() to move - if _, busy := m.archiving[id]; busy { + if _, busy := m.archiving[id]; busy || now.Before(sb.ArchiveRetryAt) { continue } m.archiving[id] = struct{}{} @@ -154,7 +154,7 @@ func (m *Manager) reapOnce(ctx context.Context) { td := m.quiesceVolumes(ctx, v.sb) td.holds = v.holds m.out.Disarm(v.id, m.removeOrRetry(ctx, v.vmName, v.id, td)) - m.dropSnap(ctx, v.snap) + m.dropSnapAfterTransition(ctx, v.sb, v.snap) m.counters.reaps.Add(1) m.recordUsage(ctx, usage.Event{Event: reapEvent, ID: v.id, VMName: v.vmName, Tenant: v.tenant, Deadline: v.deadline}) logger.Infof(ctx, "reaped expired sandbox %s (%s)", v.id, v.vmName) @@ -177,7 +177,12 @@ func (m *Manager) pauseArchive(ctx context.Context, sb *types.Sandbox) error { } sb.Transition.Unlock() if !live || err != nil { - m.untrack(m.archiving, sb.ID) + m.mu.Lock() + delete(m.archiving, sb.ID) + if err != nil { + sb.ArchiveRetryAt = time.Now().Add(archiveRetryDelay) + } + m.mu.Unlock() return cmp.Or(err, errWokeMeanwhile) } return m.archive(ctx, sb) diff --git a/sandboxd/pool/reconcile.go b/sandboxd/pool/reconcile.go index c0a74a48b..6cc51e872 100644 --- a/sandboxd/pool/reconcile.go +++ b/sandboxd/pool/reconcile.go @@ -236,7 +236,6 @@ func (m *Manager) removeStaleVM(ctx context.Context, name string, rec types.VMRe m.queueStaleCreate(name, rec.TapDevice()) return false } - // Not-creating: the record moved on under the lock; remove normally. } if m.removeVM(ctx, name) { return true @@ -313,7 +312,6 @@ func (m *Manager) resyncEgress(ctx context.Context, live map[string]types.VMReco keep[sb.TAP] = true } } - // Keep every still-running VM's tap; sweep only a confirmed-gone VM's table. for name, rec := range live { if tap := rec.TapDevice(); tap != "" && !removed[name] { keep[tap] = true diff --git a/sandboxd/pool/refill.go b/sandboxd/pool/refill.go index 9c86abe7f..50c6e4a3b 100644 --- a/sandboxd/pool/refill.go +++ b/sandboxd/pool/refill.go @@ -28,7 +28,7 @@ func (m *Manager) refillOnce(ctx context.Context) { inFlight := 0 for _, p := range m.pools { inFlight += p.refilling - // SetPools leaves a removed pool in place while a build, refill or claim clone is in flight + // applyPools leaves a removed pool in place while a build, refill or claim clone is in flight if p.removed { if p.settled() { m.dropPoolLocked(ctx, p) diff --git a/sandboxd/pool/remove.go b/sandboxd/pool/remove.go index 5480cc8b0..dba0c9550 100644 --- a/sandboxd/pool/remove.go +++ b/sandboxd/pool/remove.go @@ -11,6 +11,7 @@ import ( "github.com/projecteru2/core/log" "github.com/cocoonstack/sandbox/sandboxd/engine" + "github.com/cocoonstack/sandbox/sandboxd/types" ) func (m *Manager) removeVM(ctx context.Context, name string) bool { @@ -117,6 +118,22 @@ func (m *Manager) destroyAll(ctx context.Context, names []string) *sync.WaitGrou return m.runBounded(ctx, len(names), func(ctx context.Context, i int) { m.destroy(ctx, names[i]) }) } +func (m *Manager) dropSnapAfterTransition(ctx context.Context, sb *types.Sandbox, snap string) { + if snap == "" { + return + } + if sb.Transition.TryLock() { + defer sb.Transition.Unlock() + m.dropSnap(ctx, snap) + return + } + go func() { + sb.Transition.Lock() + defer sb.Transition.Unlock() + m.dropSnap(ctx, snap) + }() +} + func (m *Manager) dropSnap(ctx context.Context, snap string) { if snap == "" { return diff --git a/sandboxd/pool/replica.go b/sandboxd/pool/replica.go index 68ffb2837..829ac48a9 100644 --- a/sandboxd/pool/replica.go +++ b/sandboxd/pool/replica.go @@ -96,6 +96,7 @@ func (m *Manager) replicate(ctx context.Context, sb *types.Sandbox) error { m.mu.Unlock() defer m.untrack(m.pendingCks, ckID) if _, _, err := m.publishCheckpoint(ctx, sb, ckID, "replica", sb.Tenant, "", true); err != nil { + m.deleteOrphanArchiveCk(ctx, ckID) return err } var prev string diff --git a/sandboxd/pool/replica_test.go b/sandboxd/pool/replica_test.go index 0893bc129..9eabe968d 100644 --- a/sandboxd/pool/replica_test.go +++ b/sandboxd/pool/replica_test.go @@ -662,8 +662,8 @@ func TestReplicaPassSkipsWhileFenced(t *testing.T) { synctest.Test(t, func(t *testing.T) { c := newReplicaCluster(t) a := c.node(t, "a", t.TempDir(), keepAnHour) - a.tripFence() sb := mustClaim(t, a, testKey) + a.tripFence() a.replicateOnce(t.Context()) time.Sleep(replicateEvery * time.Second) a.replicateOnce(t.Context()) diff --git a/sandboxd/pool/setpools.go b/sandboxd/pool/setpools.go index f6b746856..d287d9afe 100644 --- a/sandboxd/pool/setpools.go +++ b/sandboxd/pool/setpools.go @@ -40,6 +40,9 @@ func (m *Manager) desiredPools(specs []config.PoolSpec) (map[types.PoolKey]confi if err := spec.ValidateLimits(); err != nil { return nil, fmt.Errorf("%w: pool %s: %w", ErrBadConfig, spec.Template, err) } + if m.view.Load().poolReaders[spec.PoolKey] && spec.Warm < 1 { + return nil, fmt.Errorf("%w: pool %s: reader needs warm of at least 1", ErrBadConfig, spec.Template) + } for _, f := range spec.ConfigOwned() { if !f.Value.IsZero() { return nil, fmt.Errorf("%w: pool %q: %s is set in the config file, not via the API", ErrBadKey, spec.Template, f.Name) diff --git a/sandboxd/pool/telemetry_test.go b/sandboxd/pool/telemetry_test.go index 4c0230011..da6ebbb08 100644 --- a/sandboxd/pool/telemetry_test.go +++ b/sandboxd/pool/telemetry_test.go @@ -392,7 +392,10 @@ func TestSandboxesIndexOmitsTokens(t *testing.T) { if len(list) != 1 || list[0].ID != sb.ID || list[0].ClaimRef != "ns/workload" || list[0].Hibernated { t.Fatalf("index %+v", list) } - raw, _ := json.Marshal(list) + raw, err := json.Marshal(list) + if err != nil { + t.Fatalf("marshal the index: %v", err) + } if strings.Contains(string(raw), sb.Token) { t.Error("index leaked a token") } diff --git a/sandboxd/pool/tun.go b/sandboxd/pool/tun.go index f2b8dbaa1..bed5da481 100644 --- a/sandboxd/pool/tun.go +++ b/sandboxd/pool/tun.go @@ -105,7 +105,7 @@ func (m *Manager) assignTun(ctx context.Context, sbs []*types.Sandbox) { } v := m.view.Load().out for _, sb := range sbs { - if !v.Tun(sb.PolicyKey()) { + if !v.Tun(sb.PolicyKey()) || sb.NoEgress { continue } addr, err := m.tunAddrs.Assign(ctx, sb.ID) @@ -230,7 +230,7 @@ func (m *Manager) planTun(sb *types.Sandbox, locked bool, enabled *bool, profile if !locked && (!next.off || next.profile != prev.profile) { return tunPlan{}, ErrPaused } - assign := !next.off && !prev.addr.IsValid() && m.tunAddrs != nil && m.view.Load().out.Tun(sb.PolicyKey()) + assign := !next.off && !prev.addr.IsValid() && !sb.NoEgress && m.tunAddrs != nil && m.view.Load().out.Tun(sb.PolicyKey()) return tunPlan{prev: prev, next: next, assign: assign}, nil } diff --git a/sandboxd/pool/volume.go b/sandboxd/pool/volume.go index 7819f789a..7e7c19ceb 100644 --- a/sandboxd/pool/volume.go +++ b/sandboxd/pool/volume.go @@ -26,7 +26,6 @@ type resolvedVolume struct { applied types.Volume } -// volumeTeardown captures marker paths at quiesce time; a later config can re-point the entry. type volumeTeardown struct { holds []types.Volume clears []string @@ -115,9 +114,9 @@ func (m *Manager) resolveVolumes(ctx context.Context, tenant string, requested [ if len(requested) == 0 { return nil, nil } - applied, err := types.ValidateVolumes(requested) + applied, err := normalizeVolumes(requested) if err != nil { - return nil, fmt.Errorf("%w: %w", ErrBadVolume, err) + return nil, err } catalog := m.catalogSnapshot() resolved := make([]resolvedVolume, 0, len(applied)) diff --git a/sandboxd/pool/volume_mutation.go b/sandboxd/pool/volume_mutation.go index 68a6c81f8..9cb03f7de 100644 --- a/sandboxd/pool/volume_mutation.go +++ b/sandboxd/pool/volume_mutation.go @@ -46,7 +46,7 @@ func (m *Manager) PatchVolumes(ctx context.Context, id, tenant string, patch typ if err := validateDetachNames(patch.Remove); err != nil { return nil, err } - additions, err := normalizeLiveVolumes(patch.Add) + additions, err := normalizeVolumes(patch.Add) if err != nil { return nil, err } @@ -91,7 +91,7 @@ func (m *Manager) changeVolumes(ctx context.Context, id, tenant string, change v if err != nil { return view, err } - target, err = normalizeLiveVolumes(target) + target, err = normalizeVolumes(target) if err != nil { return view, err } @@ -430,7 +430,7 @@ func (m *Manager) clearVolumeError(sb *types.Sandbox, run *volumeRun, name strin } } -func normalizeLiveVolumes(volumes []types.Volume) ([]types.Volume, error) { +func normalizeVolumes(volumes []types.Volume) ([]types.Volume, error) { out, err := types.ValidateVolumes(volumes) if err != nil { return nil, fmt.Errorf("%w: %w", ErrBadVolume, err) diff --git a/sandboxd/server/preview.go b/sandboxd/server/preview.go index 8e97a2cba..c92bebb08 100644 --- a/sandboxd/server/preview.go +++ b/sandboxd/server/preview.go @@ -93,13 +93,15 @@ func (p *PreviewServer) serve(w http.ResponseWriter, r *http.Request) { return } // The token names the owner by its advertise address at mint time; holding the claim is what counts - if _, held := p.mgr.Sandbox(claims.ID); !held && claims.Owner != p.owner { - if r.Header.Get(forwardedHeader) != "" { - http.Error(w, "preview owner is not this node", http.StatusBadGateway) + if claims.Owner != p.owner { + if _, held := p.mgr.Sandbox(claims.ID); !held { + if r.Header.Get(forwardedHeader) != "" { + http.Error(w, "preview owner is not this node", http.StatusBadGateway) + return + } + p.forward(w, r, claims.Owner) return } - p.forward(w, r, claims.Owner) - return } p.proxyLocal(w, r, claims) } diff --git a/sandboxd/server/reader.go b/sandboxd/server/reader.go index 12dd40424..a70431739 100644 --- a/sandboxd/server/reader.go +++ b/sandboxd/server/reader.go @@ -181,10 +181,7 @@ func (r *router) fail(err error) { default: log.WithFunc("server.router.fail").Errorf(r.ctx, err, "agent socket for %s", r.id) } - frame, encodeErr := wire.EncodeResponse(&wire.ErrorResp{Kind: kind, Message: msg}) - if encodeErr != nil { - return - } + frame, _ := wire.EncodeResponse(&wire.ErrorResp{Kind: kind, Message: msg}) _, _ = r.client.Write(append(frame, '\n')) } diff --git a/sandboxd/server/server.go b/sandboxd/server/server.go index fa7473325..ad5d80244 100644 --- a/sandboxd/server/server.go +++ b/sandboxd/server/server.go @@ -479,8 +479,12 @@ func (s *Server) redirectVolumeClaim(ctx context.Context, w http.ResponseWriter, if localVolumes && (!promoted || localTemplate || pooled) { return false, nil } + missing := pool.ErrVolumeUnavailable + if localVolumes { + missing = pool.ErrUnknownTemplate + } if s.placer == nil || req.NoRedirect { - return false, pool.ErrVolumeUnavailable + return false, missing } var owners []string @@ -495,7 +499,7 @@ func (s *Server) redirectVolumeClaim(ctx context.Context, w http.ResponseWriter, owners = s.placer.VolumeOwners(names) } if len(owners) == 0 { - return false, pool.ErrVolumeUnavailable + return false, missing } writeJSON(w, http.StatusOK, ClaimResponse{Redirect: s.clientAddrs(owners), RequirePromoted: promoted}) return true, nil @@ -924,7 +928,10 @@ func (s *Server) requireToken(next http.HandlerFunc) http.HandlerFunc { writeErr(w, http.StatusUnauthorized, "invalid api token") return } - next(w, r.WithContext(withTenant(r.Context(), tenant))) + if tenant != nil { + r = r.WithContext(withTenant(r.Context(), tenant)) + } + next(w, r) } } diff --git a/sandboxd/server/server_test.go b/sandboxd/server/server_test.go index 76b2ea132..4a5d19dda 100644 --- a/sandboxd/server/server_test.go +++ b/sandboxd/server/server_test.go @@ -1442,7 +1442,7 @@ func TestVolumeClaimUsesVolumeAndTemplateIntersections(t *testing.T) { { name: "redirect target refuses missing promoted template before provisioning", localVolumes: true, noRedirect: true, requirePromoted: true, - wantStatus: http.StatusBadRequest, wantPromoted: true, + wantStatus: http.StatusNotFound, wantPromoted: true, }, { name: "promoted template refuses when no volume holder exists", diff --git a/sandboxd/tunnet/addrs.go b/sandboxd/tunnet/addrs.go index 177947db2..a942aea7f 100644 --- a/sandboxd/tunnet/addrs.go +++ b/sandboxd/tunnet/addrs.go @@ -143,7 +143,7 @@ func (l *Local) Release(_ context.Context, claimID string, addr netip.Addr) erro return fmt.Errorf("open %s: %w", freedFile, err) } _, err = f.Write(appendFreed(nil, addr, now)) - err = cmp.Or(err, f.Close()) + err = cmp.Or(err, f.Sync(), f.Close()) if err != nil { return fmt.Errorf("append %s: %w", freedFile, err) } diff --git a/sandboxd/tunnet/gro.go b/sandboxd/tunnet/gro.go index 5cc0921a3..8b9ce6687 100644 --- a/sandboxd/tunnet/gro.go +++ b/sandboxd/tunnet/gro.go @@ -31,8 +31,12 @@ type coalescer struct { // add takes an admitted frame held in bp; false leaves both with the caller to enqueue unmerged. func (c *coalescer) add(r *Router, l *link, bp *[]byte, frame []byte) bool { + if !l.merge { + c.flush(r) + return false + } seg, ok := segmentOf(frame) - if !ok || !l.merge { + if !ok { c.flush(r) return false } diff --git a/sandboxd/types/types.go b/sandboxd/types/types.go index e52ba7c0c..4a38cd206 100644 --- a/sandboxd/types/types.go +++ b/sandboxd/types/types.go @@ -340,6 +340,8 @@ type Sandbox struct { // 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:"-"` + ArchiveRetryAt time.Time `json:"-"` + // lastActivity is unix-nanos of the last data-plane connection, stamped lock-free. lastActivity atomic.Int64 // open counts the data-plane connections held right now. diff --git a/sandboxd/usage/ship.go b/sandboxd/usage/ship.go index 348d1ff83..532a6a351 100644 --- a/sandboxd/usage/ship.go +++ b/sandboxd/usage/ship.go @@ -139,7 +139,7 @@ func (s *shipper) locate(ctx context.Context, live, backup *os.File) (*os.File, } } if found == nil { - log.WithFunc("usage.shipper.locate").Warn(ctx, "the last landed usage line rotated out of the journal; the lines between it and the oldest kept line are lost") + log.WithFunc("usage.shipper.locate").Warn(ctx, "the last landed usage line is gone from the journal; shipping resumes at the oldest kept line, and lines rotated out unlanded are lost") return oldest, 0 } return found, s.at.end diff --git a/scripts/archive-e2e.sh b/scripts/archive-e2e.sh old mode 100644 new mode 100755 diff --git a/sdk/go/client_test.go b/sdk/go/client_test.go index 5755ade51..193a59cef 100644 --- a/sdk/go/client_test.go +++ b/sdk/go/client_test.go @@ -204,6 +204,24 @@ func TestAMovedSandboxRetriesAStreamCallAtItsNewOwner(t *testing.T) { } } +func TestARebindThatLosesTheRaceToAnotherCallStillRetries(t *testing.T) { + var sb *Sandbox + var adopter *httptest.Server + adopter = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path == "/v1/sandboxes/sb_9/owner" { + sb.mu.Lock() + sb.owner = addrOf(adopter) + sb.mu.Unlock() + _ = json.MarshalWrite(w, map[string]string{"owner_addr": addrOf(adopter)}) + } + })) + t.Cleanup(adopter.Close) + sb = &Sandbox{ID: "sb_9", c: testClient(t, adopter), token: "tok9", owner: "127.0.0.1:1"} + if !sb.rebind(t.Context(), &ownerGoneError{err: errors.New("refused"), id: "sb_9", addr: "127.0.0.1:1"}) { + t.Error("a call that failed at the old owner was not retried after another call rebound the handle") + } +} + func TestAWriteToAGoneSandboxLooksItsOwnerUpOnce(t *testing.T) { var lookups atomic.Int32 node := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { diff --git a/sdk/go/keepalive.go b/sdk/go/keepalive.go index 8b73be550..3ed3d2b4f 100644 --- a/sdk/go/keepalive.go +++ b/sdk/go/keepalive.go @@ -102,7 +102,6 @@ func (c *agentConn) quiet() bool { return c.sock != nil && c.sock.Read(c.peekFn) == nil && c.peeked } -// sent sends req on c and hands it back, closing it on failure. func (c *agentConn) sent(req wire.Request) (*agentConn, error) { if err := c.Send(req); err != nil { _ = c.Close() diff --git a/sdk/go/port.go b/sdk/go/port.go index 680f3b6ea..b752e849e 100644 --- a/sdk/go/port.go +++ b/sdk/go/port.go @@ -102,6 +102,7 @@ func (s *Sandbox) ProxyPort(ctx context.Context, localAddr string, port uint16) for { local, err := l.Accept() if err != nil { + _ = l.Close() return } go s.proxyConn(ctx, local, port) diff --git a/sdk/go/sandbox.go b/sdk/go/sandbox.go index 7b85e326c..42ab1de38 100644 --- a/sdk/go/sandbox.go +++ b/sdk/go/sandbox.go @@ -16,9 +16,10 @@ import ( ) const ( - releaseTimeout = 30 * time.Second - killWait = 5 * time.Second - stdinChunk = 32 * 1024 + releaseTimeout = 30 * time.Second + agentDialTimeout = 30 * time.Second + killWait = 5 * time.Second + stdinChunk = 32 * 1024 ) // Cmd describes a streaming Run. @@ -215,7 +216,9 @@ func (s *Sandbox) Close() error { s.pool.drain() ctx, cancel := context.WithTimeout(context.Background(), releaseTimeout) defer cancel() - found, err := s.releaseAt(ctx, s.Owner()) + first, cancelFirst := context.WithTimeout(ctx, releaseTimeout/2) + found, err := s.releaseAt(first, s.Owner()) + cancelFirst() if _, refused := errors.AsType[*APIError](err); refused || found { return err } @@ -304,13 +307,13 @@ func (s *Sandbox) rebind(ctx context.Context, err error) bool { return false } s.mu.Lock() - changed := moved.owner != s.owner + rebound := moved.owner != s.owner s.owner = moved.owner s.mu.Unlock() - if changed { + if rebound { s.pool.drain() } - return changed + return moved.owner != g.addr } // retryMoved runs rpc once more on the new owner when nothing of it reached the old one; rpc's request must be its whole input. @@ -350,7 +353,6 @@ func (s *Sandbox) call(ctx context.Context, req wire.Request) (*silkd.Conn, *lea return c.Conn, &lease{s: s, conn: c, stop: context.AfterFunc(ctx, func() { _ = c.Close() })}, nil } -// connect sends req on a parked connection or a fresh dial. func (s *Sandbox) connect(ctx context.Context, req wire.Request) (*agentConn, error) { if err := ctx.Err(); err != nil { return nil, err @@ -382,7 +384,6 @@ func markGone(err error, id, addr string) error { return err } -// pumpStdin chunks the reader into stdin frames. func pumpStdin(conn *silkd.Conn, r io.Reader) error { buf := make([]byte, stdinChunk) for { diff --git a/sdk/go/silkd/silkdtest/fake.go b/sdk/go/silkd/silkdtest/fake.go index 2d0fff1d6..156d715c3 100644 --- a/sdk/go/silkd/silkdtest/fake.go +++ b/sdk/go/silkd/silkdtest/fake.go @@ -22,7 +22,7 @@ import ( // Fake is a stateful silkd fake: fs verbs run on a real directory and sessions are tracked, so a write-then-read round-trips. type Fake struct { - Root string + root string mu sync.Mutex sessions map[string]bool @@ -33,7 +33,7 @@ type Fake struct { // NewFake returns a Fake serving files under root. func NewFake(root string) *Fake { - return &Fake{Root: root, sessions: map[string]bool{}, branches: []string{"main"}, current: "main"} + return &Fake{root: root, sessions: map[string]bool{}, branches: []string{"main"}, current: "main"} } // ServeConn serves RPCs on an already-open connection (e.g. after an HTTP hijack) until the peer closes it. @@ -102,7 +102,7 @@ func (f *Fake) fsWrite(conn net.Conn, r *bufio.Reader, req *wire.FsWrite) { func (f *Fake) fsRead(conn net.Conn, path string) { data, err := os.ReadFile(f.abs(path)) if err != nil { - errFrame(conn, wire.KindNotFound, err.Error()) + fsErr(conn, err) return } sendChunked(conn, data) @@ -111,7 +111,7 @@ func (f *Fake) fsRead(conn net.Conn, path string) { func (f *Fake) fsList(conn net.Conn, path string) { ents, err := os.ReadDir(f.abs(path)) if err != nil { - errFrame(conn, wire.KindNotFound, err.Error()) + fsErr(conn, err) return } var entries []wire.DirEntry @@ -134,7 +134,7 @@ func (f *Fake) fsList(conn net.Conn, path string) { func (f *Fake) fsStat(conn net.Conn, path string) { info, err := os.Stat(f.abs(path)) if err != nil { - errFrame(conn, wire.KindNotFound, err.Error()) + fsErr(conn, err) return } kind := wire.FileKindFile @@ -150,13 +150,17 @@ func (f *Fake) fsStat(conn net.Conn, path string) { func (f *Fake) fsRm(conn net.Conn, req *wire.FsRm) { if req.Recursive { + if _, err := os.Lstat(f.abs(req.Path)); err != nil { + fsErr(conn, err) + return + } f.done(conn, os.RemoveAll(f.abs(req.Path))) } else { f.done(conn, os.Remove(f.abs(req.Path))) } } -// fsPush extracts the uploaded tar under dest, close enough to silkd for SDK round trips. +// fsPush extracts only directories and regular files from the uploaded tar under dest. func (f *Fake) fsPush(conn net.Conn, r *bufio.Reader, dest string) { data, err := drainUpload(r) if err != nil { @@ -220,7 +224,7 @@ func (f *Fake) fsPull(conn net.Conn, path string) { return err }) if err != nil { - errFrame(conn, wire.KindNotFound, err.Error()) + fsErr(conn, err) return } _ = tw.Close() @@ -241,7 +245,7 @@ func (f *Fake) sessionCreate(conn net.Conn, req *wire.SessionCreate) { func (f *Fake) sessionList(conn net.Conn) { f.mu.Lock() - ids := slices.Collect(maps.Keys(f.sessions)) + ids := slices.Sorted(maps.Keys(f.sessions)) f.mu.Unlock() send(conn, &wire.Sessions{Sessions: ids}) } @@ -289,7 +293,7 @@ func (f *Fake) fsFind(conn net.Conn, req *wire.FsFind) { for i, line := range strings.Split(strings.TrimSuffix(string(body), "\n"), "\n") { if re.MatchString(line) { send(conn, &wire.Match{ - File: strings.TrimPrefix(path, f.Root), + File: strings.TrimPrefix(path, f.root), Line: uint64(i) + 1, Content: line, }) @@ -313,7 +317,7 @@ func (f *Fake) fsReplace(conn net.Conn, req *wire.FsReplace) { for _, file := range req.Files { body, err := os.ReadFile(f.abs(file)) if err != nil { - errFrame(conn, wire.KindNotFound, err.Error()) + fsErr(conn, err) return } count := len(re.FindAll(body, -1)) @@ -331,7 +335,7 @@ func (f *Fake) fsReplace(conn net.Conn, req *wire.FsReplace) { // fsWatch answers ready and one synthetic created event, then holds the stream until the client disconnects. func (f *Fake) fsWatch(conn net.Conn, r *bufio.Reader, req *wire.FsWatch) { if _, err := os.Stat(f.abs(req.Path)); err != nil { - errFrame(conn, wire.KindNotFound, err.Error()) + fsErr(conn, err) return } send(conn, wire.Ready{}) @@ -367,15 +371,15 @@ func (f *Fake) gitBranch(conn net.Conn, req *wire.GitBranch) { func (f *Fake) done(conn net.Conn, err error) { if err != nil { - errFrame(conn, wire.KindInternal, err.Error()) + fsErr(conn, err) return } send(conn, wire.Done{}) } -// abs keeps a request path inside Root; an absolute request path is rebased. +// abs keeps a request path inside root; an absolute request path is rebased. func (f *Fake) abs(path string) string { - return filepath.Join(f.Root, filepath.Clean("/"+path)) + return filepath.Join(f.root, filepath.Clean("/"+path)) } // portEcho refuses port 1 as not_found and echoes Data frames on any other port until DataEnd (Done) or disconnect. @@ -429,6 +433,14 @@ func globRegexp(glob string) *regexp.Regexp { return regexp.MustCompile("^" + pat + "$") } +func fsErr(conn net.Conn, err error) { + kind := wire.KindInternal + if errors.Is(err, fs.ErrNotExist) { + kind = wire.KindNotFound + } + errFrame(conn, kind, err.Error()) +} + func errFrame(conn net.Conn, kind, msg string) { send(conn, &wire.ErrorResp{Kind: kind, Message: msg}) } diff --git a/sdk/go/silkd/silkdtest/silkdtest.go b/sdk/go/silkd/silkdtest/silkdtest.go index 49667a9c0..f3d153168 100644 --- a/sdk/go/silkd/silkdtest/silkdtest.go +++ b/sdk/go/silkd/silkdtest/silkdtest.go @@ -12,13 +12,6 @@ import ( "github.com/cocoonstack/sandbox/protocol/wire" ) -type hijackedConn struct { - net.Conn - r *bufio.Reader -} - -func (c *hijackedConn) Read(p []byte) (int, error) { return c.r.Read(p) } - // Serve accepts connections until l closes, serving RPCs back to back on each like silkd. func Serve(l net.Listener) { acceptLoop(l, ServeConn) @@ -64,6 +57,13 @@ func ListenHybrid(sockPath string, port int) (io.Closer, error) { return l, nil } +type hijackedConn struct { + net.Conn + r *bufio.Reader +} + +func (c *hijackedConn) Read(p []byte) (int, error) { return c.r.Read(p) } + // serveConn answers info and dispatches the other requests until the peer closes; input frames outside an RPC are dropped as silkd drops them. func serveConn(conn net.Conn, r *bufio.Reader, serve func(net.Conn, *bufio.Reader, wire.Request)) { defer func() { _ = conn.Close() }() @@ -110,6 +110,9 @@ func serveExec(conn net.Conn, r *bufio.Reader, req *wire.Exec) { return } send(conn, &wire.Started{PID: 4242}) + if req.Detach { + return + } switch req.Argv[0] { case "echo": send(conn, &wire.Stdout{Data: []byte(strings.Join(req.Argv[1:], " ") + "\n")}) diff --git a/sdk/go/upgrade.go b/sdk/go/upgrade.go index 5252f5851..854121281 100644 --- a/sdk/go/upgrade.go +++ b/sdk/go/upgrade.go @@ -63,7 +63,7 @@ func (c *Client) dialAgent(ctx context.Context, addr, id, token string) (*upgrad return nil, fmt.Errorf("agent upgrade: id or token contains a control character") } scheme, authority, target := agentEndpoint(addr, c.scheme) - var d net.Dialer + d := net.Dialer{Timeout: agentDialTimeout} raw, err := d.DialContext(ctx, "tcp", target) if err != nil { return nil, markGone(err, id, addr) diff --git a/sdk/go/utils.go b/sdk/go/utils.go index 7009c6b7c..74166f076 100644 --- a/sdk/go/utils.go +++ b/sdk/go/utils.go @@ -225,7 +225,7 @@ func recv(ctx context.Context, conn *silkd.Conn) (wire.Response, error) { return nil, ctxErr } if errors.Is(err, io.EOF) { - return nil, fmt.Errorf("connection closed before a terminal frame") + return nil, fmt.Errorf("connection closed before a terminal frame: %w", io.ErrUnexpectedEOF) } return nil, err } diff --git a/sdk/langchain/cocoonsandbox_langchain/toolkit.py b/sdk/langchain/cocoonsandbox_langchain/toolkit.py index b4527bb48..e6fd1dfe7 100644 --- a/sdk/langchain/cocoonsandbox_langchain/toolkit.py +++ b/sdk/langchain/cocoonsandbox_langchain/toolkit.py @@ -106,6 +106,7 @@ def close(self) -> None: self._closed = True if sb is not None: sb.close() + self._client.close() def sandbox(self, deadline: float | None = None) -> Sandbox: """The claimed sandbox, claiming (or branching) on first use by deadline, else within CALL_TIMEOUT.""" diff --git a/sdk/python/cocoonsandbox/client.py b/sdk/python/cocoonsandbox/client.py index 469c0f37d..795865b1b 100644 --- a/sdk/python/cocoonsandbox/client.py +++ b/sdk/python/cocoonsandbox/client.py @@ -96,11 +96,12 @@ def probe(addr: str) -> Sandbox: ) return Sandbox(client=self, id=id, token=token, owner=reply.get("owner_addr") or addr) - addrs = [self.addr, *self._peers()] + peers, peers_err = self._peers() try: - return _scatter(addrs, probe) + return _scatter([self.addr, *peers], probe) except APIError: - raise APIError("lookup", 404, f"no owner found for {id}") from None + reason = f"no owner found for {id}" if peers_err is None else f"no owner found for {id}; {peers_err}" + raise APIError("lookup", 404, reason) from None def attach(self, owner_addr: str, id: str, token: str) -> Sandbox: """Binds a handle to a known owner without a lookup.""" @@ -176,16 +177,14 @@ def post(peer: str) -> dict[str, Any]: owner, reply = _redirect_fallback(addr, redirect, post, verb) return self._handle_from(owner, reply, verb) - def _peers(self) -> list[str]: + def _peers(self) -> tuple[list[str], APIError | None]: try: - return ( - self._request( - self.addr, "GET", "/v1/peers", None, "peers", timeout=min(_PEERS_TIMEOUT, self.timeout) - ).get("peers") - or [] + reply = self._request( + self.addr, "GET", "/v1/peers", None, "peers", timeout=min(_PEERS_TIMEOUT, self.timeout) ) - except APIError: - return [] + except APIError as exc: + return [], exc + return reply.get("peers") or [], None def _handle_from(self, dialed: str, reply: dict[str, Any], verb: str = "claim") -> Sandbox: return Sandbox( diff --git a/sdk/python/cocoonsandbox/conn.py b/sdk/python/cocoonsandbox/conn.py index 697d8cf2f..6ebfef63d 100644 --- a/sdk/python/cocoonsandbox/conn.py +++ b/sdk/python/cocoonsandbox/conn.py @@ -203,11 +203,6 @@ def dial_agent( except OSError as exc: err = _handshake_error(f"dial {addr}", exc) raise (_mark_gone(err, tag) if isinstance(exc, ConnectionRefusedError) else err) from exc - sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1) - sock.setsockopt(socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1) - for opt, setting in _TCP_KEEPALIVE.items(): - if hasattr(socket, opt): - sock.setsockopt(socket.IPPROTO_TCP, getattr(socket, opt), setting) request = ( f"GET /v1/sandboxes/{sandbox_id}/agent HTTP/1.1\r\n" f"Host: {endpoint.netloc}\r\n" @@ -218,6 +213,11 @@ def dial_agent( ).encode() reader = None try: + sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1) + sock.setsockopt(socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1) + for opt, setting in _TCP_KEEPALIVE.items(): + if hasattr(socket, opt): + sock.setsockopt(socket.IPPROTO_TCP, getattr(socket, opt), setting) if endpoint.scheme != "https": sock.settimeout(None) return Conn(sock, _buffered_reader(sock), request, timeout, tag) diff --git a/silkd/README.md b/silkd/README.md index 233029d43..ad168bda5 100644 --- a/silkd/README.md +++ b/silkd/README.md @@ -19,7 +19,7 @@ verdict that decides whether an exec inherits the proxy variables (`src/net.rs`). - `src/proto.rs` — frame types + caps; the golden corpus in - `../protocol/wire/fixtures/v1/` is round-tripped by the Rust, Go, and Python + `../protocol/wire/fixtures/v1/` is checked by the Rust, Go, and Python sides, so wire drift fails CI - `src/server.rs` — dispatch; one module per verb family - `tests/` — e2e suites driving the daemon in-process