diff --git a/README.md b/README.md index 04468e8d9..a69644d05 100644 --- a/README.md +++ b/README.md @@ -35,7 +35,7 @@ performance) — source in - `sandboxd/` — per-node control plane (Go): warm pools refilled from golden snapshot exports (online-retunable), claim/release/renew/hibernate/fork/ promote/checkpoint HTTP API, operator-catalog dataset volumes (read-only or - writable; read-only ones change on a running claim), signed preview URLs, + writable, mounted or attach-only, with live add and remove), signed preview URLs, the HTTP-upgrade byte relays to silkd and to any guest port, read-only file access to hibernated or archived claims from a warm VM of their pool (opt-in per pool `reader`), claim env with diff --git a/docs/deploy.md b/docs/deploy.md index 1ba503659..154b29c5b 100644 --- a/docs/deploy.md +++ b/docs/deploy.md @@ -145,7 +145,7 @@ gets the intersection of the two. | `reader` (pool entry) | `false` | Serve read-only file calls on this pool's hibernated or archived claims from a warm VM of the same pool instead of waking them; needs `warm` of at least 1 and a cocoon whose `vm disk attach` takes `--from-snapshot` and `--from-dir`. See [Reading sleeping claims](#reading-sleeping-claims-without-waking-them). Config-owned like `capture_trim`: `PUT /v1/pools` rejects it | | `egress_classes` | unset | named tenant egress layers: `[{"name": "desk", "egress": {…}, "egress_upstream_env": "DESK_UPSTREAM"}]`. A tenant names one with `egress_class` through the [tenant API](sandboxd-api.md#tenants-v1tenants) and its claims take that policy intersected with the pool's. `egress` is required; `intercept` stays a pool-rule flag. Reloadable: a class's policy reaches live claims on their next request. A class removed while tenants still name it leaves them with no egress and is logged at load and at reload | | `egress_upstream_env` (pool or egress class entry) | unset | names a node environment variable holding the default upstream proxy URL for the pool's or class's claims (class before pool, a claim's own entry before both); needs `egress_upstream`, and the URL must be on its `allow`. Read and checked at startup. Config-owned like `egress`: `PUT /v1/pools` rejects it | -| `replicate_seconds` (pool entry) | 0 (off) | snapshot every running claim of this pool to the checkpoint store this often and point the claim's registry row at the snapshot, so another node restores the claim, memory and processes included, when this node is lost (see [running claims across node loss](cluster.md#running-claims-across-node-loss)). Each round pauses the guest for the length of a memory and disk capture. A restore resumes from the last committed replica and loses everything after it: up to the longer of the interval and one replication pass, plus the 5-second tick, plus one more pass, since a claim can run first in one pass and last in the next. A pass covers all the node's due claims of the replicating pools with bounded concurrency, so it grows with claim count and guest memory; the first round of a claim after a wake or adoption uploads the whole claim, so that pass is the long pass. The claim's row carries no retention deadline while it runs. Requires `archive_failover` and a `none`-lane pool, both checked at config load; a claim with volumes is skipped. Config-owned like `egress`: `PUT /v1/pools` rejects it. Restart only: a claim is replicated by its pool's setting from the moment it is claimed, so a reload that changes it is refused (a pool restarted with 0 drops its claims' rows and replicas on the first pass) | +| `replicate_seconds` (pool entry) | 0 (off) | snapshot every running claim of this pool to the checkpoint store this often and point the claim's registry row at the snapshot, so another node restores the claim, memory and processes included, when this node is lost (see [running claims across node loss](cluster.md#running-claims-across-node-loss)). Each round pauses the guest for the length of a memory and disk capture. A restore resumes from the last committed replica and loses everything after it: up to the longer of the interval and one replication pass, plus the 5-second tick, plus one more pass, since a claim can run first in one pass and last in the next. A pass covers all the node's due claims of the replicating pools with bounded concurrency, so it grows with claim count and guest memory; the first round of a claim after a wake or adoption uploads the whole claim, so that pass is the long pass. The claim's row carries no retention deadline while it runs. Requires `archive_failover` and a `none`-lane pool, both checked at config load; a claim with volumes is skipped. Config-owned like `egress`: `PUT /v1/pools` rejects it. Restart only: a claim is replicated by its pool's setting from the moment it is claimed, so a reload that changes it is refused (a pool restarted with 0 starts retiring its replicas while the registry remains enabled; see [Disabling archive failover](#disabling-archive-failover)) | | `storage` (pool entry) | unset (cocoon's default, 10G) | size of the copy-on-write disk the pool's golden is cold-booted with, in cocoon's own `--storage` spelling (`"40G"`, `"40GiB"`, `"40Gi"`: Docker/Kubernetes units, all binary) and passed to it verbatim; every clone, fork, checkpoint branch and promoted template of the pool inherits it. At least cocoon's default of 10G. Equal sizes spelled differently build the same golden; a changed size rebuilds the golden on the next build, and claims already out keep their disk. Config-owned like `egress`: `PUT /v1/pools` rejects it | | `max_claims` | 0 (unlimited) | node-wide cap on live claims; claim/fork/branch requests beyond it answer 429 with the pool state unharmed (on a cluster a non-volume claim tries a warm-peer redirect first; a volume claim answers the 429 with no redirect) | | `audit_log` | false | append every relayed frame that opens an RPC — its op + addressing fields, never payloads — to `/audit.jsonl`; continuation `data`, `data_end`, `stdin`, and `stdin_close` frames pass through unrecorded. The file is size-rotated with one `.1` backup. Records are `{t, id, op}` plus whichever addressing fields the op carries (`argv`, `path`, `dest`, `from`, `to`, `url`, `session`, `port`), plus `method` (`GET`, `CONNECT`, `SOCKS5`, …), `decision` (`allow`, `deny`, `rate_limited`), `secret` (the ref name, never its value), `upstream` (the [upstream proxy](egress.md#upstream-proxies)'s `host:port`, when one carried it), `tenant`, and on an allowed connection `src_addr` and `dst_addr` (the node-side and destination `ip:port` of the upstream TCP connection, the tunnel target through an upstream proxy) on `egress` records, which a relayed frame cannot set; preview accesses record as op `preview`, one per request; a tenant change records as op `tenants` with the `added`, `changed` and `removed` names, never a token. An opening frame whose first line exceeds 4 KiB records as op `oversized` with no addressing fields | @@ -248,9 +248,13 @@ the pool had `reader`, and its reads wake the claim as without the feature. ### Dataset volumes -Running claims can change mounted read-only catalog entries through the +Running claims can add or remove mounted and attach-only catalog entries in +read-only or writable mode through the [GET, PUT, and PATCH volume API](sandboxd-api.md#get-v1sandboxesidvolumes) -and both SDKs. The image must already exist on the owner node. Live changes +and both SDKs. Changing a held entry's descriptor requires successful removal +first. The [live volume contract](sandboxd-api.md#put-v1sandboxesidvolumes) +defines admission policy and dirty-marker handling. +The image must already exist on the owner node. Live changes require Cocoon v0.7.2 or newer for idempotent disk attach and detach. PUT and PATCH reject an older release, or a runtime whose version cannot be queried, with 503 before any intent, hold, or device change; a dev build is assumed @@ -337,7 +341,8 @@ below. With `directio=off`, readers share the host page cache; use `directio=on` when cache interference matters. A dataset mounted into an egress-lane sandbox can be uploaded wherever that tenant's egress policy permits, so treat the ACL and egress policy as one access -decision. Image replication, detach, and refcounting remain out of scope. +decision. Operators remain responsible for image distribution and cross-node +ownership. sandboxd tracks local holds and handles live detach through the volume API. #### Writable dataset volumes @@ -705,6 +710,36 @@ drains the warm pools; poll `GET /v1/info` until `claimed` reaches zero (or let the leases expire), then stop sandboxd. `DELETE /v1/drain` uncordons. The drain is not persisted — a restarted node serves again. +### Disabling archive failover + +Stop admitting claims that use `on_expire: archive`. Set the node and affected +pools' `archive_after_seconds` to `0`, and use +[`renew`](sandboxd-api.md#post-v1sandboxesidrenew) to switch existing active claims +to `on_expire: destroy` while keeping their leases valid. These changes stop +idle-time and expiry-triggered archives. Before setting `archive_failover` to `false`: + +1. Set every affected pool's `replicate_seconds` to `0` and restart its nodes + with failover and the registry still enabled. Keep the registry reachable + while replica retirement retries. +2. Check each node's `/claims.json` for affected claims. Wait until + `replica_ck` and `replica_pending` are absent after successful journal writes. + Use `sync_claims: true` when completion must survive power loss. +3. Wake or release existing registered archives while failover remains enabled. + Wait for `archive_pending` to clear in the journal. Verify that the affected + IDs have no rows in PostgreSQL's + `sandboxd_archives` table for your `meta_store.cell`. A consumed row is a + non-adoptable tombstone; wait for its cleanup retries to finish too. +4. Set `archive_failover` to `false` and restart the nodes. + +The public API omits replica journal fields. A fixed delay or an absence of +warnings does not prove retirement. Disabling failover first stops replica +retirement and leaves residual references that reject new live RW attachments. +A local archive can still wake and clear `archive_ck`, but that does not clear a +retained `replica_pending` or retire an old shared row. Release can remove local +state without reconciling that row. Re-enabling failover against stale rows can +fence or drop a local claim if another node has adopted it. Resolve registry +ownership before serving that claim again. + ### Shared tenant database With `meta_store` set, every node reads tenants from one PostgreSQL table diff --git a/docs/live-volumes.md b/docs/live-volumes.md index 589e504d0..1d42f0038 100644 --- a/docs/live-volumes.md +++ b/docs/live-volumes.md @@ -10,7 +10,8 @@ v0.7.2 or newer and set `SANDBOX_ADDR` and `SANDBOX_API_TOKEN` for the client. A sandbox token alone cannot manage volumes. Entries [registered at runtime](sandboxd-api.md#put-v1volumesname) work the -same way. +same way. The API also supports +[writable and attach-only entries](sandboxd-api.md#put-v1sandboxesidvolumes). Pause application reads before replacement. Close files, mmap mappings, cwd, and bind mount references on the old disk. Replacement unmounts the old disk @@ -61,9 +62,10 @@ with client.new("rt:24.04", ttl_seconds=900) as sb: A successful run reads `v1`, then `v2`, and preserves every identity assertion. The repeated full target verifies that a settled mount can be retained. -This example starts without other volumes. If your sandbox has writable or -attach-only volumes, retain them in a full PUT or use PATCH to replace only -the dataset names. `None`, a nil Go slice, and a missing `volumes` field do not +This example starts without other volumes. Retain any volumes you intend to +keep in a full PUT, or use PATCH to replace only the dataset names. All mounted +and attach-only entries support live removal in either mode. +`None`, a nil Go slice, and a missing `volumes` field do not mean an empty target. Go uses the same owning handle and client API token. diff --git a/docs/sandboxd-api.md b/docs/sandboxd-api.md index 3262ad0e2..783da5c97 100644 --- a/docs/sandboxd-api.md +++ b/docs/sandboxd-api.md @@ -87,13 +87,12 @@ Auth: `Authorization: Bearer ` (when configured). defaults to `/volumes/`; a custom value must be absolute and clean, outside the guest OS tree, unique, and non-nesting within the request. `mode` is `"ro"` (default, omitted) or `"rw"`; `"rw"` requires the catalog - entry's `writable: true` (see [deploy](deploy.md#dataset-volumes)) -- `volumes_attach_only` (default `false`) attaches every requested volume - without mounting it, handing the whole mount contract to the workload. It - requires at least one volume, and rejects any entry carrying a `mount` — - the path is meaningless when the caller mounts the device. Everything below - describes the default, eager behaviour, which is completely unchanged; the - attach-only contract is in [its own section](#attach-only-volumes) + entry's `writable: true` (see [deploy](deploy.md#dataset-volumes)). + `attach_only: true` attaches that entry without mounting it and requires an + empty or omitted `mount`. Mounted and attach-only entries can share a list +- `volumes_attach_only` (default `false`) is the legacy claim-wide form of + `attach_only`. The server expands it onto every entry before validation. + It requires at least one volume and rejects a nonempty `mount`. Success: @@ -135,12 +134,14 @@ an existing populated guest directory for the claim's life. ### Attach-only volumes -`"volumes_attach_only": true` stops after the attach. The device appears in -the guest, nothing is mounted, and the echoed entries carry no `mount` key: +`"attach_only": true` stops after the attach for that entry. The device appears +in the guest without a mount. Responses preserve `attach_only: true` and omit +`mount`: ```json {"id": "sb_…", "token": "…", - "volumes": [{"name": "imagenet"}, {"name": "scratch-db", "mode": "rw"}]} + "volumes": [{"name": "imagenet", "attach_only": true}, + {"name": "scratch-db", "mode": "rw", "attach_only": true}]} ``` Find each device by its virtio serial, which is the catalog name — poll this, @@ -221,7 +222,7 @@ an unknown and a forbidden volume return the same error text. Errors: 400 unknown template axis, metadata over a bound (the message names it), an unknown `on_expire`, invalid/duplicate volumes, -`volumes_attach_only` with no volumes or with an entry carrying a `mount`, +`volumes_attach_only` with no volumes, or an attach-only entry with a nonempty `mount`, `mode: "rw"` against a non-writable entry, or a volume that is unknown or forbidden (the latter two are deliberately indistinguishable), or @@ -354,22 +355,32 @@ Replaces the full desired set and converges synchronously. Auth matches GET. The owner tenant's catalog access list applies to newly admitted names, including for root. Retaining, repairing, or removing held names does not reauthorize them. A new image must exist on the owning node. The request must include a non-null `volumes` -array. An explicit empty array removes all mutable entries. +array. An explicit empty array removes all volumes. ```json {"volumes": [{"name": "imagenet", "mount": "/datasets/training"}]} ``` -Only mounted read-only catalog entries can change live (`mode` omitted or -`ro`). A PUT must retain every existing writable or attach-only entry unchanged. -The claim-time name, mount, and eight-volume limits apply to the desired set. -An omitted mount defaults to `/volumes/`, except that a held attach-only -entry is retained by sending it as the claim returned it, without `mount`. To -change an existing name's -mount or mode, first remove that name and wait for successful convergence. -An identical target can be retried. Claims with `on_expire: archive` must change -that policy before adding a volume. - +Mounted read-only, mounted writable, attach-only read-only, and attach-only +writable entries can change live, including mixed lists. The claim-time name, +mount, and eight-volume limits apply. An omitted mount defaults to +`/volumes/` unless that entry has `attach_only: true`. Responses carry +that flag directly, so a GET target can be sent back with PUT. Omitting the flag +or sending `false` does not inherit an existing attachment's shape. + +To change a held name's mount, mode, or `attach_only`, first remove the name and +wait for successful convergence. An identical target can be retried. Claims +with `on_expire: archive` must change that policy before adding a volume. + +New live writable attachments require a non-replicating pool, including a +promoted template's source pool. A replicating pool returns 409 before changing +intent, holds, or devices. Initial claim-time writable volumes remain supported. +An existing or pending replica, a pending archive, or this claim's unacknowledged checkpoint +retirement, returns 503 with `Retry-After`. A replica or archive intent left +with `archive_failover` disabled returns 409 because it cannot retire through +the registry. A retained local checkpoint reference alone still returns 503. +Volume mutation does not evict remote replicas. Read-only additions, exact +retries of held writers, and removals remain available under these restrictions. The sandbox must be running. Paused, archived, and failed sandboxes return 409 without waking. A failed sandbox holding volumes is release-only after its VMM exits; PUT with an empty target cannot restore its processes. Missing, invalid, @@ -390,13 +401,30 @@ unchanged. An empty patch retries the current desired set. ``` Adding an identical entry or removing an absent valid name is harmless. -Duplicate names, a name in both arrays, or changes to writable or attach-only -entries return 400. Removal refuses busy or foreign mounts and uses neither +Duplicate names, a name in both arrays, or changes to a held entry's mount, +mode, or `attach_only` return 400. Removal refuses busy or foreign mounts and uses neither lazy nor forced unmount. A foreign mount, another bind mount, or a completed ordinary unmount with nonzero exit returns 409 with the guest mount path and a safe per-volume reason. Guest transport failures remain server errors; timeouts return 504 and cancellation returns 408. +A mounted writable attachment marks the image dirty before the guest can write. +Live removal clears the marker only when this call observes and normally +unmounts the exact writable whole-device filesystem, finds no residual device +mounts, and confirms host detach. Both mount and superblock options must be +writable. A missing device or pre-unmounted filesystem does not prove clean +unmount. The marker remains on uncertainty and blocks subsequent readers. +Attach-only writers have no automatic dirty-marker protection. The caller owns +mounting, flushing, unmounting, recovery, and fencing before reuse. + +The per-entry field requires coordinated daemon and SDK releases. Older daemons +ignore `attach_only` in claims and reject it in live PUT and PATCH with 400. +Older Python SDKs reject the field when +returned descriptors are reused in requests. Responses without the field do +not safely roundtrip attach-only intent through new clients. The server still +accepts the legacy claim-wide request flag and normalizes old persisted empty +mounts when loading claims. + ### Partial failure and recovery PUT and PATCH return the GET response shape on success. A convergence error diff --git a/docs/sdk-python.md b/docs/sdk-python.md index d726d3664..ac483ebd7 100644 --- a/docs/sdk-python.md +++ b/docs/sdk-python.md @@ -161,13 +161,31 @@ peer advertising both the template and every volume; when that intersection is empty, one volume holder may self-verify access to a shared template store before provisioning. -## Changing read-only datasets +## Changing live volumes Use `sb.get_volumes()`, `sb.set_volumes([...])`, and `sb.patch_volumes(add=[...], remove=[...])` on the same running sandbox. These methods use the client's API token on the owning node; a sandbox token cannot -manage datasets. `set_volumes([])` clears mutable entries; `None` is rejected. -PUT must retain writable and attach-only entries. PATCH retains omitted names. +manage volumes. `set_volumes([])` detaches all entries; `None` is rejected. +PATCH retains omitted names. Parameters take lists of names or `Volume` mappings. +The SDK sends volume policy to sandboxd and preserves `attach_only` directly. +Use matching daemon and SDK releases for attach-only GET-to-PUT retries. + +```python +from __future__ import annotations + +from cocoonsandbox import Volume + +volumes: list[str | Volume] = [ + {"name": "dataset", "mount": "/datasets"}, + {"name": "work", "mode": "rw", "attach_only": True}, +] +view = sb.set_volumes(volumes) +``` + +The caller mounts, flushes, and unmounts attach-only devices. See the +[live volume contract](sandboxd-api.md#put-v1sandboxesidvolumes) for +writer exclusion, replica refusals, and dirty-marker limits. Each method takes `timeout=` (default: the client timeout, 120 s); a large replacement can need longer, and a call that times out cancels the server run, so retry the same target. diff --git a/docs/sdk.md b/docs/sdk.md index 2d66f2aae..278ea9d63 100644 --- a/docs/sdk.md +++ b/docs/sdk.md @@ -234,16 +234,17 @@ checkpoint branches support neither volumes nor a claim reference in this versio `Checkpoint.New`: the labels follow the [claim bounds](sandboxd-api.md#post-v1claim), and a branch never inherits its source's labels or expiry action. -For datasets chosen after the sandbox starts, set the desired read-only mounts -on the same claim: +For datasets chosen after the sandbox starts, set the desired volumes on +the same claim: ```go view, err := sb.SetVolumes(ctx, []sandbox.Volume{ {Name: "imagenet", Mount: "/datasets/training"}, + {Name: "work", Mode: "rw", AttachOnly: true}, }) view, err = sb.GetVolumes(ctx) view, err = sb.PatchVolumes(ctx, sandbox.VolumePatch{Remove: []string{"imagenet"}}) -// An explicit empty slice clears mutable mounts. A nil slice is rejected. +// An explicit empty slice detaches all volumes. A nil slice is rejected. view, err = sb.SetVolumes(ctx, []sandbox.Volume{}) ``` @@ -256,6 +257,13 @@ view = sb.patch_volumes(remove=["imagenet"]) view = sb.set_volumes([]) # None is rejected. ``` +`Volume.AttachOnly` leaves mounting and unmounting to the caller. Each entry +can independently use `Mode: "rw"` and `AttachOnly: true`. The SDK preserves +these fields directly and leaves name, mode, and mount policy to sandboxd. +Use matching daemon and SDK releases for attach-only GET-to-PUT retries. +See the [live volume contract](sandboxd-api.md#put-v1sandboxesidvolumes) for +writer exclusion, replica refusals, and dirty-marker limits. + Each response contains `volumes_want`, `volumes`, `states`, and `converging`. Go exposes them as `SandboxVolumesResponse.VolumesWant`, `.Volumes`, `.States`, and `.Converging`. `VolumePhase` has `VolumeAttached`, `VolumePending`, and @@ -274,12 +282,15 @@ the handle retains its last known sets. Retry an interrupted operation with the same full PUT target, or set a new target to recover. `Client.Sandboxes` exposes `VolumesWant` and `Volumes` for inspection with an API token. -A PUT must retain existing writable and attach-only entries unchanged. PATCH -cannot remove or change them. Paused sandboxes are not woken, and images must -exist on the owner node. Admission holds cover the union of desired and applied -sets. A settled retained entry receives no new attach or mount calls. `converging` -is true only during an active request. Pending or detaching entries with false -require an explicit retry. Daemon restart replaces unfinished diagnostics with +PUT and PATCH can add or remove mounted and attach-only entries in either mode. +Changing a held entry's mount, mode, or `attach_only` requires successful +removal first. See the [live volume contract](sandboxd-api.md#put-v1sandboxesidvolumes) +for admission policy and dirty-marker handling. Paused sandboxes are not woken, +and images must exist on the owner node. Admission holds cover the union of +desired and applied sets. A settled retained entry receives no new attach or +mount calls. `converging` is true while a volume-change request is running. +Pending or detaching entries with `converging: false` require an explicit retry. +Daemon restart replaces unfinished diagnostics with an interrupted message. Capture is refused while either set is nonempty. Release cancels convergence and hands held volumes to teardown. See the [API recovery contract](sandboxd-api.md#partial-failure-and-recovery) and diff --git a/e2e/e2e_test.go b/e2e/e2e_test.go index cf305baa6..a97279b64 100644 --- a/e2e/e2e_test.go +++ b/e2e/e2e_test.go @@ -764,7 +764,7 @@ func TestAttachOnlyVolumeEndToEnd(t *testing.T) { if err != nil { t.Fatalf("attach-only claim: %v", err) } - if want := []sandbox.Volume{{Name: "scratch", Mode: "rw"}}; !slices.Equal(sb.Volumes, want) { + if want := []sandbox.Volume{{Name: "scratch", Mode: "rw", AttachOnly: true}}; !slices.Equal(sb.Volumes, want) { t.Errorf("claim volumes %+v, want %+v", sb.Volumes, want) } applied := []string{"attach:scratch:rw"} @@ -800,7 +800,7 @@ func TestAttachOnlyVolumeWireShape(t *testing.T) { { "attach-only omits the mount", `{"template":"rt:24.04","volumes_attach_only":true,"volumes":[{"name":"scratch","mode":"rw"}]}`, - map[string]any{"name": "scratch", "mode": "rw"}, + map[string]any{"name": "scratch", "mode": "rw", "attach_only": true}, }, { "eager claim is unchanged", diff --git a/e2e/fakeengine_test.go b/e2e/fakeengine_test.go index 9bbc9113a..79eccfe12 100644 --- a/e2e/fakeengine_test.go +++ b/e2e/fakeengine_test.go @@ -215,14 +215,14 @@ func (f *fakeEngine) SyncGuest(_ context.Context, _ string) error { return nil } -func (f *fakeEngine) DetachVolume(ctx context.Context, _, sock, name, mount string) error { +func (f *fakeEngine) DetachVolume(ctx context.Context, _, sock, name, mount string) (bool, error) { if err := f.UnmountVolume(ctx, sock, mount); err != nil { - return err + return false, err } f.mu.Lock() defer f.mu.Unlock() f.volumeOps = append(f.volumeOps, "detach:"+name) - return nil + return mount != "", nil } func (f *fakeEngine) volumeOpsLog() []string { diff --git a/e2e/volume_mutation_test.go b/e2e/volume_mutation_test.go index 5fcedef86..1ad074dae 100644 --- a/e2e/volume_mutation_test.go +++ b/e2e/volume_mutation_test.go @@ -52,7 +52,7 @@ func TestRejectedVolumeTargetExplainsRefusalWithoutChangingState(t *testing.T) { t.Fatal(checkErr) } _, err = sb.SetVolumes(t.Context(), []sandbox.Volume{{Name: "dataset"}, {Name: "dataset"}}) - if refusal, ok := errors.AsType[*sandbox.VolumeError](err); !ok || refusal.Status != http.StatusBadRequest || !strings.Contains(refusal.Message, "duplicate volume") { + if refusal, ok := errors.AsType[*sandbox.VolumeError](err); !ok || refusal.Status != http.StatusBadRequest || !strings.Contains(refusal.Message, "duplicates name") { t.Errorf("invalid target explanation: %v", err) } view, err := sb.GetVolumes(t.Context()) @@ -77,3 +77,57 @@ func TestEmptyVolumePatchChangesNothing(t *testing.T) { t.Fatalf("empty patch: %+v err=%v", got, err) } } + +func TestMixedLiveVolumeModesThroughSDK(t *testing.T) { + volumes := []sandbox.Volume{ + {Name: "mounted-ro"}, + {Name: "mounted-rw", Mode: "rw"}, + {Name: "device-ro", AttachOnly: true}, + {Name: "device-rw", Mode: "rw", AttachOnly: true}, + } + var catalog []config.VolumeSpec + for _, volume := range volumes { + catalog = append(catalog, config.VolumeSpec{ + Name: volume.Name, Path: writeVolumeImage(t, volume.Name+".img", "seed"), Writable: volume.Mode == "rw", + }) + } + stack := startTenantStack(t, "node-token", nil, catalog) + sb, err := stack.client.New(t.Context(), "rt:24.04") + if err != nil { + t.Fatal(err) + } + defer sb.Close() + view, err := sb.PatchVolumes(t.Context(), sandbox.VolumePatch{Add: volumes}) + if err != nil { + t.Fatal(err) + } + want := slices.Clone(volumes) + want[0].Mount, want[1].Mount = "/volumes/mounted-ro", "/volumes/mounted-rw" + if !slices.Equal(view.Volumes, want) || !slices.Equal(sb.Volumes, want) { + t.Fatalf("mixed SDK attachment = %+v, handle=%+v, want %+v", view, sb.Volumes, want) + } + status, body := rawJSON(t, stack, http.MethodGet, "/v1/sandboxes/"+sb.ID+"/volumes", "") + if status != http.StatusOK || strings.Count(string(body), `"attach_only":true`) != 4 { + t.Fatalf("mixed HTTP view lost explicit attach_only in desired or applied entries: status=%d body=%s", status, body) + } + view, err = sb.GetVolumes(t.Context()) + if err != nil { + t.Fatal(err) + } + view, err = sb.SetVolumes(t.Context(), view.VolumesWant) + if err != nil || !slices.Equal(view.Volumes, want) || view.Converging { + t.Fatalf("mixed GET-to-PUT = %+v, err=%v", view, err) + } + view, err = sb.SetVolumes(t.Context(), []sandbox.Volume{}) + if err != nil || len(view.Volumes) != 0 || len(view.VolumesWant) != 0 || view.Converging { + t.Fatalf("mixed removal = %+v, err=%v", view, err) + } + initial, err := stack.client.New(t.Context(), "rt:24.04", sandbox.WithVolumes(volumes...)) + if err != nil { + t.Fatalf("mixed initial claim: %v", err) + } + defer initial.Close() + if !slices.Equal(initial.Volumes, want) { + t.Fatalf("mixed initial claim = %+v, want %+v", initial.Volumes, want) + } +} diff --git a/sandboxd/engine/volume_live.go b/sandboxd/engine/volume_live.go index e1033a845..02b625186 100644 --- a/sandboxd/engine/volume_live.go +++ b/sandboxd/engine/volume_live.go @@ -31,7 +31,7 @@ func (e *LiveVolumeUnsupported) Error() string { } type volumeMount struct { - devNum, root, path, options string + devNum, root, path, options, superOptions string } // CheckLiveVolumeSupport rejects runtimes without idempotent live disk attach and detach. @@ -62,39 +62,50 @@ func (e *Engine) MountLiveVolume(ctx context.Context, vsockSocket, name, mount s return e.mountDevice(ctx, vsockSocket, device, name, mount, rw) } -// DetachVolume unmounts a guest volume before confirming host removal. -func (e *Engine) DetachVolume(ctx context.Context, vmName, vsockSocket, name, mount string) error { +// DetachVolume unmounts and detaches a guest volume; true means a writable mount closed cleanly. +func (e *Engine) DetachVolume(ctx context.Context, vmName, vsockSocket, name, mount string) (bool, error) { ctx, cancel := context.WithTimeout(ctx, VolumeMutationTimeout) defer cancel() device, found, err := e.findVolumeDevice(ctx, vsockSocket, name) if err != nil { - return err + return false, err } + clean := false if found { - if err = e.unmountLiveVolume(ctx, vsockSocket, device, mount); err != nil { - return err + if clean, err = e.unmountLiveVolume(ctx, vsockSocket, device, mount); err != nil { + return false, err } } - _, err = e.run(ctx, "vm", "disk", "detach", vmName, argName, name) - return err + if _, err = e.run(ctx, "vm", "disk", "detach", vmName, argName, name); err != nil { + return false, err + } + return clean, nil } -func (e *Engine) unmountLiveVolume(ctx context.Context, vsockSocket, device, mount string) error { +func (e *Engine) unmountLiveVolume(ctx context.Context, vsockSocket, device, mount string) (bool, error) { + if mount == "" { + return false, e.refuseOtherMounts(ctx, vsockSocket, device) + } devNum, mounts, err := e.volumeMounts(ctx, vsockSocket, device) if err != nil { - return err + return false, err } target := e.resolveMount(ctx, vsockSocket, mount, mounts) atTarget := slices.DeleteFunc(mounts, func(m volumeMount) bool { return m.path != target }) + clean := false if len(atTarget) == 1 && atTarget[0].devNum == devNum && atTarget[0].root == "/" { if err = e.UnmountVolume(ctx, vsockSocket, mount); err != nil { if _, ok := errors.AsType[*guestExitError](err); ok { - return &VolumeMountConflict{Message: fmt.Sprintf("cannot unmount %q; close guest references and retry", mount)} + return false, &VolumeMountConflict{Message: fmt.Sprintf("cannot unmount %q; close guest references and retry", mount)} } - return err + return false, err } + clean = writableMountOptions(atTarget[0].options) && writableMountOptions(atTarget[0].superOptions) + } + if err = e.refuseOtherMounts(ctx, vsockSocket, device); err != nil { + return false, err } - return e.refuseOtherMounts(ctx, vsockSocket, device) + return clean, nil } func (e *Engine) volumeMounts(ctx context.Context, vsockSocket, device string) (string, []volumeMount, error) { @@ -106,15 +117,8 @@ func (e *Engine) volumeMounts(ctx context.Context, vsockSocket, device string) ( if err != nil { return "", nil, err } - var mounts []volumeMount - for line := range strings.SplitSeq(strings.TrimSpace(string(raw)), "\n") { - fields := strings.Fields(line) - if len(fields) < 10 || !slices.Contains(fields[6:], "-") { - return "", nil, errors.New("invalid guest mountinfo") - } - mounts = append(mounts, volumeMount{fields[2], mountUnescaper.Replace(fields[3]), mountUnescaper.Replace(fields[4]), fields[5]}) - } - return strings.TrimSpace(string(devNum)), mounts, nil + mounts, err := parseVolumeMounts(string(raw)) + return strings.TrimSpace(string(devNum)), mounts, err } func (e *Engine) volumeMounted(ctx context.Context, vsockSocket, device, mount string, rw bool) (bool, error) { @@ -164,6 +168,31 @@ func (e *Engine) refuseOtherMounts(ctx context.Context, vsockSocket, device stri return nil } +func parseVolumeMounts(raw string) ([]volumeMount, error) { + var mounts []volumeMount + for line := range strings.SplitSeq(strings.TrimSpace(raw), "\n") { + mount, filesystem, ok := strings.Cut(line, " - ") + fields := strings.Fields(mount) + fsFields := strings.Split(filesystem, " ") + if !ok || len(fields) < 6 || len(fsFields) != 3 || fsFields[0] == "" || fsFields[2] == "" { + return nil, errors.New("invalid guest mountinfo") + } + mounts = append(mounts, volumeMount{ + devNum: fields[2], + root: mountUnescaper.Replace(fields[3]), + path: mountUnescaper.Replace(fields[4]), + options: fields[5], + superOptions: fsFields[2], + }) + } + return mounts, nil +} + +func writableMountOptions(options string) bool { + parts := strings.Split(options, ",") + return slices.Contains(parts, "rw") && !slices.Contains(parts, "ro") && !slices.Contains(parts, "") +} + func bindConflict(path string) *VolumeMountConflict { return &VolumeMountConflict{Message: fmt.Sprintf("volume device has another mount at %q; remove guest mount references before retrying", path)} } diff --git a/sandboxd/engine/volume_live_test.go b/sandboxd/engine/volume_live_test.go index a731f2e1c..4569f0c7b 100644 --- a/sandboxd/engine/volume_live_test.go +++ b/sandboxd/engine/volume_live_test.go @@ -54,7 +54,7 @@ func TestLiveVolumeConvergesMountAndDetach(t *testing.T) { t.Fatal(err) } } - if err := e.DetachVolume(t.Context(), "sbx-1", sock, spec.Name, tt.mount); err != nil { + if _, err := e.DetachVolume(t.Context(), "sbx-1", sock, spec.Name, tt.mount); err != nil { t.Fatal(err) } fake.mu.Lock() @@ -191,7 +191,7 @@ func TestLiveDetachKeepsBusyAndUnobservableDevices(t *testing.T) { fake.serial["/proc/self/mountinfo"] = []byte(rootMountInfo + "2 1 8:32 / /data ro - ext4 /dev/vdc ro\n3 2 8:32 /sub /data ro - ext4 /dev/vdc ro\n") } fake.mu.Unlock() - err := New("cocoon", nil, nil, false, false, "").DetachVolume(t.Context(), "sbx-1", sock, spec.Name, "/data") + _, err := New("cocoon", nil, nil, false, false, "").DetachVolume(t.Context(), "sbx-1", sock, spec.Name, "/data") if err == nil { t.Fatal("detach must fail closed") } @@ -222,7 +222,7 @@ func TestLiveDetachMissingGuestStillChecksHost(t *testing.T) { sock := sockPath(t) serveFakeSilkd(t, sock) spec := liveDiskCLI(t, hostErr) - err := New("cocoon", nil, nil, false, false, "").DetachVolume(t.Context(), "sbx-1", sock, spec.Name, "/data") + _, err := New("cocoon", nil, nil, false, false, "").DetachVolume(t.Context(), "sbx-1", sock, spec.Name, "/data") wantErr := hostErr != "" if (err != nil) != wantErr { t.Fatalf("got %v, want error=%v", err, wantErr) @@ -240,7 +240,7 @@ func TestLiveDetachIgnoresAnotherDeviceAtThePath(t *testing.T) { fake.serial["/sys/block/vdc/dev"] = []byte("8:32\n") fake.serial["/proc/self/mountinfo"] = []byte(rootMountInfo + "2 1 8:48 / /data ro - ext4 /dev/vdd ro\n") fake.mu.Unlock() - if err := New("cocoon", nil, nil, false, false, "").DetachVolume(t.Context(), "sbx-1", sock, spec.Name, "/data"); err != nil { + if _, err := New("cocoon", nil, nil, false, false, "").DetachVolume(t.Context(), "sbx-1", sock, spec.Name, "/data"); err != nil { t.Fatalf("detach of an unmounted device blocked by another device's mount: %v", err) } fake.mu.Lock() @@ -250,6 +250,63 @@ func TestLiveDetachIgnoresAnotherDeviceAtThePath(t *testing.T) { } } +func TestLiveDetachCleanEvidenceRequiresWritableUnmountAndHostSuccess(t *testing.T) { + for _, tc := range []struct { + name, options, superOptions, mount, hostErr string + missing, unmounted, busy, residual, wantClean, wantErr bool + }{ + {name: "normal", options: "rw", superOptions: "rw", mount: "/data", wantClean: true}, + {name: "unordered options", options: "relatime,rw", superOptions: "errors=remount-ro,rw", mount: "/data", wantClean: true}, + {name: "readonly", options: "ro", superOptions: "ro", mount: "/data"}, + {name: "mount readonly", options: "ro", superOptions: "rw", mount: "/data"}, + {name: "superblock readonly", options: "rw", superOptions: "ro", mount: "/data"}, + {name: "extra superblock fields", options: "rw", superOptions: "rw extra", mount: "/data", wantErr: true}, + {name: "pre-unmounted", unmounted: true, mount: "/data"}, + {name: "missing device", missing: true, unmounted: true, mount: "/data"}, + {name: "host failure", options: "rw", superOptions: "rw", mount: "/data", hostErr: "host removal failed", wantErr: true}, + {name: "busy", options: "rw", superOptions: "rw", mount: "/data", busy: true, wantErr: true}, + {name: "residual mount", options: "rw", superOptions: "rw", mount: "/data", residual: true, wantErr: true}, + {name: "attach-only mounted", options: "rw", superOptions: "rw", wantErr: true}, + } { + t.Run(tc.name, func(t *testing.T) { + sock := sockPath(t) + fake := serveFakeSilkd(t, sock) + if !tc.missing { + configureVolumeDevices(fake) + } + spec := liveDiskCLI(t, tc.hostErr) + fake.mu.Lock() + fake.serial["/sys/block/vdc/dev"] = []byte("8:32\n") + base := rootMountInfo + "501 84 0:62 / /empty-source rw,relatime - tmpfs rw,size=1024k,inode64\n" + raw := base + if !tc.unmounted { + raw += "2 1 8:32 / /data " + tc.options + " shared:4 - ext4 /dev/vdc " + tc.superOptions + "\n" + } + fake.serial["/proc/self/mountinfo"] = []byte(raw) + if tc.busy { + fake.execCode = 32 + } + fake.execHook = func(argv []string) { + if argv[0] == "umount" && !tc.residual && !tc.busy { + fake.serial["/proc/self/mountinfo"] = []byte(base) + } + } + fake.mu.Unlock() + clean, err := New("cocoon", nil, nil, false, false, "").DetachVolume(t.Context(), "sbx-1", sock, spec.Name, tc.mount) + if (err != nil) != tc.wantErr || clean != tc.wantClean { + t.Fatalf("clean=%t err=%v", clean, err) + } + if tc.mount == "" { + fake.mu.Lock() + defer fake.mu.Unlock() + if len(guestChanges(fake.execCalls)) != 0 { + t.Fatal("unmounted a caller-managed filesystem") + } + } + }) + } +} + func liveDiskCLI(t *testing.T, detachErr string) VolumeSpec { t.Helper() dir := t.TempDir() diff --git a/sandboxd/pool/claims.go b/sandboxd/pool/claims.go index 2f07ecaff..aaeb92417 100644 --- a/sandboxd/pool/claims.go +++ b/sandboxd/pool/claims.go @@ -116,6 +116,11 @@ func (s *claimStore) load() (map[string]*types.Sandbox, error) { s.mu.Lock() defer s.mu.Unlock() for _, sb := range claims { + for _, volumes := range [][]types.Volume{sb.Volumes, sb.VolumesWant} { + for i := range volumes { + volumes[i].AttachOnly = volumes[i].Mount == "" + } + } if sb.VolumesWant == nil { sb.VolumesWant = sb.Volumes } @@ -340,6 +345,12 @@ func (s *claimStore) checkpointReferences() map[string]struct{} { return refs } +func (s *claimStore) hasCheckpointReferences(owner string) bool { + s.mu.Lock() + defer s.mu.Unlock() + return s.refs[owner] != [4]string{} || len(s.retained[owner]) != 0 +} + func encodeRow(dto *claimDTO) ([]byte, error) { raw, err := json.Marshal(map[string]*claimDTO{dto.ID: dto}) if err != nil { diff --git a/sandboxd/pool/claims_test.go b/sandboxd/pool/claims_test.go index f7e193111..8a6db5a1e 100644 --- a/sandboxd/pool/claims_test.go +++ b/sandboxd/pool/claims_test.go @@ -103,9 +103,10 @@ func TestStoreRoundTrip(t *testing.T) { func TestClaimStoreSnapshotDetachesVolumes(t *testing.T) { s := newClaimStore(t.TempDir(), false) - sb := &types.Sandbox{ID: "sb_a", Volumes: []types.Volume{{Name: "dataset-a", Mount: "/datasets/a"}}} + sb := &types.Sandbox{ID: "sb_a", Volumes: []types.Volume{{Name: "dataset-a", Mount: "/datasets/a"}}, VolumesWant: []types.Volume{{Name: "dataset-b", Mount: "/datasets/b"}}} snap := s.set(sb) sb.Volumes[0].Mount = "/mutated" + sb.VolumesWant[0].Mount = "/mutated-wanted" if err := s.commit(snap); err != nil { t.Fatalf("commit: %v", err) } @@ -120,6 +121,9 @@ func TestClaimStoreSnapshotDetachesVolumes(t *testing.T) { if !slices.Equal(loaded.Volumes, []types.Volume{{Name: "dataset-a", Mount: "/datasets/a"}}) { t.Errorf("volumes %v, want detached snapshot", loaded.Volumes) } + if !slices.Equal(loaded.VolumesWant, []types.Volume{{Name: "dataset-b", Mount: "/datasets/b"}}) { + t.Errorf("wanted volumes %v, want detached snapshot", loaded.VolumesWant) + } } func TestClaimRowOmitsUnsetFlags(t *testing.T) { diff --git a/sandboxd/pool/failover.go b/sandboxd/pool/failover.go index da3c8c058..bd5f63314 100644 --- a/sandboxd/pool/failover.go +++ b/sandboxd/pool/failover.go @@ -131,6 +131,10 @@ func (m *Manager) putRow(ctx context.Context, sb *types.Sandbox, ck string, repl return 0, nil } m.mu.Lock() + if replica && m.hasVolumeHolds(sb) { + m.mu.Unlock() + return 0, ErrVolumeCapture + } dto, deadline, held := dtoOf(sb), sb.Deadline, sb.ArchiveEpoch switch { case replica: @@ -300,7 +304,9 @@ func (m *Manager) reputArchive(ctx context.Context, sb *types.Sandbox, held int6 epoch, err = m.putReplica(ctx, sb, sb.ReplicaCk) } if err != nil { - log.WithFunc("pool.reputArchive").Warnf(ctx, "re-put archive row %s: %v", sb.ID, err) + if !errors.Is(err, ErrVolumeCapture) { + log.WithFunc("pool.reputArchive").Warnf(ctx, "re-put archive row %s: %v", sb.ID, err) + } return } _, _ = m.commitIfLive(ctx, sb, func() { sb.ArchiveEpoch = epoch }) diff --git a/sandboxd/pool/pool.go b/sandboxd/pool/pool.go index f6a4a2957..184a13744 100644 --- a/sandboxd/pool/pool.go +++ b/sandboxd/pool/pool.go @@ -106,6 +106,10 @@ var ( ErrVolumeRegistrationConflict = errors.New("volume registration conflicts with the catalog") ErrVolumeCatalogUnavailable = errors.New("volume catalog requires a successful persistence retry") + ErrReplicaPresent = errors.New("live writable attachment requires replica and recovery checkpoint retirement") + ErrReplicatingPool = errors.New("live writable volumes require a non-replicating pool") + ErrReplicaStranded = errors.New("live writable attachment is blocked by retained replica or archive state with archive_failover disabled") + // Replaying a journal takes a writable mount, so readers stay out. ErrVolumeNeedsRecovery = errors.New("volume needs recovery by a writable claim") ErrArchived = errors.New("sandbox is archived; an exec or file call wakes it first") @@ -149,7 +153,7 @@ type Engine interface { TrimCow(ctx context.Context, vsockSocket string) error CheckLiveVolumeSupport(ctx context.Context) error DiskAttach(ctx context.Context, vmName string, spec engine.VolumeSpec) error - DetachVolume(ctx context.Context, vmName, sock, name, mount string) error + DetachVolume(ctx context.Context, vmName, sock, name, mount string) (cleanRW bool, err error) MountVolume(ctx context.Context, vsockSocket, name, mount string, rw bool) error MountLiveVolume(ctx context.Context, vsockSocket, name, mount string, rw bool) error UnmountVolume(ctx context.Context, vsockSocket, mount string) error diff --git a/sandboxd/pool/pool_test.go b/sandboxd/pool/pool_test.go index 56e4148f6..513c05dee 100644 --- a/sandboxd/pool/pool_test.go +++ b/sandboxd/pool/pool_test.go @@ -1253,6 +1253,8 @@ type fakeEngine struct { mountVolumeErr error unmountVolumeErr error detachVolumeErr error + detachUnclean bool + liveVolumeSupportErr error liveVolumeHook func(string) error stopped map[string]bool disks map[string][]types.VMDisk @@ -1616,7 +1618,7 @@ func (f *fakeEngine) TrimCow(_ context.Context, sock string) error { return f.trimErr } -func (f *fakeEngine) CheckLiveVolumeSupport(context.Context) error { return nil } +func (f *fakeEngine) CheckLiveVolumeSupport(context.Context) error { return f.liveVolumeSupportErr } func (f *fakeEngine) DiskAttach(ctx context.Context, _ string, spec engine.VolumeSpec) error { if f.liveVolumeHook != nil { @@ -1662,24 +1664,27 @@ func (f *fakeEngine) MountLiveVolume(ctx context.Context, sock, name, mount stri return f.MountVolume(ctx, sock, name, mount, rw) } -func (f *fakeEngine) DetachVolume(ctx context.Context, _, sock, name, mount string) error { +func (f *fakeEngine) DetachVolume(ctx context.Context, _, sock, name, mount string) (cleanRW bool, err error) { if f.liveVolumeHook != nil { if err := f.liveVolumeHook("detach"); err != nil { - return err + return false, err } } if err := ctx.Err(); err != nil { - return err + return false, err } if mount != "" { if err := f.UnmountVolume(ctx, sock, mount); err != nil { - return err + return false, err } } f.mu.Lock() defer f.mu.Unlock() f.volumeOps = append(f.volumeOps, "detach:"+name) - return f.detachVolumeErr + if f.detachVolumeErr != nil { + return false, f.detachVolumeErr + } + return !f.detachUnclean && slices.ContainsFunc(f.volumeMounts, func(v types.Volume) bool { return v.Name == name && v.Mount == mount && v.RW() }), nil } func (f *fakeEngine) UnmountVolume(_ context.Context, _, mount string) error { diff --git a/sandboxd/pool/replica.go b/sandboxd/pool/replica.go index 6524154d9..b27c22be6 100644 --- a/sandboxd/pool/replica.go +++ b/sandboxd/pool/replica.go @@ -112,16 +112,21 @@ func (m *Manager) replicate(ctx context.Context, sb *types.Sandbox) error { sb.Transition.Lock() defer sb.Transition.Unlock() m.mu.Lock() - ok := m.claimed[sb.ID] == sb && sb.ArchiveCk == "" && sb.ArchivePending == "" var intent claimSnapshot - if ok { + var refusal error + switch { + case m.claimed[sb.ID] != sb || sb.ArchiveCk != "" || sb.ArchivePending != "": + refusal = errWokeMeanwhile + case m.hasVolumeHolds(sb): + refusal = ErrVolumeCapture + default: sb.ReplicaPending = ckID intent = m.store.set(sb) } m.mu.Unlock() - if !ok { + if refusal != nil { m.deleteOrphanArchiveCk(ctx, ckID) - return errWokeMeanwhile + return refusal } if err := m.store.commit(intent); err != nil { _ = m.clearReplicaIntent(ctx, sb, ckID) diff --git a/sandboxd/pool/volume.go b/sandboxd/pool/volume.go index 7467696c0..fc3ec2baf 100644 --- a/sandboxd/pool/volume.go +++ b/sandboxd/pool/volume.go @@ -136,8 +136,7 @@ func (m *Manager) resolveVolumes(ctx context.Context, tenant string, requested [ if len(requested) == 0 { return nil, nil } - // an attach-only claim has no mounts to default; re-defaulting here would mount them - applied, err := types.ValidateVolumes(requested, types.VolumesAttachOnly(requested)) + applied, err := types.ValidateVolumes(requested) if err != nil { return nil, fmt.Errorf("%w: %w", ErrBadVolume, err) } @@ -171,7 +170,6 @@ func (m *Manager) resolveCatalogEntry(ctx context.Context, volume types.Volume, } disk := entry.disk disk.RW = volume.RW() - volume.AttachOnly = false // past validation the empty mount carries it return resolvedVolume{entry: entry, disk: disk, applied: volume}, nil } @@ -204,7 +202,7 @@ func (m *Manager) applyVolumes(ctx context.Context, sb *types.Sandbox, volumes [ // applyVolume keeps one volume's steps strictly ordered; siblings overlap freely. func (m *Manager) applyVolume(ctx context.Context, sb *types.Sandbox, volume resolvedVolume, live bool) error { // attach-only mounts nothing, so there is no umount to verify and no marker. - attachOnly := volume.applied.Mount == "" + attachOnly := volume.applied.AttachOnly // write-ahead: the marker must be durable before any guest write can be. if volume.disk.RW && !attachOnly { if err := markVolumeDirty(volume.disk.Path); err != nil { @@ -281,7 +279,6 @@ func (m *Manager) finishVolumeTeardown(ctx context.Context, td volumeTeardown) { m.unreserveVolumes(td.holds) } -// adoptVolumes counts volumes an adopted claim already holds; callers hold m.mu. func (m *Manager) adoptVolumes(volumes []types.Volume) { for _, volume := range volumes { holders := m.catalog.holders[volume.Name] @@ -294,7 +291,6 @@ func (m *Manager) adoptVolumes(volumes []types.Volume) { } } -// releaseVolumes drops one claim's admission holds; callers hold m.mu. func (m *Manager) releaseVolumes(volumes []types.Volume) { for _, volume := range volumes { holders := m.catalog.holders[volume.Name] @@ -356,7 +352,7 @@ func writableMounts(volumes []types.Volume) int { } func needsUnmount(volume types.Volume) bool { - return volume.RW() && volume.Mount != "" + return volume.RW() && !volume.AttachOnly } func appliedVolumes(volumes []resolvedVolume) []types.Volume { diff --git a/sandboxd/pool/volume_attach_test.go b/sandboxd/pool/volume_attach_test.go index 7fc14e75a..427b7e867 100644 --- a/sandboxd/pool/volume_attach_test.go +++ b/sandboxd/pool/volume_attach_test.go @@ -21,7 +21,7 @@ func TestAttachOnlyClaimAttachesWithoutMounting(t *testing.T) { if err != nil { t.Fatalf("ClaimProvision: %v", err) } - want := []types.Volume{{Name: "scratch", Mode: types.VolumeModeRW}} + want := []types.Volume{{Name: "scratch", Mode: types.VolumeModeRW, AttachOnly: true}} if !slices.Equal(sb.Volumes, want) { t.Errorf("applied volumes=%v, want %v", sb.Volumes, want) } @@ -78,13 +78,8 @@ func TestAttachOnlyClaimAttachesEveryVolumeConcurrently(t *testing.T) { if err != nil { t.Fatalf("ClaimProvision: %v", err) } - want := []types.Volume{ - {Name: "scratch", Mode: types.VolumeModeRW}, - {Name: "dataset"}, - {Name: "cache", Mode: types.VolumeModeRW}, - } - if !slices.Equal(sb.Volumes, want) { - t.Errorf("volumes=%v, want the request order %v", sb.Volumes, want) + if !slices.Equal(sb.Volumes, requested) { + t.Errorf("volumes=%v, want the request order %v", sb.Volumes, requested) } wantOps := []string{"attach:cache", "attach:dataset", "attach:scratch", "probe", "provision"} if ops := slices.Sorted(slices.Values(eng.volumeOpsLog())); !slices.Equal(ops, wantOps) { @@ -380,7 +375,7 @@ func TestQuiesceVolumesSkipsAttachOnlyEntries(t *testing.T) { sb := &types.Sandbox{ ID: "sb_mixed", VMName: "sbx-mixed", Key: testKey, VsockSocket: "/vsock/mixed", Volumes: []types.Volume{ - {Name: "scratch", Mode: types.VolumeModeRW}, + {Name: "scratch", Mode: types.VolumeModeRW, AttachOnly: true}, {Name: "mounted", Mount: "/volumes/mounted", Mode: types.VolumeModeRW}, }, } @@ -406,11 +401,11 @@ func TestWritableMountsCountsOnlyQuiescableEntries(t *testing.T) { }{ {"none", nil, 0}, {"read-only mount", []types.Volume{{Name: "a", Mount: "/a"}}, 0}, - {"attach-only writer", []types.Volume{{Name: "a", Mode: types.VolumeModeRW}}, 0}, + {"attach-only writer", []types.Volume{{Name: "a", Mode: types.VolumeModeRW, AttachOnly: true}}, 0}, {"writable mount", []types.Volume{{Name: "a", Mount: "/a", Mode: types.VolumeModeRW}}, 1}, {"mixed", []types.Volume{ {Name: "a", Mount: "/a", Mode: types.VolumeModeRW}, - {Name: "b", Mode: types.VolumeModeRW}, + {Name: "b", Mode: types.VolumeModeRW, AttachOnly: true}, {Name: "c", Mount: "/c"}, }, 1}, } { diff --git a/sandboxd/pool/volume_convergence_test.go b/sandboxd/pool/volume_convergence_test.go index 3913c596b..db53172ac 100644 --- a/sandboxd/pool/volume_convergence_test.go +++ b/sandboxd/pool/volume_convergence_test.go @@ -287,7 +287,7 @@ func TestUnsupportedRuntimeDoesNotChangeVolumeIntent(t *testing.T) { eng := newFakeEngine() m := newVolumeManager(t, eng, liveCatalog(t)) sb := mustClaim(t, m, testKey) - m.eng = &unsupportedVolumeEngine{eng} + eng.liveVolumeSupportErr = &engine.LiveVolumeUnsupported{} before, err := os.ReadFile(m.store.path) if err != nil { t.Fatal(err) @@ -420,12 +420,6 @@ func TestUncertainTargetCommitReleasesOnlyConfirmedPrecleanedHolds(t *testing.T) } } -type unsupportedVolumeEngine struct{ *fakeEngine } - -func (e *unsupportedVolumeEngine) CheckLiveVolumeSupport(context.Context) error { - return &engine.LiveVolumeUnsupported{} -} - func unmountConflict() *engine.VolumeMountConflict { return &engine.VolumeMountConflict{Message: `cannot unmount "/volumes/data"; close guest references and retry`} } diff --git a/sandboxd/pool/volume_mutation.go b/sandboxd/pool/volume_mutation.go index 6b7caf2bf..7be4a25bd 100644 --- a/sandboxd/pool/volume_mutation.go +++ b/sandboxd/pool/volume_mutation.go @@ -40,13 +40,13 @@ func (m *Manager) SetVolumes(ctx context.Context, id, tenant string, want []type return m.changeVolumes(ctx, id, tenant, func([]types.Volume) ([]types.Volume, error) { return slices.Clone(want), nil }) } -// PatchVolumes adds and removes a running claim's read-only mounted volumes by name. +// PatchVolumes adds and removes a running claim's volumes by name. func (m *Manager) PatchVolumes(ctx context.Context, id, tenant string, patch types.VolumePatch) (types.SandboxVolumesResponse, error) { return m.changeVolumes(ctx, id, tenant, func(current []types.Volume) ([]types.Volume, error) { if err := validateDetachNames(patch.Remove); err != nil { return nil, err } - additions, err := normalizeLiveVolumes(patch.Add, current) + additions, err := normalizeLiveVolumes(patch.Add) if err != nil { return nil, err } @@ -56,7 +56,7 @@ func (m *Manager) PatchVolumes(ctx context.Context, id, tenant string, patch typ } if i := volumeIndex(current, v.Name); i >= 0 { if current[i] != v { - return nil, fmt.Errorf("%w: volume %q already has a different mount or mode", ErrBadVolume, v.Name) + return nil, fmt.Errorf("%w: volume %q already has a different mount, mode, or attach_only", ErrBadVolume, v.Name) } } else { current = append(current, v) @@ -91,7 +91,7 @@ func (m *Manager) changeVolumes(ctx context.Context, id, tenant string, change v if err != nil { return view, err } - target, err = normalizeLiveVolumes(target, held) + target, err = normalizeLiveVolumes(target) if err != nil { return view, err } @@ -108,7 +108,13 @@ func (m *Manager) changeVolumes(ctx context.Context, id, tenant string, change v if err != nil { return view, err } - if err = m.checkVolumeRun(ctx, sb, run); err != nil { + m.mu.Lock() + err = m.checkVolumeRunLocked(ctx, sb, run) + if err == nil && slices.ContainsFunc(additions, types.Volume.RW) { + err = m.liveWriterErrorLocked(sb) + } + m.mu.Unlock() + if err != nil { return view, err } sb.Touch() @@ -259,21 +265,11 @@ func (m *Manager) stageVolumeTarget(ctx context.Context, sb *types.Sandbox, run sb.Transition.Lock() defer sb.Transition.Unlock() m.mu.Lock() - if err := m.checkVolumeRunLocked(ctx, sb, run); err != nil { - m.mu.Unlock() - return claimSnapshot{}, false, err - } - if sb.OnExpire == types.ExpireArchive && len(target) > 0 { - m.mu.Unlock() - return claimSnapshot{}, false, ErrVolumeCapture - } - if err := m.checkResolvedLocked(resolved); err != nil { - m.mu.Unlock() + err := m.holdLiveAdditionsLocked(ctx, sb, run, target, additions, resolved) + m.mu.Unlock() + if err != nil { return claimSnapshot{}, false, err } - m.adoptVolumes(additions) - run.holds = append(run.holds, additions...) - m.mu.Unlock() cleanErr := m.confirmVolumesClean(resolved) m.mu.Lock() defer m.mu.Unlock() @@ -296,14 +292,69 @@ func (m *Manager) stageVolumeTarget(ctx context.Context, sb *types.Sandbox, run return m.store.set(sb), true, nil } +func (m *Manager) holdLiveAdditionsLocked(ctx context.Context, sb *types.Sandbox, run *volumeRun, target, additions []types.Volume, resolved []resolvedVolume) error { + if err := m.checkVolumeRunLocked(ctx, sb, run); err != nil { + return err + } + if sb.OnExpire == types.ExpireArchive && len(target) > 0 { + return ErrVolumeCapture + } + if slices.ContainsFunc(additions, types.Volume.RW) { + if err := m.liveWriterErrorLocked(sb); err != nil { + return err + } + } + if err := m.checkResolvedLocked(resolved); err != nil { + return err + } + m.adoptVolumes(additions) + run.holds = append(run.holds, additions...) + return nil +} + +func (m *Manager) liveWriterErrorLocked(sb *types.Sandbox) error { + if m.view.Load().poolReplicate[sb.PolicyKey()] > 0 { + return ErrReplicatingPool + } + if m.archives == nil && (sb.HasReplica() || sb.ArchivePending != "") { + return ErrReplicaStranded + } + if sb.HasReplica() || sb.ArchivePending != "" || m.store.hasCheckpointReferences(sb.ID) { + return ErrReplicaPresent + } + return nil +} + func (m *Manager) detachVolumeRun(ctx context.Context, sb *types.Sandbox, run *volumeRun, v types.Volume) error { - if err := m.checkVolumeRun(ctx, sb, run); err != nil { + m.mu.Lock() + err := m.checkVolumeRunLocked(ctx, sb, run) + entry := m.catalog.entries[v.Name] + m.mu.Unlock() + if err != nil { return err } - if err := m.eng.DetachVolume(ctx, sb.VMName, sb.VsockSocket, v.Name, v.Mount); err != nil { + cleanRW, err := m.eng.DetachVolume(ctx, sb.VMName, sb.VsockSocket, v.Name, v.Mount) + if err != nil { return fmt.Errorf("detach volume %q: %w", v.Name, err) } - return m.checkVolumeRun(ctx, sb, run) + clearMarker := needsUnmount(v) && cleanRW && entry != nil + hold := [1]types.Volume{v} + m.mu.Lock() + err = m.checkVolumeRunLocked(ctx, sb, run) + if err == nil && clearMarker { + m.adoptVolumes(hold[:]) + } + m.mu.Unlock() + if err != nil { + return err + } + if clearMarker { + defer m.unreserveVolumes(hold[:]) + if err = clearVolumeDirty(entry.disk.Path); err != nil { + return fmt.Errorf("clear dirty marker for volume %q: %w", v.Name, err) + } + } + return nil } func (m *Manager) commitVolumeApplied(ctx context.Context, sb *types.Sandbox, run *volumeRun, v types.Volume, attached bool) error { @@ -413,30 +464,9 @@ func (m *Manager) clearVolumeError(sb *types.Sandbox, run *volumeRun, name strin } } -func normalizeLiveVolumes(volumes, held []types.Volume) ([]types.Volume, error) { - if len(volumes) > types.MaxClaimVolumes { - return nil, fmt.Errorf("%w: too many volumes", ErrBadVolume) - } - out := make([]types.Volume, 0, len(volumes)) - for _, v := range volumes { - attachOnly := v.AttachOnly - if i := volumeIndex(held, v.Name); i >= 0 && held[i].Mount == "" && v.Mount == "" { - attachOnly = true - } - normalized, err := types.ValidateVolumes([]types.Volume{v}, attachOnly) - if err != nil { - return nil, fmt.Errorf("%w: %w", ErrBadVolume, err) - } - if volumeIndex(out, v.Name) >= 0 { - return nil, fmt.Errorf("%w: duplicate volume %q", ErrBadVolume, v.Name) - } - if attachOnly { - normalized[0].AttachOnly = false - } - out = append(out, normalized[0]) - } - mounted := slices.DeleteFunc(slices.Clone(out), func(v types.Volume) bool { return v.Mount == "" }) - if _, err := types.ValidateVolumes(mounted, false); err != nil { +func normalizeLiveVolumes(volumes []types.Volume) ([]types.Volume, error) { + out, err := types.ValidateVolumes(volumes) + if err != nil { return nil, fmt.Errorf("%w: %w", ErrBadVolume, err) } return out, nil @@ -446,20 +476,12 @@ func validateLiveTarget(held, target []types.Volume) error { for _, v := range held { i := volumeIndex(target, v.Name) if i >= 0 && target[i] != v { - return fmt.Errorf("%w: remove volume %q before changing its mount or mode", ErrBadVolume, v.Name) + return fmt.Errorf("%w: remove volume %q before changing its mount, mode, or attach_only", ErrBadVolume, v.Name) } - if immutableVolume(v) && i < 0 { - return fmt.Errorf("%w: writable and attach-only volumes cannot change live", ErrBadVolume) - } - } - if slices.ContainsFunc(volumeDifference(target, held), immutableVolume) { - return fmt.Errorf("%w: live additions require read-only mounts", ErrBadVolume) } return nil } -func immutableVolume(v types.Volume) bool { return v.RW() || v.AttachOnly || v.Mount == "" } - func volumeIndex(volumes []types.Volume, name string) int { return slices.IndexFunc(volumes, func(v types.Volume) bool { return v.Name == name }) } diff --git a/sandboxd/pool/volume_mutation_test.go b/sandboxd/pool/volume_mutation_test.go index e1925af35..19ce01c78 100644 --- a/sandboxd/pool/volume_mutation_test.go +++ b/sandboxd/pool/volume_mutation_test.go @@ -1,6 +1,7 @@ package pool import ( + "bytes" "context" "errors" "fmt" @@ -18,6 +19,75 @@ import ( "github.com/cocoonstack/sandbox/sandboxd/types" ) +func TestLiveVolumeFourModesAndMixedRoundtrip(t *testing.T) { + eng := newFakeEngine() + catalog := []config.VolumeSpec{} + request := []types.Volume{{Name: "mounted"}, {Name: "writer", Mode: types.VolumeModeRW}, {Name: "raw", AttachOnly: true}, {Name: "raw-writer", Mode: types.VolumeModeRW, AttachOnly: true}} + for _, v := range request { + catalog = append(catalog, config.VolumeSpec{Name: v.Name, Path: writeVolumeImage(t, v.Name, "seed"), Writable: true}) + } + m := newVolumeManager(t, eng, catalog) + sb := mustClaim(t, m, testKey) + view, err := m.PatchVolumes(t.Context(), sb.ID, "", types.VolumePatch{Add: request}) + if err != nil { + t.Fatal(err) + } + if len(view.Volumes) != 4 || len(eng.volumeMounts) != 2 { + t.Fatalf("mixed modes: view=%+v mounts=%+v", view, eng.volumeMounts) + } + for _, v := range view.Volumes { + if v.AttachOnly != (v.Mount == "") { + t.Fatalf("lost canonical shape: %+v", v) + } + } + before := slices.Clone(eng.volumeOpsLog()) + if _, err = m.SetVolumes(t.Context(), sb.ID, "", view.VolumesWant); err != nil { + t.Fatal(err) + } + if !slices.Equal(before, eng.volumeOpsLog()) { + t.Fatal("GET-to-PUT roundtrip changed devices") + } + for _, change := range []types.Volume{{Name: "raw", Mount: "/new"}, {Name: "raw-writer", AttachOnly: true}, {Name: "writer", Mount: "/other", Mode: types.VolumeModeRW}} { + if _, err = m.PatchVolumes(t.Context(), sb.ID, "", types.VolumePatch{Add: []types.Volume{change}}); !errors.Is(err, ErrBadVolume) { + t.Fatalf("held shape change accepted: %+v, %v", change, err) + } + } + legacy := slices.Clone(view.VolumesWant) + for i := range legacy { + legacy[i].AttachOnly = false + } + if _, err = m.SetVolumes(t.Context(), sb.ID, "", legacy); !errors.Is(err, ErrBadVolume) { + t.Fatalf("explicit false inherited attach-only: %v", err) + } + assertLiveJournal(t, m, sb) + raw, err := os.ReadFile(m.store.path) + if err != nil { + t.Fatal(err) + } + if err = os.WriteFile(m.store.path, bytes.ReplaceAll(raw, []byte(`,"attach_only":true`), nil), 0o600); err != nil { + t.Fatal(err) + } + restarted := newVolumeManagerAt(t, eng, m.dataDir, catalog) + if err = restarted.Reconcile(t.Context()); err != nil { + t.Fatal(err) + } + recovered, err := restarted.GetVolumes(sb.ID, "") + if err != nil || !slices.Equal(recovered.Volumes, view.Volumes) { + t.Fatalf("mixed restart: %+v, %v", recovered, err) + } + if _, err = restarted.ClaimProvision(t.Context(), testKey, ClaimOptions{Volumes: []types.Volume{{Name: "raw-writer", AttachOnly: true}}}); !errors.Is(err, ErrVolumeBusy) { + t.Fatalf("restart lost writer exclusion: %v", err) + } + if _, err = restarted.SetVolumes(t.Context(), sb.ID, "", []types.Volume{}); err != nil { + t.Fatal(err) + } + for _, spec := range catalog { + if volumeDirty(spec.Path) || volumeHoldersOf(restarted, spec.Name) != (volumeHolders{}) { + t.Fatalf("detach leaked marker or hold: %s", spec.Name) + } + } +} + func TestLiveVolumesCommitAndRetry(t *testing.T) { eng := newFakeEngine() m := newVolumeManager(t, eng, liveCatalog(t)) @@ -68,8 +138,8 @@ func TestLiveAttachPreservesWritableAndAttachOnlyVolumes(t *testing.T) { if _, err := m.SetVolumes(t.Context(), sb.ID, sb.Tenant, []types.Volume{{Name: "other", Mount: original.Mount}}); !errors.Is(err, ErrBadVolume) { t.Fatalf("accepted mode change: %v", err) } - if _, err := m.PatchVolumes(t.Context(), sb.ID, sb.Tenant, types.VolumePatch{Remove: []string{"data", "other"}}); !errors.Is(err, ErrBadVolume) || len(sb.Volumes) != 2 { - t.Fatalf("writable detach changed the batch: %v", err) + if _, err := m.PatchVolumes(t.Context(), sb.ID, sb.Tenant, types.VolumePatch{Remove: []string{"data", "other"}}); err != nil || len(sb.Volumes) != 0 { + t.Fatalf("writable detach failed: %v", err) } }) } @@ -80,8 +150,8 @@ func TestLiveVolumeValidationPrecedesDeviceCalls(t *testing.T) { m := newVolumeManager(t, eng, liveCatalog(t)) sb := mustClaim(t, m, testKey) for _, request := range [][]types.Volume{ - {{Name: "data", Mode: "rw"}}, - {{Name: "data", AttachOnly: true}}, + {{Name: "data", Mode: "unknown"}}, + {{Name: "data", AttachOnly: true, Mount: "/data"}}, {{Name: "data", Mount: "/etc"}}, {{Name: "data"}, {Name: "data"}}, {{Name: "data", Mount: "/a"}, {Name: "other", Mount: "/a/b"}}, @@ -479,6 +549,7 @@ func TestDesiredVolumesJournalMigration(t *testing.T) { want []types.Volume }{ {"legacy", "", []types.Volume{{Name: "data", Mount: "/data"}}}, + {"legacy attach-only desired", `,"volumes_want":[{"name":"raw","mode":"rw"}]`, []types.Volume{{Name: "raw", Mode: types.VolumeModeRW, AttachOnly: true}}}, {"explicit empty", `,"volumes_want":[]`, nil}, {"desired", `,"volumes_want":[{"name":"other","mount":"/other"}]`, []types.Volume{{Name: "other", Mount: "/other"}}}, } { @@ -849,7 +920,7 @@ func TestLivenessRefusesAdmittedVolumeRun(t *testing.T) { m.finishVolumeRun(sb, run) } -func TestRemovedTenantCanRemoveReadOnlyVolumeBesideImmutableVolume(t *testing.T) { +func TestRemovedTenantCanRemoveReadOnlyVolumeBesideWritableVolume(t *testing.T) { eng := newFakeEngine() m, sb := newTenantVolumeManager(t, eng, liveCatalog(t), types.Volume{Name: "other", Mode: "rw"}) if _, err := m.PatchVolumes(t.Context(), sb.ID, sb.Tenant, types.VolumePatch{Add: []types.Volume{{Name: "data"}}}); err != nil { @@ -863,7 +934,7 @@ func TestRemovedTenantCanRemoveReadOnlyVolumeBesideImmutableVolume(t *testing.T) t.Fatalf("removed tenant cleanup=%+v %v", view, err) } if volumeHoldersOf(m, "data") != (volumeHolders{}) || volumeHoldersOf(m, "other").writers != 1 { - t.Fatal("cleanup altered immutable hold") + t.Fatal("cleanup altered writable hold") } } diff --git a/sandboxd/pool/volume_replica_test.go b/sandboxd/pool/volume_replica_test.go new file mode 100644 index 000000000..7124f2357 --- /dev/null +++ b/sandboxd/pool/volume_replica_test.go @@ -0,0 +1,217 @@ +package pool + +import ( + "errors" + "slices" + "sync" + "testing" + "time" + + "github.com/cocoonstack/sandbox/sandboxd/config" + "github.com/cocoonstack/sandbox/sandboxd/types" +) + +func TestReplicaRecoveryDoesNotRepublishReadOnlyHold(t *testing.T) { + m, c := newReplicaVolumeManager(t) + sb := mustClaim(t, m, testKey) + replicateNow(t, m, sb) + if view, err := m.SetVolumes(t.Context(), sb.ID, "", []types.Volume{{Name: "data"}}); err != nil || len(view.Volumes) != 1 { + t.Fatalf("read-only live attachment: %+v, %v", view, err) + } + if _, err := c.reg.Consume(t.Context(), sb.ID, sb.ArchiveEpoch); err != nil { + t.Fatal(err) + } + rec, ok, err := c.reg.Get(t.Context(), sb.ID) + if err != nil || !ok || !rec.Consumed { + t.Fatalf("setup: %+v, %t, %v", rec, ok, err) + } + m.reconcileOwnRow(t.Context(), rec, time.Now()) + if rec, ok, err = c.reg.Get(t.Context(), sb.ID); err != nil || !ok || !rec.Consumed { + t.Fatalf("recovery republished read-only hold: %+v, %t, %v", rec, ok, err) + } +} + +func TestReplicaPublicationRefusesVolumeAdmittedDuringExport(t *testing.T) { + m, c := newReplicaVolumeManager(t) + disableReplication(m) + sb := mustClaim(t, m, testKey) + stalled := &stallingStore{Store: m.ckpts, published: make(chan string), release: make(chan struct{})} + m.ckpts = stalled + result := make(chan error, 1) + done := make(chan struct{}) + go func() { defer close(done); result <- m.replicate(t.Context(), sb) }() + ck := <-stalled.published + release := sync.OnceFunc(func() { close(stalled.release) }) + t.Cleanup(func() { release(); <-done }) + writer := []types.Volume{{Name: "data", Mode: types.VolumeModeRW, AttachOnly: true}} + view, err := m.SetVolumes(t.Context(), sb.ID, "", writer) + if err != nil || len(view.Volumes) != 1 { + t.Fatalf("writer during checkpoint publication: %+v, %v", view, err) + } + release() + if err = <-result; !errors.Is(err, ErrVolumeCapture) { + t.Fatalf("stale replica registered after writer admission: %v", err) + } + if _, ok, getErr := c.reg.Get(t.Context(), sb.ID); getErr != nil || ok || sb.HasReplica() || ckExists(t, m, ck) { + t.Fatalf("stale replica remains recoverable: row=%t, claim=%t, checkpoint=%t, err=%v", ok, sb.HasReplica(), ckExists(t, m, ck), getErr) + } + if got := volumeHoldersOf(m, "data"); got.writers != 1 { + t.Fatalf("refusing replica lost writer hold: %+v", got) + } + if err = m.replicate(t.Context(), sb); !errors.Is(err, ErrVolumeCapture) { + t.Fatalf("snapshot admission while writer is held: %v", err) + } +} + +func TestReplicaCaptureRefusesPendingLiveWriter(t *testing.T) { + m, c := newReplicaVolumeManager(t) + disableReplication(m) + sb := mustClaim(t, m, testKey) + eng := m.eng.(*fakeEngine) + entered, resume := make(chan struct{}), make(chan struct{}) + release := sync.OnceFunc(func() { close(resume) }) + eng.liveVolumeHook = func(verb string) error { + if verb == "attach" { + close(entered) + <-resume + } + return nil + } + result := make(chan error, 1) + done := make(chan struct{}) + go func() { + defer close(done) + _, err := m.SetVolumes(t.Context(), sb.ID, "", []types.Volume{{Name: "data", Mode: types.VolumeModeRW}}) + result <- err + }() + t.Cleanup(func() { release(); <-done }) + <-entered + view, err := m.GetVolumes(sb.ID, "") + if err != nil || len(view.Volumes) != 0 || len(view.VolumesWant) != 1 || !view.Converging { + t.Fatalf("writer did not reach pending attachment: %+v, %v", view, err) + } + if err = m.replicate(t.Context(), sb); !errors.Is(err, ErrVolumeCapture) { + t.Fatalf("captured pending writer: %v", err) + } + if _, ok, getErr := c.reg.Get(t.Context(), sb.ID); getErr != nil || ok || len(eng.snapSaves) != 0 { + t.Fatalf("pending writer produced snapshot or row: snapshots=%v, row=%t, err=%v", eng.snapSaves, ok, getErr) + } + release() + if err = <-result; err != nil { + t.Fatalf("pending writer did not complete: %v", err) + } +} + +func TestLiveWriterAdmissionAgainstReplicaState(t *testing.T) { + for _, tc := range []struct { + name string + prepare func(*testing.T, *Manager, *types.Sandbox) + want error + }{ + {"replicating pool", func(*testing.T, *Manager, *types.Sandbox) {}, ErrReplicatingPool}, + {"source pool replicates", func(_ *testing.T, _ *Manager, sb *types.Sandbox) { + sb.PolicySource = testKey + sb.Key.Template = "promoted:template" + }, ErrReplicatingPool}, + {"source pool does not replicate", func(_ *testing.T, _ *Manager, sb *types.Sandbox) { + sb.PolicySource = testKey + sb.PolicySource.Template = "other:template" + }, nil}, + {"published replica", func(t *testing.T, m *Manager, sb *types.Sandbox) { disableReplication(m); replicateNow(t, m, sb) }, ErrReplicaPresent}, + {"pending replica", func(_ *testing.T, m *Manager, sb *types.Sandbox) { + disableReplication(m) + sb.ReplicaPending = "pending" + m.store.set(sb) + }, ErrReplicaPresent}, + {"registry disabled", func(t *testing.T, m *Manager, sb *types.Sandbox) { + disableReplication(m) + replicateNow(t, m, sb) + m.archives = nil + }, ErrReplicaStranded}, + {"archive pending without registry", func(_ *testing.T, m *Manager, sb *types.Sandbox) { + disableReplication(m) + sb.ArchivePending = "pending" + m.archives = nil + }, ErrReplicaStranded}, + {"unrelated checkpoint", func(t *testing.T, m *Manager, _ *types.Sandbox) { + disableReplication(m) + other := mustClaim(t, m, testKey) + other.ReplicaCk = "unrelated" + m.store.set(other) + }, nil}, + } { + t.Run(tc.name, func(t *testing.T) { + m, _ := newReplicaVolumeManager(t) + sb := mustClaim(t, m, testKey) + tc.prepare(t, m, sb) + before, ops := m.store.mark(), m.eng.(*fakeEngine).volumeOpsLog() + view, err := m.SetVolumes(t.Context(), sb.ID, "", []types.Volume{{Name: "data", Mode: types.VolumeModeRW, AttachOnly: true}}) + if !errors.Is(err, tc.want) { + t.Fatalf("writer refusal = %v, want %v", err, tc.want) + } + if tc.want != nil && (len(view.VolumesWant) != 0 || len(view.Volumes) != 0 || view.Converging || m.store.mark() != before || volumeHoldersOf(m, "data") != (volumeHolders{}) || !slices.Equal(ops, m.eng.(*fakeEngine).volumeOpsLog())) { + t.Fatalf("refusal changed claim or device: %+v", view) + } + }) + } +} + +func TestInitialWriterAndRetiredReplicaAllowLiveWork(t *testing.T) { + m, _ := newReplicaVolumeManager(t) + sb, err := m.ClaimProvision(t.Context(), testKey, ClaimOptions{Volumes: []types.Volume{{Name: "data", Mode: types.VolumeModeRW}}}) + if err != nil { + t.Fatal(err) + } + if _, err = m.SetVolumes(t.Context(), sb.ID, "", sb.Volumes); err != nil { + t.Fatalf("held writer retry: %v", err) + } + if _, err = m.SetVolumes(t.Context(), sb.ID, "", nil); err != nil { + t.Fatalf("writer cleanup: %v", err) + } + replicateNow(t, m, sb) + disableReplication(m) + m.store.recommitting.Store(true) + restore := failClaimWrites(t, m.store) + m.retireReplica(t.Context(), sb) + writer := []types.Volume{{Name: "data", Mode: types.VolumeModeRW}} + if sb.HasReplica() || !m.store.hasCheckpointReferences(sb.ID) { + t.Fatal("failed clear lost checkpoint uncertainty") + } + if _, err = m.SetVolumes(t.Context(), sb.ID, "", writer); !errors.Is(err, ErrReplicaPresent) { + t.Fatalf("unacknowledged retirement: %v", err) + } + restore() + if err = m.store.commit(m.store.mark()); err != nil { + t.Fatal(err) + } + if _, err = m.SetVolumes(t.Context(), sb.ID, "", writer); err != nil { + t.Fatalf("acknowledged retirement: %v", err) + } +} + +func newReplicaVolumeManager(t *testing.T) (*Manager, *failoverCluster) { + t.Helper() + c := newReplicaCluster(t) + spec := archivePool(keepAnHour) + spec.ReplicateSeconds = c.replicate + cfg := &config.Config{ + APIToken: "root", DataDir: t.TempDir(), CheckpointDir: c.ckDir, + Pools: []config.PoolSpec{spec}, Mesh: &config.MeshConfig{NodeID: "a"}, + Volumes: []config.VolumeSpec{{Name: "data", Path: writeVolumeImage(t, "data", "original"), Writable: true}}, + } + backends := testBackends(t, cfg) + backends.Archives = c.reg + m, err := NewManager(t.Context(), cfg, newFakeEngine(), testSecrets(t), backends) + if err != nil { + t.Fatal(err) + } + c.renew(t, "a") + m.liftFence() + return m, c +} + +func disableReplication(m *Manager) { + view := *m.view.Load() + view.poolReplicate = nil + m.view.Store(&view) +} diff --git a/sandboxd/pool/volume_rw_test.go b/sandboxd/pool/volume_rw_test.go index 7fc072947..f950f40d3 100644 --- a/sandboxd/pool/volume_rw_test.go +++ b/sandboxd/pool/volume_rw_test.go @@ -16,6 +16,139 @@ import ( "github.com/cocoonstack/sandbox/sandboxd/types" ) +func TestLiveWritableDetachRequiresCleanEvidence(t *testing.T) { + for _, tc := range []struct { + name string + unclean bool + }{ + {"clean unmount", false}, + {"unclean unmount", true}, + } { + t.Run(tc.name, func(t *testing.T) { + base := newFakeEngine() + catalog := liveCatalog(t) + m := newVolumeManager(t, base, catalog) + base.detachUnclean = tc.unclean + sb := mustClaim(t, m, testKey) + if _, err := m.SetVolumes(t.Context(), sb.ID, "", []types.Volume{{Name: "data", Mode: types.VolumeModeRW}}); err != nil { + t.Fatal(err) + } + path := catalog[0].Path + if !volumeDirty(path) || !base.attachDirty["data"] { + t.Fatal("writer was attached before dirty marker") + } + if _, err := m.SetVolumes(t.Context(), sb.ID, "", []types.Volume{}); err != nil { + t.Fatal(err) + } + if volumeDirty(path) != tc.unclean { + t.Fatalf("marker disagrees with clean evidence: unclean=%t dirty=%t", tc.unclean, volumeDirty(path)) + } + _, err := m.ClaimProvision(t.Context(), testKey, ClaimOptions{Volumes: []types.Volume{{Name: "data"}}}) + if !tc.unclean && err != nil || tc.unclean && !errors.Is(err, ErrVolumeNeedsRecovery) { + t.Fatalf("reader admission after detach: %v", err) + } + }) + } +} + +func TestLiveWritableFailedHostDetachKeepsMarkerAndHold(t *testing.T) { + eng := newFakeEngine() + catalog := liveCatalog(t) + m := newVolumeManager(t, eng, catalog) + sb := mustClaim(t, m, testKey) + if _, err := m.SetVolumes(t.Context(), sb.ID, "", []types.Volume{{Name: "data", Mode: types.VolumeModeRW}}); err != nil { + t.Fatal(err) + } + eng.detachVolumeErr = errors.New("host removal uncertain") + if _, err := m.SetVolumes(t.Context(), sb.ID, "", []types.Volume{}); err == nil { + t.Fatal("failed host detach reported success") + } + if !volumeDirty(catalog[0].Path) || volumeHoldersOf(m, "data").writers != 1 { + t.Fatal("failed host detach lost dirty marker or writer") + } + if _, err := m.Checkpoint(t.Context(), sb.ID, Cred{Token: sb.Token}, "uncertain", ""); !errors.Is(err, ErrVolumeCapture) { + t.Fatalf("capture after uncertain detach: %v", err) + } + eng.detachVolumeErr = nil + eng.detachUnclean = true + if _, err := m.SetVolumes(t.Context(), sb.ID, "", []types.Volume{}); err != nil { + t.Fatal(err) + } + if !volumeDirty(catalog[0].Path) { + t.Fatal("absent-device retry invented clean evidence") + } +} + +func TestLiveWritableMarkerClearFailureRetainsOnlyClaimHold(t *testing.T) { + eng := newFakeEngine() + catalog := liveCatalog(t) + m := newVolumeManager(t, eng, catalog) + sb := mustClaim(t, m, testKey) + if _, err := m.SetVolumes(t.Context(), sb.ID, "", []types.Volume{{Name: "data", Mode: types.VolumeModeRW}}); err != nil { + t.Fatal(err) + } + marker := volumeDirtyPath(catalog[0].Path) + if err := os.Remove(marker); err != nil { + t.Fatal(err) + } + if err := os.Mkdir(marker, 0o700); err != nil { + t.Fatal(err) + } + child := filepath.Join(marker, "block-unlink") + if err := os.WriteFile(child, nil, 0o600); err != nil { + t.Fatal(err) + } + if _, err := m.SetVolumes(t.Context(), sb.ID, "", []types.Volume{}); err == nil || !strings.Contains(err.Error(), "clear dirty marker") { + t.Fatalf("marker unlink failure: %v", err) + } + if !volumeDirty(catalog[0].Path) || volumeHoldersOf(m, "data") != (volumeHolders{writers: 1}) { + t.Fatal("failed marker clear lost the marker or leaked an operation hold") + } + eng.detachUnclean = true + if _, err := m.SetVolumes(t.Context(), sb.ID, "", []types.Volume{}); err != nil { + t.Fatal(err) + } + if !volumeDirty(catalog[0].Path) || volumeHoldersOf(m, "data") != (volumeHolders{}) { + t.Fatal("absent-device retry cleared the marker or retained an admission hold") + } + if _, err := m.ClaimProvision(t.Context(), testKey, ClaimOptions{Volumes: []types.Volume{{Name: "data"}}}); !errors.Is(err, ErrVolumeNeedsRecovery) { + t.Fatalf("reader admission after failed marker clear: %v", err) + } + if err := os.Remove(child); err != nil { + t.Fatal(err) + } + if err := os.Remove(marker); err != nil { + t.Fatal(err) + } + if err := markVolumeDirty(catalog[0].Path); err != nil { + t.Fatal(err) + } + if _, err := m.ClaimProvision(t.Context(), testKey, ClaimOptions{Volumes: []types.Volume{{Name: "data", Mode: types.VolumeModeRW}}}); err != nil { + t.Fatalf("recovery writer admission: %v", err) + } +} + +func TestLiveWritableUncertainAdditionClearsOnlyAfterDetach(t *testing.T) { + eng := newFakeEngine() + catalog := liveCatalog(t) + m := newVolumeManager(t, eng, catalog) + sb := mustClaim(t, m, testKey) + eng.mountVolumeErr = errors.New("mount response lost") + if _, err := m.SetVolumes(t.Context(), sb.ID, "", []types.Volume{{Name: "data", Mode: types.VolumeModeRW}}); err == nil { + t.Fatal("expected unknown mount result") + } + if len(sb.Volumes) != 0 || len(sb.VolumesWant) != 1 || !volumeDirty(catalog[0].Path) { + t.Fatal("pending writable attachment lost intent or marker") + } + eng.mountVolumeErr = nil + if _, err := m.SetVolumes(t.Context(), sb.ID, "", []types.Volume{}); err != nil { + t.Fatal(err) + } + if volumeDirty(catalog[0].Path) || volumeHoldersOf(m, "data") != (volumeHolders{}) { + t.Fatal("uncertain addition cleanup retained marker or hold") + } +} + func TestWritableClaimMarksDirtyBeforeAttachAndClearsOnRelease(t *testing.T) { path := writeVolumeImage(t, "scratch.img", "data") eng := newFakeEngine() diff --git a/sandboxd/server/api.go b/sandboxd/server/api.go index 6e26e0c94..f26864d24 100644 --- a/sandboxd/server/api.go +++ b/sandboxd/server/api.go @@ -1,6 +1,8 @@ package server import ( + "errors" + "slices" "time" "github.com/cocoonstack/sandbox/sandboxd/types" @@ -42,6 +44,20 @@ func (r ClaimRequest) Key() types.PoolKey { return types.PoolKey{Template: r.Template, Net: r.Net, Size: r.Size}.Defaulted() } +func (r ClaimRequest) normalizedVolumes() ([]types.Volume, error) { + if !r.VolumesAttachOnly { + return types.ValidateVolumes(r.Volumes) + } + if len(r.Volumes) == 0 { + return nil, errors.New("volumes_attach_only requires at least one volume") + } + volumes := slices.Clone(r.Volumes) + for i := range volumes { + volumes[i].AttachOnly = true + } + return types.ValidateVolumes(volumes) +} + // ClaimResponse is the wire reply of POST /v1/claim; success and Redirect are exclusive. type ClaimResponse struct { ID string `json:"id,omitempty"` diff --git a/sandboxd/server/server.go b/sandboxd/server/server.go index 286abcddd..e3d6c4de2 100644 --- a/sandboxd/server/server.go +++ b/sandboxd/server/server.go @@ -65,6 +65,9 @@ var poolErrHTTP = []struct { {pool.ErrVolumeCapture, http.StatusConflict, ""}, {pool.ErrVolumeBusy, http.StatusConflict, ""}, {pool.ErrVolumeRegistrationConflict, http.StatusConflict, pool.ErrVolumeRegistrationConflict.Error()}, + {pool.ErrReplicaPresent, http.StatusServiceUnavailable, ""}, + {pool.ErrReplicatingPool, http.StatusConflict, ""}, + {pool.ErrReplicaStranded, http.StatusConflict, ""}, {pool.ErrVolumeNeedsRecovery, http.StatusConflict, ""}, {pool.ErrArchived, http.StatusConflict, ""}, {pool.ErrPaused, http.StatusConflict, ""}, @@ -356,7 +359,7 @@ func (s *Server) handleClaim(w http.ResponseWriter, r *http.Request) { } func (s *Server) handleVolumeClaim(w http.ResponseWriter, r *http.Request, req ClaimRequest, key types.PoolKey, hash, tenant string) { - volumes, err := types.ValidateVolumes(req.Volumes, req.VolumesAttachOnly) + volumes, err := req.normalizedVolumes() if err != nil { writeErr(w, http.StatusBadRequest, fmt.Sprintf("%v: %v", pool.ErrBadVolume, err)) return diff --git a/sandboxd/server/volume_mutation.go b/sandboxd/server/volume_mutation.go index 1e4f86f9a..86dd63983 100644 --- a/sandboxd/server/volume_mutation.go +++ b/sandboxd/server/volume_mutation.go @@ -60,7 +60,7 @@ func (s *Server) handleSandboxVolumes(w http.ResponseWriter, r *http.Request) { code, view.Error = mapped, msg } } - if code >= http.StatusInternalServerError { + if code >= http.StatusInternalServerError && !errors.Is(err, pool.ErrReplicaPresent) { log.WithFunc("server.handleSandboxVolumes").Errorf(r.Context(), err, "converge volumes for %s", id) } } diff --git a/sandboxd/server/volume_mutation_test.go b/sandboxd/server/volume_mutation_test.go index 48ff46b6b..4cd91e174 100644 --- a/sandboxd/server/volume_mutation_test.go +++ b/sandboxd/server/volume_mutation_test.go @@ -153,7 +153,7 @@ func TestLiveVolumeRoutesHideOtherTenants(t *testing.T) { func TestLiveVolumeRoutesReportRefusalsOverRecordedDiagnostics(t *testing.T) { const cause = "cannot unmount /data; close guest references and retry" - for _, refusal := range []error{pool.ErrVolumeBusy, pool.ErrVolumeCapture} { + for _, refusal := range []error{pool.ErrVolumeBusy, pool.ErrVolumeCapture, pool.ErrReplicaPresent, pool.ErrReplicatingPool, pool.ErrReplicaStranded} { t.Run(refusal.Error(), func(t *testing.T) { mgr := &fakeManager{setVolumes: func(string, string, []types.Volume) (types.SandboxVolumesResponse, error) { return types.SandboxVolumesResponse{VolumesWant: []types.Volume{{Name: "old"}}, States: []types.VolumeState{{Name: "old", State: "pending", Error: cause}}, Error: "volume change failed; retry the desired target"}, fmt.Errorf("%w: volume %q", refusal, "data") @@ -165,9 +165,45 @@ func TestLiveVolumeRoutesReportRefusalsOverRecordedDiagnostics(t *testing.T) { if err := json.UnmarshalRead(resp.Body, &got); err != nil { t.Fatal(err) } - if resp.StatusCode != http.StatusConflict || !strings.HasPrefix(got.Error, refusal.Error()) || len(got.States) != 1 || got.States[0].Error != cause { + wantStatus := http.StatusConflict + if errors.Is(refusal, pool.ErrReplicaPresent) { + wantStatus = http.StatusServiceUnavailable + if resp.Header.Get("Retry-After") == "" { + t.Fatal("temporary replica retirement refusal has no Retry-After") + } + } else if resp.Header.Get("Retry-After") != "" { + t.Fatal("permanent conflict advertises a retry delay") + } + if resp.StatusCode != wantStatus || !strings.HasPrefix(got.Error, refusal.Error()) || len(got.States) != 1 || got.States[0].Error != cause { t.Fatalf("status=%d view=%+v", resp.StatusCode, got) } }) } } + +func TestClaimVolumeCompatibilityExpansion(t *testing.T) { + for _, tt := range []struct { + name string + req ClaimRequest + want []types.Volume + err bool + }{ + {"default mount", ClaimRequest{Volumes: []types.Volume{{Name: "data"}}}, []types.Volume{{Name: "data", Mount: "/volumes/data"}}, false}, + {"legacy all devices", ClaimRequest{VolumesAttachOnly: true, Volumes: []types.Volume{{Name: "data"}, {Name: "work", Mode: types.VolumeModeRW}}}, []types.Volume{{Name: "data", AttachOnly: true}, {Name: "work", Mode: types.VolumeModeRW, AttachOnly: true}}, false}, + {"mixed", ClaimRequest{Volumes: []types.Volume{{Name: "data"}, {Name: "work", AttachOnly: true}}}, []types.Volume{{Name: "data", Mount: "/volumes/data"}, {Name: "work", AttachOnly: true}}, false}, + {"mixed reversed", ClaimRequest{Volumes: []types.Volume{{Name: "work", AttachOnly: true}, {Name: "data"}}}, []types.Volume{{Name: "work", AttachOnly: true}, {Name: "data", Mount: "/volumes/data"}}, false}, + {"empty legacy", ClaimRequest{VolumesAttachOnly: true}, nil, true}, + {"legacy mount conflict", ClaimRequest{VolumesAttachOnly: true, Volumes: []types.Volume{{Name: "data", Mount: "/data"}}}, nil, true}, + } { + t.Run(tt.name, func(t *testing.T) { + before := slices.Clone(tt.req.Volumes) + got, err := tt.req.normalizedVolumes() + if (err != nil) != tt.err || !slices.Equal(got, tt.want) { + t.Fatalf("normalized volumes=%+v err=%v, want %+v error=%t", got, err, tt.want, tt.err) + } + if !slices.Equal(tt.req.Volumes, before) { + t.Fatal("legacy expansion mutated caller input") + } + }) + } +} diff --git a/sandboxd/types/types.go b/sandboxd/types/types.go index 4cc8d3077..1c537ead5 100644 --- a/sandboxd/types/types.go +++ b/sandboxd/types/types.go @@ -752,8 +752,8 @@ type Volume struct { Name string `json:"name"` Mount string `json:"mount,omitempty"` Mode string `json:"mode,omitempty"` - // AttachOnly carries the claim's flag into validation; past it the empty Mount is the signal. - AttachOnly bool `json:"-"` + // AttachOnly leaves mounting and unmounting to the caller. + AttachOnly bool `json:"attach_only,omitzero"` } // RW reports whether the entry asks for write access. @@ -803,19 +803,11 @@ func VolumeRWNames(volumes []Volume) []string { return names } -// VolumesAttachOnly reports whether the caller mounts the devices itself. -func VolumesAttachOnly(volumes []Volume) bool { - return slices.ContainsFunc(volumes, func(v Volume) bool { return v.AttachOnly }) -} - // ValidateVolumes returns detached entries with defaults filled and modes normalized. -func ValidateVolumes(volumes []Volume, attachOnly bool) ([]Volume, error) { +func ValidateVolumes(volumes []Volume) ([]Volume, error) { if len(volumes) > MaxClaimVolumes { return nil, fmt.Errorf("volumes must contain at most %d entries, got %d", MaxClaimVolumes, len(volumes)) } - if attachOnly && len(volumes) == 0 { - return nil, errors.New("volumes_attach_only requires at least one volume") - } applied := make([]Volume, len(volumes)) names := make(map[string]struct{}, len(volumes)) for i, volume := range volumes { @@ -834,7 +826,7 @@ func ValidateVolumes(volumes []Volume, attachOnly bool) ([]Volume, error) { default: return nil, fmt.Errorf("volumes[%d] mode %q must be %s or %s", i, mode, VolumeModeRO, VolumeModeRW) } - if attachOnly { + if volume.AttachOnly { if volume.Mount != "" { return nil, fmt.Errorf("volumes[%d] mount %q is meaningless when the caller mounts the device itself", i, volume.Mount) } @@ -847,6 +839,9 @@ func ValidateVolumes(volumes []Volume, attachOnly bool) ([]Volume, error) { } for j := range i { other := applied[j].Mount + if other == "" { + continue + } switch { case mount == other: return nil, fmt.Errorf("volumes[%d] mount %q duplicates volumes[%d]", i, mount, j) diff --git a/sandboxd/types/volume_test.go b/sandboxd/types/volume_test.go index 9549af49c..79947d597 100644 --- a/sandboxd/types/volume_test.go +++ b/sandboxd/types/volume_test.go @@ -31,7 +31,7 @@ func TestValidVolumeName(t *testing.T) { func TestValidateVolumes(t *testing.T) { volumes := []Volume{{Name: "dataset"}, {Name: "weights-1", Mount: "/models"}} wantInput := slices.Clone(volumes) - got, err := ValidateVolumes(volumes, false) + got, err := ValidateVolumes(volumes) if err != nil { t.Fatalf("ValidateVolumes: %v", err) } @@ -54,7 +54,7 @@ func TestValidateVolumes(t *testing.T) { {Name: "f"}, {Name: "g"}, {Name: "h"}, - }, false); err != nil || len(got) != MaxClaimVolumes { + }); err != nil || len(got) != MaxClaimVolumes { t.Errorf("ValidateVolumes at limit: got=%v err=%v", got, err) } @@ -75,7 +75,7 @@ func TestValidateVolumes(t *testing.T) { {"uppercase-mode", []Volume{{Name: "dataset", Mode: "RW"}}}, } { t.Run(tt.name, func(t *testing.T) { - if _, err := ValidateVolumes(tt.volumes, false); err == nil { + if _, err := ValidateVolumes(tt.volumes); err == nil { t.Fatal("ValidateVolumes succeeded") } }) @@ -87,7 +87,7 @@ func TestValidateVolumesNormalizesMode(t *testing.T) { {Name: "shared"}, {Name: "explicit", Mode: VolumeModeRO}, {Name: "writable", Mode: VolumeModeRW}, - }, false) + }) if err != nil { t.Fatalf("ValidateVolumes: %v", err) } @@ -112,10 +112,10 @@ func TestValidateVolumesNormalizesMode(t *testing.T) { func TestValidateVolumesAttachOnly(t *testing.T) { got, err := ValidateVolumes([]Volume{ - {Name: "dataset"}, - {Name: "scratch", Mode: VolumeModeRW}, - {Name: "explicit", Mode: VolumeModeRO}, - }, true) + {Name: "dataset", AttachOnly: true}, + {Name: "scratch", Mode: VolumeModeRW, AttachOnly: true}, + {Name: "explicit", Mode: VolumeModeRO, AttachOnly: true}, + }) if err != nil { t.Fatalf("ValidateVolumes: %v", err) } @@ -127,49 +127,43 @@ func TestValidateVolumesAttachOnly(t *testing.T) { if !slices.Equal(got, want) { t.Errorf("volumes %v, want %v", got, want) } - if !VolumesAttachOnly(got) { - t.Error("validated attach-only set does not report itself") - } - if VolumesAttachOnly(nil) || VolumesAttachOnly([]Volume{{Name: "dataset", Mount: "/volumes/dataset"}}) { - t.Error("mounted set reports itself attach-only") - } - - for _, tt := range []struct { - name string - volumes []Volume - }{ - {"no volumes", nil}, - {"default mount spelled out", []Volume{{Name: "dataset", Mount: "/volumes/dataset"}}}, - {"custom mount", []Volume{{Name: "dataset", Mount: "/datasets"}}}, - {"one entry of many", []Volume{{Name: "dataset"}, {Name: "scratch", Mount: "/scratch"}}}, - {"duplicate name", []Volume{{Name: "dataset"}, {Name: "dataset"}}}, - {"invalid name", []Volume{{Name: "bad/name"}}}, - {"unknown mode", []Volume{{Name: "dataset", Mode: "readwrite"}}}, - } { - t.Run(tt.name, func(t *testing.T) { - if _, err := ValidateVolumes(tt.volumes, true); err == nil { - t.Fatal("ValidateVolumes succeeded") - } - }) + if _, err = ValidateVolumes([]Volume{{Name: "dataset", Mount: "/datasets", AttachOnly: true}}); err == nil { + t.Fatal("accepted attach-only with a mount") } } func TestValidateVolumesRejectsGuestOSMounts(t *testing.T) { for _, root := range append([]string{"/"}, guestOSMountRoots...) { t.Run(root, func(t *testing.T) { - if _, err := ValidateVolumes([]Volume{{Name: "dataset", Mount: root}}, false); err == nil { + if _, err := ValidateVolumes([]Volume{{Name: "dataset", Mount: root}}); err == nil { t.Fatalf("accepted OS mount %q", root) } if root != "/" { - if _, err := ValidateVolumes([]Volume{{Name: "dataset", Mount: root + "/child"}}, false); err == nil { + if _, err := ValidateVolumes([]Volume{{Name: "dataset", Mount: root + "/child"}}); err == nil { t.Fatalf("accepted mount under OS root %q", root) } } }) } for _, mount := range []string{"/data", "/home/dataset", "/opt-data", "/usrdata"} { - if _, err := ValidateVolumes([]Volume{{Name: "dataset", Mount: mount}}, false); err != nil { + if _, err := ValidateVolumes([]Volume{{Name: "dataset", Mount: mount}}); err != nil { t.Errorf("rejected allowed mount %q: %v", mount, err) } } } + +func TestValidateMixedVolumesPreservesPerEntryIntent(t *testing.T) { + request := []Volume{{Name: "raw", AttachOnly: true}, {Name: "mounted"}, {Name: "raw-writer", Mode: VolumeModeRW, AttachOnly: true}, {Name: "writer", Mode: VolumeModeRW}} + got, err := ValidateVolumes(request) + if err != nil { + t.Fatal(err) + } + want := []Volume{{Name: "raw", AttachOnly: true}, {Name: "mounted", Mount: "/volumes/mounted"}, {Name: "raw-writer", Mode: VolumeModeRW, AttachOnly: true}, {Name: "writer", Mount: "/volumes/writer", Mode: VolumeModeRW}} + if !slices.Equal(got, want) { + t.Fatalf("mixed intent changed: got %+v, want %+v", got, want) + } + again, err := ValidateVolumes(got) + if err != nil || !slices.Equal(again, want) { + t.Fatalf("second validation changed mixed intent: %+v, %v", again, err) + } +} diff --git a/sdk/go/client.go b/sdk/go/client.go index ccb932fd1..587f2422c 100644 --- a/sdk/go/client.go +++ b/sdk/go/client.go @@ -110,18 +110,6 @@ func (r claimRequest) rejectPinnedAxes() error { return nil } -func (r claimRequest) validateVolumes() error { - for _, v := range r.Volumes { - if v.Mode != "" && v.Mode != volumeModeRW { - return fmt.Errorf("volume %q: mode must be \"\", %q, or %q, got %q", v.Name, volumeModeRO, volumeModeRW, v.Mode) - } - if r.VolumesAttachOnly && v.Mount != "" { - return fmt.Errorf("volume %q: mount %q is meaningless with WithVolumesAttachOnly, which leaves mounting to the caller", v.Name, v.Mount) - } - } - return nil -} - type claimResponse struct { ID string `json:"id"` Token string `json:"token"` @@ -172,9 +160,6 @@ func (c *Client) New(ctx context.Context, template string, opts ...Option) (*San for _, opt := range opts { opt(&claim) } - if err := claim.validateVolumes(); err != nil { - return nil, err - } addr, cr, err := claimFollow(c.addr, "claim", func(noRedirect, requirePromoted bool) ([]byte, error) { claim.NoRedirect, claim.RequirePromoted = noRedirect, requirePromoted return encodeBody("claim", claim) diff --git a/sdk/go/options.go b/sdk/go/options.go index 20b75401d..a1bca0e6d 100644 --- a/sdk/go/options.go +++ b/sdk/go/options.go @@ -17,9 +17,6 @@ const ( Large Size = "large" XLarge Size = "xlarge" XXLarge Size = "2xlarge" - - volumeModeRO = "ro" - volumeModeRW = "rw" ) // NetShape selects the sandbox network lane. @@ -35,6 +32,8 @@ type Volume struct { Mount string `json:"mount,omitempty"` // Mode is "rw" for a writable mount; empty (or "ro") means read-only. Mode string `json:"mode,omitempty"` + // AttachOnly leaves mounting and unmounting to the caller. + AttachOnly bool `json:"attach_only,omitzero"` } // VolumeInfo describes one visible fleet catalog entry. @@ -80,18 +79,10 @@ func WithSize(s Size) Option { // read-only, set Volume.Mode to "rw" for a writable mount. func WithVolumes(volumes ...Volume) Option { volumes = slices.Clone(volumes) - for i, v := range volumes { - if v.Mode == volumeModeRO { - volumes[i].Mode = "" - } - } return func(r *claimRequest) { r.Volumes = volumes } } -// WithVolumesAttachOnly attaches the claim's volumes without mounting them; -// the workload finds each device by its serial and owns the mount — and its -// flush — from there: release without a clean self-umount discards unsynced -// guest pages, since the sandbox performs no sync of its own. +// WithVolumesAttachOnly is the legacy claim-wide form of Volume.AttachOnly. func WithVolumesAttachOnly() Option { return func(r *claimRequest) { r.VolumesAttachOnly = true } } diff --git a/sdk/go/template.go b/sdk/go/template.go index 34ee31f7a..c7f45c7bd 100644 --- a/sdk/go/template.go +++ b/sdk/go/template.go @@ -29,9 +29,6 @@ func (t *Template) New(ctx context.Context, opts ...Option) (*Sandbox, error) { if err := claim.rejectPinnedAxes(); err != nil { return nil, err } - if err := claim.validateVolumes(); err != nil { - return nil, err - } claim.Net, claim.Size = t.net, t.size noVolumes := len(claim.Volumes) == 0 addr, cr, err := claimFollow(t.addr, "claim", func(noRedirect, _ bool) ([]byte, error) { diff --git a/sdk/go/volume_mutation_test.go b/sdk/go/volume_mutation_test.go index 7f65427c7..761fe55d5 100644 --- a/sdk/go/volume_mutation_test.go +++ b/sdk/go/volume_mutation_test.go @@ -13,31 +13,28 @@ import ( ) func TestLiveVolumesUseOwnerAndAPIToken(t *testing.T) { - for _, method := range []string{http.MethodGet, http.MethodPut, http.MethodPatch} { - t.Run(method, func(t *testing.T) { + for _, request := range []struct{ method, body string }{ + {http.MethodGet, ""}, + {http.MethodPut, `{"volumes":[{"name":"data"}]}`}, + {http.MethodPatch, `{"add":[{"name":"data"}],"remove":["old"]}`}, + } { + t.Run(request.method, func(t *testing.T) { ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { - if r.Method != method || r.URL.Path != "/v1/sandboxes/sb_1/volumes" || r.Header.Get("Authorization") != "Bearer operator-token" { + if r.Method != request.method || r.URL.Path != "/v1/sandboxes/sb_1/volumes" || r.Header.Get("Authorization") != "Bearer operator-token" { t.Errorf("request: %s %s %s", r.Method, r.URL.Path, r.Header.Get("Authorization")) } body, _ := io.ReadAll(r.Body) - want := "" - switch method { - case http.MethodPut: - want = `{"volumes":[{"name":"data"}]}` - case http.MethodPatch: - want = `{"add":[{"name":"data"}],"remove":["old"]}` - } - if string(body) != want { - t.Errorf("body=%s want=%s", body, want) + if string(body) != request.body { + t.Errorf("body=%s want=%s", body, request.body) } - _, _ = io.WriteString(w, `{"volumes_want":[{"name":"data","mount":"/data"}],"volumes":[{"name":"data","mount":"/data"}],"converging":true,"states":[{"name":"data","state":"attached"}]}`) + _, _ = io.WriteString(w, `{"volumes_want":[{"name":"data","attach_only":true}],"volumes":[{"name":"data","attach_only":true}],"converging":true,"states":[{"name":"data","state":"attached"}]}`) })) t.Cleanup(ts.Close) c := testClient(t, ts, WithAPIToken("operator-token")) sb := &Sandbox{ID: "sb_1", token: "sandbox-token", owner: c.addr, c: c} c.addr = "unused.invalid:7777" - got, err := callVolumes(t, sb, method) - if err != nil || !slices.Equal(got.Volumes, []Volume{{Name: "data", Mount: "/data"}}) || !slices.Equal(got.VolumesWant, got.Volumes) || got.States[0].State != VolumeAttached || !got.Converging { + got, err := callVolumes(t, sb, request.method) + if err != nil || !slices.Equal(got.Volumes, []Volume{{Name: "data", AttachOnly: true}}) || !slices.Equal(got.VolumesWant, got.Volumes) || got.States[0].State != VolumeAttached || !got.Converging { t.Fatalf("got=%+v err=%v", got, err) } got.Volumes[0].Name, got.VolumesWant[0].Name = "caller copy", "caller copy" @@ -51,7 +48,7 @@ func TestLiveVolumesUseOwnerAndAPIToken(t *testing.T) { func TestLiveVolumeFailureRetainsPartialView(t *testing.T) { ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { w.WriteHeader(http.StatusInternalServerError) - _, _ = io.WriteString(w, `{"volumes_want":[{"name":"new","mount":"/new"}],"volumes":[{"name":"old","mount":"/old"}],"converging":false,"states":[{"name":"new","state":"pending","error":"retry"},{"name":"old","state":"detaching"}],"error":"convergence failed"}`) + _, _ = io.WriteString(w, `{"volumes_want":[{"name":"new","mode":"rw","attach_only":true}],"volumes":[{"name":"old","attach_only":true}],"converging":false,"states":[{"name":"new","state":"pending","error":"retry"},{"name":"old","state":"detaching"}],"error":"convergence failed"}`) })) t.Cleanup(ts.Close) c := testClient(t, ts) @@ -63,7 +60,10 @@ func TestLiveVolumeFailureRetainsPartialView(t *testing.T) { if !isVolume || !isAPI || api.Status != http.StatusInternalServerError || failure.Response == nil { t.Fatalf("%s error lost response or APIError: %v", method, err) } - if !reflect.DeepEqual(got, *failure.Response) || got.States[0].State != "pending" || got.States[1].State != "detaching" || sb.VolumesWant[0].Name != "new" || sb.Volumes[0].Name != "old" { + if !reflect.DeepEqual(got, *failure.Response) || got.States[0].State != "pending" || got.States[1].State != "detaching" { + t.Fatalf("partial view=%+v", got) + } + if !slices.Equal(sb.VolumesWant, []Volume{{Name: "new", Mode: "rw", AttachOnly: true}}) || !slices.Equal(sb.Volumes, []Volume{{Name: "old", AttachOnly: true}}) { t.Fatalf("partial view=%+v handle=%+v", got, sb) } failure.Response.Volumes[0].Name = "error caller" @@ -177,6 +177,30 @@ func TestLiveVolumeIncompleteViewPreservesHandle(t *testing.T) { } } +func TestLiveVolumeMutationEncodesDeviceFields(t *testing.T) { + const want = `{"volumes":[{"name":"device","mode":"rw","attach_only":true},{"name":"mounted","mount":"/work"}]}` + ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + body, err := io.ReadAll(r.Body) + if err != nil { + t.Error(err) + } + if r.Method != http.MethodPut || string(body) != want { + t.Errorf("request=%s %s want=PUT %s", r.Method, body, want) + } + _, _ = io.WriteString(w, `{"volumes_want":[],"volumes":[],"states":[],"converging":false}`) + })) + t.Cleanup(ts.Close) + c := testClient(t, ts) + sb := &Sandbox{ID: "sb_1", owner: c.addr, c: c} + _, err := sb.SetVolumes(t.Context(), []Volume{ + {Name: "device", Mode: "rw", AttachOnly: true}, + {Name: "mounted", Mount: "/work"}, + }) + if err != nil { + t.Fatal(err) + } +} + func callVolumes(t *testing.T, sb *Sandbox, method string) (SandboxVolumesResponse, error) { t.Helper() switch method { diff --git a/sdk/go/volumes.go b/sdk/go/volumes.go index 69e1dea41..4d7e39108 100644 --- a/sdk/go/volumes.go +++ b/sdk/go/volumes.go @@ -55,7 +55,7 @@ func (s *Sandbox) GetVolumes(ctx context.Context) (SandboxVolumesResponse, error return s.requestVolumes(ctx, http.MethodGet, nil, "get volumes") } -// SetVolumes converges the full desired set; an explicit empty slice clears mutable mounts, and nil is rejected. +// SetVolumes converges the full desired set; an explicit empty slice detaches all volumes, and nil is rejected. func (s *Sandbox) SetVolumes(ctx context.Context, volumes []Volume) (SandboxVolumesResponse, error) { if volumes == nil { return SandboxVolumesResponse{}, errors.New("set volumes: use an explicit empty slice to clear volumes") @@ -65,7 +65,7 @@ func (s *Sandbox) SetVolumes(ctx context.Context, volumes []Volume) (SandboxVolu }{Volumes: volumes}, "set volumes") } -// PatchVolumes adds or removes read-only mounted entries by name, retaining immutable entries. +// PatchVolumes adds or removes volumes by name. func (s *Sandbox) PatchVolumes(ctx context.Context, patch VolumePatch) (SandboxVolumesResponse, error) { return s.requestVolumes(ctx, http.MethodPatch, patch, "patch volumes") } diff --git a/sdk/go/volumes_test.go b/sdk/go/volumes_test.go index d8a2c14e4..57dda6cee 100644 --- a/sdk/go/volumes_test.go +++ b/sdk/go/volumes_test.go @@ -196,7 +196,7 @@ func TestWithVolumesEncodesMode(t *testing.T) { mode string }{ {"empty stays omitted", ""}, - {"ro normalizes to omitted", "ro"}, + {"ro stays explicit", "ro"}, {"rw rides the wire", "rw"}, } for _, tt := range tests { @@ -215,9 +215,9 @@ func TestWithVolumesEncodesMode(t *testing.T) { t.Fatalf("New: %v", err) } got, present := raw.Volumes[0]["mode"] - if tt.mode == volumeModeRW { - if !present || got != volumeModeRW { - t.Errorf("wire mode = %v (present=%v), want %q", got, present, volumeModeRW) + if tt.mode != "" { + if !present || got != tt.mode { + t.Errorf("wire mode = %v (present=%v), want %q", got, present, tt.mode) } } else if present { t.Errorf("wire mode = %v, want omitted", got) @@ -231,7 +231,7 @@ func TestWithVolumesAttachOnlySendsFlagAndDecodesMountlessEcho(t *testing.T) { Volumes []map[string]any `json:"volumes"` VolumesAttachOnly bool `json:"volumes_attach_only"` } - echo := []Volume{{Name: "imagenet"}, {Name: "scratch", Mode: "rw"}} + echo := []Volume{{Name: "imagenet", AttachOnly: true}, {Name: "scratch", Mode: "rw", AttachOnly: true}} ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { if err := json.UnmarshalRead(r.Body, &raw); err != nil { t.Errorf("decode body: %v", err) @@ -259,20 +259,7 @@ func TestWithVolumesAttachOnlySendsFlagAndDecodesMountlessEcho(t *testing.T) { } } -func TestWithVolumesAttachOnlyRejectsMountLocally(t *testing.T) { - ts := httptest.NewServer(http.HandlerFunc(func(http.ResponseWriter, *http.Request) { - t.Error("server should not be contacted for a locally invalid claim") - })) - t.Cleanup(ts.Close) - - _, err := testClient(t, ts).New(t.Context(), "rt:24.04", - WithVolumes(Volume{Name: "imagenet", Mount: "/datasets"}), WithVolumesAttachOnly()) - if err == nil || !strings.Contains(err.Error(), "meaningless with WithVolumesAttachOnly") { - t.Errorf("err = %v, want local mount rejection", err) - } -} - -func TestRejectsInvalidVolumeModeLocally(t *testing.T) { +func TestVolumePolicyErrorsComeFromServer(t *testing.T) { tests := []struct { name string claim func(ctx context.Context, c *Client) (*Sandbox, error) @@ -287,14 +274,15 @@ func TestRejectsInvalidVolumeModeLocally(t *testing.T) { } for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { - ts := httptest.NewServer(http.HandlerFunc(func(http.ResponseWriter, *http.Request) { - t.Error("server should not be contacted for a locally invalid mode") + ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.WriteHeader(http.StatusBadRequest) + _, _ = w.Write([]byte(`{"error":"mode must be ro or rw"}`)) })) t.Cleanup(ts.Close) _, err := tt.claim(t.Context(), testClient(t, ts)) if err == nil || !strings.Contains(err.Error(), "mode must be") { - t.Errorf("err = %v, want local mode rejection", err) + t.Errorf("err = %v, want server mode rejection", err) } }) } diff --git a/sdk/python/cocoonsandbox/client.py b/sdk/python/cocoonsandbox/client.py index aa3c8fd19..cdf8587ec 100644 --- a/sdk/python/cocoonsandbox/client.py +++ b/sdk/python/cocoonsandbox/client.py @@ -9,7 +9,7 @@ import threading import time import urllib.parse -from collections.abc import Callable, Iterable, Mapping, Sequence +from collections.abc import Callable, Iterable, Sequence from typing import Any, TypeVar, cast from .checkpoint import Checkpoint @@ -18,6 +18,7 @@ from .errors import APIError, SandboxTimeout, require_field from .sandbox import Sandbox from .template import _claim_body, _template_query +from .volumes import Volume T = TypeVar("T") @@ -60,7 +61,7 @@ def new( size: str = "", ttl_seconds: int = 0, claim_ref: str = "", - volumes: list[str | Mapping[str, str]] | None = None, + volumes: list[str | Volume] | None = None, mount: bool = True, *, deadline: float | None = None, diff --git a/sdk/python/cocoonsandbox/sandbox.py b/sdk/python/cocoonsandbox/sandbox.py index 1a83fa869..b3368d440 100644 --- a/sdk/python/cocoonsandbox/sandbox.py +++ b/sdk/python/cocoonsandbox/sandbox.py @@ -7,7 +7,7 @@ import threading import time from collections import deque -from collections.abc import Callable, Iterator, Mapping +from collections.abc import Callable, Iterator from typing import TYPE_CHECKING, Any, cast from .checkpoint import Checkpoint @@ -45,7 +45,7 @@ def __init__( deadline: str = "", from_checkpoint: str = "", template_digest: str = "", - volumes: list[dict[str, str]] | None = None, + volumes: list[Volume] | None = None, ) -> None: self._client = client self.id = id @@ -54,8 +54,8 @@ def __init__( self.deadline = deadline self.from_checkpoint = from_checkpoint self.template_digest = template_digest - self.volumes = [dict(volume) for volume in volumes or []] - self.volumes_want = [dict(volume) for volume in self.volumes] + self.volumes = [volume.copy() for volume in volumes or []] + self.volumes_want = [volume.copy() for volume in self.volumes] self._pool = ConnPool(client.keep_alive) self._proto = 0 @@ -73,29 +73,27 @@ def get_volumes(self, *, timeout: float | None = None) -> SandboxVolumesResponse """Reads the desired and applied volumes from the owning node and records them on this handle.""" return self._volumes_request("GET", None, "get volumes", timeout) - def set_volumes( - self, volumes: list[str | Mapping[str, str] | Volume], *, timeout: float | None = None - ) -> SandboxVolumesResponse: - """Replaces the desired set and converges; an explicit empty list clears mutable mounts.""" - if volumes is None or isinstance(volumes, str): + def set_volumes(self, volumes: list[str | Volume], *, timeout: float | None = None) -> SandboxVolumesResponse: + """Replaces the desired set and converges; an explicit empty list detaches all volumes.""" + if not isinstance(volumes, list): raise TypeError("set volumes requires an explicit list; use [] to clear volumes") return self._volumes_request( - "PUT", {"volumes": [_volume_body(volume, True) for volume in volumes]}, "set volumes", timeout + "PUT", {"volumes": [_volume_body(volume) for volume in volumes]}, "set volumes", timeout ) def patch_volumes( self, *, - add: list[str | Mapping[str, str] | Volume] | None = None, + add: list[str | Volume] | None = None, remove: list[str] | None = None, timeout: float | None = None, ) -> SandboxVolumesResponse: - """Adds or removes mounted read-only entries by name; omitted arguments leave the set unchanged.""" - if isinstance(add, str) or isinstance(remove, str): + """Adds or removes volumes by name; omitted arguments leave the set unchanged.""" + if (add is not None and not isinstance(add, list)) or (remove is not None and not isinstance(remove, list)): raise TypeError("patch volumes takes lists of volumes and names") body: dict[str, Any] = {} if add is not None: - body["add"] = [_volume_body(volume, True) for volume in add] + body["add"] = [_volume_body(volume) for volume in add] if remove is not None: body["remove"] = list(remove) return self._volumes_request("PATCH", body, "patch volumes", timeout) @@ -445,8 +443,8 @@ def _volumes_request( return view def _record_volumes(self, view: SandboxVolumesResponse) -> None: - self.volumes_want = [cast(dict[str, str], volume.copy()) for volume in view["volumes_want"]] - self.volumes = [cast(dict[str, str], volume.copy()) for volume in view["volumes"]] + self.volumes_want = [volume.copy() for volume in view["volumes_want"]] + self.volumes = [volume.copy() for volume in view["volumes"]] def _dial(self, deadline: float | None = None) -> Conn: return self._client._dial(self.owner, self.id, self.token, deadline) diff --git a/sdk/python/cocoonsandbox/template.py b/sdk/python/cocoonsandbox/template.py index 5e468aa08..94db0da64 100644 --- a/sdk/python/cocoonsandbox/template.py +++ b/sdk/python/cocoonsandbox/template.py @@ -3,10 +3,9 @@ from __future__ import annotations import urllib.parse -from collections.abc import Mapping from typing import TYPE_CHECKING, Any -from .volumes import _volume_body +from .volumes import Volume, _volume_body if TYPE_CHECKING: from .client import Client @@ -27,7 +26,7 @@ def __init__(self, client: Client, addr: str, name: str, net: str, size: str, co def new( self, ttl_seconds: int = 0, - volumes: list[str | Mapping[str, str]] | None = None, + volumes: list[str | Volume] | None = None, mount: bool = True, *, deadline: float | None = None, @@ -55,7 +54,7 @@ def _claim_body( net: str, size: str, ttl_seconds: int, - volumes: list[str | Mapping[str, str]] | None = None, + volumes: list[str | Volume] | None = None, mount: bool = True, claim_ref: str = "", ) -> dict[str, Any]: @@ -66,8 +65,10 @@ def _claim_body( claim["size"] = size if ttl_seconds: claim["ttl_seconds"] = ttl_seconds + if volumes is not None and not isinstance(volumes, list): + raise TypeError("volumes must be a list") if volumes: - claim["volumes"] = [_volume_body(volume, mount) for volume in volumes] + claim["volumes"] = [_volume_body(volume) for volume in volumes] if not mount: claim["volumes_attach_only"] = True if claim_ref: diff --git a/sdk/python/cocoonsandbox/volumes.py b/sdk/python/cocoonsandbox/volumes.py index ce156554d..209aaa50e 100644 --- a/sdk/python/cocoonsandbox/volumes.py +++ b/sdk/python/cocoonsandbox/volumes.py @@ -16,10 +16,11 @@ class _VolumeName(TypedDict): class Volume(_VolumeName, total=False): - """A catalog volume by name, with its guest mount point and an `rw` mode for writable claims.""" + """A catalog volume with a guest mount or caller-managed device attachment.""" mount: str mode: str + attach_only: bool class _VolumeState(TypedDict): @@ -51,29 +52,15 @@ def _volume_response(reply: dict[str, Any] | None) -> SandboxVolumesResponse | N reply is None or not all(isinstance(reply.get(key), list) for key in ("volumes_want", "volumes", "states")) or not isinstance(reply.get("converging"), bool) + or any(not isinstance(entry, dict) for key in ("volumes_want", "volumes") for entry in reply[key]) ): return None return cast(SandboxVolumesResponse, reply) -def _volume_body(volume: str | Mapping[str, str] | Volume, mount: bool) -> dict[str, str]: +def _volume_body(volume: str | Volume) -> Volume: if isinstance(volume, str): return {"name": volume} if not isinstance(volume, Mapping): raise TypeError("volume must be a name string or mapping") - unknown = sorted(set(volume) - {"name", "mount", "mode"}) - if unknown: - raise TypeError( - f"volume mapping accepts only name, mount, and mode, got unexpected key(s): {', '.join(unknown)}" - ) - if not mount and "mount" in volume: - raise TypeError("volume mount is meaningless with mount=False, which leaves mounting to the caller") - body = dict(cast(Mapping[str, str], volume)) - mode = body.get("mode") - if mode in (None, "", "ro"): - body.pop("mode", None) - elif mode != "rw": - name = volume.get("name") - prefix = f"volume {name!r}: " if name is not None else "" - raise TypeError(f"{prefix}mode must be 'rw' or 'ro', got {mode!r}") - return body + return cast(Volume, dict(volume)) diff --git a/sdk/python/tests/test_client.py b/sdk/python/tests/test_client.py index 90b799dca..bd44278f1 100644 --- a/sdk/python/tests/test_client.py +++ b/sdk/python/tests/test_client.py @@ -71,17 +71,21 @@ def test_claim_sends_volumes(node): ] +def test_claim_rejects_invalid_volumes(node): + with pytest.raises(TypeError, match="name string or mapping"): + Client(node).new("rt:24.04", volumes=[("imagenet", "/datasets/imagenet")]) + + @pytest.mark.parametrize( - ("volumes", "match"), + "new", [ - ([("imagenet", "/datasets/imagenet")], "name string or mapping"), - ([{"name": "imagenet", "mode": "rwx"}], "volume 'imagenet': mode must be 'rw' or 'ro', got 'rwx'"), - ([{"name": "imagenet", "bogus": "x"}], r"unexpected key\(s\): bogus"), + lambda node: Client(node).new("rt:24.04", volumes="data"), + lambda node: Template(Client(node), node, "task:v1", "none", "small").new(volumes="data"), ], ) -def test_claim_rejects_invalid_volumes(node, volumes, match): - with pytest.raises(TypeError, match=match): - Client(node).new("rt:24.04", volumes=volumes) +def test_claim_rejects_a_bare_volume_string(node, new): + with pytest.raises(TypeError, match="volumes must be a list"): + new(node) def test_claim_sends_volume_mode_rw(node): @@ -100,11 +104,12 @@ def test_claim_sends_volume_mode_rw(node): assert sb.volumes == [{"name": "scratch", "mount": "/data", "mode": "rw"}] -def test_claim_omits_volume_mode_ro(node): +def test_claim_preserves_explicit_volume_mode(node): seen = recording_claim({"id": "sb_1", "token": "tok"}) Client(node).new("rt:24.04", volumes=[{"name": "imagenet", "mode": "ro"}]) Client(node).new("rt:24.04", volumes=[{"name": "imagenet", "mode": ""}]) - assert seen[0]["volumes"] == seen[1]["volumes"] == [{"name": "imagenet"}], seen + assert seen[0]["volumes"] == [{"name": "imagenet", "mode": "ro"}] + assert seen[1]["volumes"] == [{"name": "imagenet", "mode": ""}] def test_claim_attaches_volumes_without_mounting(node): @@ -130,13 +135,12 @@ def test_template_claim_attaches_volumes_without_mounting(node): assert sb.volumes == [{"name": "imagenet"}] -def test_claim_rejects_mount_without_mounting(node): - with pytest.raises(TypeError, match="meaningless with mount=False"): - Client(node).new("rt:24.04", volumes=[{"name": "imagenet", "mount": "/datasets"}], mount=False) - with pytest.raises(TypeError, match="meaningless with mount=False"): - Template(Client(node), node, "task:v1", "none", "small").new( - volumes=[{"name": "imagenet", "mount": "/datasets"}], mount=False - ) +def test_claim_leaves_volume_policy_to_server(node): + seen = recording_claim({"id": "sb_1", "token": "tok"}) + entry = {"name": "imagenet", "mount": "/datasets", "attach_only": True, "mode": "future-mode"} + Client(node).new("rt:24.04", volumes=[entry], mount=False) + Template(Client(node), node, "task:v1", "none", "small").new(volumes=[entry], mount=False) + assert [request["volumes"] for request in seen] == [[entry], [entry]] def test_claim_keeps_mounting_by_default(node): diff --git a/sdk/python/tests/test_volumes.py b/sdk/python/tests/test_volumes.py index 3226b82d8..e7ddabb5f 100644 --- a/sdk/python/tests/test_volumes.py +++ b/sdk/python/tests/test_volumes.py @@ -1,5 +1,7 @@ """Live volume HTTP contract and partial convergence recovery.""" +from collections import UserDict + import pytest from conftest import FakeNode, sandbox_at @@ -14,8 +16,31 @@ } -@pytest.mark.parametrize("method", ["GET", "PUT", "PATCH"]) -def test_volume_requests_use_owner_and_api_token(spawn_node, method): +def test_volume_mapping_is_json_encoded(spawn_node): + seen = [] + + def reply(body, path): + seen.append(body) + return 200, VIEW + + sb = sandbox_at(spawn_node({("PUT", PATH): reply})) + sb.set_volumes([UserDict({"name": "data", "mount": "/data"})]) + assert seen == [{"volumes": [{"name": "data", "mount": "/data"}]}] + + +@pytest.mark.parametrize( + "method,body,call", + [ + ("GET", {}, lambda sb: sb.get_volumes()), + ("PUT", {"volumes": [{"name": "data"}]}, lambda sb: sb.set_volumes(["data"])), + ( + "PATCH", + {"add": [{"name": "data"}], "remove": ["old"]}, + lambda sb: sb.patch_volumes(add=["data"], remove=["old"]), + ), + ], +) +def test_volume_requests_use_owner_and_api_token(spawn_node, method, body, call): seen = [] def reply(body, path): @@ -25,15 +50,8 @@ def reply(body, path): addr = spawn_node({(method, PATH): reply}) sb = sandbox_at(addr, api_token="operator-token") sb._client.addr = "unused.invalid:7777" - if method == "GET": - got = sb.get_volumes() - assert seen == [{}] - elif method == "PUT": - got = sb.set_volumes(["data"]) - assert seen == [{"volumes": [{"name": "data"}]}] - else: - got = sb.patch_volumes(add=["data"], remove=["old"]) - assert seen == [{"add": [{"name": "data"}], "remove": ["old"]}] + got = call(sb) + assert seen == [body] assert FakeNode.last_headers["Authorization"] == "Bearer operator-token" assert got == VIEW assert sb.volumes == VIEW["volumes"] @@ -43,11 +61,18 @@ def reply(body, path): assert sb.volumes[0]["name"] == sb.volumes_want[0]["name"] == "data" -@pytest.mark.parametrize("method", ["GET", "PUT", "PATCH"]) -def test_volume_error_preserves_typed_partial_view(spawn_node, method): +@pytest.mark.parametrize( + "method,call", + [ + ("GET", lambda sb: sb.get_volumes()), + ("PUT", lambda sb: sb.set_volumes(["new"])), + ("PATCH", lambda sb: sb.patch_volumes(remove=["old"])), + ], +) +def test_volume_error_preserves_typed_partial_view(spawn_node, method, call): partial = { - "volumes_want": [{"name": "new", "mount": "/new"}], - "volumes": [{"name": "old", "mount": "/old"}], + "volumes_want": [{"name": "new", "mode": "rw", "attach_only": True}], + "volumes": [{"name": "old", "attach_only": True}], "converging": False, "states": [ {"name": "new", "state": "pending", "error": "retry"}, @@ -57,12 +82,7 @@ def test_volume_error_preserves_typed_partial_view(spawn_node, method): } sb = sandbox_at(spawn_node({(method, PATH): lambda body, path: (500, partial)})) with pytest.raises(VolumeError) as caught: - if method == "GET": - sb.get_volumes() - elif method == "PUT": - sb.set_volumes(["new"]) - else: - sb.patch_volumes(remove=["old"]) + call(sb) assert isinstance(caught.value, APIError) assert caught.value.status == 500 assert caught.value.response == partial @@ -139,9 +159,16 @@ def test_volume_methods_reject_non_positive_timeout(black_hole, timeout): @pytest.mark.parametrize("status", [200, 409]) -def test_incomplete_volume_view_preserves_handle(spawn_node, status): - incomplete = {"volumes_want": [], "volumes": [], "states": []} - sb = sandbox_at(spawn_node({("GET", PATH): lambda body, path: (status, incomplete)})) +@pytest.mark.parametrize( + "payload", + [ + {"volumes_want": [], "volumes": [], "states": [], "error": "storage unavailable"}, + {**VIEW, "error": "storage unavailable", "volumes": [None]}, + {**VIEW, "error": "storage unavailable", "volumes_want": [None]}, + ], +) +def test_incomplete_volume_view_preserves_handle(spawn_node, status, payload): + sb = sandbox_at(spawn_node({("GET", PATH): lambda body, path: (status, payload)})) sb.volumes = [{"name": "old"}] sb.volumes_want = [{"name": "wanted"}] with pytest.raises(APIError) as caught: @@ -151,6 +178,8 @@ def test_incomplete_volume_view_preserves_handle(spawn_node, status): if status == 409: assert isinstance(caught.value, VolumeError) assert caught.value.response is None + assert caught.value.message == "storage unavailable" + assert caught.value.reply == payload @pytest.mark.parametrize( @@ -164,3 +193,20 @@ def test_incomplete_volume_view_preserves_handle(spawn_node, status): def test_volume_methods_reject_a_bare_string(black_hole, call): with pytest.raises(TypeError): call(sandbox_at(black_hole)) + + +def test_volume_mutations_encode_explicit_device_fields(spawn_node): + entries = [ + {"name": "device", "mount": "", "mode": "rw", "attach_only": True}, + {"name": "mounted", "attach_only": False}, + ] + seen = [] + + def reply(body, path): + seen.append(body) + return 200, VIEW + + sb = sandbox_at(spawn_node({("PUT", PATH): reply, ("PATCH", PATH): reply})) + sb.set_volumes(entries) + sb.patch_volumes(add=entries) + assert seen == [{"volumes": entries}, {"add": entries}]