diff --git a/.github/workflows/build-boot.yml b/.github/workflows/build-boot.yml index 9c6f8ca56..d8c5d4bef 100644 --- a/.github/workflows/build-boot.yml +++ b/.github/workflows/build-boot.yml @@ -116,7 +116,7 @@ jobs: id: ref # Same 7-char short SHA docker/metadata-action's type=sha emits, so # the ref matches the tag actually pushed below. - run: echo "image=ghcr.io/${{ github.repository }}/boot:$(cat boot/kernel/VERSION)-${GITHUB_SHA::7}" >> "$GITHUB_OUTPUT" + run: echo "image=ghcr.io/${{ github.repository }}/boot:${{ steps.ver.outputs.kernel }}-${GITHUB_SHA::7}" >> "$GITHUB_OUTPUT" - name: Download digests uses: actions/download-artifact@v8.0.2 @@ -149,7 +149,7 @@ jobs: working-directory: /tmp/digests run: | docker buildx imagetools create \ - $(jq -cr '.tags | map("-t " + .) | join(" ")' <<< '${{ steps.meta.outputs.json }}') \ + $(jq -cr '.tags | map("-t " + .) | join(" ")' <<< "$DOCKER_METADATA_OUTPUT_JSON") \ $(printf 'ghcr.io/${{ github.repository }}/boot@sha256:%s ' *) - name: Inspect diff --git a/.github/workflows/build-os-images.yml b/.github/workflows/build-os-images.yml index bddc30b7b..ae179d2d5 100644 --- a/.github/workflows/build-os-images.yml +++ b/.github/workflows/build-os-images.yml @@ -1,7 +1,7 @@ # Reusable: the images gating workflow calls this after the boot/silkd # carriers build, passing which changed and their sha-pinned refs. It detects # changed image dirs for the matrix, builds base then flavors, and pushes to -# ghcr; base gets its same-dir *.sh files as buildx secrets. +# ghcr; base gets install-agent.sh as a buildx secret. # - a changed carrier (boot or silkd) refolds into every image; otherwise # only the git-changed image dirs rebuild; # - a base change rebuilds every flavor (they are FROM base); @@ -87,7 +87,6 @@ jobs: if [[ "$CARRIER_CHANGED" == "true" || "$MANUAL" == "true" || -z "$BASE" ]] || git diff --name-only "$BASE" "$SHA" -- "$wf/build-os-images.yml" "$wf/images.yml" | grep -q .; then all_names - all_changed="" else all_changed=$(git diff --name-only "$BASE" "$SHA" -- 'os-image/' ':!*.md' | sort -u) declare -A names @@ -103,29 +102,13 @@ jobs: fi fi - # Build matrix entries with a per-entry no-cache flag. Cache must be - # dropped when any non-Dockerfile input changed: shared files - # directly under os-image/{name}/ (depth 3) or tag-specific files - # next to the Dockerfile (depth 4) — both are injected as buildx - # secrets, which never bust docker layer cache by content. + no_cache=false + if [[ "$MANUAL_NO_CACHE" == "true" ]]; then + no_cache=true + fi base_entries=() flavor_entries=() for name in "${!names[@]}"; do - needs_no_cache="false" - if [[ "$MANUAL_NO_CACHE" == "true" ]]; then - needs_no_cache="true" - elif [[ -n "$all_changed" ]]; then - while IFS= read -r f; do - [[ -z "$f" ]] && continue - depth=$(echo "$f" | awk -F'/' '{print NF}') - base_name=$(basename "$f") - if [[ "$depth" -eq 3 ]] || { [[ "$depth" -eq 4 ]] && [[ "$base_name" != "Dockerfile" ]]; }; then - needs_no_cache="true" - break - fi - done <<< "$(echo "$all_changed" | grep "^os-image/$name/")" - fi - platforms="linux/amd64,linux/arm64" if [[ -f "os-image/$name/platforms" ]]; then platforms=$(cat "os-image/$name/platforms") @@ -135,7 +118,7 @@ jobs: [[ -z "$dockerfile" ]] && continue tag=$(echo "$dockerfile" | cut -d'/' -f3) if [[ -n "$tag" ]]; then - entry="{\"name\":\"$name\",\"tag\":\"$tag\",\"no_cache\":\"$needs_no_cache\",\"platforms\":\"$platforms\"}" + entry="{\"name\":\"$name\",\"tag\":\"$tag\",\"no_cache\":\"$no_cache\",\"platforms\":\"$platforms\"}" if [[ "$name" == "base" ]]; then base_entries+=("$entry") else @@ -205,29 +188,6 @@ jobs: hash=$(find "os-image/${{ matrix.name }}" -type f ! -name Dockerfile ! -name '*.md' -print0 | sort -z | xargs -0 sha256sum | sha256sum | cut -d' ' -f1) echo "hash=$hash" >> "$GITHUB_OUTPUT" - - name: Discover secret files - id: secrets - run: | - base="os-image/${{ matrix.name }}" - tag="${{ matrix.tag }}" - { - echo "files<> "$GITHUB_OUTPUT" - - name: Build and push uses: docker/build-push-action@v7.4.0 with: @@ -245,7 +205,7 @@ jobs: INPUTS_HASH=${{ steps.inputs.outputs.hash }} tags: ${{ steps.meta.outputs.tags }} labels: ${{ steps.meta.outputs.labels }} - secret-files: ${{ steps.secrets.outputs.files }} + secret-files: sandbox_install_agent=os-image/base/install-agent.sh no-cache: ${{ matrix.no_cache == 'true' }} cache-from: type=gha,scope=${{ matrix.name }}-${{ matrix.tag }} cache-to: type=gha,mode=max,scope=${{ matrix.name }}-${{ matrix.tag }} diff --git a/.github/workflows/build-silkd.yml b/.github/workflows/build-silkd.yml index e3f9a7574..a4f6fe57f 100644 --- a/.github/workflows/build-silkd.yml +++ b/.github/workflows/build-silkd.yml @@ -142,7 +142,7 @@ jobs: working-directory: /tmp/digests run: | docker buildx imagetools create \ - $(jq -cr '.tags | map("-t " + .) | join(" ")' <<< '${{ steps.meta.outputs.json }}') \ + $(jq -cr '.tags | map("-t " + .) | join(" ")' <<< "$DOCKER_METADATA_OUTPUT_JSON") \ $(printf 'ghcr.io/${{ github.repository }}/silkd@sha256:%s ' *) - name: Inspect diff --git a/.github/workflows/silkd.yml b/.github/workflows/silkd.yml index be5575cce..427e79949 100644 --- a/.github/workflows/silkd.yml +++ b/.github/workflows/silkd.yml @@ -36,4 +36,4 @@ jobs: - name: root-only tests working-directory: silkd - run: sudo -E env "PATH=$PATH" "HOME=$HOME" cargo test --locked --test exec_e2e --test git_e2e --test fs_e2e --test find_e2e --test tun_e2e -- --ignored --skip throughput + run: sudo -E env "PATH=$PATH" "HOME=$HOME" cargo test --locked --test '*' -- --ignored --skip throughput diff --git a/README.md b/README.md index ab913bc52..30900c689 100644 --- a/README.md +++ b/README.md @@ -99,15 +99,14 @@ sandboxd](docs/deploy.md) covers release binaries, the full config and systemd. `boot/**`, `silkd/**`, or `os-image/**` it builds the changed carriers (via `build-boot.yml` / `build-silkd.yml`, `workflow_call`) then os-images, in order, so the chain is deterministic -- `build-os-images.yml` — bakes base + flavors FROM the sha-pinned carriers +- `build-os-images.yml` — bakes base + flavors FROM the carriers, sha-pinned + when rebuilt in the same run - `release.yml` — on a version tag, checks the tag against silkd's version, then builds the release binaries and archives - `publish-pypi.yml` — on an `sdk-*-v*` tag, builds and publishes the matching package via PyPI Trusted Publishing (OIDC, per-package environment) -On a fresh repo run build-boot first — images build `FROM` the boot artifact. - ## Boot chain ``` diff --git a/boot/Dockerfile b/boot/Dockerfile index c6db6e393..b711310a8 100644 --- a/boot/Dockerfile +++ b/boot/Dockerfile @@ -10,17 +10,10 @@ RUN apt-get update && apt-get install -y --no-install-recommends \ xz-utils curl ca-certificates python3 \ && rm -rf /var/lib/apt/lists/* ARG KERNEL_VERSION -# Mirror chain: KERNEL_MIRROR (cdn.kernel.org by default), then kernel.org's -# edge mirror and an rsync mirror. Download to a file first — a mid-stream -# failure must not leave a half-extracted tree — and verify against the pinned SHA256 -# (kernel/SHA256, updated together with VERSION): the fallback mirrors are -# third parties and this kernel boots untrusted-code sandboxes. ARG KERNEL_MIRROR=https://cdn.kernel.org/pub/linux/kernel WORKDIR /linux COPY kernel/SHA256 /tmp/SHA256 -# The hash is taken from the exact line naming this kernel version, so a -# multi-line SHA256 file (e.g. keeping the previous kernel's entry during a -# bump) can never pair the tarball with the wrong hash. +# Mirrors are third parties, so the tarball is checked against this version's kernel/SHA256 line before extraction. RUN set -e; \ hash=$(awk -v f="linux-${KERNEL_VERSION}.tar.xz" '$2==f{print $1; exit}' /tmp/SHA256); \ [ -n "$hash" ] || { echo "kernel/SHA256 has no entry for linux-${KERNEL_VERSION}.tar.xz"; exit 1; }; \ @@ -36,10 +29,7 @@ RUN set -e; \ ARG TARGETARCH COPY kernel/sandbox.config /tmp/sandbox.config COPY kernel/sandbox-arm64.config /tmp/sandbox-arm64.config -# merge_config only warns and olddefconfig silently drops renamed or -# dependency-orphaned symbols — assert the boot- and container-critical set -# so a kernel bump cannot lose one with a green build. Boot entry and console -# are per-arch: amd64 = PVH, arm64 = direct Image boot + PL031 RTC. +# olddefconfig silently drops renamed or orphaned symbols, so assert the boot- and container-critical set. RUN set -e; \ case "$TARGETARCH" in \ amd64) make x86_64_defconfig kvm_guest.config; \ diff --git a/boot/kernel/sandbox.config b/boot/kernel/sandbox.config index bf99bee11..bc2cb3908 100644 --- a/boot/kernel/sandbox.config +++ b/boot/kernel/sandbox.config @@ -1,5 +1,5 @@ -# sandbox guest kernel fragment, applied over `x86_64_defconfig` + -# `kvm_guest.config` (see boot/Dockerfile). Everything the boot path touches +# sandbox guest kernel fragment, applied over the arch defconfig (plus +# `kvm_guest.config` on x86_64; see boot/Dockerfile). Everything the boot path touches # is built in (=y): no modules to load, no initramfs modprobe, no udev settle. ## No module support at all — the strongest "nothing to wait for" guarantee. @@ -196,6 +196,5 @@ CONFIG_TRANSPARENT_HUGEPAGE_MADVISE=y # CONFIG_KEXEC is not set # CONFIG_FTRACE is not set CONFIG_DEBUG_INFO_NONE=y -# Boot-time W+X page-table scan: debug-only cost on every boot (it also -# flagged 6 W+X pages on the PVH path — tracked separately). +# Boot-time W+X page-table scan: debug-only cost on every boot. # CONFIG_DEBUG_WX is not set diff --git a/docs/cluster.md b/docs/cluster.md index 76242b35c..4618ca017 100644 --- a/docs/cluster.md +++ b/docs/cluster.md @@ -312,7 +312,9 @@ release and adoption an evacuation uses: `disk_reserve`, under `memory_limit` and under `max_running`. - When a peer adopts, the owner forgets the claim and the wake answers 404, as after an evacuation. The client's `Lookup` finds the peer, which wakes the - claim there. The hand-off counts in `sandboxd_wake_handoffs_total`. + claim there. The hand-off counts in `sandboxd_wake_handoffs_total`. A preview + URL minted before the hand-off stops working until it is minted again, as + after an evacuation. - When no peer adopts (no peer, or every peer full or closed), the owner takes the row back and the wake answers 429 as before. A node runs one hand-off at a time, and after one fails it asks no peer for 30 seconds, so clients diff --git a/docs/deploy.md b/docs/deploy.md index dad09ed26..d8e7d3d82 100644 --- a/docs/deploy.md +++ b/docs/deploy.md @@ -656,7 +656,7 @@ golden fail the same way, it is retired again at most once every 30 s. The systemd unit is [`packaging/sandboxd.service`](https://github.com/cocoonstack/sandbox/blob/main/packaging/sandboxd.service); -install it as shipped. Its `KillMode` and `TimeoutStopSec` are explained below. +install it as shipped. The unit runs `/usr/local/bin/sandboxd -config /etc/sandboxd/config.json`, and its `KillMode` and `TimeoutStopSec` are explained below. `LimitCORE=0` keeps a crash from writing sandboxd's memory, which holds every running claim's host-only env values, to a core file. For the same reason, run sandboxd hosts with swap off or encrypted. @@ -1148,7 +1148,7 @@ all at once or not at all. Every field falls into one of three classes: - per pool: `egress`, `warmup`, `capture_trim`, `storage`, `reader`, `egress_upstream_env`, `tun`, for existing keys and for keys new in the file; - `tun`: `exits`, `mtu`, `port` and `limits`; its `subnet`, `listen`, `vni`, `senders` and `reuse_after_seconds`, and adding or removing the block, need a restart; - `egress_classes`: each class's `egress` and `egress_upstream_env`; - - node-wide: `egress_internal_allow`, `secrets`, `egress_upstream`, `egress_usage_bytes`, `egress_usage_interval_seconds`, `egress_conn_rate`, `disk_reserve`, `memory_limit`, `max_claims`, `max_running`, `move_concurrency`, `max_fork_count`, `release_delay_seconds`, `vmm_restart`, `sync_claims`, `checkpoint_ttl_hours`, and the unpooled defaults `idle_hibernate_seconds`, `archive_after_seconds` and `archive_delete_after_seconds`. + - node-wide: `egress_internal_allow`, `secrets`, `egress_upstream`, `egress_usage_bytes`, `egress_usage_interval_seconds`, `egress_conn_rate`, `disk_reserve`, `memory_limit`, `balloon`, `max_claims`, `max_running`, `move_concurrency`, `max_fork_count`, `release_delay_seconds`, `vmm_restart`, `sync_claims`, `checkpoint_ttl_hours`, and the unpooled defaults `idle_hibernate_seconds`, `archive_after_seconds` and `archive_delete_after_seconds`. - **Left to their API**: a pool's targets (`warm`, `warm_max`, idle and archive durations) belong to `PUT /v1/pools`. A reload never changes them and lists them as `ignored`. A pool key new in the file gets its settings at once and stays diff --git a/docs/performance.md b/docs/performance.md index a6234549f..45f79b485 100644 --- a/docs/performance.md +++ b/docs/performance.md @@ -68,7 +68,7 @@ keep an idle relay for 30 seconds and send later RPCs on it; use mode C of ## Relay connection reuse Every data-plane RPC used to pay a TCP dial, an HTTP Upgrade and, behind a -TLS edge, a handshake. From proto 2 silkd serves RPCs back to back, and both +TLS edge, a handshake. silkd now serves RPCs back to back on one connection, and both SDKs park up to 8 idle relay connections per handle for a 30s idle window. `e2e/cmd/rpcbench`, warm `rt:24.04` sandbox, `net=none`, n=200 `fs_stat` diff --git a/docs/sandboxd-api.md b/docs/sandboxd-api.md index e48179979..fd671d813 100644 --- a/docs/sandboxd-api.md +++ b/docs/sandboxd-api.md @@ -120,8 +120,8 @@ or the operator token, which claims as root but never with `volumes` (403). 0, and the next call that reaches the guest restores it with a fresh lease of the claim's length, so its next expiry archives it again; a call that wakes it between the hibernate and the export grants that fresh lease too. - A failed hibernate or export at expiry keeps the claim and retries on every - reap tick (5 s); it never falls back to destroy. A failed claim (`failed` set, see + A failed hibernate or export at expiry keeps the claim and retries it a + minute later; it never falls back to destroy. A failed claim (`failed` set, see [dead VMMs](deploy.md#dead-vmms-and-host-restarts)) is destroyed at expiry even under `archive`. A volume claim or an egress-lane claim cannot hibernate, so `archive` on one answers 409 with the same error as a @@ -962,7 +962,7 @@ has no `preview_listen`. The signed token embeds the sandbox id, port, and owner `advertise_addr`, so any node's preview listener can serve it (forwarding to the owner's main listener) and a released sandbox's URL answers 404 — no revocation list. A browser request to the minted URL answers 503 -`lease lost; retry` for a replicated claim while its node's lease is overdue, 503 `sandbox moving to another node; retry` while an [evacuation](#post-v1drain) has yet to move it, and 429 when it has to wake a sandbox while the disk reserve is low or the node is at `max_running`. See [deploy](deploy.md#preview-urls). +`lease lost; retry` for a replicated claim while its node's lease is overdue, 503 `sandbox moving to another node; retry` while an [evacuation](#post-v1drain) has yet to move it, and 429 when it has to wake a sandbox while the disk reserve is low or the node is at `max_running`; a wake that [hands an archived claim to a peer](cluster.md#waking-an-archived-claim-on-a-full-node) answers 404, as after a move. See [deploy](deploy.md#preview-urls). ## POST /v1/sandboxes/{id}/checkpoint @@ -1273,8 +1273,8 @@ archive/unarchive/archive-delete counters, `sandboxd_dns_malformed_total` (guest for average latency, `sandboxd_claim_duration_seconds{tier}`, a histogram of pool claims by tier, `sandboxd_claim_redirects_total`, claims this node handed to a peer, `sandboxd_wakes_refused_total{reason}`, wakes `max_running` (`max_running`), -`disk_reserve` (`disk`) refused with 429, and wakes `memory_limit` (`memory`) -refused here whether or not a hand-off ran, `sandboxd_wake_handoffs_total`, +`disk_reserve` (`disk`) and `memory_limit` (`memory`) +refused here, whether or not a hand-off ran, `sandboxd_wake_handoffs_total`, archived claims this node [handed to a peer with room](cluster.md#waking-an-archived-claim-on-a-full-node) instead, and `sandboxd_wake_memory_fallbacks_total`, archived wakes over `memory_limit` this node woke itself because no hand-off ran or no peer adopted. `sandboxd_balloon_inflations_total`, `sandboxd_balloon_deflations_total` and `sandboxd_balloon_failures_total` count the balloon sweep's changes and the ones that failed, which retry a minute later. On a node with a `tun` block it adds the packet path's counters: diff --git a/docs/sdk.md b/docs/sdk.md index 9df07129b..3a1c6656d 100644 --- a/docs/sdk.md +++ b/docs/sdk.md @@ -347,7 +347,7 @@ out, err := sb.Exec(ctx, "cat", "/tmp/state") // sessions & memory intact guest does can fall between the snapshot point and the stop. The handle stays valid: the first call that reaches the guest restores the VM (adding roughly a restore's latency, tens of milliseconds on bare metal). On a node at -[`max_running`](deploy.md#configuration) that call returns a 429 `*APIError`, +[`max_running`](deploy.md#configuration), or below `disk_reserve`, that call returns a 429 `*APIError`, which the SDK does not retry, and the sandbox stays hibernated. The TTL keeps running — a hibernated sandbox is still reaped at its deadline, so claim with a `WithTimeout` that covers the idle period or `Renew` before it @@ -705,7 +705,7 @@ sb = client.Attach(ownerAddr, id, token) // bind a known handle, no lookup roun err = client.PutTenant(ctx, sandbox.TenantSpec{Name: "u:42", Token: tok, MaxClaims: 10, EgressClass: "desk"}) err = client.DeleteTenant(ctx, "u:42") -res, _ = client.PutTenantCluster(ctx, sandbox.TenantSpec{Name: "u:42", MaxClaims: 20}) // token kept +res, _ = client.PutTenantCluster(ctx, sandbox.TenantSpec{Name: "u:42", MaxClaims: 20, EgressClass: "desk"}) // token kept res, _ = client.DeleteTenantCluster(ctx, "u:42") page, _ := client.Tenants(ctx, "", 0) // page.Tenants: names, caps, live claims, removed tenants; never tokens; page.Next pages page, _ = client.SetTenants(ctx, specs) // replace the whole set diff --git a/docs/silkd.md b/docs/silkd.md index fc5dac33c..40b101e91 100644 --- a/docs/silkd.md +++ b/docs/silkd.md @@ -61,7 +61,7 @@ cannot parse fails CI. | pty | `pty_open {cols, rows, cwd?, env?, user?}` / `pty_resize {pid, cols, rows}` | `pty_open` answers `started{pid}`, then a shell under a pseudo-terminal: output as `stdout` frames, input as `stdin` frames, `exit{code}` terminal; `pty_resize` answers `done`. PTYs register in the proc table like any exec. A client that disconnects hangs the shell up (SIGHUP, so it hangs up its jobs, as a closed ssh session does), and a shell still alive 2 s later is killed | | git | `git_clone {url, path, branch?, depth?, auth?}` / `git_status {path}` / `git_add {path, files}` / `git_commit {path, message, author}` / `git_push {path, auth?}` / `git_pull {path, auth?}` / `git_branch {path, action, name?}` | `git_status_result` (porcelain v2), `git_commit_result{hash}`, `git_branches{current, branches}` for `git_branch {action: "list"}`; every other verb terminates with `done`. A status past about 1 MiB of entries carries `truncated: true` with the head of the list, so one frame never exceeds the cap. `auth` is injected as an in-memory header, never written to guest disk | | port | `port_forward {port}` | relays guest TCP 127.0.0.1:port over this connection: `ready` once connected, then `data` both ways. Each direction ends on its own: `data_end` half-closes the guest socket, `done` reports the guest closing its write side, and the RPC ends once both have. A client that only drops the connection leaves the relay open until the guest side closes or writes. Works on both lanes — the no-network lane's only way in | -| lsp | `lsp_start {language, root?}` / `lsp_request {server_id}` / `lsp_stop {server_id}` | a broker for the language server a flavor image ships: `lsp_start` spawns the argv named in `/etc/silkd/lsp.d/` (absent on the base image → `not_found` naming the flavor) and answers `lsp_started{server_id}`; `lsp_request` attaches the JSON-RPC byte stream (`ready`, then `data` both ways — silkd pipes bytes, it never parses LSP; `data_end` half-closes the server's stdin) and the stream ending reaps the server (v1 single-shot); `lsp_stop` kills it early, and a server nothing attaches within 5 minutes is reaped | +| lsp | `lsp_start {language, root?}` / `lsp_request {server_id}` / `lsp_stop {server_id}` | a broker for the language server a flavor image ships: `lsp_start` spawns the argv named in `/etc/silkd/lsp.d/` (absent on the base image → `not_found` naming the manifest path a flavor provides) and answers `lsp_started{server_id}`; `lsp_request` attaches the JSON-RPC byte stream (`ready`, then `data` both ways — silkd pipes bytes, it never parses LSP; `data_end` half-closes the server's stdin) and the stream ending reaps the server (v1 single-shot); `lsp_stop` kills it early, and a server nothing attaches within 5 minutes is reaped | | tun | `tun_up` / `tun_down` | the [advanced network mode](advanced-network.md) switch: `tun_up` creates `tun0` and starts its packet stream on vsock 2051, `tun_down` removes `tun0` and its routes; both answer `done`, `tun_up` an `error` when `tun0` cannot be created. Advisory: the host enforces | | misc | `info` | `info{version, uptime_secs, procs, sessions}` — sandboxd uses it for readiness; there is no public SDK method. Distinct from the control plane's `GET /v1/info`, which reports node pools and claims | @@ -71,7 +71,7 @@ A failed verb terminates with `error {kind, message}`: | kind | meaning | |---|---| -| `bad_request` | malformed frame, empty argv, invalid pattern, an exec argv or cwd the guest cannot spawn (missing, not a directory, not executable), an exec or pty `user` the guest cannot resolve (`unknown user`, or `look up user: ` when the name service itself fails) | +| `bad_request` | malformed frame, empty argv, invalid pattern, an exec argv or cwd the guest cannot spawn (missing, not a directory, not executable), an exec or pty `user` the guest cannot resolve (`unknown user ""`, or `look up user "": ` when the name service itself fails) | | `not_found` | unknown pid/session/path, a `port_forward` port nothing listens on, a language with no LSP manifest | | `unimplemented` | reserved in the protocol; no verb answers it today | | `internal` | everything else (other spawn failures, git errors, I/O) | diff --git a/e2e/reader_test.go b/e2e/reader_test.go index df447fc98..ad8a2da46 100644 --- a/e2e/reader_test.go +++ b/e2e/reader_test.go @@ -26,7 +26,7 @@ func TestReaderServesAHibernatedClaimUntilAWriteWakesIt(t *testing.T) { if err != nil { t.Fatalf("New: %v", err) } - defer func() { _ = sb.Close() }() + defer sb.Close() if err = sb.Hibernate(t.Context()); err != nil { t.Fatalf("Hibernate: %v", err) } diff --git a/mcp/server_test.go b/mcp/server_test.go index 8febe938d..729f51a22 100644 --- a/mcp/server_test.go +++ b/mcp/server_test.go @@ -122,10 +122,7 @@ func TestExecKeepsOutputWhenTheGuestDrops(t *testing.T) { func TestReadFileStopsAtTheCap(t *testing.T) { chunk := `{"type":"data","data":"` + base64.StdEncoding.EncodeToString(make([]byte, 256<<10)) + `"}` + "\n" - srv := newTestServer(t, agentRoute(t, func(req string) (string, bool) { - if strings.Contains(req, `"op":"fs_stat"`) { - return `{"type":"stat","info":{"kind":"file","size":0}}` + "\n", false - } + srv := newTestServer(t, agentRoute(t, func(string) (string, bool) { return strings.Repeat(chunk, 8) + `{"type":"done"}` + "\n", false })) replies := serveLines(t, srv, @@ -154,10 +151,7 @@ func TestLogsKeepsAtMostTheCap(t *testing.T) { func TestToolRepliesReplaceInvalidUTF8(t *testing.T) { lossy := base64.StdEncoding.EncodeToString([]byte("a\xffb")) srv := newTestServer(t, agentRoute(t, func(req string) (string, bool) { - switch { - case strings.Contains(req, `"op":"fs_stat"`): - return `{"type":"stat","info":{"kind":"file","size":3}}` + "\n", false - case strings.Contains(req, `"op":"logs"`): + if strings.Contains(req, `"op":"logs"`) { return `{"type":"stdout","data":"` + lossy + `"}` + "\n" + `{"type":"done"}` + "\n", false } return `{"type":"data","data":"` + lossy + `"}` + "\n" + `{"type":"done"}` + "\n", false diff --git a/os-image/base/install-agent.sh b/os-image/base/install-agent.sh index 6fc34e407..37f79340c 100755 --- a/os-image/base/install-agent.sh +++ b/os-image/base/install-agent.sh @@ -1,11 +1,6 @@ #!/bin/sh -# Bake cocoon-agent (vsock exec daemon) into a sandbox image. -# Mirrors cocoon/os-image/ubuntu/install-agent.sh with two deltas: -# - the unit starts at sysinit (DefaultDependencies=no): the agent listening -# on vsock is cocoon's VM readiness signal, so it must not wait for -# network or multi-user; the sandbox kernel builds vsock in (no modprobe). -# - sshd stays socket-activated (Dockerfile enables ssh.socket), so there is -# no `systemctl enable ssh` here. +# Bake cocoon-agent into a sandbox image, as cocoon/os-image/ubuntu/install-agent.sh does. +# Differences: the unit starts at sysinit (its vsock listener is cocoon's readiness signal), and sshd stays socket-activated. set -eu AGENT_VERSION="${COCOON_AGENT_VERSION:-0.2.5}" @@ -35,10 +30,7 @@ cat > /etc/systemd/system/cocoon-agent.service <<'EOF' [Unit] Description=Cocoon agent (vsock command exec) Documentation=https://github.com/cocoonstack/cocoon-agent -# cocoon's VM readiness signal: start as early as systemd allows. -# No After= at all: sandbox-init hands over a fully-assembled rootfs before -# PID 1 runs and /dev/vsock is devtmpfs — measured, waiting for -# local-fs.target cost ~140ms of pure ordering delay on the critical chain. +# No After=: the rootfs is whole before PID 1 and /dev/vsock is devtmpfs; waiting for local-fs.target cost ~140ms. DefaultDependencies=no Before=basic.target diff --git a/protocol/README.md b/protocol/README.md index 2c7fc756b..81ea74f18 100644 --- a/protocol/README.md +++ b/protocol/README.md @@ -11,8 +11,9 @@ implementations of the silkd protocol: `req_*.json` are client→server frames, `resp_*.json` server→client. A frame that the Go side cannot round-trip, or that the Rust side cannot parse (a -request) or round-trip (a response), is a protocol drift bug. The Rust side also -pins the corpus size, so adding a verb without its fixture fails CI. +request) or round-trip (a response), is a protocol drift bug. The Go and Rust +tests pin the corpus size, and the Go test checks that every verb it decodes has +a fixture, so adding a verb without its fixture fails CI. `enums.json` pins each wire enum's full value set (error/event/file kinds, git branch actions). Frame fixtures carry only one representative value per diff --git a/sandboxd/egress/proxy_test.go b/sandboxd/egress/proxy_test.go index f0c86f26b..5ac6d65bc 100644 --- a/sandboxd/egress/proxy_test.go +++ b/sandboxd/egress/proxy_test.go @@ -707,7 +707,7 @@ func TestAChunkedResponseEndsCleanlyThroughTheProxy(t *testing.T) { } } -func (d DialFunc) Route() string { return "" } +func (DialFunc) Route() string { return "" } func (d DialFunc) Dial(ctx context.Context, _, network, addr string) (net.Conn, error) { return d(ctx, network, addr) @@ -840,10 +840,7 @@ func readStatus(t *testing.T, br *bufio.Reader) string { break } } - if len(line) >= 2 { - line = line[:len(line)-2] - } - return line + return strings.TrimSuffix(line, "\r\n") } func echoServer(t *testing.T) string { diff --git a/sandboxd/egress/socks_test.go b/sandboxd/egress/socks_test.go index 2d5116a3f..4b44c00a6 100644 --- a/sandboxd/egress/socks_test.go +++ b/sandboxd/egress/socks_test.go @@ -115,9 +115,9 @@ func TestSocksRefusesWhatItCannotServe(t *testing.T) { } func TestSocksDeniesAndRecordsAZonedOrNonASCIIName(t *testing.T) { - for _, name := range []string{"2606:4700:4700::1111%.example.com", "x.exampl\u0130.com"} { + for _, name := range []string{"2606:4700:4700::1111%.example.com", "x.ap\u0130.internal"} { events := make(chan Event, 4) - _, addr := socksProxy(t, Policy{Allow: []Rule{{Host: "*.example.com"}}}, nil, fixedDial(echoServer(t)), events, nil) + _, addr := socksProxy(t, Policy{Allow: []Rule{{Host: "*.example.com"}, {Host: "*.api.internal"}}}, nil, fixedDial(echoServer(t)), events, nil) if _, reply := socksExchange(t, addr, socksGreeting, socksNameRequest(name, 443)); reply[1] != socksDenied { t.Errorf("reply code for %q = %#x, want denied", name, reply[1]) } diff --git a/sandboxd/egress/splice_linux_test.go b/sandboxd/egress/splice_linux_test.go index d32f73084..0b4340adf 100644 --- a/sandboxd/egress/splice_linux_test.go +++ b/sandboxd/egress/splice_linux_test.go @@ -1,5 +1,3 @@ -//go:build linux - package egress import ( diff --git a/sandboxd/engine/engine.go b/sandboxd/engine/engine.go index a479dd4d8..6197e6d63 100644 --- a/sandboxd/engine/engine.go +++ b/sandboxd/engine/engine.go @@ -1,4 +1,4 @@ -// Package engine drives VM lifecycle through the cocoon CLI, its only stable contract, and dials the in-guest silkd over hybrid vsock. +// Package engine drives VM lifecycle through the cocoon CLI, follows the cocoon daemon's event stream, and dials the in-guest silkd over hybrid vsock. package engine import ( diff --git a/sandboxd/engine/fakesilkd_test.go b/sandboxd/engine/fakesilkd_test.go index ad59fdccd..4b9d33fcf 100644 --- a/sandboxd/engine/fakesilkd_test.go +++ b/sandboxd/engine/fakesilkd_test.go @@ -123,11 +123,6 @@ func (f *fakeSilkd) serve(conn net.Conn) { if json.Unmarshal([]byte(line), &req) != nil { return } - if req.Op == "exec" { - f.mu.Lock() - f.execDetach = append(f.execDetach, req.Detach) - f.mu.Unlock() - } switch req.Op { case "info": writeFakeSilkdResponse(conn, &wire.InfoResp{Version: "test"}) @@ -140,7 +135,7 @@ func (f *fakeSilkd) serve(conn net.Conn) { case "fs_stat": f.handleStat(conn, req.Path) case "exec": - f.handleExec(conn, req.Argv, req.Env) + f.handleExec(conn, req.Argv, req.Env, req.Detach) case "tun_up", "tun_down": f.handleTun(conn, req.Op) } @@ -176,8 +171,9 @@ func (f *fakeSilkd) handleWrite(conn net.Conn, r *bufio.Reader, path string, mod writeFakeSilkdResponse(conn, &wire.Done{}) } -func (f *fakeSilkd) handleExec(conn net.Conn, argv []string, env map[string]string) { +func (f *fakeSilkd) handleExec(conn net.Conn, argv []string, env map[string]string, detach bool) { f.mu.Lock() + f.execDetach = append(f.execDetach, detach) f.execArgv = argv f.execCalls = append(f.execCalls, argv) f.execEnv = env diff --git a/sandboxd/engine/silkd.go b/sandboxd/engine/silkd.go index 12e4b26b1..dc64bc136 100644 --- a/sandboxd/engine/silkd.go +++ b/sandboxd/engine/silkd.go @@ -160,7 +160,7 @@ func (e *Engine) silkdWriteFile(ctx context.Context, vsockSocket, path string, m defer s.close() // A send racing the guest's early answer+close hits EPIPE; prefer the buffered verdict. sendErr := func() error { - if serr := s.send(wire.FsWrite{Path: path, Mode: &mode}); serr != nil { + if serr := s.send(wire.FsWrite{Path: path, Mode: new(mode)}); serr != nil { return serr } for chunk := range slices.Chunk(data, wire.BulkChunk) { diff --git a/sandboxd/engine/volume_test.go b/sandboxd/engine/volume_test.go index ee40b25d2..78bb42ebe 100644 --- a/sandboxd/engine/volume_test.go +++ b/sandboxd/engine/volume_test.go @@ -185,11 +185,13 @@ func TestUmountVolumeExecsUmount(t *testing.T) { func TestUmountVolumeWaitsOutAFlushPastTheCallTimeout(t *testing.T) { path := sockPath(t) fake := serveFakeSilkd(t, path) - fake.execHook = func(argv []string) { - if argv[0] == "umount" { - time.Sleep(syncTimeout + 250*time.Millisecond) + fake.set(func(f *fakeSilkd) { + f.execHook = func(argv []string) { + if argv[0] == "umount" { + time.Sleep(syncTimeout + 250*time.Millisecond) + } } - } + }) if err := newTestEngine().umountVolume(t.Context(), path, "/data"); err != nil { t.Errorf("umount whose flush outlasts %s: %v", syncTimeout, err) } diff --git a/sandboxd/outbound/claim_test.go b/sandboxd/outbound/claim_test.go index 98eb7fa27..534966d16 100644 --- a/sandboxd/outbound/claim_test.go +++ b/sandboxd/outbound/claim_test.go @@ -422,27 +422,6 @@ type tunnel struct { seen []string } -func tunnelProxy(t *testing.T) *tunnel { - t.Helper() - ln, err := net.Listen("tcp", "127.0.0.1:0") - if err != nil { - t.Fatalf("listen: %v", err) - } - t.Cleanup(func() { _ = ln.Close() }) - p := &tunnel{addr: ln.Addr().String()} - go func() { - for { - conn, err := ln.Accept() - if err != nil { - return - } - p.accepted.Add(1) - go p.serve(conn) - } - }() - return p -} - func (p *tunnel) targets() []string { p.mu.Lock() defer p.mu.Unlock() @@ -470,6 +449,27 @@ func (p *tunnel) serve(conn net.Conn) { _, _ = io.Copy(conn, target) } +func tunnelProxy(t *testing.T) *tunnel { + t.Helper() + ln, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("listen: %v", err) + } + t.Cleanup(func() { _ = ln.Close() }) + p := &tunnel{addr: ln.Addr().String()} + go func() { + for { + conn, err := ln.Accept() + if err != nil { + return + } + p.accepted.Add(1) + go p.serve(conn) + } + }() + return p +} + func pipeDial(n *atomic.Int32) egress.DialFunc { return func(context.Context, string, string) (net.Conn, error) { n.Add(1) diff --git a/sandboxd/outbound/doors_test.go b/sandboxd/outbound/doors_test.go index 233cbbb91..c3ecdbdd7 100644 --- a/sandboxd/outbound/doors_test.go +++ b/sandboxd/outbound/doors_test.go @@ -105,7 +105,7 @@ func TestEgressDoorCapsConcurrentConnections(t *testing.T) { client.Timeout = 500 * time.Millisecond resp, err := client.Get(origin.URL + "/") if err == nil { - resp.Body.Close() + _ = resp.Body.Close() t.Fatal("a request past the door cap was served") } if netErr, ok := errors.AsType[net.Error](err); !ok || !netErr.Timeout() { @@ -118,7 +118,7 @@ func TestEgressDoorCapsConcurrentConnections(t *testing.T) { if err != nil { t.Fatalf("request after a slot freed: %v", err) } - resp.Body.Close() + _ = resp.Body.Close() if resp.StatusCode != http.StatusOK { t.Errorf("status %d after a slot freed, want 200", resp.StatusCode) } diff --git a/sandboxd/outbound/outbound_test.go b/sandboxd/outbound/outbound_test.go index f1c6db604..2dcfa7470 100644 --- a/sandboxd/outbound/outbound_test.go +++ b/sandboxd/outbound/outbound_test.go @@ -49,17 +49,20 @@ func TestEgressProxyInjectsAndGates(t *testing.T) { t.Fatalf("allowed request: %v", err) } body, _ := io.ReadAll(resp.Body) - resp.Body.Close() + _ = resp.Body.Close() if string(body) != "ok" || resp.Header.Get("X-Seen-Auth") != "s3cr3t" { t.Errorf("allowed request body=%q injected=%q, want ok/s3cr3t", body, resp.Header.Get("X-Seen-Auth")) } - req, _ := http.NewRequest(http.MethodGet, "http://blocked.example/", nil) + req, err := http.NewRequestWithContext(t.Context(), http.MethodGet, "http://blocked.example/", nil) + if err != nil { + t.Fatalf("build denied request: %v", err) + } deny, err := client.Do(req) if err != nil { t.Fatalf("denied request: %v", err) } - deny.Body.Close() + _ = deny.Body.Close() if deny.StatusCode != http.StatusForbidden { t.Errorf("denied status %d, want 403", deny.StatusCode) } diff --git a/sandboxd/pool/archive.go b/sandboxd/pool/archive.go index 06f997dfd..5c883f45f 100644 --- a/sandboxd/pool/archive.go +++ b/sandboxd/pool/archive.go @@ -210,7 +210,7 @@ func (m *Manager) wakeArchived(ctx context.Context, sb *types.Sandbox) (string, if errors.Is(err, ErrQuota) && m.handOffForRoom(ctx, sb) { return "", fmt.Errorf("wake %s: %w", sb.ID, ErrUnknownSandbox) } - // memory only spreads load, so a wake no hand-off placed runs here + // Memory only spreads load, so a wake no hand-off placed runs here if !errors.Is(err, errMemoryLow) { return "", fmt.Errorf("wake %s: %w", sb.ID, err) } diff --git a/sandboxd/pool/archive_test.go b/sandboxd/pool/archive_test.go index 069971025..64a2b8d9f 100644 --- a/sandboxd/pool/archive_test.go +++ b/sandboxd/pool/archive_test.go @@ -6,6 +6,7 @@ import ( "fmt" "os" "path/filepath" + "runtime" "slices" "strings" "sync" @@ -25,7 +26,7 @@ const ( func TestArchivePublishWindowPinsCheckpoint(t *testing.T) { eng := newFakeEngine() - m := newTestManager(t, eng, archivePool(3600)) + m := newTestManager(t, eng, archivePool(keepAnHour)) sb := mustClaim(t, m, testKey) hibernateClaim(t, m, sb) stall := &stallingStore{Store: m.ckpts, published: make(chan string, 1), release: make(chan struct{})} @@ -52,7 +53,7 @@ func TestArchivePublishWindowPinsCheckpoint(t *testing.T) { func TestArchivePublishesUnderTheSandboxLineage(t *testing.T) { eng := newFakeEngine() - m := newTestManager(t, eng, archivePool(3600)) + m := newTestManager(t, eng, archivePool(keepAnHour)) sb := mustClaim(t, m, testKey) hibernateClaim(t, m, sb) rec := &lineageStore{Store: m.ckpts} @@ -67,7 +68,7 @@ func TestArchivePublishesUnderTheSandboxLineage(t *testing.T) { func TestArchiveWakeRehibernateWindowAborts(t *testing.T) { eng := newFakeEngine() - m := newTestManager(t, eng, archivePool(3600)) + m := newTestManager(t, eng, archivePool(keepAnHour)) sb := mustClaim(t, m, testKey) hibernateClaim(t, m, sb) stall := &stallingStore{Store: m.ckpts, published: make(chan string, 1), release: make(chan struct{})} @@ -110,7 +111,7 @@ func TestArchiveWakeRehibernateWindowAborts(t *testing.T) { func TestArchiveNeverCapturesAGuestThatWokeBeforeItsExport(t *testing.T) { eng := newFakeEngine() - m := newTestManager(t, eng, archivePool(3600)) + m := newTestManager(t, eng, archivePool(keepAnHour)) sb := mustClaim(t, m, testKey) hibernateClaim(t, m, sb) m.ckpts = &stagingStore{Store: m.ckpts, onStage: func() { @@ -136,7 +137,7 @@ func TestArchiveCkHiddenAcrossNodes(t *testing.T) { mkMgr := func() *Manager { cfg := &config.Config{ DataDir: t.TempDir(), CheckpointDir: shared, - Pools: []config.PoolSpec{archivePool(3600)}, + Pools: []config.PoolSpec{archivePool(keepAnHour)}, } return mustManager(t, cfg, newFakeEngine()) } @@ -185,7 +186,7 @@ func TestReconcileReclaimsOrphanArchiveCk(t *testing.T) { func TestClaimCheckpointRefusesArchive(t *testing.T) { eng := newFakeEngine() - m := newTestManager(t, eng, archivePool(3600)) + m := newTestManager(t, eng, archivePool(keepAnHour)) sb := mustClaim(t, m, testKey) mustArchive(t, m, sb) @@ -200,7 +201,7 @@ func TestClaimCheckpointRefusesArchive(t *testing.T) { func TestArchiveLifecycleRoundTrip(t *testing.T) { synctest.Test(t, func(t *testing.T) { eng := newFakeEngine() - m := newTestManager(t, eng, archivePool(3600)) + m := newTestManager(t, eng, archivePool(keepAnHour)) sb := mustClaim(t, m, testKey) id, token, origVM := sb.ID, sb.Token, sb.VMName @@ -251,7 +252,7 @@ func TestArchiveLifecycleRoundTrip(t *testing.T) { func TestArchivedClaimRefusesCaptureVerbs(t *testing.T) { eng := newFakeEngine() - m := newTestManager(t, eng, archivePool(3600)) + m := newTestManager(t, eng, archivePool(keepAnHour)) sb := mustClaim(t, m, testKey) mustArchive(t, m, sb) cred := Cred{Token: sb.Token} @@ -275,7 +276,7 @@ func TestArchivedClaimRefusesCaptureVerbs(t *testing.T) { func TestSweepExpiredCheckpointsIsANoopWithoutTTL(t *testing.T) { eng := newFakeEngine() - m := newTestManager(t, eng, archivePool(3600)) + m := newTestManager(t, eng, archivePool(keepAnHour)) sb := mustClaim(t, m, testKey) ck, err := m.Checkpoint(t.Context(), sb.ID, Cred{Token: sb.Token}, "", "") if err != nil { @@ -290,7 +291,7 @@ func TestSweepExpiredCheckpointsIsANoopWithoutTTL(t *testing.T) { func TestArchiveTTLSweepSparesLiveCheckpoint(t *testing.T) { eng := newFakeEngine() - m := newTestManager(t, eng, archivePool(3600)) + m := newTestManager(t, eng, archivePool(keepAnHour)) editView(m, func(v *configView) { v.ckptTTL = time.Nanosecond }) sb := mustClaim(t, m, testKey) id, token := sb.ID, sb.Token @@ -333,7 +334,7 @@ func TestArchiveKeepsForeverWhenDeleteZero(t *testing.T) { func TestArchiveRetentionPurge(t *testing.T) { synctest.Test(t, func(t *testing.T) { eng := newFakeEngine() - m := newTestManager(t, eng, archivePool(3600)) + m := newTestManager(t, eng, archivePool(keepAnHour)) sb := claimTenant(t, m, "acme") mustArchive(t, m, sb) ck := sb.ArchiveCk @@ -347,8 +348,8 @@ func TestArchiveRetentionPurge(t *testing.T) { if ckExists(t, m, ck) { t.Error("purge did not delete the archived checkpoint") } - if got := usageEventsOf(t, m, sb.ID); !strings.HasSuffix(got, "archive,archive_delete,reap") { - t.Errorf("events %s, want the purge to close the claim with a reap", got) + if got := usageEventsOf(t, m, sb.ID); !strings.HasSuffix(got, "archive,reap,archive_delete") { + t.Errorf("events %s, want the purge to end the archived span with its reap", got) } m.mu.Lock() @@ -362,7 +363,7 @@ func TestArchiveRetentionPurge(t *testing.T) { func TestArchiveWakeRegrantsTheClaimsLease(t *testing.T) { eng := newFakeEngine() - m := newTestManager(t, eng, archivePool(3600)) + m := newTestManager(t, eng, archivePool(keepAnHour)) setTenantCaps(t, m, map[string]int{"acme": 0}) sb, err := m.ClaimProvision(t.Context(), testKey, ClaimOptions{TTL: 2 * time.Hour, Tenant: "acme"}) if err != nil { @@ -380,7 +381,7 @@ func TestArchiveWakeRegrantsTheClaimsLease(t *testing.T) { func TestReleaseArchivedDeletesCheckpoint(t *testing.T) { eng := newFakeEngine() - m := newTestManager(t, eng, archivePool(3600)) + m := newTestManager(t, eng, archivePool(keepAnHour)) sb := mustClaim(t, m, testKey) id, token := sb.ID, sb.Token mustArchive(t, m, sb) @@ -407,7 +408,7 @@ func TestArchiveDeleteRetryAfterRestart(t *testing.T) { synctest.Test(t, func(t *testing.T) { eng := newFakeEngine() dataDir := t.TempDir() - m := newTestManagerAt(t, eng, dataDir, archivePool(3600)) + m := newTestManagerAt(t, eng, dataDir, archivePool(keepAnHour)) sb := mustClaim(t, m, testKey) id, token := sb.ID, sb.Token mustArchive(t, m, sb) @@ -442,7 +443,7 @@ func TestArchiveDeleteRetryAfterRestart(t *testing.T) { t.Fatal("failed deletion did not retain the checkpoint and cleanup marker") } - m2 := newTestManagerAt(t, eng, dataDir, archivePool(3600)) + m2 := newTestManagerAt(t, eng, dataDir, archivePool(keepAnHour)) if err := m2.Reconcile(t.Context()); err != nil { t.Fatalf("Reconcile: %v", err) } @@ -459,7 +460,7 @@ func TestArchiveDeleteRetryAfterRestart(t *testing.T) { func TestArchiveWakeDeleteFailureRetries(t *testing.T) { eng := newFakeEngine() - m := newTestManager(t, eng, archivePool(3600)) + m := newTestManager(t, eng, archivePool(keepAnHour)) sb := mustClaim(t, m, testKey) mustArchive(t, m, sb) ck := sb.ArchiveCk @@ -491,7 +492,7 @@ func TestArchiveWakeDeleteFailureRetries(t *testing.T) { func TestArchiveWakeRequiresDeleteMarker(t *testing.T) { eng := newFakeEngine() - m := newTestManager(t, eng, archivePool(3600)) + m := newTestManager(t, eng, archivePool(keepAnHour)) sb := mustClaim(t, m, testKey) mustArchive(t, m, sb) ck := sb.ArchiveCk @@ -533,7 +534,7 @@ func TestArchiveRemovalCommitPinsCheckpoint(t *testing.T) { for _, action := range []string{testArchiveRelease, testArchiveReap} { t.Run(action, func(t *testing.T) { eng := newFakeEngine() - m := newTestManager(t, eng, archivePool(3600)) + m := newTestManager(t, eng, archivePool(keepAnHour)) sb := mustClaim(t, m, testKey) mustArchive(t, m, sb) id, token, ck := sb.ID, sb.Token, sb.ArchiveCk @@ -592,7 +593,7 @@ func TestArchiveRemovalRequiresDeleteMarker(t *testing.T) { for _, action := range []string{testArchiveRelease, testArchiveReap} { t.Run(action, func(t *testing.T) { eng := newFakeEngine() - m := newTestManager(t, eng, archivePool(3600)) + m := newTestManager(t, eng, archivePool(keepAnHour)) sb := mustClaim(t, m, testKey) id, token := sb.ID, sb.Token mustArchive(t, m, sb) @@ -633,7 +634,7 @@ func TestArchiveRemovalRequiresDeleteMarker(t *testing.T) { func TestArchiveDeleteRetryRechecksWakeRollback(t *testing.T) { eng := newFakeEngine() - m := newTestManager(t, eng, archivePool(3600)) + m := newTestManager(t, eng, archivePool(keepAnHour)) sb := mustClaim(t, m, testKey) mustArchive(t, m, sb) id, token, ck := sb.ID, sb.Token, sb.ArchiveCk @@ -690,7 +691,7 @@ func TestArchiveDeleteRetryRechecksWakeRollback(t *testing.T) { func TestIdleOnceSkipsArchived(t *testing.T) { synctest.Test(t, func(t *testing.T) { eng := newFakeEngine() - m := newTestManager(t, eng, archivePool(3600)) + m := newTestManager(t, eng, archivePool(keepAnHour)) sb := mustClaim(t, m, testKey) mustArchive(t, m, sb) hibBefore := eng.hibernateCount() @@ -708,7 +709,7 @@ func TestIdleOnceSkipsArchived(t *testing.T) { func TestDeleteCheckpointCannotBrickArchive(t *testing.T) { eng := newFakeEngine() - m := newTestManager(t, eng, archivePool(3600)) + m := newTestManager(t, eng, archivePool(keepAnHour)) sb := mustClaim(t, m, testKey) id, token := sb.ID, sb.Token mustArchive(t, m, sb) @@ -731,7 +732,7 @@ func TestDeleteCheckpointCannotBrickArchive(t *testing.T) { func TestArchivePersistFailureRollsBack(t *testing.T) { synctest.Test(t, func(t *testing.T) { eng := newFakeEngine() - m := newTestManager(t, eng, archivePool(3600)) + m := newTestManager(t, eng, archivePool(keepAnHour)) sb := mustClaim(t, m, testKey) hibernateClaim(t, m, sb) snap, vm := sb.HibernateSnap, sb.VMName @@ -804,7 +805,7 @@ func TestArchivedClaimRollsBackOnPersistFailure(t *testing.T) { for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { synctest.Test(t, func(t *testing.T) { - m := newTestManager(t, newFakeEngine(), archivePool(3600)) + m := newTestManager(t, newFakeEngine(), archivePool(keepAnHour)) sb := mustClaim(t, m, testKey) mustArchive(t, m, sb) ck := sb.ArchiveCk @@ -826,7 +827,7 @@ func TestArchivedClaimRollsBackOnPersistFailure(t *testing.T) { func TestArchiveOnceSkipsInFlight(t *testing.T) { synctest.Test(t, func(t *testing.T) { eng := newFakeEngine() - m := newTestManager(t, eng, archivePool(3600)) + m := newTestManager(t, eng, archivePool(keepAnHour)) sb := mustClaim(t, m, testKey) hibernateClaim(t, m, sb) backdate(m, sb, time.Hour) @@ -849,7 +850,7 @@ func TestArchiveOnceSkipsInFlight(t *testing.T) { func TestArchiveWakeEvictsRecLock(t *testing.T) { eng := newFakeEngine() - m := newTestManager(t, eng, archivePool(3600)) + m := newTestManager(t, eng, archivePool(keepAnHour)) sb := mustClaim(t, m, testKey) hibernateClaim(t, m, sb) base := lockCount(m) @@ -884,8 +885,13 @@ func TestMarkArchiveCksKeepsOnlyTheIDsItCouldNotMark(t *testing.T) { func BenchmarkReapPurgesArchivedClaims(b *testing.B) { m := newTestManager(b, newFakeEngine()) b.ReportAllocs() + var purged uint64 for b.Loop() { b.StopTimer() + for m.counters.reaps.Load() < purged { + runtime.Gosched() + } + purged += 500 past := time.Now().Add(-time.Minute) m.mu.Lock() for i := range 500 { diff --git a/sandboxd/pool/balloon.go b/sandboxd/pool/balloon.go index 9b2d72f51..d295aacbb 100644 --- a/sandboxd/pool/balloon.go +++ b/sandboxd/pool/balloon.go @@ -66,9 +66,6 @@ func (m *Manager) balloonTo(ctx context.Context, sb *types.Sandbox, want string) if skip || !due { return } - if want != "" { - m.balloonSeen.Store(true) - } defer func() { m.mu.Lock() sb.BalloonBusy = false @@ -142,7 +139,7 @@ func (m *Manager) balloonSkipLocked(sb *types.Sandbox) bool { sb.Balloon == types.BalloonUnsupported || sb.BalloonBusy || time.Now().Before(sb.BalloonRetryAt) || m.midMoveLocked(sb) } -// balloonDue reports whether sb's balloon needs a call to reach want, an idle target, or the boot target when want is empty; callers hold m.mu. +// balloonDue reports whether sb's balloon needs a call to reach want; callers hold m.mu. func balloonDue(sb *types.Sandbox, want string) bool { if want != "" { return sb.Balloon != types.BalloonInflated || sb.BalloonTarget != want diff --git a/sandboxd/pool/balloon_test.go b/sandboxd/pool/balloon_test.go index f3aac08ab..fdcab73d8 100644 --- a/sandboxd/pool/balloon_test.go +++ b/sandboxd/pool/balloon_test.go @@ -309,23 +309,6 @@ func TestReclaimTurnedOnAfterARestartRepairsAnEarlierRunsInflate(t *testing.T) { } } -func TestAPreHibernateInflateKeepsTheSweepOn(t *testing.T) { - eng := newFakeEngine() - m := balloonManager(t, eng, t.TempDir(), archivePool(keepAnHour)) - sb := mustClaim(t, m, testKey) - m.balloonSeen.Store(false) - backdate(m, sb, 5*time.Second) - sweepIdle(t, m) - editView(m, func(v *configView) { v.balloonIdle, v.balloonTarget = 0, "" }) - if err := wakeClaim(t, m, sb); err != nil { - t.Fatalf("wake: %v", err) - } - sweepBalloons(t, m) - if calls := balloonCalls(eng); len(calls) != 2 || m.Counters().BalloonDeflations != 1 { - t.Errorf("balloon calls %q, want the pre-hibernate inflate deflated after the wake once reclaim is off", calls) - } -} - func TestTurningBalloonOffDeflatesTheInflatedClaims(t *testing.T) { eng := newFakeEngine() m := balloonManager(t, eng, t.TempDir(), config.PoolSpec{PoolKey: testKey}) diff --git a/sandboxd/pool/checkpoint.go b/sandboxd/pool/checkpoint.go index 77e757851..c42e353ff 100644 --- a/sandboxd/pool/checkpoint.go +++ b/sandboxd/pool/checkpoint.go @@ -318,7 +318,7 @@ func (m *Manager) sweepExpiredCheckpoints(ctx context.Context) { // Checkpoints hides archive images, so the TTL never reaches one; reapPurge retires them ckpts, err := m.Checkpoints(ctx, "") if err != nil { - logger.Error(ctx, err, "list for retention") + logger.Warnf(ctx, "list for retention: %v", err) return } cutoff := time.Now().Add(-ttl) @@ -330,7 +330,7 @@ func (m *Manager) sweepExpiredCheckpoints(ctx context.Context) { err := m.ckpts.Delete(ctx, ckpt.ID) unlock() if err != nil { - logger.Errorf(ctx, err, "expire %s", ckpt.ID) + logger.Warnf(ctx, "expire %s: %v", ckpt.ID, err) } } } diff --git a/sandboxd/pool/checkpoint_cleanup_test.go b/sandboxd/pool/checkpoint_cleanup_test.go index e7d19811e..ce97248fb 100644 --- a/sandboxd/pool/checkpoint_cleanup_test.go +++ b/sandboxd/pool/checkpoint_cleanup_test.go @@ -198,11 +198,8 @@ func TestFailoverDisabledExitKeepsReplicaCheckpoint(t *testing.T) { APIToken: "root", DataDir: old.dataDir, CheckpointDir: c.ckDir, Pools: []config.PoolSpec{{PoolKey: testKey}}, } - m, err := NewManager(t.Context(), cfg, old.eng, testSecrets(t), testBackends(t, cfg)) - if err != nil { - t.Fatalf("manager: %v", err) - } - if err = m.Reconcile(t.Context()); err != nil { + m := mustManager(t, cfg, old.eng) + if err := m.Reconcile(t.Context()); err != nil { t.Fatalf("reconcile: %v", err) } restored, ok := m.byID(sb.ID) diff --git a/sandboxd/pool/checkpoint_heal_test.go b/sandboxd/pool/checkpoint_heal_test.go index 1399fc9ef..1bc11baa5 100644 --- a/sandboxd/pool/checkpoint_heal_test.go +++ b/sandboxd/pool/checkpoint_heal_test.go @@ -223,7 +223,7 @@ func TestFetchCheckpointNeverPulls(t *testing.T) { func TestHasCheckpointAnswersOnlyForABranchableRecord(t *testing.T) { eng := newFakeEngine() - m := newTestManager(t, eng, archivePool(3600)) + m := newTestManager(t, eng, archivePool(keepAnHour)) src := mustClaim(t, m, testKey) ckpt, err := m.Checkpoint(t.Context(), src.ID, Cred{Token: src.Token}, "", "") diff --git a/sandboxd/pool/cordon_test.go b/sandboxd/pool/cordon_test.go index d080ecb04..fb417acfb 100644 --- a/sandboxd/pool/cordon_test.go +++ b/sandboxd/pool/cordon_test.go @@ -141,7 +141,7 @@ func TestCordonRefusesForksAndCheckpointClaims(t *testing.T) { } func TestACordonedNodeKeepsServingItsOwnClaims(t *testing.T) { - m := newTestManager(t, newFakeEngine(), archivePool(3600)) + m := newTestManager(t, newFakeEngine(), archivePool(keepAnHour)) sb := mustClaim(t, m, testKey) gone := mustClaim(t, m, testKey) cred := Cred{Token: sb.Token} @@ -245,14 +245,11 @@ func TestCordonOutlivesAReloadAndAnUndrain(t *testing.T) { func TestCordonIsAudited(t *testing.T) { dir := t.TempDir() cfg := &config.Config{DataDir: dir, AuditLog: true} - m, err := NewManager(t.Context(), cfg, newFakeEngine(), testSecrets(t), testBackends(t, cfg)) - if err != nil { - t.Fatalf("setup manager: %v", err) - } - if err = m.Cordon(t.Context()); err != nil { + m := mustManager(t, cfg, newFakeEngine()) + if err := m.Cordon(t.Context()); err != nil { t.Fatalf("cordon: %v", err) } - if err = m.Uncordon(t.Context()); err != nil { + if err := m.Uncordon(t.Context()); err != nil { t.Fatalf("uncordon: %v", err) } raw, err := os.ReadFile(filepath.Join(dir, "audit.jsonl")) diff --git a/sandboxd/pool/egress_test.go b/sandboxd/pool/egress_test.go index d08b8bd34..e0ec88560 100644 --- a/sandboxd/pool/egress_test.go +++ b/sandboxd/pool/egress_test.go @@ -63,7 +63,7 @@ func TestGoldenStampGatesAdoptionOnLane(t *testing.T) { if err := os.MkdirAll(g, 0o750); err != nil { t.Fatalf("stage golden: %v", err) } - if err := os.WriteFile(g+goldenStampSuffix, []byte(m.goldenStamp(m.view.Load(), testKey, "")), 0o644); err != nil { + if err := os.WriteFile(g+goldenStampSuffix, []byte(m.goldenStamp(m.view.Load(), testKey, "")), 0o600); err != nil { t.Fatalf("write stamp: %v", err) } m.adoptGolden(p, "") diff --git a/sandboxd/pool/failover.go b/sandboxd/pool/failover.go index e080f3743..6cf48e851 100644 --- a/sandboxd/pool/failover.go +++ b/sandboxd/pool/failover.go @@ -624,7 +624,6 @@ func (m *Manager) purgeArchiveRow(ctx context.Context, rec archives.Record) { if !rec.Consumed { m.counters.archiveDeletes.Add(1) m.recordUsage(ctx, usage.Event{Event: "archive_delete", ID: rec.ID, Reference: rec.Checkpoint, Tenant: rec.Tenant}) - m.releaseRowTun(ctx, rec) } logger.Infof(ctx, "purged archive %s of %s", rec.ID, rec.Owner) } diff --git a/sandboxd/pool/failover_test.go b/sandboxd/pool/failover_test.go index 3d55dfa93..dcc9b5e1d 100644 --- a/sandboxd/pool/failover_test.go +++ b/sandboxd/pool/failover_test.go @@ -2535,19 +2535,6 @@ func (r *hookRegistry) write(do func() (int64, error)) (int64, error) { return epoch, err } -func fire(hook *func()) { - if h := *hook; h != nil { - *hook = nil - h() - } -} - -func lost(flag *bool) bool { - was := *flag - *flag = false - return was -} - type unreadableStore struct { store.Store } @@ -2584,6 +2571,19 @@ func (r *countingRegistry) Sweepable(ctx context.Context, now time.Time, owner s return r.Registry.Sweepable(ctx, now, owner) } +func fire(hook *func()) { + if h := *hook; h != nil { + *hook = nil + h() + } +} + +func lost(flag *bool) bool { + was := *flag + *flag = false + return was +} + func assertReleasedForGood(t *testing.T, c *failoverCluster, sb *types.Sandbox, nodes ...*Manager) { t.Helper() if rec, ok, _ := c.reg.Get(t.Context(), sb.ID); ok && !rec.Consumed { diff --git a/sandboxd/pool/hibernate.go b/sandboxd/pool/hibernate.go index ec6f8793e..9d6d974e8 100644 --- a/sandboxd/pool/hibernate.go +++ b/sandboxd/pool/hibernate.go @@ -80,7 +80,7 @@ func (m *Manager) hibernateLocked(ctx context.Context, sb *types.Sandbox) error return ErrVolumeCapture } if sb.Key.Net == types.NetEgress { - // restore resumes the guest inside the call, before sandboxd could re-lock its tap + // Restore resumes the guest inside the call, before sandboxd could re-lock its tap return ErrNoEgressHibernate } if err := m.settleCapture(ctx, sb); err != nil { diff --git a/sandboxd/pool/liveness_test.go b/sandboxd/pool/liveness_test.go index 7cd5ab310..2062226e2 100644 --- a/sandboxd/pool/liveness_test.go +++ b/sandboxd/pool/liveness_test.go @@ -613,7 +613,7 @@ func TestReconcileKeepsRunningClaimsTheHostLost(t *testing.T) { } func TestAVMEventFindsAWokenArchivedClaimByItsNewVM(t *testing.T) { - m := newTestManager(t, newFakeEngine(), archivePool(3600)) + m := newTestManager(t, newFakeEngine(), archivePool(keepAnHour)) sb := mustClaim(t, m, testKey) hibernateClaim(t, m, sb) oldVM := sb.VMName diff --git a/sandboxd/pool/memory_test.go b/sandboxd/pool/memory_test.go index 6f6e082b0..8859e1364 100644 --- a/sandboxd/pool/memory_test.go +++ b/sandboxd/pool/memory_test.go @@ -98,7 +98,7 @@ func TestMemoryOverLimitKeepsFreeingAndOwnedWakesWorking(t *testing.T) { func TestMemoryOverLimitStillArchives(t *testing.T) { synctest.Test(t, func(t *testing.T) { - m := newTestManager(t, newFakeEngine(), archivePool(3600)) + m := newTestManager(t, newFakeEngine(), archivePool(keepAnHour)) sb := mustClaim(t, m, testKey) m.memoryLow.Store(true) diff --git a/sandboxd/pool/move_in.go b/sandboxd/pool/move_in.go index 26c45b95c..1f569edca 100644 --- a/sandboxd/pool/move_in.go +++ b/sandboxd/pool/move_in.go @@ -211,11 +211,11 @@ func (m *Manager) markMoveStarted(ctx context.Context, id string, epoch int64) { } func (m *Manager) dropMoveIn(ctx context.Context, sb *types.Sandbox) { + _ = os.RemoveAll(m.moveDir(sb.ID)) if holds, ok := m.forgetClaim(ctx, sb); ok { m.out.Disarm(sb.ID, true) m.removeOrRetry(ctx, sb.VMName, "", volumeTeardown{holds: holds}) } - _ = os.RemoveAll(m.moveDir(sb.ID)) } func (m *Manager) settleMoveIn(ctx context.Context, sb *types.Sandbox) { diff --git a/sandboxd/pool/move_test.go b/sandboxd/pool/move_test.go index 7ce8c87ac..4bfb42b98 100644 --- a/sandboxd/pool/move_test.go +++ b/sandboxd/pool/move_test.go @@ -350,7 +350,7 @@ func TestMoveTargetRefusesARoundThatOutlivedItsReservation(t *testing.T) { reserveMove(t, mc.t, "sb_r", "acme") src := t.TempDir() if err := os.WriteFile(filepath.Join(src, "state.json"), []byte("{}"), 0o600); err != nil { - t.Fatal(err) + t.Fatalf("write state: %v", err) } var captured bytes.Buffer capture := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { @@ -387,7 +387,7 @@ func TestMoveTargetLetsTheSameSourceRetryAndRefusesAnother(t *testing.T) { first, _ := reservationOf(mc.t, "sb_r") stale := filepath.Join(mc.t.moveDir("sb_r"), "state.json") if err := os.WriteFile(stale, []byte("a crashed source's round"), 0o600); err != nil { - t.Fatal(err) + t.Fatalf("write stale round: %v", err) } if _, err := mc.t.AdmitMove(t.Context(), "sb_r", MoveAdmission{Tenant: "acme", Key: testKey, From: "other"}); !errors.Is(err, ErrNotMovable) { t.Errorf("admission from another source: %v, want ErrNotMovable", err) @@ -422,7 +422,7 @@ func TestMoveTargetKeepsARetrysStagingDirWhenTheCrashedSourcesRoundEnds(t *testi reserveMove(t, mc.t, "sb_r", "acme") retried := filepath.Join(mc.t.moveDir("sb_r"), "state.json") if err := os.WriteFile(retried, []byte("the retry's round"), 0o600); err != nil { - t.Fatal(err) + t.Fatalf("write retry round: %v", err) } _ = stream.CloseWithError(errors.New("the crashed source's connection ended")) if err := <-done; !errors.Is(err, ErrUnknownSandbox) { @@ -447,7 +447,7 @@ func TestMoveCommitOfACrashedSourceDoesNotTakeTheRetrysReservation(t *testing.T) reserveMove(t, mc.t, sb.ID, "") retry, _ := reservationOf(mc.t, sb.ID) if err := os.WriteFile(filepath.Join(mc.t.moveDir(sb.ID), "state.json"), []byte("{}"), 0o600); err != nil { - t.Fatal(err) + t.Fatalf("write state: %v", err) } close(eng.receiveStall) if err := <-done; !errors.Is(err, ErrNotMovable) { @@ -468,6 +468,42 @@ func TestMoveCommitOfACrashedSourceDoesNotTakeTheRetrysReservation(t *testing.T) } } +func TestMoveTargetDroppingALostCommitKeepsTheRetrysStagingDir(t *testing.T) { + mc := newMoveCluster(t) + sb := mustClaim(t, mc.s, testKey) + reserveMove(t, mc.t, sb.ID, "") + if err := os.WriteFile(filepath.Join(mc.t.moveDir(sb.ID), "state.json"), []byte("{}"), 0o600); err != nil { + t.Fatalf("write state: %v", err) + } + eng := fakeOf(mc.t) + stall := make(chan struct{}) + eng.removeStall = stall + done := make(chan error, 1) + go func() { + done <- mc.t.CommitMove(context.WithoutCancel(t.Context()), sb.ID, []byte(`{"id":"`+sb.ID+`","key":{"template":"rt:24.04","net":"none","size":"small"}}`), 1) + }() + waitFor(t, func() bool { + eng.mu.Lock() + received := len(eng.receives) == 1 + eng.mu.Unlock() + _, reserved := reservationOf(mc.t, sb.ID) + _, claimed := mc.t.byID(sb.ID) + return received && !reserved && !claimed + }) + reserveMove(t, mc.t, sb.ID, "") + retried := filepath.Join(mc.t.moveDir(sb.ID), "state.json") + if err := os.WriteFile(retried, []byte("the retry's round"), 0o600); err != nil { + t.Fatalf("write retry round: %v", err) + } + close(stall) + if err := <-done; !errors.Is(err, ErrNotMovable) { + t.Fatalf("a commit the source aborted: %v, want ErrNotMovable", err) + } + if _, err := os.Stat(retried); err != nil { + t.Errorf("dropping the lost commit removed the retry's files: %v", err) + } +} + func TestMoveOfAReplicatedClaimDropsItsRowBeforeTheFinalRound(t *testing.T) { mc := newMoveClusterWith(t, func(c *failoverCluster) { c.replicate = replicateEvery }) third := mc.moveNode(t, "c") @@ -1014,7 +1050,7 @@ func TestMoveRestartResolvesAJournaledMoveThroughItsRecord(t *testing.T) { dataDir := t.TempDir() seedClaims(t, dataDir, map[string]*types.Sandbox{sb.ID: sb}) if err := os.MkdirAll(filepath.Join(dataDir, movesDir, sb.ID), 0o700); err != nil { - t.Fatal(err) + t.Fatalf("stage move dir: %v", err) } eng := newFakeEngine() eng.vms[vm], eng.paused[vm], eng.stopped[vm] = "/vsock/"+vm, sb.MoveIn, tt.vmmLost @@ -1835,17 +1871,12 @@ func failRound(n int) moveHook { } func abortRecord(t *testing.T, mc *moveCluster) { - t.Helper() - for _, rec := range mustStale(t, mc) { - _, _ = mc.moveReg.Advance(t.Context(), rec.ID, rec.Epoch, rec.State, moves.Aborted) - } -} - -func mustStale(t *testing.T, mc *moveCluster) []moves.Record { t.Helper() recs, err := mc.moveReg.Stale(t.Context(), time.Now().Add(time.Hour)) if err != nil { t.Errorf("stale move records: %v", err) } - return recs + for _, rec := range recs { + _, _ = mc.moveReg.Advance(t.Context(), rec.ID, rec.Epoch, rec.State, moves.Aborted) + } } diff --git a/sandboxd/pool/reap.go b/sandboxd/pool/reap.go index 2628e96dc..7ed26dfc8 100644 --- a/sandboxd/pool/reap.go +++ b/sandboxd/pool/reap.go @@ -140,10 +140,10 @@ func (m *Manager) reapOnce(ctx context.Context) { case reapPurge: m.out.Disarm(v.id, true) m.releaseTun(ctx, v.id, v.tun) + m.recordUsage(ctx, usage.Event{Event: reapEvent, ID: v.id, Tenant: v.tenant, Deadline: v.deadline}) m.purgeArchiveCk(ctx, v.id, v.ck, v.tenant) m.untrack(m.pendingCks, v.ck) m.counters.reaps.Add(1) - m.recordUsage(ctx, usage.Event{Event: reapEvent, ID: v.id, Tenant: v.tenant, Deadline: v.deadline}) logger.Infof(ctx, "purged archived sandbox %s", v.id) case reapArchive: logSweepResult(ctx, logger, m.archive(ctx, v.sb), "archived expired sandbox "+v.id, "archive expired sandbox "+v.id) diff --git a/sandboxd/pool/replica_test.go b/sandboxd/pool/replica_test.go index dba31ba6a..bbe5470cf 100644 --- a/sandboxd/pool/replica_test.go +++ b/sandboxd/pool/replica_test.go @@ -912,6 +912,20 @@ func TestReconcileWithTheRegistryDownRecordsOnlyTheEndNoOtherNodeCanHold(t *test if got, want := usageEventsOf(t, restarted, plain.ID), "claim,reap"; got != want { t.Errorf("events of the unreplicated claim %s, want %s", got, want) } + + restarted.sweepOrphanArchives(t.Context()) + rec, ok, err := c.reg.Get(t.Context(), replicated.ID) + if err != nil || !ok || rec.Deadline.IsZero() { + t.Fatalf("row once the registry returns: %+v %t %v, want it kept with a retention deadline", rec, ok, err) + } + rec.Deadline = time.Now().Add(-time.Second) + if _, err = c.reg.Replace(t.Context(), rec, rec.Epoch); err != nil { + t.Fatalf("age the row: %v", err) + } + restarted.sweepOrphanArchives(t.Context()) + if got, want := usageEventsOf(t, restarted, replicated.ID), "claim,archive_delete"; got != want { + t.Errorf("events after the retention purge %s, want %s: a reap would end the running span at the retention deadline", got, want) + } } func TestReplicaRestartDropsAClaimAdoptedElsewhere(t *testing.T) { diff --git a/sandboxd/pool/telemetry_test.go b/sandboxd/pool/telemetry_test.go index c5c2b993a..02d4c01ac 100644 --- a/sandboxd/pool/telemetry_test.go +++ b/sandboxd/pool/telemetry_test.go @@ -230,7 +230,7 @@ func TestEveryClaimScopedUsageEventCarriesTheTenant(t *testing.T) { } func TestEveryEventThatMovesTheDeadlineStatesIt(t *testing.T) { - m := newTestManager(t, newFakeEngine(), archivePool(3600)) + m := newTestManager(t, newFakeEngine(), archivePool(keepAnHour)) ctx := t.Context() sb := mustClaim(t, m, testKey) cred := Cred{Token: sb.Token} @@ -261,16 +261,16 @@ func TestEveryEventThatMovesTheDeadlineStatesIt(t *testing.T) { } func TestAPurgeReapStatesTheRetentionEnd(t *testing.T) { - m := newTestManager(t, newFakeEngine(), archivePool(3600)) + m := newTestManager(t, newFakeEngine(), archivePool(keepAnHour)) sb := mustClaim(t, m, testKey) mustArchive(t, m, sb) expire(m, sb) want := m.deadlineOf(sb) m.reapOnce(t.Context()) - waitFor(t, func() bool { return strings.HasSuffix(usageEventsOf(t, m, sb.ID), ",reap") }) + waitFor(t, func() bool { return strings.HasSuffix(usageEventsOf(t, m, sb.ID), ",reap,archive_delete") }) evs := usageJournal(t, m) - if last := evs[len(evs)-1]; !last.Deadline.Equal(want) { - t.Errorf("purge reap deadline %v, want %v", last.Deadline, want) + if reap := evs[len(evs)-2]; reap.Event != "reap" || !reap.Deadline.Equal(want) { + t.Errorf("purge reap %s deadline %v, want reap at %v", reap.Event, reap.Deadline, want) } } @@ -287,7 +287,7 @@ func TestAnArchiveKeptForeverStatesNoDeadline(t *testing.T) { func TestArchiveEventPrecedesAWakeDuringItsTeardown(t *testing.T) { synctest.Test(t, func(t *testing.T) { eng := newFakeEngine() - m := newTestManager(t, eng, archivePool(3600)) + m := newTestManager(t, eng, archivePool(keepAnHour)) ctx := t.Context() sb := mustClaim(t, m, testKey) hibernateClaim(t, m, sb) diff --git a/sandboxd/pool/tun.go b/sandboxd/pool/tun.go index e338e7ee9..90d4a3af1 100644 --- a/sandboxd/pool/tun.go +++ b/sandboxd/pool/tun.go @@ -2,7 +2,6 @@ package pool import ( "context" - "encoding/json/v2" "errors" "fmt" "maps" @@ -244,18 +243,6 @@ func (m *Manager) confirmAdoptedTun(ctx context.Context, sb *types.Sandbox) erro return nil } -func (m *Manager) releaseRowTun(ctx context.Context, rec archives.Record) { - if m.tunAddrs == nil { - return - } - sb := &types.Sandbox{} - if err := json.Unmarshal(rec.Claim, sb); err != nil { - log.WithFunc("pool.releaseRowTun").Warnf(ctx, "decode archive row %s: %v", rec.ID, err) - return - } - m.releaseTun(ctx, rec.ID, sb.TunAddr) -} - // foldTunBytes ends the live flows, so their records are written, and copies every armed claim's live tun byte count into its record, so the next commit persists it. func (m *Manager) foldTunBytes() { if m.tunRouter == nil { diff --git a/sandboxd/pool/tun_test.go b/sandboxd/pool/tun_test.go index a343d8f88..538cd953b 100644 --- a/sandboxd/pool/tun_test.go +++ b/sandboxd/pool/tun_test.go @@ -52,7 +52,7 @@ func TestATunPoolClaimGetsAnAddressAndAPacketDoor(t *testing.T) { t.Error("a claim of a pool without tun has a packet door") } if got := tunVerbsOf(eng); len(got) != 1 { - t.Errorf("tun verbs %q, want none for the claim without tun", got) + t.Errorf("tun verbs %q, want only the tun claim's up verb", got) } } @@ -290,10 +290,10 @@ func TestFlushClaimsPersistsTheLiveTunBytes(t *testing.T) { stream := openStream(t, sb) pkt := udpFrame(netip.MustParseAddr("93.184.216.34")) if _, err := stream.Write(binary.LittleEndian.AppendUint32(nil, uint32(len(pkt)))); err != nil { - t.Fatal(err) + t.Fatalf("write frame header: %v", err) } if _, err := stream.Write(pkt); err != nil { - t.Fatal(err) + t.Fatalf("write frame: %v", err) } waitFor(t, func() bool { used, _ := m.out.TunUsed(sb.ID); return used > 0 }) mustFlush(t, m) diff --git a/sandboxd/pool/volume_cleanup_test.go b/sandboxd/pool/volume_cleanup_test.go index 519157d6a..69c77e162 100644 --- a/sandboxd/pool/volume_cleanup_test.go +++ b/sandboxd/pool/volume_cleanup_test.go @@ -1,7 +1,6 @@ package pool import ( - "context" "errors" "sync" "testing" @@ -23,7 +22,7 @@ func TestDispatchedCleanupRetainsCatalogReference(t *testing.T) { if err = m.Release(t.Context(), sb.ID, Cred{Token: sb.Token}); err == nil { t.Fatal("expected survivor") } - wg := dispatchBlocked(t.Context(), t, m) + wg := dispatchBlocked(t, m) err = m.UnregisterVolume(t.Context(), "data") unblock(m, wg) if !errors.Is(err, ErrVolumeRegistrationConflict) { @@ -59,12 +58,12 @@ func TestOwnershipTransferReleasesRunningVolumeHoldAfterRemoval(t *testing.T) { } } -func dispatchBlocked(ctx context.Context, t *testing.T, m *Manager) *sync.WaitGroup { +func dispatchBlocked(t *testing.T, m *Manager) *sync.WaitGroup { t.Helper() for range cap(m.maintSem) { m.maintSem <- struct{}{} } - wg := m.retryRemovals(ctx) + wg := m.retryRemovals(t.Context()) m.mu.Lock() defer m.mu.Unlock() if len(m.pendingRemovals) != 0 { diff --git a/sandboxd/server/api_test.go b/sandboxd/server/api_test.go deleted file mode 100644 index 0fd70d024..000000000 --- a/sandboxd/server/api_test.go +++ /dev/null @@ -1,12 +0,0 @@ -package server - -import ( - "testing" - "time" -) - -func TestTTLFieldReachesTheClampForAHugeCount(t *testing.T) { - if got := (TTLField{TTLSeconds: 9223372037}).TTL(); got <= 24*time.Hour { - t.Errorf("TTL() = %v for a huge ttl_seconds, want it to reach the 24h clamp", got) - } -} diff --git a/sandboxd/server/metrics.go b/sandboxd/server/metrics.go index e1fcf0cff..2516ed943 100644 --- a/sandboxd/server/metrics.go +++ b/sandboxd/server/metrics.go @@ -40,7 +40,7 @@ func (s *Server) handleMetrics(w http.ResponseWriter, _ *http.Request) { flag("cordoned", "1 while the node is cordoned", g.Cordoned) scalar("evacuating", "claims an evacuation has yet to move to a peer", g.Evacuating) flag("memory_low", "1 while memory_limit refuses claims and refills", g.MemoryLow) - scalar("anon_bytes", "node anon memory, 0 when memory_limit has no anon limit", g.AnonBytes) + scalar("anon_bytes", "node anon memory plus used swap, 0 when memory_limit has no anon limit", g.AnonBytes) scalar("balloon_inflated", "running claims whose balloon holds the idle target", g.BalloonInflated) if tenants := s.mgr.TenantClaims(); len(tenants) > 0 { diff --git a/sandboxd/server/move_test.go b/sandboxd/server/move_test.go index 1a5ef35e0..0d16e2fa2 100644 --- a/sandboxd/server/move_test.go +++ b/sandboxd/server/move_test.go @@ -25,7 +25,7 @@ func TestMoveRouteResolvesThePeerAndMapsTheOutcome(t *testing.T) { srv := newMeshTestServer(t, "root", mgr, &fakePlacer{addrs: []string{"10.0.0.2:7777"}}, nil) resp := doReq(t, http.MethodPost, srv.URL+"/v1/sandboxes/sb_1/move", "root", `{"to":"10.0.0.2:7777","max_downtime_ms":1500}`) - defer func() { _ = resp.Body.Close() }() + defer resp.Body.Close() var body MoveResponse if err := json.UnmarshalRead(resp.Body, &body); err != nil || resp.StatusCode != http.StatusOK { t.Fatalf("move: %d %v", resp.StatusCode, err) @@ -67,7 +67,7 @@ func TestMoveRouteRefusesABudgetTheStartAllowanceAloneExceedsBeforeAnyMoveStarts }} srv := newMeshTestServer(t, "root", mgr, &fakePlacer{addrs: []string{"10.0.0.2:7777"}}, nil) resp := doReq(t, http.MethodPost, srv.URL+"/v1/sandboxes/sb_1/move", "root", `{"to":"10.0.0.2:7777","max_downtime_ms":400}`) - defer func() { _ = resp.Body.Close() }() + defer resp.Body.Close() body, err := io.ReadAll(resp.Body) if err != nil || resp.StatusCode != http.StatusBadRequest || !strings.Contains(string(body), "above 400") { t.Errorf("answer %d %s %v, want a 400 naming the floor", resp.StatusCode, body, err) diff --git a/sandboxd/store/dir/dir_test.go b/sandboxd/store/dir/dir_test.go index 42bf7eb5c..6845b0639 100644 --- a/sandboxd/store/dir/dir_test.go +++ b/sandboxd/store/dir/dir_test.go @@ -358,14 +358,11 @@ func TestSweepSparesPublishingGeneration(t *testing.T) { } func TestSweepReclaimsOnlyAgedUncommittedGenerations(t *testing.T) { - const ( - orphanID = "ck_00000000000000aa" - publishID = "ck_00000000000000bb" - ) + const publishID = "ck_00000000000000bb" root := t.TempDir() sweeper := newStore(t, root) writer := newStore(t, root) - orphan := filepath.Join(root, orphanID, store.ExportDir+"-0011223344556677") + orphan := filepath.Join(root, testID, store.ExportDir+"-0011223344556677") if mkdirErr := os.MkdirAll(orphan, 0o750); mkdirErr != nil { t.Fatalf("seed orphan: %v", mkdirErr) } @@ -390,7 +387,7 @@ func TestSweepReclaimsOnlyAgedUncommittedGenerations(t *testing.T) { t.Fatalf("sweep old: %v", sweepErr) } assertGonef(t, orphan, "aged uncommitted generation survived") - assertPresentf(t, filepath.Join(root, orphanID), "orphan record shell was removed") + assertPresentf(t, filepath.Join(root, testID), "orphan record shell was removed") assertPresentf(t, fresh, "publishing generation was swept") if removeErr := os.RemoveAll(commitBlocker); removeErr != nil { diff --git a/sandboxd/store/peer/broadcast_test.go b/sandboxd/store/peer/broadcast_test.go index 88d0e89d1..ca92c771f 100644 --- a/sandboxd/store/peer/broadcast_test.go +++ b/sandboxd/store/peer/broadcast_test.go @@ -59,10 +59,7 @@ func TestBroadcastDeleteDedupsRepeatedAddrs(t *testing.T) { } func TestBroadcastDeleteSwallowsFailures(t *testing.T) { - fail := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { - w.WriteHeader(http.StatusInternalServerError) - })) - defer fail.Close() + fail := statusServer(t, http.StatusInternalServerError) ok, reached := countingServer(t, http.StatusNoContent) b := &Broadcaster{Peers: func() []string { return []string{fail.URL, ok.URL, "127.0.0.1:1"} }} diff --git a/sandboxd/store/peer/transport_test.go b/sandboxd/store/peer/transport_test.go index 18fbe752b..b6fec65ac 100644 --- a/sandboxd/store/peer/transport_test.go +++ b/sandboxd/store/peer/transport_test.go @@ -337,7 +337,6 @@ func TestUntarRejectsDamagedChunk(t *testing.T) { func tarDir(src string, w io.Writer) error { tw := tar.NewWriter(w) - defer func() { _ = tw.Close() }() if err := tarInto(src, "", tw); err != nil { return err } diff --git a/sandboxd/store/s3/s3_test.go b/sandboxd/store/s3/s3_test.go index 591412378..b64084395 100644 --- a/sandboxd/store/s3/s3_test.go +++ b/sandboxd/store/s3/s3_test.go @@ -348,7 +348,7 @@ func TestFetchPrunesSupersededGenerationsIdlePastTheGrace(t *testing.T) { } old := time.Now().Add(-store.GenerationGrace - time.Minute) if err := os.Chtimes(first, old, old); err != nil { - t.Fatal(err) + t.Fatalf("age the first generation: %v", err) } fetchGen(3) if _, err := os.Stat(first); !errors.Is(err, os.ErrNotExist) { @@ -369,7 +369,7 @@ func TestFetchHitRefreshesTheGenerationsUse(t *testing.T) { } old := time.Now().Add(-2 * store.GenerationGrace) if err = os.Chtimes(filepath.Dir(export), old, old); err != nil { - t.Fatal(err) + t.Fatalf("age the export: %v", err) } if _, _, _, err = st.Fetch(t.Context(), id); err != nil { t.Fatalf("fetch hit: %v", err) @@ -541,7 +541,7 @@ func TestPublishDigestedHashesHolesLikeZeros(t *testing.T) { func TestPublishDigestedSplitsChunksAtTheFixedBoundary(t *testing.T) { const id = "ck_00000000000000a5" _, st := newFakeStore(t) - content := string(bytes.Repeat([]byte{'x'}, int(store.DigestChunkSize))) + "y" + content := strings.Repeat("x", int(store.DigestChunkSize)) + "y" staging := stageRecord(t, st, id, metaJSON(id), content) digest, err := storetest.PublishDigested(t.Context(), st, staging, id) if err != nil { @@ -901,7 +901,11 @@ func TestOnlyALabeledStoreReadsLabelsWhenListing(t *testing.T) { for _, id := range ids { gets += fake.requestCount(http.MethodGet, "ck/"+id+"/"+store.LabelsFile) } - if want := map[bool]int{true: len(ids), false: 0}[labeled]; gets != want { + want := 0 + if labeled { + want = len(ids) + } + if gets != want { t.Errorf("labeled=%v: %d label GETs, want %d", labeled, gets, want) } } diff --git a/sandboxd/store/sparse_test.go b/sandboxd/store/sparse_test.go index c9228c7b8..6e47b0c45 100644 --- a/sandboxd/store/sparse_test.go +++ b/sandboxd/store/sparse_test.go @@ -70,8 +70,9 @@ func TestScanFileDoesNotReadHoles(t *testing.T) { if len(digest.Chunks) != 6 { t.Fatalf("%d chunks, want 6", len(digest.Chunks)) } + zero := zeroSum(DigestChunkSize) for i, sum := range digest.Chunks { - if sum != zeroSum(DigestChunkSize) { + if sum != zero { t.Errorf("hole chunk %d does not carry the zero-chunk sum", i) } } diff --git a/sandboxd/tunnet/fence_test.go b/sandboxd/tunnet/fence_test.go index 28e2a6913..bb99df68a 100644 --- a/sandboxd/tunnet/fence_test.go +++ b/sandboxd/tunnet/fence_test.go @@ -26,7 +26,7 @@ func TestFenceStopsPacketsAndARPAndResumesWithoutResettingQuota(t *testing.T) { u.inject(t, reply(remote)) readFrame(t, guest) synctest.Wait() - before, bytes := r.Stats(), used.Load() + before, charged := r.Stats(), used.Load() arp := appendARP(appendHeader(nil, 7, broadcast, exitMAC, etherARP), arpReq, exitMAC, vtep, [6]byte{}, claimAddr) for range 2 { @@ -41,8 +41,8 @@ func TestFenceStopsPacketsAndARPAndResumesWithoutResettingQuota(t *testing.T) { if got.DgramsOut != before.DgramsOut || got.FramesOut != before.FramesOut || got.ARPSent != before.ARPSent || got.ARPAnswered != before.ARPAnswered { t.Errorf("fenced traffic: before %+v, after %+v", before, got) } - if used.Load() != bytes || len(u.out) != 0 { - t.Errorf("fenced claim charged %d bytes and sent %d datagrams", used.Load()-bytes, len(u.out)) + if used.Load() != charged || len(u.out) != 0 { + t.Errorf("fenced claim charged %d bytes and sent %d datagrams", used.Load()-charged, len(u.out)) } } @@ -68,8 +68,8 @@ func TestFenceStopsPacketsAndARPAndResumesWithoutResettingQuota(t *testing.T) { if got := r.Stats(); got.ARPSent != before.ARPSent+1 || got.ARPAnswered != before.ARPAnswered+1 || got.DgramsOut != before.DgramsOut+1 || got.FramesOut != before.FramesOut+1 { t.Errorf("unfenced traffic: before %+v, after %+v", before, got) } - if got := used.Load(); got != bytes*2 { - t.Errorf("quota after recovery %d, want %d", got, bytes*2) + if got := used.Load(); got != charged*2 { + t.Errorf("quota after recovery %d, want %d", got, charged*2) } }) } diff --git a/sdk/go/info.go b/sdk/go/info.go index 7c7e7c104..48e57a65b 100644 --- a/sdk/go/info.go +++ b/sdk/go/info.go @@ -7,7 +7,7 @@ import ( "time" ) -// peersTimeout bounds peer discovery so one slow entry node cannot stall the caller. +// peersTimeout bounds peer discovery and each owner probe, so one slow node cannot stall the caller. const peersTimeout = 5 * time.Second // NodeInfo is one node's operational state from GET /v1/info. diff --git a/sdk/go/keepalive.go b/sdk/go/keepalive.go index 2f21e3f2f..749519cd7 100644 --- a/sdk/go/keepalive.go +++ b/sdk/go/keepalive.go @@ -117,7 +117,7 @@ type lease struct { over atomic.Bool } -// done ends the RPC: a terminal frame, error frames included, parks the connection; any other failure drops it. +// done parks the connection after nil or an unwrapped error frame, and drops it otherwise. func (l *lease) done(err error) error { if _, frame := err.(*wire.ErrorResp); err == nil || frame { l.reuse() diff --git a/sdk/go/sandbox.go b/sdk/go/sandbox.go index 8a3388d3a..5aa224486 100644 --- a/sdk/go/sandbox.go +++ b/sdk/go/sandbox.go @@ -238,7 +238,7 @@ func (s *Sandbox) run(ctx context.Context, cmd Cmd) (int, error) { defer l.close() stdinDone := make(chan struct{}) - var stdinErr error + stdinErr := make(chan error, 1) if cmd.Stdin == nil { // A guest that already answered and closed fails this send; the frames it sent still come back below _ = conn.Send(wire.StdinClose{}) @@ -246,16 +246,15 @@ func (s *Sandbox) run(ctx context.Context, cmd Cmd) (int, error) { } else { go func() { defer close(stdinDone) - stdinErr = pumpStdin(conn, cmd.Stdin) + pumpStdin(conn, cmd.Stdin, stdinErr) }() } pid, code, exited, err := pumpStdio(ctx, conn, cmd.Stdout, cmd.Stderr) // A pump still parked in Read would write into the next RPC, so its connection is not kept - pumped := false select { case <-stdinDone: - err, pumped = l.done(err), true + err = l.done(err) default: } if err != nil { @@ -269,10 +268,12 @@ func (s *Sandbox) run(ctx context.Context, cmd Cmd) (int, error) { if !exited { return 0, unexpected(&wire.Done{}) } - if pumped && stdinErr != nil { - return int(code), &callerError{fmt.Errorf("read stdin: %w", stdinErr)} + select { + case err := <-stdinErr: + return int(code), &callerError{fmt.Errorf("read stdin: %w", err)} + default: + return int(code), nil } - return int(code), nil } func (s *Sandbox) releaseAt(ctx context.Context, addr string) (found bool, err error) { @@ -384,21 +385,21 @@ func markGone(err error, id, addr string) error { return err } -func pumpStdin(conn *silkd.Conn, r io.Reader) error { +func pumpStdin(conn *silkd.Conn, r io.Reader, readErr chan<- error) { buf := make([]byte, stdinChunk) for { n, err := r.Read(buf) if n > 0 { if sendErr := conn.Send(&wire.Stdin{Data: buf[:n]}); sendErr != nil { - return nil + return } } if err != nil { - _ = conn.Send(wire.StdinClose{}) - if errors.Is(err, io.EOF) { - return nil + if !errors.Is(err, io.EOF) { + readErr <- err } - return err + _ = conn.Send(wire.StdinClose{}) + return } } } diff --git a/sdk/go/sandbox_test.go b/sdk/go/sandbox_test.go index c8f0f09ca..6d17c4734 100644 --- a/sdk/go/sandbox_test.go +++ b/sdk/go/sandbox_test.go @@ -74,14 +74,16 @@ func TestRunReportsAStdinReadFailure(t *testing.T) { sb := testSandbox(t, newAgentServer(t, silkdtest.ServeConn)) broken := errors.New("upstream reset") - var out strings.Builder - code, err := sb.Run(t.Context(), Cmd{ - Argv: []string{"cat"}, - Stdin: io.MultiReader(strings.NewReader("partial"), iotest.ErrReader(broken)), - Stdout: &out, - }) - if !errors.Is(err, broken) || code != 0 || out.String() != "partial" { - t.Errorf("code=%d out=%q err=%v, want the exit code with the stdin failure", code, out.String(), err) + for range 100 { + var out strings.Builder + code, err := sb.Run(t.Context(), Cmd{ + Argv: []string{"cat"}, + Stdin: io.MultiReader(strings.NewReader("partial"), iotest.ErrReader(broken)), + Stdout: &out, + }) + if !errors.Is(err, broken) || code != 0 || out.String() != "partial" { + t.Fatalf("code=%d out=%q err=%v, want the exit code with the stdin failure", code, out.String(), err) + } } } diff --git a/sdk/go/volumes.go b/sdk/go/volumes.go index bd9f90706..8d04e43d8 100644 --- a/sdk/go/volumes.go +++ b/sdk/go/volumes.go @@ -93,22 +93,22 @@ func (s *Sandbox) volumesAt(ctx context.Context, method, addr string, body []byt return out, err } defer func() { _ = resp.Body.Close() }() - var wire struct { + var reply struct { VolumesWant *[]Volume `json:"volumes_want"` Volumes *[]Volume `json:"volumes"` States *[]VolumeState `json:"states"` Converging *bool `json:"converging"` Error string `json:"error"` } - decodeErr := json.UnmarshalRead(resp.Body, &wire) - complete := decodeErr == nil && wire.VolumesWant != nil && wire.Volumes != nil && wire.States != nil && wire.Converging != nil + decodeErr := json.UnmarshalRead(resp.Body, &reply) + complete := decodeErr == nil && reply.VolumesWant != nil && reply.Volumes != nil && reply.States != nil && reply.Converging != nil if complete { - out = SandboxVolumesResponse{VolumesWant: *wire.VolumesWant, Volumes: *wire.Volumes, States: *wire.States, Converging: *wire.Converging, Error: wire.Error} + out = SandboxVolumesResponse{VolumesWant: *reply.VolumesWant, Volumes: *reply.Volumes, States: *reply.States, Converging: *reply.Converging, Error: reply.Error} s.VolumesWant = slices.Clone(out.VolumesWant) s.Volumes = slices.Clone(out.Volumes) } if resp.StatusCode < 200 || resp.StatusCode >= 300 { - failure := &VolumeError{APIError: &APIError{Verb: verb, Status: resp.StatusCode, Message: wire.Error}} + failure := &VolumeError{APIError: &APIError{Verb: verb, Status: resp.StatusCode, Message: reply.Error}} if complete { view := out view.VolumesWant, view.Volumes, view.States = slices.Clone(out.VolumesWant), slices.Clone(out.Volumes), slices.Clone(out.States) diff --git a/sdk/langchain/cocoonsandbox_langchain/toolkit.py b/sdk/langchain/cocoonsandbox_langchain/toolkit.py index e6fd1dfe7..f149c6c04 100644 --- a/sdk/langchain/cocoonsandbox_langchain/toolkit.py +++ b/sdk/langchain/cocoonsandbox_langchain/toolkit.py @@ -104,9 +104,11 @@ def close(self) -> None: with self._lock: sb, self._sb = self._sb, None self._closed = True - if sb is not None: - sb.close() - self._client.close() + try: + if sb is not None: + sb.close() + finally: + 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/langchain/tests/test_toolkit.py b/sdk/langchain/tests/test_toolkit.py index 4e882fcfd..eea4ae071 100644 --- a/sdk/langchain/tests/test_toolkit.py +++ b/sdk/langchain/tests/test_toolkit.py @@ -167,6 +167,17 @@ def test_a_failed_release_never_hides_the_block_error(monkeypatch): kit.get_tools()[0].invoke({"command": "x"}) +def test_a_failed_release_still_closes_the_client(monkeypatch): + kit, fake = hooked(monkeypatch) + closed = [] + monkeypatch.setattr(kit._client, "close", lambda: closed.append(True)) + fake.close_error = APIError("release", 500, "boom") + kit.get_tools()[0].invoke({"command": "x"}) + with pytest.raises(APIError): + kit.close() + assert closed == [True], closed + + def test_a_stalled_agent_upgrade_comes_back_as_a_tool_error(): server = socket.create_server(("127.0.0.1", 0)) addr = f"127.0.0.1:{server.getsockname()[1]}" diff --git a/sdk/openai/cocoonsandbox_openai/adapter.py b/sdk/openai/cocoonsandbox_openai/adapter.py index 048666d17..a5cad4c6f 100644 --- a/sdk/openai/cocoonsandbox_openai/adapter.py +++ b/sdk/openai/cocoonsandbox_openai/adapter.py @@ -144,7 +144,11 @@ async def create( options: CocoonSandboxClientOptions, ) -> SandboxSession: client = Client(options.addr, api_token=options.api_token) - sb = await asyncio.to_thread(client.new, options.template, net=options.net, ttl_seconds=options.ttl_seconds) + try: + sb = await asyncio.to_thread(client.new, options.template, net=options.net, ttl_seconds=options.ttl_seconds) + except BaseException: + client.close() + raise try: if manifest is None: manifest = Manifest(root="/workspace") @@ -162,6 +166,7 @@ async def create( except BaseException: with contextlib.suppress(Exception): # the construction error is the one worth reading await asyncio.to_thread(sb.close) + client.close() raise session = CocoonSandboxSession.from_state(state) session._client, session._sb = client, sb @@ -172,9 +177,11 @@ async def delete(self, session: SandboxSession) -> SandboxSession: if not isinstance(inner, CocoonSandboxSession): raise TypeError("CocoonSandboxClient.delete expects a CocoonSandboxSession") await inner._shutdown_backend() - await asyncio.to_thread(inner._sandbox().close) - if inner._client is not None: - inner._client.close() + try: + await asyncio.to_thread(inner._sandbox().close) + finally: + if inner._client is not None: + inner._client.close() return session async def resume(self, state: SandboxSessionState) -> SandboxSession: diff --git a/sdk/openai/tests/test_adapter.py b/sdk/openai/tests/test_adapter.py index eda52dd9f..1ec76fb07 100644 --- a/sdk/openai/tests/test_adapter.py +++ b/sdk/openai/tests/test_adapter.py @@ -9,7 +9,7 @@ from pathlib import Path import pytest -from cocoonsandbox import ProtocolError, SilkdError +from cocoonsandbox import APIError, ProtocolError, SilkdError from cocoonsandbox_openai import CocoonSandboxClient, CocoonSandboxClientOptions, CocoonSandboxSessionState CLAIMS: list = [] @@ -208,8 +208,10 @@ def close(self): if release_fails: raise RuntimeError("node gone") + closed = [] monkeypatch.setattr("cocoonsandbox_openai.adapter.resolve_snapshot", bad_snapshot) monkeypatch.setattr("cocoonsandbox.Sandbox.close", close) + monkeypatch.setattr("cocoonsandbox.Client.close", lambda self: closed.append(self)) async def go(): client = CocoonSandboxClient() @@ -218,6 +220,24 @@ async def go(): asyncio.run(go()) assert released == ["sb_1"] + assert len(closed) == 1, closed + + +def test_create_closes_its_client_when_the_claim_fails(node, monkeypatch): + closed = [] + + def refuse(self, *args, **kwargs): + raise APIError("claim", 503, "draining") + + monkeypatch.setattr("cocoonsandbox.Client.new", refuse) + monkeypatch.setattr("cocoonsandbox.Client.close", lambda self: closed.append(self)) + + async def go(): + with pytest.raises(APIError): + await CocoonSandboxClient().create(options=CocoonSandboxClientOptions(addr=node)) + + asyncio.run(go()) + assert len(closed) == 1, closed async def _session_with(node, monkeypatch, fake): diff --git a/sdk/python/cocoonsandbox/conn.py b/sdk/python/cocoonsandbox/conn.py index 6ebfef63d..4b92afcf3 100644 --- a/sdk/python/cocoonsandbox/conn.py +++ b/sdk/python/cocoonsandbox/conn.py @@ -274,7 +274,7 @@ def _buffered_reader(sock: socket.socket) -> BinaryIO: def _handshake_error(what: str, exc: OSError) -> SandboxError: - if isinstance(exc, (TimeoutError, socket.timeout)): + if isinstance(exc, TimeoutError): return SandboxTimeout(f"{what} timed out") return ProtocolError(f"{what}: {exc}") diff --git a/sdk/python/cocoonsandbox/sandbox.py b/sdk/python/cocoonsandbox/sandbox.py index 3b2df8b36..e031590c4 100644 --- a/sdk/python/cocoonsandbox/sandbox.py +++ b/sdk/python/cocoonsandbox/sandbox.py @@ -803,20 +803,21 @@ def _pump_stdio( if t == "started" and on_started: on_started(cast(int, _need(frame, "pid"))) elif t == "stdout" and on_stdout: - _deliver(on_stdout, _need(frame, "data")) + _deliver(conn, on_stdout, _need(frame, "data")) elif t == "stderr" and on_stderr: - _deliver(on_stderr, _need(frame, "data")) + _deliver(conn, on_stderr, _need(frame, "data")) elif t == "exit": return cast(int, _need(frame, "code")) return None -def _deliver(callback: Callable[[bytes], object], data: bytes) -> None: - """Runs a caller's output callback; an SDK error it raises never sends the running call to another node.""" +def _deliver(conn: Conn, callback: Callable[[bytes], object], data: bytes) -> None: + """Runs a caller's output callback; an SDK error it raises ends the stream and never moves the call.""" try: callback(data) except SandboxError as exc: exc._owner_gone = None + conn.abort() raise diff --git a/sdk/python/tests/test_keepalive.py b/sdk/python/tests/test_keepalive.py index 536e997bd..78a59942f 100644 --- a/sdk/python/tests/test_keepalive.py +++ b/sdk/python/tests/test_keepalive.py @@ -91,6 +91,18 @@ def test_a_refused_stream_open_keeps_its_connection(relay): assert agent.upgrades == 1 +def test_a_callback_error_drops_the_stream(relay): + agent, sb = relay() + + def refuse(_: bytes) -> None: + raise SilkdError("io", "push refused") + + with pytest.raises(SilkdError): + sb.run(["echo", "42"], on_stdout=refuse) + assert sb.exec("echo", "42") == "42\n" + assert agent.upgrades == 2, agent.upgrades + + def test_keep_alive_off_dials_per_call(relay): agent, sb = relay(keep_alive=0) for _ in range(3): diff --git a/silkd/Dockerfile b/silkd/Dockerfile index 0ec35cc0f..4f1f3820a 100644 --- a/silkd/Dockerfile +++ b/silkd/Dockerfile @@ -4,9 +4,7 @@ FROM rust:1.99.1-slim-bookworm@sha256:24707e70a4f1e23d9a7b25202ec43126d28a52294b96aad5c670df1fd1f1d400 AS build ARG TARGETARCH WORKDIR /silkd -# Before any rustup/cargo call: the pin, not the base image's default, is the -# compiler. A digest bump that moves ahead of the pin then costs a toolchain -# download instead of silently building the guest binary with another rustc. +# The pin, not the base image's default, is the compiler: copy it before any rustup or cargo call. COPY rust-toolchain.toml ./ RUN sed -i '/^components/d;/^targets/d' rust-toolchain.toml RUN case "$TARGETARCH" in \ diff --git a/silkd/src/net.rs b/silkd/src/net.rs index f211a1cdc..801d01dfc 100644 --- a/silkd/src/net.rs +++ b/silkd/src/net.rs @@ -10,7 +10,7 @@ const LANE_FILE: &str = "/etc/silkd-lane"; /// The lane file's verdict once read: -1 not yet present, 0 direct, 1 relay. static LANE: AtomicI8 = AtomicI8::new(-1); -// probed once: the NIC set is fixed for the guest's life +/// Probed once: the NIC set is fixed for the guest's life. static DEVICE_BACKED: LazyLock = LazyLock::new(|| { let Ok(entries) = std::fs::read_dir("/sys/class/net") else { return true; @@ -18,11 +18,12 @@ static DEVICE_BACKED: LazyLock = LazyLock::new(|| { entries.flatten().any(|e| e.path().join("device").exists()) }); /// SILKD_NET decoded once: operator config, fixed at service start. -static SILKD_NET: LazyLock = LazyLock::new(|| match std::env::var("SILKD_NET").as_deref() { - Ok("none") => 0, - Ok("egress") => 1, - _ => -1, -}); +static SILKD_NET: LazyLock> = + LazyLock::new(|| match std::env::var("SILKD_NET").as_deref() { + Ok("none") => Some(false), + Ok("egress") => Some(true), + _ => None, + }); /// Reports whether execs route directly: a NIC the host did not mark as relayed. pub fn routes_directly() -> bool { @@ -41,12 +42,7 @@ pub fn routes_directly() -> bool { /// Reports whether the guest can reach a network; only a `device`-backed interface counts, since the kernel auto-creates virtual tunnels. fn has_egress() -> bool { - match *SILKD_NET { - 0 => return false, - 1 => return true, - _ => {} - } - *DEVICE_BACKED + SILKD_NET.unwrap_or_else(|| *DEVICE_BACKED) } /// Reads the lane file: `Some(true)` for relay, `Some(false)` for direct, `None` while the host has not written it. diff --git a/silkd/src/net_egress.rs b/silkd/src/net_egress.rs index 33bc97c77..6b850bb70 100644 --- a/silkd/src/net_egress.rs +++ b/silkd/src/net_egress.rs @@ -80,12 +80,12 @@ pub async fn serve_datagrams(listen: SocketAddrV4, host_vsock_port: u32) -> io:: continue; } }; - let (Ok(slot), Ok(len)) = (slots.clone().try_acquire_owned(), u16::try_from(n)) else { + let (Ok(slot), Ok(len)) = (Arc::clone(&slots).try_acquire_owned(), u16::try_from(n)) else { continue; }; buf[..2].copy_from_slice(&len.to_be_bytes()); let framed = buf[..2 + n].to_vec(); - let sock = sock.clone(); + let sock = Arc::clone(&sock); tokio::spawn(async move { let _slot = slot; let exchange = async { diff --git a/silkd/src/sysutil.rs b/silkd/src/sysutil.rs index bd801ef55..cd16b9c43 100644 --- a/silkd/src/sysutil.rs +++ b/silkd/src/sysutil.rs @@ -1,4 +1,4 @@ -//! Small OS helpers; the crate's unsafe work lives here. +//! Small OS helpers and syscall wrappers. use std::borrow::Cow; use std::io::Read; @@ -16,16 +16,6 @@ use nix::fcntl::{FcntlArg, OFlag, fcntl}; use tokio::io::unix::AsyncFd; use tokio::process::Command; -static TMP_SEQ: AtomicU64 = AtomicU64::new(0); - -/// The FORWARDED values in silkd's own environment, read once: the unit environment is fixed. -static PROXY_VARS: LazyLock> = - LazyLock::new(|| snapshot_proxy(|key| std::env::var(key).ok())); -static DIRECT: LazyLock> = - LazyLock::new(|| compose_env(true, &PROXY_VARS)); -static RELAYED: LazyLock> = - LazyLock::new(|| compose_env(false, &PROXY_VARS)); - /// Forwarded into every exec that has no direct route: the loopback proxy relay is its only way out. const FORWARDED: [&str; 6] = [ "http_proxy", @@ -39,6 +29,16 @@ const FORWARDED: [&str; 6] = [ /// The claim's environment the host writes, in systemd EnvironmentFile syntax. const CLAIM_ENV_PATH: &str = "/run/silkd.env"; +static TMP_SEQ: AtomicU64 = AtomicU64::new(0); + +/// The FORWARDED values in silkd's own environment, read once: the unit environment is fixed. +static PROXY_VARS: LazyLock> = + LazyLock::new(|| snapshot_proxy(|key| std::env::var(key).ok())); +static DIRECT: LazyLock> = + LazyLock::new(|| compose_env(true, &PROXY_VARS)); +static RELAYED: LazyLock> = + LazyLock::new(|| compose_env(false, &PROXY_VARS)); + /// A suffix unique within this process, enough to name a temp file; not cryptographic. pub fn tmp_suffix() -> String { format!( diff --git a/silkd/tests/common/mod.rs b/silkd/tests/common/mod.rs index f566be0f7..fdf6bbaa4 100644 --- a/silkd/tests/common/mod.rs +++ b/silkd/tests/common/mod.rs @@ -20,8 +20,9 @@ pub const DEADLINE: Duration = Duration::from_secs(30); pub type FrameWriter = WriteHalf; pub type FrameLines = Lines>>; +pub type Connection = (FrameWriter, FrameLines, JoinHandle>); -pub fn connect(state: &Arc) -> (FrameWriter, FrameLines, JoinHandle>) { +pub fn connect(state: &Arc) -> Connection { let (client, server) = tokio::io::duplex(1 << 20); let state = Arc::clone(state); let (sr, sw) = tokio::io::split(server); @@ -30,10 +31,7 @@ pub fn connect(state: &Arc) -> (FrameWriter, FrameLines, JoinHandle, - first: Value, -) -> (FrameWriter, FrameLines, JoinHandle>) { +pub async fn connect_with(state: &Arc, first: Value) -> Connection { let (mut cw, lines, handle) = connect(state); send(&mut cw, first).await; (cw, lines, handle) diff --git a/silkd/tests/lsp_e2e.rs b/silkd/tests/lsp_e2e.rs index 590f5b882..3533a9e22 100644 --- a/silkd/tests/lsp_e2e.rs +++ b/silkd/tests/lsp_e2e.rs @@ -13,8 +13,8 @@ use tokio::io::AsyncWriteExt; use tokio::time::timeout; use common::{ - DEADLINE, FrameLines, FrameWriter, alive, assert_error, b64, connect_with, decode, - expect_frame, one, send, type_of, under_deadline, + Connection, DEADLINE, alive, assert_error, b64, connect_with, decode, expect_frame, one, send, + type_of, under_deadline, }; const FAKE_SERVER: &str = r#"#!/bin/sh @@ -54,14 +54,7 @@ async fn started_server(lang: &str, server_body: &str) -> (TempDir, Arc, (env, state, server_id) } -async fn lsp_stream( - state: &Arc, - server_id: &str, -) -> ( - FrameWriter, - FrameLines, - tokio::task::JoinHandle>, -) { +async fn lsp_stream(state: &Arc, server_id: &str) -> Connection { let (cw, mut out, handle) = connect_with(state, json!({"op":"lsp_request","server_id":server_id})).await; expect_frame(&mut out, "ready").await; diff --git a/silkd/tests/pty_e2e.rs b/silkd/tests/pty_e2e.rs index 586c1fe84..a0bbe8431 100644 --- a/silkd/tests/pty_e2e.rs +++ b/silkd/tests/pty_e2e.rs @@ -18,7 +18,7 @@ use tokio::io::{AsyncWrite, AsyncWriteExt}; use tokio::sync::mpsc; use common::{ - FrameLines, FrameWriter, alive, assert_error, b64, connect_with, decode, frames_until, one, + Connection, FrameLines, alive, assert_error, b64, connect_with, decode, frames_until, one, proc_entry, request, send, type_of, wait_for_path, }; @@ -47,13 +47,7 @@ async fn read_until(lines: &mut FrameLines, pred: impl Fn(&Value) -> bool) -> Va frames_until(lines, pred).await.pop().unwrap() } -async fn open_pty( - state: &Arc, -) -> ( - FrameWriter, - FrameLines, - tokio::task::JoinHandle>, -) { +async fn open_pty(state: &Arc) -> Connection { connect_with(state, json!({"op":"pty_open","cols":80,"rows":24})).await } diff --git a/silkd/tests/tun_e2e.rs b/silkd/tests/tun_e2e.rs index 71d2b45ec..66c04ceec 100644 --- a/silkd/tests/tun_e2e.rs +++ b/silkd/tests/tun_e2e.rs @@ -12,13 +12,12 @@ use std::thread::JoinHandle; use std::time::{Duration, Instant}; use nix::sched::{CloneFlags, unshare}; -use tokio::io::{AsyncBufReadExt, AsyncReadExt, AsyncWriteExt, BufReader}; -use tokio::net::{TcpListener, TcpStream, UdpSocket}; - use silkd::sysutil; use silkd::tun::{ self, FLAG_ON, Flow, HELLO_LEN, Hello, Io, PEER_ADDR, Tun, VERSION, VNET_HDR_LEN, }; +use tokio::io::{AsyncBufReadExt, AsyncReadExt, AsyncWriteExt, BufReader}; +use tokio::net::{TcpListener, TcpStream, UdpSocket}; const CLAIM_ADDR: Ipv4Addr = Ipv4Addr::new(100, 65, 0, 7); const UPLOAD: u16 = 5001; @@ -63,9 +62,10 @@ fn tcp_udp_and_gso_cross_a_rewriting_host() { for io in [Io::Uring, Io::Epoll] { let up = Arc::new(Stats::default()); let down = Arc::new(Stats::default()); - let (seen, accepts) = with_link(io, true, Some((up.clone(), down.clone())), |_| { - guest_workload() - }); + let (seen, accepts) = + with_link(io, true, Some((Arc::clone(&up), Arc::clone(&down))), |_| { + guest_workload() + }); assert_eq!(seen.upload_peer, CLAIM_ADDR.to_string(), "{io:?}"); assert_eq!(seen.echo_peers, vec![CLAIM_ADDR.to_string(); 16], "{io:?}"); assert_eq!(seen.downloaded, BULK, "{io:?}"); @@ -107,9 +107,12 @@ fn full_checksums_cross_when_offloads_are_off() { for io in [Io::Uring, Io::Epoll] { let up = Arc::new(Stats::default()); let down = Arc::new(Stats::default()); - let (seen, _) = with_link(io, false, Some((up.clone(), down.clone())), |_| { - guest_workload() - }); + let (seen, _) = with_link( + io, + false, + Some((Arc::clone(&up), Arc::clone(&down))), + |_| guest_workload(), + ); assert_eq!(seen.upload_peer, CLAIM_ADDR.to_string(), "{io:?}"); assert_eq!(seen.echo_peers, vec![CLAIM_ADDR.to_string(); 16], "{io:?}"); assert_eq!(seen.downloaded, BULK, "{io:?}"); @@ -226,7 +229,7 @@ fn with_link( io, offloads, relay_threads.is_empty(), - stop.clone(), + Arc::clone(&stop), ); let out = in_netns(move || { let tun = Tun::open("tun0").unwrap();