diff --git a/docs/deploy.md b/docs/deploy.md index 8266a7db3..1e2450078 100644 --- a/docs/deploy.md +++ b/docs/deploy.md @@ -140,7 +140,7 @@ the intersection of the two. | `preview_secret` | — | cluster-shared HMAC secret signing preview tokens (all nodes share one) | | `preview_advertise` | = `preview_listen` | the browser-facing preview base URL, minted into every preview URL, so it must name a routable host (a wildcard `preview_listen` needs it set explicitly); nodes behind one TLS proxy may share it, while signed tokens route internally through each owner's `advertise_addr` | | `checkpoint_dir` | `/checkpoints` | where checkpoints and promoted templates live. Point it at a shared FUSE mount (JuiceFS over object storage, NFS) and every node sharing the mount can branch every checkpoint — records are generation-addressed with `meta.json` as the atomic commit pointer, so no cross-node locking is required of the filesystem. A re-publish retains its superseded export generation for at least ~1h and until a following hourly sweep so an in-flight clone that resolved the old metadata can finish; budget the current generation plus every generation retained across that grace-and-sweep window. An explicit delete can make a concurrent clone fail visibly. One contract on any shared root (mount or bucket): a template key has a single writer — promotes go to the sandbox's owner node, and operators must not race promotes of one name from different nodes (checkpoint ids are node-generated and never collide) | -| `checkpoint_store` | dir | checkpoint AND promoted-template backend (both live in one store root, id-namespaced ck_/tp_): `{"kind": "s3", "s3": {"bucket": "…", "prefix": "ck/", "endpoint": "…", "region": "…", "force_path_style": true}}` stores checkpoints in object storage (any node claims any checkpoint, no shared mount needed). Credentials come from the standard AWS chain (env/IAM role), never this file. An export file is stored as 16 MiB chunks keyed by SHA-256 under `chunks//`, with one `manifests//.json` per published generation naming each file's size and present chunks, and `/meta.json` uploaded last as the commit marker (the record prefix exists only while the record is live). Holes and all-zero chunks are never uploaded, and a fetch recreates the file at its full size and writes only the present chunks, so a mostly-empty disk costs its data, not its size, on the wire and on the fetching node. A lineage is the dedup scope: it defaults to the record id, and every checkpoint and archive of one claim shares the claim's sandbox id, so consecutive captures upload only the chunks that changed (the first capture after a wake uploads everything again when no other live record of that claim remains, because the wake deletes the record it restored from); a publish into a shared lineage writes a zero-byte marker `lineages//` after its manifest and before `meta.json`, and reads only the records those markers list; a crash between manifest and marker only costs a missed dedup, and Delete removes the record's marker right away. Re-publish retains the prior generation's manifest so an in-flight fetch that selected old metadata can finish. sandboxd never deletes a chunk or a manifest from the bucket: Delete removes `meta.json`, the labels and the lineage markers and re-stamps the manifests, so a fetch that already selected the record still completes, and reclaiming storage is the operator's job, see [Reclaiming S3 storage](#reclaiming-s3-storage). A publish or fetch is aborted after 30 minutes. Do not put an expiry lifecycle rule on the prefix: it would delete chunks a live record still names. Budget storage for every generation and chunk until your reclaim runs. When caching a new generation, each node prunes other cached generations of that record unused for at least an hour. Absent = the dir backend at `checkpoint_dir` | +| `checkpoint_store` | dir | checkpoint AND promoted-template backend (both live in one store root, id-namespaced ck_/tp_): `{"kind": "s3", "s3": {"bucket": "…", "prefix": "ck/", "endpoint": "…", "region": "…", "force_path_style": true}}` stores checkpoints in object storage (any node claims any checkpoint, no shared mount needed). Credentials come from the standard AWS chain (env/IAM role), never this file. An export file is stored as 16 MiB chunks keyed by SHA-256 under `chunks//`, with one `manifests//.json` per published generation naming each file's size and present chunks, and `/meta.json` uploaded last as the commit marker (the record prefix exists only while the record is live). Holes and all-zero chunks are never uploaded, and a fetch recreates the file at its full size and writes only the present chunks, so a mostly-empty disk costs its data, not its size, on the wire and on the fetching node. A lineage is the dedup scope: it defaults to the record id, and every checkpoint and archive of one claim shares the claim's sandbox id, so consecutive captures upload only the chunks that changed (the first capture after a wake uploads everything again when no other live record of that claim remains, because the wake deletes the record it restored from); a publish into a shared lineage writes a zero-byte marker `lineages//` after its manifest and before `meta.json`, and reads only the records those markers list; a crash between manifest and marker only costs a missed dedup, and Delete removes the record's marker right away. Re-publish retains the prior generation's manifest so an in-flight fetch that selected old metadata can finish. sandboxd never deletes a chunk or a manifest from the bucket: Delete removes `meta.json`, the labels and the lineage markers and re-stamps the manifests, so a fetch that already selected the record still completes, and reclaiming storage is the operator's job, see [Reclaiming S3 storage](#reclaiming-s3-storage). A publish or fetch is aborted after 30 minutes. Do not put an expiry lifecycle rule on the prefix: it would delete chunks a live record still names. Budget storage for every generation and chunk until your reclaim runs. When caching a new generation, each node prunes other cached generations of that record unused for at least an hour. The periodic generation sweep drops a node's cache of a record once the record is gone from the bucket and the cache is unused for at least an hour. Absent = the dir backend at `checkpoint_dir` | | `checkpoint_ttl_hours` | 0 (keep forever) | ages out checkpoints older than this; the sweep runs hourly and at startup. Explicit deletes never wait for it. Must be nonzero and match fleet-wide when `checkpoint_peer_heal` is on — it is the expiry eligibility point for a healed replica a delete broadcast missed, after which its next successful hourly sweep removes it; persistent sweep failure extends retention until one succeeds, so it is not a hard ceiling | | `checkpoint_peer_heal` | false | on a cluster, lets a node pull a checkpoint it lacks from a peer — found via a live probe, not gossip — rather than failing the branch; see [placement lifecycle](cluster.md#checkpoints-on-a-cluster). Three requirements, all enforced at config load: a nonempty `api_token` (the blob transfer between peers authenticates with it; without one the raw record stream would be open), `mesh.cluster_key` set (the pull presents the fleet `api_token` to an address learned from the peer probe, so the gossip layer carrying that address must itself be authenticated), and `checkpoint_ttl_hours` nonzero (a replica a delete broadcast missed becomes eligible for expiry after it, and its next successful hourly sweep removes it — so it is the finite eligibility point, not an exact ceiling). A shared checkpoint store (`checkpoint_store` kind `s3`) ignores this setting — every node already resolves every checkpoint directly, so there is nothing to heal | | `warm_max` (pool entry) | 0 (static) | turns on the demand-adaptive watermark for that pool: the warm target rises from `warm` toward `warm_max` while claims arrive faster than the measured provision lead covers, and decays back over ~a minute of silence. The pool follows the target down: warm VMs above it are destroyed once the count has stayed above it for a full decay period, so a burst does not leave the pool parked at its high-water mark | @@ -677,9 +677,9 @@ With `cocoond_socket` set, the node follows `cocoon daemon`'s event stream pidfd, so an exit reaches sandboxd within a daemon pass and a recycled PID never reads as a live VMM. The node then lists cocoon's VMs only every 60 s as a backstop for a dropped event, and every 10 s while the stream is down (it -reconnects with backoff) or when no socket is configured. Each list runs only -while the node holds a running claim; each suspect is then checked with a -single-VM `cocoon vm inspect` under its lock. A claim +reconnects with backoff) or when no socket is configured. Each suspect is +then checked with a single-VM `cocoon vm inspect` under its lock; a warm VM +found down is destroyed and replaced. A claim whose VMM is gone — killed, crashed, or lost with the host — is handled per `vmm_restart`. A VMM that exits because the guest powered off is treated the same way. A claim whose lease has already lapsed is left to the reap, unless it diff --git a/docs/sandboxd-api.md b/docs/sandboxd-api.md index 8cfaaf277..ec4055369 100644 --- a/docs/sandboxd-api.md +++ b/docs/sandboxd-api.md @@ -413,7 +413,9 @@ residual device mounts. Live removal also needs a confirmed host detach. Both mount and superblock options must be writable; ext4's `emergency_ro` after a filesystem error counts as read-only. 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 +blocks subsequent readers. Release waits up to 30 seconds for each writable +volume's unmount, and a guest that does not answer within that keeps every +remaining marker. Attach-only writers have no automatic dirty-marker protection. The caller owns mounting, flushing, unmounting, recovery, and fencing before reuse. diff --git a/sandboxd/engine/capacity_test.go b/sandboxd/engine/capacity_test.go index 5595fabec..e9373f341 100644 --- a/sandboxd/engine/capacity_test.go +++ b/sandboxd/engine/capacity_test.go @@ -2,6 +2,7 @@ package engine import ( "errors" + "fmt" "testing" ) @@ -15,6 +16,7 @@ func TestOnlyNodeWideFailuresCarryACapacitySignature(t *testing.T) { {"disk exhaustion", errors.New("write overlay: no space left on device"), true}, {"one VM failed to boot", errors.New("cocoon vm clone: exit status 1: Error: guest did not respond"), false}, {"missing golden", errors.New("cocoon vm clone: exit status 1: Error: snapshot not found"), false}, + {"guest command out of disk", fmt.Errorf("warmup: %w", &guestExitError{code: 1, output: "dd: no space left on device"}), false}, {"no error", nil, false}, } { t.Run(tc.name, func(t *testing.T) { diff --git a/sandboxd/engine/engine.go b/sandboxd/engine/engine.go index 0ebe4cdeb..66f8c8f3c 100644 --- a/sandboxd/engine/engine.go +++ b/sandboxd/engine/engine.go @@ -8,6 +8,7 @@ import ( "bytes" "context" "encoding/json/v2" + "errors" "fmt" "hash/fnv" "io" @@ -295,7 +296,7 @@ func (e *Engine) DialGuestPort(ctx context.Context, vsockSocket string, port uin return nil, fmt.Errorf("write port_forward: %w", err) } // The forwarded stream's own reader takes the handshake reply, so an over-read stays buffered. - r := bufio.NewReaderSize(conn, portReadBuf) + r := bufio.NewReaderSize(conn, portForwardMax) line, readErr := readBufLine(r, portForwardMax) if readErr != nil { _ = conn.Close() @@ -492,7 +493,7 @@ func SocksSocketPath(vsockSocket string) string { // CapacitySignature returns the node-capacity signature err carries, or "". func CapacitySignature(err error) string { - if err == nil { + if _, guest := errors.AsType[*guestExitError](err); err == nil || guest { return "" } msg := strings.ToLower(err.Error()) diff --git a/sandboxd/engine/portconn.go b/sandboxd/engine/portconn.go index 99796ec40..6ac222a5e 100644 --- a/sandboxd/engine/portconn.go +++ b/sandboxd/engine/portconn.go @@ -89,6 +89,10 @@ func (g *guestPortConn) Write(p []byte) (int, error) { func (g *guestPortConn) readFrame() ([]byte, error) { line, err := g.r.ReadSlice('\n') + if errors.Is(err, bufio.ErrBufferFull) && g.r.Size() < portReadBuf { + g.r = bufio.NewReaderSize(io.MultiReader(bytes.NewReader(line), g.Conn), portReadBuf) + line, err = g.r.ReadSlice('\n') + } if !errors.Is(err, bufio.ErrBufferFull) { return line, err } diff --git a/sandboxd/engine/portconn_test.go b/sandboxd/engine/portconn_test.go index 03c8c728b..8e0d80e3e 100644 --- a/sandboxd/engine/portconn_test.go +++ b/sandboxd/engine/portconn_test.go @@ -149,6 +149,31 @@ func TestPortReadBufHoldsOneBulkFrame(t *testing.T) { } } +func TestGuestPortConnReadsAFirstFrameThatIsAFullBulkFrame(t *testing.T) { + silk, gc := newTestGuestPortConn(t) + payload := patterned(wire.BulkChunk) + + go writeDataFrames(t, silk, payload) + + if got, err := io.ReadAll(gc); err != nil || !bytes.Equal(got, payload) { + t.Fatalf("read %d bytes (err %v), want the %d-byte bulk payload", len(got), err, len(payload)) + } +} + +func TestGuestPortConnReadsSmallFramesThenBulkFrames(t *testing.T) { + silk, gc := newTestGuestPortConn(t) + chunks := [][]byte{[]byte("HTTP/1.1 200 OK\r\n"), patterned(100), patterned(3000), patterned(wire.BulkChunk), patterned(wire.BulkChunk - 1), []byte("tail")} + + go writeDataFrames(t, silk, chunks...) + + if got, err := io.ReadAll(gc); err != nil || !bytes.Equal(got, bytes.Join(chunks, nil)) { + t.Fatalf("read %d bytes (err %v), want %d", len(got), err, len(bytes.Join(chunks, nil))) + } + if size := gc.r.Size(); size != portReadBuf { + t.Errorf("reader size %d after a bulk frame, want %d", size, portReadBuf) + } +} + func TestGuestPortConnRefusesAFrameOverMaxFrame(t *testing.T) { for name, tail := range map[string]string{"unterminated": "", "terminated": "\n"} { t.Run(name, func(t *testing.T) { @@ -169,5 +194,26 @@ func newTestGuestPortConn(t *testing.T) (net.Conn, *guestPortConn) { t.Helper() client, silk := net.Pipe() t.Cleanup(func() { _ = client.Close(); _ = silk.Close() }) - return silk, newGuestPortConn(client, bufio.NewReaderSize(client, portReadBuf)) + return silk, newGuestPortConn(client, bufio.NewReaderSize(client, portForwardMax)) +} + +func writeDataFrames(t *testing.T, silk net.Conn, chunks ...[]byte) { + var stream []byte + for _, chunk := range chunks { + frame, err := wire.EncodeResponse(wire.DataResp{Data: chunk}) + if err != nil { + t.Error(err) + return + } + stream = append(append(stream, frame...), '\n') + } + _, _ = silk.Write(append(stream, `{"type":"done"}`+"\n"...)) +} + +func patterned(n int) []byte { + out := make([]byte, n) + for i := range out { + out[i] = byte(i*31 + n) + } + return out } diff --git a/sandboxd/engine/volume.go b/sandboxd/engine/volume.go index e7ba47dff..b0f24840b 100644 --- a/sandboxd/engine/volume.go +++ b/sandboxd/engine/volume.go @@ -82,8 +82,6 @@ func (e *Engine) SyncGuest(ctx context.Context, vsockSocket string) error { } func (e *Engine) umountVolume(ctx context.Context, vsockSocket, mount string) error { - ctx, cancel := context.WithTimeout(ctx, VolumeCallTimeout) - defer cancel() if err := e.silkdExec(ctx, vsockSocket, "umount", "--", mount); err != nil { return fmt.Errorf("unmount volume %s: %w", mount, err) } diff --git a/sandboxd/engine/volume_live.go b/sandboxd/engine/volume_live.go index 00936ef89..a82d0d155 100644 --- a/sandboxd/engine/volume_live.go +++ b/sandboxd/engine/volume_live.go @@ -64,8 +64,6 @@ func (e *Engine) MountLiveVolume(ctx context.Context, vsockSocket, name, mount s // 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, map[string]bool{}) if err != nil { return false, err diff --git a/sandboxd/engine/volume_test.go b/sandboxd/engine/volume_test.go index 9e7d02f6a..8a9f71c46 100644 --- a/sandboxd/engine/volume_test.go +++ b/sandboxd/engine/volume_test.go @@ -165,10 +165,7 @@ func TestSettleGuestWaitsForInitAndTakesAnyExitAsFinished(t *testing.T) { } } -func TestUmountVolumeExecsBoundedUmount(t *testing.T) { - if VolumeCallTimeout != 2*time.Second { - t.Fatalf("volume call timeout = %s, want 2s", VolumeCallTimeout) - } +func TestUmountVolumeExecsUmount(t *testing.T) { path := sockPath(t) fake := serveFakeSilkd(t, path) e := New("cocoon", nil, nil, false, false, "") @@ -189,6 +186,19 @@ func TestUmountVolumeExecsBoundedUmount(t *testing.T) { } } +func TestUmountVolumeWaitsOutAFlushPastTheCallTimeout(t *testing.T) { + path := sockPath(t) + fake := serveFakeSilkd(t, path) + fake.execHook = func(argv []string) { + if argv[0] == "umount" { + time.Sleep(VolumeCallTimeout + 250*time.Millisecond) + } + } + if err := New("cocoon", nil, nil, false, false, "").umountVolume(t.Context(), path, "/data"); err != nil { + t.Errorf("umount whose flush outlasts %s: %v", VolumeCallTimeout, err) + } +} + func TestSyncGuestExecsPlainSync(t *testing.T) { path := sockPath(t) fake := serveFakeSilkd(t, path) diff --git a/sandboxd/pool/checkpoint.go b/sandboxd/pool/checkpoint.go index 8ec390c6a..e5f6eab58 100644 --- a/sandboxd/pool/checkpoint.go +++ b/sandboxd/pool/checkpoint.go @@ -130,6 +130,7 @@ func (m *Manager) DeleteCheckpoint(ctx context.Context, ckptID, tenant string, s if err := m.ckpts.Delete(ctx, ckptID); err != nil { return fmt.Errorf("delete checkpoint: %w", err) } + m.vetoIfHealPending(ckptID) // A shared backend has no per-node replicas to chase. if scope == DeleteFleet && m.peerDelete != nil && !m.ckptsShared { m.peerDelete(context.WithoutCancel(ctx), ckptID) diff --git a/sandboxd/pool/checkpoint_heal_test.go b/sandboxd/pool/checkpoint_heal_test.go index e49dd1b05..d85cde2f4 100644 --- a/sandboxd/pool/checkpoint_heal_test.go +++ b/sandboxd/pool/checkpoint_heal_test.go @@ -127,6 +127,42 @@ func TestDeleteVetoesInFlightHeal(t *testing.T) { }) } +func TestDeleteOfAPublishedRecordVetoesInFlightHeal(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + id := "ck_00000000000000bb" + ckpt := types.Checkpoint{ID: id, Key: testKey, CreatedAt: time.Now()} + m, puller := newHealManager(t, ckpt, []string{"peer-a:7777"}) + puller.release = make(chan struct{}) + + healDone := make(chan error, 1) + go func() { + _, err := m.ClaimCheckpointHeal(t.Context(), id, ClaimOptions{TTL: time.Hour}) + healDone <- err + }() + waitFor(t, func() bool { return puller.count() >= 1 }) + staging, err := m.ckpts.Stage(id) + if err != nil { + t.Fatalf("stage: %v", err) + } + plantHealedRecord(t, staging, ckpt) + if err := m.ckpts.Publish(t.Context(), staging, id, ""); err != nil { + t.Fatalf("publish: %v", err) + } + + if err := m.DeleteCheckpoint(t.Context(), id, "", DeleteLocal); err != nil { + t.Fatalf("delete a published record while a heal is staging: %v", err) + } + + close(puller.release) + if err := <-healDone; !errors.Is(err, ErrUnknownCheckpoint) { + t.Errorf("heal after a delete: %v, want ErrUnknownCheckpoint", err) + } + if m.HasCheckpoint(t.Context(), id) { + t.Error("a heal republished a checkpoint deleted with 204") + } + }) +} + func TestClaimCheckpointHealCtxCancelReturnsPromptly(t *testing.T) { synctest.Test(t, func(t *testing.T) { id := "ck_00000000000000bb" diff --git a/sandboxd/pool/checkpoint_test.go b/sandboxd/pool/checkpoint_test.go index 447db3144..e89f3a675 100644 --- a/sandboxd/pool/checkpoint_test.go +++ b/sandboxd/pool/checkpoint_test.go @@ -166,6 +166,10 @@ func TestReconcileSweepsCheckpointStaging(t *testing.T) { if err := os.MkdirAll(stale, 0o750); err != nil { t.Fatalf("setup: %v", err) } + old := time.Now().Add(-2 * store.GenerationGrace) + if err := os.Chtimes(stale, old, old); err != nil { + t.Fatalf("age staging: %v", err) + } if err := m.Reconcile(t.Context()); err != nil { t.Fatalf("Reconcile: %v", err) } @@ -364,24 +368,16 @@ func TestRejectedCheckpointOpsLeakNoRecLock(t *testing.T) { } } -func TestSharedStoreSweepsOncePerTick(t *testing.T) { - for _, tc := range []struct { - name string - shared bool - wantCkpts int32 - wantTpls int32 - }{ - {"per-node store sweeps both views", false, 1, 1}, - {"shared store sweeps its one prefix once", true, 1, 0}, - } { - t.Run(tc.name, func(t *testing.T) { +func TestEveryTickSweepsBothStoreViewsWhetherTheStoreIsSharedOrNot(t *testing.T) { + for name, shared := range map[string]bool{"per-node store": false, "shared store": true} { + t.Run(name, func(t *testing.T) { m := newTestManager(t, newFakeEngine(), config.PoolSpec{PoolKey: testKey}) ckpts := &sweepCountingStore{Store: m.ckpts} tpls := &sweepCountingStore{Store: m.tpls} - m.ckpts, m.tpls, m.ckptsShared = ckpts, tpls, tc.shared + m.ckpts, m.tpls, m.ckptsShared = ckpts, tpls, shared m.sweepStoreGenerations(t.Context()) - if ckpts.sweeps.Load() != tc.wantCkpts || tpls.sweeps.Load() != tc.wantTpls { - t.Errorf("sweeps = checkpoints %d, templates %d, want %d, %d", ckpts.sweeps.Load(), tpls.sweeps.Load(), tc.wantCkpts, tc.wantTpls) + if ckpts.sweeps.Load() != 1 || tpls.sweeps.Load() != 1 { + t.Errorf("sweeps = checkpoints %d, templates %d, want 1 each: each view has its own fetch cache", ckpts.sweeps.Load(), tpls.sweeps.Load()) } }) } diff --git a/sandboxd/pool/claim.go b/sandboxd/pool/claim.go index 26bda1f09..a3760ba17 100644 --- a/sandboxd/pool/claim.go +++ b/sandboxd/pool/claim.go @@ -98,7 +98,7 @@ func (m *Manager) ClaimProvision(ctx context.Context, key types.PoolKey, o Claim golden.release() return nil, ErrUnknownTemplate } - sb, err := m.provision(ctx, key, golden.dir) + sb, err := m.provisionResolved(ctx, key, golden) golden.release() if err != nil { return nil, err diff --git a/sandboxd/pool/liveness.go b/sandboxd/pool/liveness.go index 9ab5164a1..490f94d0e 100644 --- a/sandboxd/pool/liveness.go +++ b/sandboxd/pool/liveness.go @@ -40,10 +40,6 @@ func (m *Manager) livenessOnce(ctx context.Context) { return } watched := m.watchedClaims("") - if len(watched) == 0 { - m.livenessSweep.Store(false) - return - } go func() { defer m.livenessSweep.Store(false) vms, err := m.eng.List(ctx) @@ -56,7 +52,9 @@ func (m *Manager) livenessOnce(ctx context.Context) { for _, vm := range vms { up[vm.Config.Name] = vm.State == vmStateRunning } - m.recoverDown(ctx, watched, up).Wait() + recovering := m.recoverDown(ctx, watched, up) + m.evictDeadWarm(ctx, up).Wait() + recovering.Wait() }() } @@ -95,12 +93,14 @@ func (m *Manager) onVMSync(ctx context.Context, vms []engine.VMStatus) { m.vmmEvents.Store(true) m.lastVMPoll.Store(time.Now().UnixNano()) m.recoverDown(ctx, m.watchedClaims(""), up) + m.evictDeadWarm(ctx, up) } func (m *Manager) onVMChange(ctx context.Context, change engine.VMChange) { if change.VM.Live && change.Kind != engine.VMDeleted { return } + m.evictDeadWarm(ctx, map[string]bool{change.VM.Name: false}) m.recoverDown(ctx, m.watchedClaims(change.VM.Name), nil) } @@ -126,6 +126,48 @@ func (m *Manager) watchedClaims(vmName string) []liveSuspect { return out } +func (m *Manager) evictDeadWarm(ctx context.Context, up map[string]bool) *sync.WaitGroup { + var suspects []liveSuspect + m.mu.Lock() + for _, p := range m.pools { + for _, sb := range p.warm { + if live, present := up[sb.VMName]; present && !live { + suspects = append(suspects, liveSuspect{sb, sb.VMName}) + } + } + } + m.mu.Unlock() + return m.runBounded(ctx, len(suspects), func(ctx context.Context, i int) { m.evictWarm(ctx, suspects[i]) }) +} + +func (m *Manager) evictWarm(ctx context.Context, suspect liveSuspect) { + logger := log.WithFunc("pool.evictWarm") + vm, found, err := m.eng.Inspect(ctx, suspect.vmName) + switch { + case err != nil: + logger.Warnf(ctx, "list %s: %v", suspect.vmName, err) + return + case found && vm.State == vmStateRunning: + return + } + m.mu.Lock() + evicted := false + if p := m.pools[suspect.sb.Key]; p != nil { + if i := slices.Index(p.warm, suspect.sb); i >= 0 { + p.warm = slices.Delete(p.warm, i, i+1) + evicted = true + } + } + m.mu.Unlock() + if !evicted { + m.recoverDown(ctx, m.watchedClaims(suspect.vmName), nil) + return + } + logger.Warnf(ctx, "warm vm %s exited; replacing it", suspect.vmName) + m.destroy(ctx, suspect.vmName) + m.kickRefill() +} + // recoverVMM re-checks one suspect under its Transition lock, then restarts it or marks it failed. func (m *Manager) recoverVMM(ctx context.Context, sb *types.Sandbox) { sb.Transition.Lock() diff --git a/sandboxd/pool/liveness_test.go b/sandboxd/pool/liveness_test.go index 83b6b3762..3f45ed85d 100644 --- a/sandboxd/pool/liveness_test.go +++ b/sandboxd/pool/liveness_test.go @@ -376,6 +376,107 @@ func TestPollBackstopsTheDaemonStream(t *testing.T) { } } +func TestLivenessReplacesADeadWarmVM(t *testing.T) { + eng := newFakeEngine() + m := newTestManager(t, eng, config.PoolSpec{PoolKey: testKey, Warm: 1}) + dead := fillOneWarm(t, m) + killVMM(eng, dead) + + runLiveness(t, m) + + if infos, _ := m.Info(); infos[0].Warm != 0 { + t.Fatalf("warm %d after its VMM exited, want 0", infos[0].Warm) + } + if removed := eng.removedNames(); !slices.Equal(removed, []string{dead}) { + t.Errorf("removed %v, want %s", removed, dead) + } + if len(m.refillKick) != 1 { + t.Errorf("pending kicks %d, want a refill kick", len(m.refillKick)) + } + if next := fillOneWarm(t, m); next == dead { + t.Errorf("refill kept the dead VM %s", dead) + } + sb, err := m.ClaimWarm(t.Context(), testKey, ClaimOptions{}) + if err != nil || sb.VMName == dead { + t.Errorf("claim got %v (err %v), want a live replacement", sb, err) + } +} + +func TestDaemonExitEventReplacesADeadWarmVM(t *testing.T) { + eng := newFakeEngine() + eng.vmEvents = make(chan fakeVMEvent) + m := newTestManager(t, eng, config.PoolSpec{PoolKey: testKey, Warm: 1}) + dead := fillOneWarm(t, m) + ctx, cancel := context.WithCancel(t.Context()) + defer cancel() + go m.watchVMMs(ctx) + eng.vmEvents <- fakeVMEvent{sync: []engine.VMStatus{{Name: dead, Live: true}}} + waitFor(t, m.vmmEvents.Load) + + killVMM(eng, dead) + eng.vmEvents <- fakeVMEvent{change: engine.VMChange{Kind: "MODIFIED", VM: engine.VMStatus{Name: dead}}} + + waitFor(t, func() bool { infos, _ := m.Info(); return infos[0].Warm == 0 }) + waitFor(t, func() bool { return slices.Equal(eng.removedNames(), []string{dead}) }) +} + +func TestAWarmVMClaimedDuringItsEvictionIsRecoveredWithoutAPoll(t *testing.T) { + eng := newFakeEngine() + eng.vmEvents = make(chan fakeVMEvent) + m := newTestManager(t, eng, config.PoolSpec{PoolKey: testKey, Warm: 1}) + dead := fillOneWarm(t, m) + ctx, cancel := context.WithCancel(t.Context()) + defer cancel() + go m.watchVMMs(ctx) + eng.vmEvents <- fakeVMEvent{sync: []engine.VMStatus{{Name: dead, Live: true}}} + waitFor(t, m.vmmEvents.Load) + lists := eng.listCalls() + killVMM(eng, dead) + claimed := make(chan *types.Sandbox, 1) + eng.mu.Lock() + eng.inspectHook = func() { + sb, err := m.ClaimWarm(t.Context(), testKey, ClaimOptions{}) + if err != nil { + t.Errorf("claim during the eviction: %v", err) + } + claimed <- sb + } + eng.mu.Unlock() + + eng.vmEvents <- fakeVMEvent{change: engine.VMChange{Kind: "MODIFIED", VM: engine.VMStatus{Name: dead}}} + + sb := <-claimed + if sb == nil { + t.Fatal("no claim during the eviction") + } + if got := lockedVMName(sb); got != dead { + t.Fatalf("claim got VM %s, want the warm VM %s", got, dead) + } + waitFor(t, func() bool { got, _ := m.Sandbox(sb.ID); return got.Restarts == 1 }) + if got := eng.listCalls(); got != lists { + t.Errorf("vm list ran %d times, want the recovery without a poll", got-lists) + } + if removed := eng.removedNames(); len(removed) != 0 { + t.Errorf("removed %v, want the claimed VM kept", removed) + } +} + +func TestStaleReportsKeepALiveWarmVM(t *testing.T) { + eng := newFakeEngine() + m := newTestManager(t, eng, config.PoolSpec{PoolKey: testKey, Warm: 1}) + name := fillOneWarm(t, m) + + m.evictDeadWarm(t.Context(), map[string]bool{}).Wait() + m.evictDeadWarm(t.Context(), map[string]bool{name: false}).Wait() + + if infos, _ := m.Info(); infos[0].Warm != 1 { + t.Errorf("warm %d, want the running VM kept", infos[0].Warm) + } + if removed := eng.removedNames(); len(removed) != 0 { + t.Errorf("removed %v, want none", removed) + } +} + func TestRestartFailedRefusesAReleasedClaim(t *testing.T) { eng := newFakeEngine() m := newTestManager(t, eng) @@ -553,3 +654,19 @@ func killVMM(eng *fakeEngine, name string) { defer eng.mu.Unlock() eng.stopped[name] = true } + +func fillOneWarm(t *testing.T, m *Manager) string { + t.Helper() + m.mu.Lock() + m.pools[testKey].goldenDir = "/goldens/x" + m.mu.Unlock() + m.refillOnce(t.Context()) + waitFor(t, func() bool { infos, _ := m.Info(); return infos[0].Warm == 1 && infos[0].Refilling == 0 }) + select { + case <-m.refillKick: + default: + } + m.mu.Lock() + defer m.mu.Unlock() + return m.pools[testKey].warm[0].VMName +} diff --git a/sandboxd/pool/pool.go b/sandboxd/pool/pool.go index 4953e52da..8d37a2bf9 100644 --- a/sandboxd/pool/pool.go +++ b/sandboxd/pool/pool.go @@ -252,8 +252,10 @@ type pool struct { claimClones atomic.Int32 // Without a streak gate a node-wide dead cause is retried every tick at full concurrency - refillFails int - nextRefill time.Time + refillFails int + nextRefill time.Time + cloneFails int + goldenRetiredAt time.Time } func newPool(key types.PoolKey) *pool { @@ -285,6 +287,18 @@ func (p *pool) noteRefillResult(now time.Time, failed bool) bool { return first } +func (p *pool) countClone(ctx context.Context, gen uint64, cloned bool, capReason string) bool { + if ctx.Err() != nil || capReason != "" || gen != p.goldenGen { + return false + } + if cloned { + p.cloneFails = 0 + } else { + p.cloneFails++ + } + return true +} + func (p *pool) applySpec(spec config.PoolSpec) { p.floor = spec.Warm p.warmMax = spec.WarmMax @@ -724,9 +738,6 @@ func (m *Manager) sweepStoreGenerations(ctx context.Context) { if err := m.ckpts.SweepGenerations(ctx); err != nil { logger.Error(ctx, err, "sweep checkpoint generations") } - if m.ckptsShared { - return - } if err := m.tpls.SweepGenerations(ctx); err != nil { logger.Error(ctx, err, "sweep template generations") } diff --git a/sandboxd/pool/pool_test.go b/sandboxd/pool/pool_test.go index 6bc3c3910..4d6a5f123 100644 --- a/sandboxd/pool/pool_test.go +++ b/sandboxd/pool/pool_test.go @@ -1253,6 +1253,7 @@ type fakeEngine struct { detachUnclean bool liveVolumeSupportErr error liveVolumeHook func(string) error + inspectHook func() stopped map[string]bool disks map[string][]types.VMDisk creating map[string]bool @@ -1529,7 +1530,12 @@ func (f *fakeEngine) List(ctx context.Context) ([]types.VMRecord, error) { func (f *fakeEngine) Inspect(ctx context.Context, name string) (types.VMRecord, bool, error) { f.mu.Lock() f.inspects++ + hook := f.inspectHook + f.inspectHook = nil f.mu.Unlock() + if hook != nil { + hook() + } vms, _ := f.list(ctx) i := slices.IndexFunc(vms, func(vm types.VMRecord) bool { return vm.Config.Name == name }) if i < 0 { @@ -1862,6 +1868,12 @@ func (f *fakeEngine) syncCount() int { return len(f.syncs) } +func (f *fakeEngine) unmountCount() int { + f.mu.Lock() + defer f.mu.Unlock() + return len(slices.DeleteFunc(slices.Clone(f.volumeOps), func(op string) bool { return !strings.HasPrefix(op, "umount:") })) +} + func (f *fakeEngine) volumeOpsLog() []string { f.mu.Lock() defer f.mu.Unlock() diff --git a/sandboxd/pool/refill.go b/sandboxd/pool/refill.go index a2d08be55..a147e6ef8 100644 --- a/sandboxd/pool/refill.go +++ b/sandboxd/pool/refill.go @@ -96,6 +96,7 @@ func (m *Manager) refillOne(ctx context.Context, p *pool, golden string, gen uin sb, err := m.startVM(ctx, p.key, func(name string) (types.VMRecord, error) { return m.eng.Clone(ctx, golden, name, p.key) }) + cloned := err == nil <-m.refillSem if err == nil { if ctx.Err() == nil { @@ -123,12 +124,13 @@ func (m *Manager) refillOne(ctx context.Context, p *pool, golden string, gen uin } // A canceled context or a capacity failure says nothing about this pool's health backedOff := false - if ctx.Err() == nil && capReason == "" { + if p.countClone(ctx, gen, cloned, capReason) { backedOff = p.noteRefillResult(now, err != nil) } if backedOff { fails, wait = p.refillFails, time.Until(p.nextRefill) } + retired := m.retireUnclonableLocked(p, gen, now) entered := capReason != "" && m.noteCapacityLocked(now, capReason) parked := m.capacityReasonLocked(now) != "" m.mu.Unlock() @@ -147,6 +149,9 @@ func (m *Manager) refillOne(ctx context.Context, p *pool, golden string, gen uin "refill %s failed %d times running; pausing refill for %s", hash, fails, wait.Round(time.Millisecond)) } + if retired { + m.rebuildRetiredGolden(ctx, hash) + } } return } @@ -226,6 +231,7 @@ func (m *Manager) buildGolden(ctx context.Context, p *pool) { return } logger.Infof(ctx, "golden ready for %s (%s)", hash, p.key.Template) + m.kickRefill() } func (m *Manager) buildGoldenSteps(ctx context.Context, v *configView, key types.PoolKey, name, snap, final, imageID string) error { @@ -304,6 +310,23 @@ func (m *Manager) dropPool(ctx context.Context, p *pool) { }() } +func (m *Manager) retireUnclonableLocked(p *pool, gen uint64, now time.Time) bool { + if p.cloneFails < refillFailStreak || p.goldenGen != gen || p.goldenDir == "" || now.Sub(p.goldenRetiredAt) < buildRetryDelay { + return false + } + p.goldenDir, p.imageID = "", "" + p.goldenGen++ + p.nextBuild, p.goldenRetiredAt = time.Time{}, now + p.refillFails, p.nextRefill, p.cloneFails = 0, time.Time{}, 0 + _ = os.Remove(filepath.Join(m.goldensDir(), p.hash) + goldenStampSuffix) + return true +} + +func (m *Manager) rebuildRetiredGolden(ctx context.Context, hash string) { + log.WithFunc("pool.rebuildRetiredGolden").Warnf(ctx, "golden of %s failed %d clones running; rebuilding it", hash, refillFailStreak) + m.kickRefill() +} + func (m *Manager) goldenStamp(v *configView, key types.PoolKey, imageID string) string { var fp string if v.out.Intercepts(key) { @@ -315,6 +338,9 @@ func (m *Manager) goldenStamp(v *configView, key types.PoolKey, imageID string) n, _ := config.StorageBytes(storage) fields = append(fields, "storage="+strconv.FormatInt(n, 10)) } + if v.noBalloon { + fields = append(fields, "no_balloon") + } return strings.Join(append(fields, v.poolWarmups[key]...), "\x00") } @@ -377,6 +403,28 @@ func (m *Manager) provision(ctx context.Context, key types.PoolKey, golden strin }) } +func (m *Manager) provisionResolved(ctx context.Context, key types.PoolKey, g goldenResolution) (*types.Sandbox, error) { + if g.held == nil { + return m.provision(ctx, key, g.dir) + } + return m.provisionVM(ctx, key, claimProbeTimeout, func(name string) (types.VMRecord, error) { + rec, err := m.eng.Clone(ctx, g.dir, name, key) + m.noteClaimClone(ctx, g, err) + return rec, err + }) +} + +func (m *Manager) noteClaimClone(ctx context.Context, g goldenResolution, err error) { + capReason := engine.CapacitySignature(err) + m.mu.Lock() + g.held.countClone(ctx, g.gen, err == nil, capReason) + retired := m.retireUnclonableLocked(g.held, g.gen, time.Now()) + m.mu.Unlock() + if retired { + m.rebuildRetiredGolden(ctx, g.held.hash) + } +} + func (m *Manager) provisionCold(ctx context.Context, key types.PoolKey) (*types.Sandbox, error) { sb, err := m.provisionVM(ctx, key, coldProbeTimeout, func(name string) (types.VMRecord, error) { return m.eng.RunCold(ctx, name, key, m.view.Load().poolStorage[key]) diff --git a/sandboxd/pool/refillbackoff_test.go b/sandboxd/pool/refillbackoff_test.go index 2b3259073..fe2bd6ca9 100644 --- a/sandboxd/pool/refillbackoff_test.go +++ b/sandboxd/pool/refillbackoff_test.go @@ -1,7 +1,9 @@ package pool import ( + "context" "errors" + "os" "testing" "testing/synctest" "time" @@ -68,6 +70,7 @@ func TestRefillOnceHonorsTheBackoff(t *testing.T) { eng.cloneErr = errors.New("clone: guest kernel panicked before vsock came up") m := newTestManager(t, eng, config.PoolSpec{PoolKey: testKey, Warm: 50}) m.pools[testKey].goldenDir = "/goldens/x" + m.pools[testKey].goldenRetiredAt = time.Now() backedOff := false for range 50 { @@ -109,3 +112,357 @@ func TestRefillOnceHonorsTheBackoff(t *testing.T) { }) }) } + +func TestAGoldenThatCannotBeClonedIsRebuiltOnce(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + eng := newFakeEngine() + eng.cloneErr = errors.New(`vm.restore -> 500 "Memory manager error","Failed to allocate a memory range"`) + m := newTestManager(t, eng, config.PoolSpec{PoolKey: testKey, Warm: 2}) + seedGolden(t, m, "") + + waitFor(t, func() bool { + m.refillOnce(t.Context()) + return eng.coldCount() == 1 + }) + eng.mu.Lock() + eng.cloneErr = nil + eng.mu.Unlock() + waitFor(t, func() bool { return refilledTo(t, m, testKey, 2) }) + + if n := eng.coldCount(); n != 1 { + t.Errorf("golden builds %d, want one rebuild of the unclonable golden", n) + } + }) +} + +func TestAGoldenThatStaysUnclonableIsRebuiltAtMostOncePerRetryDelay(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + eng := newFakeEngine() + eng.cloneErr = errors.New(`vm.restore -> 500 "Memory manager error"`) + m := newTestManager(t, eng, config.PoolSpec{PoolKey: testKey, Warm: 2}) + seedGolden(t, m, "") + start := time.Now() + + for time.Since(start) < 3*buildRetryDelay+buildRetryDelay/2 { + m.refillOnce(t.Context()) + time.Sleep(100 * time.Millisecond) + } + waitPoolSettled(t, m, testKey) + + if n, limit := eng.coldCount(), 1+int(time.Since(start)/buildRetryDelay); n < 2 || n > limit { + t.Errorf("golden builds %d over %v, want between 2 and %d", n, time.Since(start), limit) + } + }) +} + +func TestARebuiltGoldenSurvivesOneFailedRefill(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + eng := newFakeEngine() + eng.cloneErr = errors.New(`vm.restore -> 500 "Memory manager error"`) + m := newTestManager(t, eng, config.PoolSpec{PoolKey: testKey, Warm: 1}) + seedGolden(t, m, "") + waitFor(t, func() bool { + m.refillOnce(t.Context()) + return eng.coldCount() == 1 + }) + waitFor(t, func() bool { + m.mu.Lock() + defer m.mu.Unlock() + p := m.pools[testKey] + return p.goldenDir != "" && !p.building && p.refilling == 0 + }) + time.Sleep(buildRetryDelay) + + m.refillOnce(t.Context()) + waitPoolSettled(t, m, testKey) + + m.mu.Lock() + golden, fails := m.pools[testKey].goldenDir, m.pools[testKey].refillFails + m.mu.Unlock() + if golden == "" || eng.coldCount() != 1 { + t.Errorf("golden %q after %d builds: one failed refill of a rebuilt golden retired it again", golden, eng.coldCount()) + } + if fails >= refillFailStreak { + t.Errorf("refill fails %d, want the rebuilt golden's own streak, not the retired one's", fails) + } + }) +} + +func TestALateFailureOfARetiredGoldensCloneDoesNotCount(t *testing.T) { + eng := newFakeEngine() + eng.cloneErr = errors.New(`vm.restore -> 500 "Memory manager error"`) + m := newTestManager(t, eng, config.PoolSpec{PoolKey: testKey, Warm: 1}) + m.mu.Lock() + p := m.pools[testKey] + p.goldenDir, p.goldenGen = "/goldens/new", 2 + p.refillFails, p.cloneFails = refillFailStreak-1, refillFailStreak-1 + p.refilling++ + m.mu.Unlock() + m.refillSem <- struct{}{} + + m.refillOne(t.Context(), p, "/goldens/old", 1) + + m.mu.Lock() + defer m.mu.Unlock() + if p.refillFails != refillFailStreak-1 || p.cloneFails != refillFailStreak-1 || p.goldenDir != "/goldens/new" { + t.Errorf("refill fails %d clone fails %d golden %q, want %d each and the new golden kept", p.refillFails, p.cloneFails, p.goldenDir, refillFailStreak-1) + } +} + +func TestAGoldenWhoseClonesSucceedIsNeverRetiredForFailedProbes(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + eng := newFakeEngine() + eng.probeErr = errors.New("silkd not ready") + m := newTestManager(t, eng, config.PoolSpec{PoolKey: testKey, Warm: 2}) + g := seedGolden(t, m, "") + start := time.Now() + + for time.Since(start) < 2*buildRetryDelay { + m.refillOnce(t.Context()) + time.Sleep(100 * time.Millisecond) + } + waitPoolSettled(t, m, testKey) + + m.mu.Lock() + golden := m.pools[testKey].goldenDir + m.mu.Unlock() + if n := eng.probeCount(); n < refillFailStreak { + t.Fatalf("%d failed probes, want at least %d so the streak is reached", n, refillFailStreak) + } + if golden != g || eng.coldCount() != 0 { + t.Errorf("golden %q after %d builds, want %q kept: its clones succeed", golden, eng.coldCount(), g) + } + }) +} + +func TestRetiringAGoldenLeavesItsDirectoryForAClaimMidClone(t *testing.T) { + m := newTestManager(t, newFakeEngine(), config.PoolSpec{PoolKey: testKey, Warm: 1}) + g := seedGolden(t, m, "") + m.mu.Lock() + p := m.pools[testKey] + p.goldenDir, p.goldenGen, p.cloneFails = g, 1, refillFailStreak + p.claimClones.Add(1) + retired := m.retireUnclonableLocked(p, 1, time.Now()) + golden := p.goldenDir + m.mu.Unlock() + + if !retired || golden != "" { + t.Fatalf("retired %v golden %q, want the golden retired", retired, golden) + } + if _, err := os.Stat(g); err != nil { + t.Errorf("golden directory under a claim clone: %v, want it left for the rebuild to replace", err) + } + if _, err := os.Stat(g + goldenStampSuffix); !errors.Is(err, os.ErrNotExist) { + t.Errorf("golden stamp after retirement: %v, want it removed so nothing adopts it", err) + } +} + +func TestClaimClonesThatFailRetireTheGoldenOfAPoolWithNoRefill(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + eng := newFakeEngine() + eng.cloneErr = errors.New(`vm.restore -> 500 "Memory manager error"`) + m, p, g := adoptedGoldenNoWarm(t, eng) + + for i := range refillFailStreak { + if _, err := m.ClaimProvision(t.Context(), testKey, ClaimOptions{}); err == nil { + t.Fatalf("claim %d on an unclonable golden succeeded", i) + } + if i < refillFailStreak-1 && goldenOf(m, p) != g { + t.Fatalf("golden retired after %d failed claims, want %d", i+1, refillFailStreak) + } + } + if goldenOf(m, p) != "" { + t.Fatalf("golden %q after %d failed claims, want it retired", goldenOf(m, p), refillFailStreak) + } + if n := p.claimClones.Load(); n != 0 { + t.Errorf("claim clones %d after the failed claims, want 0", n) + } + if _, err := os.Stat(g); err != nil { + t.Errorf("golden directory after retirement: %v, want it left for the rebuild", err) + } + select { + case <-m.refillKick: + default: + t.Error("retirement did not kick refill, so the rebuild waits for the next tick") + } + + eng.mu.Lock() + eng.cloneErr = nil + eng.mu.Unlock() + clones := eng.cloneCount() + if _, err := m.ClaimProvision(t.Context(), testKey, ClaimOptions{}); err != nil { + t.Fatalf("claim while no golden exists: %v", err) + } + if n := eng.coldCount(); n != 1 || eng.cloneCount() != clones { + t.Errorf("cold boots %d clones %d, want the claim between golden builds to cold-boot once and clone nothing", n, eng.cloneCount()-clones) + } + + m.refillOnce(t.Context()) + waitPoolSettled(t, m, testKey) + if goldenOf(m, p) != g || eng.coldCount() != 2 { + t.Fatalf("golden %q after %d cold boots, want %q rebuilt by a second one", goldenOf(m, p), eng.coldCount(), g) + } + clones = eng.cloneCount() + if _, err := m.ClaimProvision(t.Context(), testKey, ClaimOptions{}); err != nil { + t.Fatalf("claim on the rebuilt golden: %v", err) + } + if eng.cloneCount() != clones+1 { + t.Errorf("clones %d, want the claim to clone the rebuilt golden", eng.cloneCount()-clones) + } + }) +} + +func TestClaimResultsThatSayNothingAboutTheGoldenDoNotCount(t *testing.T) { + canceled := func(t *testing.T) context.Context { + ctx, cancel := context.WithCancel(t.Context()) + cancel() + return ctx + } + for _, tt := range []struct { + name string + ctx func(*testing.T) context.Context + err error + }{ + {"canceled by its client", canceled, errors.New(`vm.restore -> 500 "Memory manager error"`)}, + {"capacity failure", (*testing.T).Context, errors.New("create tap: no space left on device")}, + } { + t.Run(tt.name, func(t *testing.T) { + eng := newFakeEngine() + eng.cloneErr = tt.err + m, p, g := adoptedGoldenNoWarm(t, eng) + clones := eng.cloneCount() + for range refillFailStreak + 1 { + if _, err := m.ClaimProvision(tt.ctx(t), testKey, ClaimOptions{}); err == nil { + t.Fatal("claim under the fault succeeded") + } + } + if n := eng.cloneCount() - clones; n != refillFailStreak+1 { + t.Fatalf("clones %d, want each claim to reach the clone", n) + } + if fails := cloneFailsOf(m, p); fails != 0 || goldenOf(m, p) != g { + t.Errorf("clone fails %d golden %q, want 0 and %q kept", fails, goldenOf(m, p), g) + } + }) + } +} + +func TestAClaimWhoseProbeFailsCountsItsCloneAsASuccess(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + eng := newFakeEngine() + eng.probeErr = errors.New("silkd not ready") + m, p, g := adoptedGoldenNoWarm(t, eng) + setCloneFails(m, p, refillFailStreak-1) + + for range 2 { + if _, err := m.ClaimProvision(t.Context(), testKey, ClaimOptions{}); err == nil { + t.Fatal("claim with a failing probe succeeded") + } + } + if fails := cloneFailsOf(m, p); fails != 0 || goldenOf(m, p) != g { + t.Errorf("clone fails %d golden %q, want 0 and %q kept: the clones succeeded", fails, goldenOf(m, p), g) + } + }) +} + +func TestAClaimCloneSuccessResetsTheStreak(t *testing.T) { + m, p, g := adoptedGoldenNoWarm(t, newFakeEngine()) + setCloneFails(m, p, refillFailStreak-2) + + if _, err := m.ClaimProvision(t.Context(), testKey, ClaimOptions{}); err != nil { + t.Fatalf("claim: %v", err) + } + if fails := cloneFailsOf(m, p); fails != 0 || goldenOf(m, p) != g { + t.Errorf("clone fails %d golden %q, want the streak reset and %q kept", fails, goldenOf(m, p), g) + } +} + +func TestALateClaimCloneResultOfARetiredGoldenDoesNotCount(t *testing.T) { + for _, tt := range []struct { + name string + err error + }{ + {"failure", errors.New(`vm.restore -> 500 "Memory manager error"`)}, + {"success", nil}, + } { + t.Run(tt.name, func(t *testing.T) { + m := newTestManager(t, newFakeEngine(), config.PoolSpec{PoolKey: testKey}) + m.mu.Lock() + p := m.pools[testKey] + p.goldenDir, p.goldenGen, p.cloneFails = "/goldens/new", 2, refillFailStreak-1 + m.mu.Unlock() + + m.noteClaimClone(t.Context(), goldenResolution{dir: "/goldens/old", held: p, gen: 1}, tt.err) + + if fails := cloneFailsOf(m, p); fails != refillFailStreak-1 || goldenOf(m, p) != "/goldens/new" { + t.Errorf("clone fails %d golden %q, want %d and the new golden kept", fails, goldenOf(m, p), refillFailStreak-1) + } + }) + } +} + +func TestAFailingCloneOfAPromotedTemplateDoesNotCount(t *testing.T) { + eng := newFakeEngine() + m := envManager(t, eng, t.TempDir()) + src, err := m.ClaimProvision(t.Context(), testKey, ClaimOptions{}) + if err != nil { + t.Fatalf("source claim: %v", err) + } + key, _, err := m.Promote(t.Context(), src.ID, Cred{Token: src.Token}, "tpl:broken", "") + if err != nil { + t.Fatalf("promote: %v", err) + } + m.mu.Lock() + p := newPool(key) + m.pools[key] = p + m.mu.Unlock() + eng.mu.Lock() + eng.cloneErr = errors.New(`vm.restore -> 500 "Memory manager error"`) + eng.mu.Unlock() + clones := eng.cloneCount() + + for range refillFailStreak { + if _, err := m.ClaimProvision(t.Context(), key, ClaimOptions{}); err == nil { + t.Fatal("claim on a failing template clone succeeded") + } + } + if n := eng.cloneCount() - clones; n != refillFailStreak { + t.Fatalf("clones %d, want each claim to clone the template", n) + } + if fails := cloneFailsOf(m, p); fails != 0 { + t.Errorf("clone fails %d on the template's pool, want 0: no pool golden was cloned", fails) + } +} + +func adoptedGoldenNoWarm(t *testing.T, eng *fakeEngine) (*Manager, *pool, string) { + t.Helper() + m := newTestManager(t, eng, config.PoolSpec{PoolKey: testKey}) + g := seedGolden(t, m, "") + m.refillOnce(t.Context()) + waitPoolSettled(t, m, testKey) + m.mu.Lock() + p := m.pools[testKey] + got := p.goldenDir + m.mu.Unlock() + if got != g { + t.Fatalf("adopted golden %q, want %q", got, g) + } + return m, p, g +} + +func goldenOf(m *Manager, p *pool) string { + m.mu.Lock() + defer m.mu.Unlock() + return p.goldenDir +} + +func cloneFailsOf(m *Manager, p *pool) int { + m.mu.Lock() + defer m.mu.Unlock() + return p.cloneFails +} + +func setCloneFails(m *Manager, p *pool, n int) { + m.mu.Lock() + defer m.mu.Unlock() + p.cloneFails = n +} diff --git a/sandboxd/pool/refillkick_test.go b/sandboxd/pool/refillkick_test.go index 7dc7552d8..6748b532c 100644 --- a/sandboxd/pool/refillkick_test.go +++ b/sandboxd/pool/refillkick_test.go @@ -1,6 +1,7 @@ package pool import ( + "context" "errors" "fmt" "testing" @@ -25,6 +26,31 @@ func TestKickRefillCoalescesAndNeverBlocks(t *testing.T) { } } +func TestAFreshGoldenStartsRefillWithoutWaitingForATick(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + eng := newFakeEngine() + m := newTestManager(t, eng, config.PoolSpec{PoolKey: testKey, Warm: 1}) + ctx, cancel := context.WithCancel(t.Context()) + start := time.Now() + + go m.Run(ctx) + synctest.Wait() + + m.mu.Lock() + golden := m.pools[testKey].goldenDir + m.mu.Unlock() + clones, waited := eng.cloneCount(), time.Since(start) + cancel() + synctest.Wait() + if golden == "" { + t.Fatal("no golden built, so the test proves nothing") + } + if clones == 0 || waited >= refillInterval { + t.Errorf("%d warm clones %v after Run started, want the first one before the %v refill tick", clones, waited, refillInterval) + } + }) +} + func TestNodeAtCapacityParksRefillInsteadOfRetrying(t *testing.T) { synctest.Test(t, func(t *testing.T) { eng := newFakeEngine() diff --git a/sandboxd/pool/storage_test.go b/sandboxd/pool/storage_test.go index 78bf6a270..6f0291b58 100644 --- a/sandboxd/pool/storage_test.go +++ b/sandboxd/pool/storage_test.go @@ -60,3 +60,18 @@ func TestSizedPoolRebuildsAnUnsizedGolden(t *testing.T) { t.Error("a pool sized to 40 GiB adopted a golden built at cocoon's default disk") } } + +func TestNoBalloonRebuildsABalloonGolden(t *testing.T) { + spec := config.PoolSpec{PoolKey: testKey, Warm: 1} + m := newTestManager(t, newFakeEngine(), spec) + seedGolden(t, m, "") + m.view.Store(newConfigView(&config.Config{NoBalloon: true, Pools: []config.PoolSpec{spec}}, testSecrets(t))) + + m.mu.Lock() + adopted := m.adoptGolden(m.pools[testKey], "") + m.mu.Unlock() + + if adopted { + t.Error("a no_balloon node adopted a golden booted with a balloon") + } +} diff --git a/sandboxd/pool/template.go b/sandboxd/pool/template.go index f6b7e5750..0bda045b8 100644 --- a/sandboxd/pool/template.go +++ b/sandboxd/pool/template.go @@ -335,6 +335,7 @@ type goldenResolution struct { promoted bool guestEnv bool held *pool + gen uint64 unlock func() } @@ -351,14 +352,15 @@ func (g goldenResolution) release() { func (m *Manager) resolveGolden(ctx context.Context, key types.PoolKey, tenant string) (goldenResolution, error) { m.mu.Lock() var dir string + var gen uint64 p, ok := m.activePool(key) if ok && p.goldenDir != "" { - dir = p.goldenDir + dir, gen = p.goldenDir, p.goldenGen p.claimClones.Add(1) } m.mu.Unlock() if dir != "" { - return goldenResolution{dir: dir, held: p}, nil + return goldenResolution{dir: dir, held: p, gen: gen}, nil } if key.Net == types.NetEgress { return goldenResolution{}, nil // never resume a live-captured template on the egress lane; cold-boot instead diff --git a/sandboxd/pool/view.go b/sandboxd/pool/view.go index 9e26058fa..6e32058ad 100644 --- a/sandboxd/pool/view.go +++ b/sandboxd/pool/view.go @@ -19,6 +19,7 @@ type configView struct { poolReaders map[types.PoolKey]bool diskReserve int64 diskDirs []string + noBalloon bool } func newConfigView(cfg *config.Config, secrets *egress.SecretStore) *configView { @@ -29,6 +30,7 @@ func newConfigView(cfg *config.Config, secrets *egress.SecretStore) *configView poolStorage: map[types.PoolKey]string{}, poolReplicate: map[types.PoolKey]time.Duration{}, poolReaders: map[types.PoolKey]bool{}, + noBalloon: cfg.NoBalloon, } if r := cfg.DiskReserve; r != nil { v.diskReserve, _ = config.StorageBytes(r.Free) diff --git a/sandboxd/pool/volume.go b/sandboxd/pool/volume.go index 8133fbf9e..828ed76c8 100644 --- a/sandboxd/pool/volume.go +++ b/sandboxd/pool/volume.go @@ -8,7 +8,6 @@ import ( "path/filepath" "slices" "strings" - "time" "github.com/projecteru2/core/log" "golang.org/x/sync/errgroup" @@ -18,11 +17,8 @@ import ( "github.com/cocoonstack/sandbox/sandboxd/utils" ) -const ( - // Sidecar, not data_dir: the marker travels with the image and survives a data_dir wipe - volumeDirtySuffix = ".dirty" - volumeQuiesceMax = 10 * time.Second -) +// Sidecar, not data_dir: the marker travels with the image and survives a data_dir wipe +const volumeDirtySuffix = ".dirty" // volumeHolders is one name's live admission state: a writer excludes every other claim. type volumeHolders struct { @@ -210,7 +206,7 @@ func (m *Manager) applyVolume(ctx context.Context, sb *types.Sandbox, volume res } } if err := m.eng.DiskAttach(ctx, sb.VMName, volume.disk); err != nil { - return fmt.Errorf("attach volume %q: %w", volume.applied.Name, err) + return fmt.Errorf("attach volume %q: %w", volume.applied.Name, withCtxErr(ctx, err)) } if attachOnly { return nil @@ -223,7 +219,7 @@ func (m *Manager) applyVolume(ctx context.Context, sb *types.Sandbox, volume res mount = m.eng.MountLiveVolume } if err := mount(ctx, sb.VsockSocket, volume.applied.Name, volume.applied.Mount, volume.disk.RW); err != nil { - return fmt.Errorf("setup volume %q: %w", volume.applied.Name, err) + return fmt.Errorf("setup volume %q: %w", volume.applied.Name, withCtxErr(ctx, err)) } return nil } @@ -231,24 +227,28 @@ func (m *Manager) applyVolume(ctx context.Context, sb *types.Sandbox, volume res // quiesceVolumes unmounts a claim's writable mounts in reverse; an unclean unmount keeps the marker. func (m *Manager) quiesceVolumes(ctx context.Context, sb *types.Sandbox) volumeTeardown { td := volumeTeardown{holds: heldVolumes(sb)} - mounts := writableMounts(sb.Volumes) - if mounts == 0 { + if writableMounts(sb.Volumes) == 0 { return td } catalog := m.catalogSnapshot() logger := log.WithFunc("pool.quiesceVolumes") // Cancellation-immune like removal: a caller hanging up must not skip the flush. - ctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), quiesceBudget(mounts)) - defer cancel() + ctx = context.WithoutCancel(ctx) stuck := false for _, volume := range slices.Backward(sb.Volumes) { if !needsUnmount(volume) { continue // attach-only: no mount to bring down, VM removal closes the device } - clean, err := m.eng.UnmountVolume(ctx, sb.VsockSocket, volume.Mount) + unmountCtx, cancel := context.WithTimeout(ctx, engine.VolumeMutationTimeout) + clean, err := m.eng.UnmountVolume(unmountCtx, sb.VsockSocket, volume.Mount) + wedged := unmountCtx.Err() != nil + cancel() if err != nil { logger.Errorf(ctx, err, "unmount volume %s of %s", volume.Name, sb.ID) stuck = true + if wedged { + break + } continue } if entry, ok := catalog[volume.Name]; ok && clean { @@ -256,10 +256,7 @@ func (m *Manager) quiesceVolumes(ctx context.Context, sb *types.Sandbox) volumeT } } if stuck { - // The unmounts may have spent the whole budget; the fallback flush gets its own - syncCtx, cancelSync := context.WithTimeout(context.WithoutCancel(ctx), engine.VolumeCallTimeout) - defer cancelSync() - if err := m.eng.SyncGuest(syncCtx, sb.VsockSocket); err != nil { + if err := m.eng.SyncGuest(ctx, sb.VsockSocket); err != nil { logger.Errorf(ctx, err, "sync guest of %s", sb.ID) } } @@ -337,11 +334,6 @@ func (m *Manager) checkResolvedLocked(volumes []resolvedVolume) error { return nil } -// quiesceBudget bounds one quiesce: a slice per writable mount plus one for the sync fallback. -func quiesceBudget(mounts int) time.Duration { - return min(time.Duration(mounts+1)*engine.VolumeCallTimeout, volumeQuiesceMax) -} - func writableMounts(volumes []types.Volume) int { mounts := 0 for _, volume := range volumes { @@ -392,3 +384,10 @@ func volumeDirty(path string) bool { func volumeDirtyPath(path string) string { return path + volumeDirtySuffix } + +func withCtxErr(ctx context.Context, err error) error { + if ctxErr := ctx.Err(); ctxErr != nil && !errors.Is(err, ctxErr) { + return fmt.Errorf("%w: %w", ctxErr, err) + } + return err +} diff --git a/sandboxd/pool/volume_mutation.go b/sandboxd/pool/volume_mutation.go index 6de24c604..bd7f4e14a 100644 --- a/sandboxd/pool/volume_mutation.go +++ b/sandboxd/pool/volume_mutation.go @@ -304,9 +304,11 @@ func (m *Manager) detachVolumeRun(ctx context.Context, sb *types.Sandbox, run *v if err != nil { return err } - cleanRW, err := m.eng.DetachVolume(ctx, sb.VMName, sb.VsockSocket, v.Name, v.Mount) + detachCtx, cancel := context.WithTimeout(ctx, engine.VolumeMutationTimeout) + defer cancel() + cleanRW, err := m.eng.DetachVolume(detachCtx, sb.VMName, sb.VsockSocket, v.Name, v.Mount) if err != nil { - return fmt.Errorf("detach volume %q: %w", v.Name, err) + return fmt.Errorf("detach volume %q: %w", v.Name, withCtxErr(detachCtx, err)) } if err = m.checkVolumeRun(ctx, sb, run); err != nil { return err diff --git a/sandboxd/pool/volume_mutation_test.go b/sandboxd/pool/volume_mutation_test.go index d41767fa2..6a9162e64 100644 --- a/sandboxd/pool/volume_mutation_test.go +++ b/sandboxd/pool/volume_mutation_test.go @@ -439,6 +439,35 @@ func TestLiveVolumeCanceledRequestRetainsIntent(t *testing.T) { } } +func TestLiveVolumeCancelHiddenByTheEngineStillReadsAsCanceled(t *testing.T) { + eng := newFakeEngine() + m := newVolumeManager(t, eng, liveCatalog(t)) + sb := mustClaim(t, m, testKey) + killed := errors.New("signal: killed") + cancelOn := func(op string) context.Context { + ctx, cancel := context.WithCancel(t.Context()) + eng.liveVolumeHook = func(got string) error { + if got != op { + return nil + } + cancel() + return killed + } + return ctx + } + + if _, err := m.SetVolumes(cancelOn("attach"), sb.ID, sb.Tenant, []types.Volume{{Name: "data"}}); !errors.Is(err, context.Canceled) { + t.Errorf("attach the engine killed on cancel: %v, want context.Canceled", err) + } + eng.liveVolumeHook = nil + if _, err := m.SetVolumes(t.Context(), sb.ID, sb.Tenant, []types.Volume{{Name: "data"}}); err != nil { + t.Fatalf("attach: %v", err) + } + if _, err := m.SetVolumes(cancelOn("detach"), sb.ID, sb.Tenant, nil); !errors.Is(err, context.Canceled) { + t.Errorf("detach the engine killed on cancel: %v, want context.Canceled", err) + } +} + func TestLiveVolumeAdmissionAndState(t *testing.T) { for _, state := range []string{"paused", "failed", "archived", "archive policy", "dirty", "writer"} { t.Run(state, func(t *testing.T) { @@ -864,6 +893,29 @@ func TestLiveAttachBoundsWholeDeviceOperation(t *testing.T) { }) } +func TestLiveDetachTimeoutHiddenByTheEngineReadsAsDeadline(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + base := newFakeEngine() + m := newVolumeManager(t, base, liveCatalog(t)) + sb := mustClaim(t, m, testKey) + if _, err := m.SetVolumes(t.Context(), sb.ID, sb.Tenant, []types.Volume{{Name: "data"}}); err != nil { + t.Fatalf("attach: %v", err) + } + eng := &stalledDetachEngine{fakeEngine: base, deadlines: make(chan time.Time, 1)} + m.eng = eng + start := time.Now() + + _, err := m.SetVolumes(t.Context(), sb.ID, sb.Tenant, nil) + + if !errors.Is(err, context.DeadlineExceeded) { + t.Errorf("detach past its budget: %v, want context.DeadlineExceeded", err) + } + if deadline := <-eng.deadlines; !deadline.Equal(start.Add(engine.VolumeMutationTimeout)) { + t.Errorf("live detach deadline=%v, want %v", deadline, start.Add(engine.VolumeMutationTimeout)) + } + }) +} + func TestLivenessRefusesAdmittedVolumeRun(t *testing.T) { eng := newFakeEngine() m := newVolumeManager(t, eng, liveCatalog(t)) @@ -926,6 +978,18 @@ func (e *deadlineVolumeEngine) DiskAttach(ctx context.Context, vmName string, sp return e.fakeEngine.DiskAttach(ctx, vmName, spec) } +type stalledDetachEngine struct { + *fakeEngine + deadlines chan time.Time +} + +func (e *stalledDetachEngine) DetachVolume(ctx context.Context, _, _, _, _ string) (bool, error) { + deadline, _ := ctx.Deadline() + e.deadlines <- deadline + <-ctx.Done() + return false, errors.New("signal: killed") +} + func liveCatalog(t *testing.T) []config.VolumeSpec { t.Helper() return []config.VolumeSpec{ diff --git a/sandboxd/pool/volume_rw_test.go b/sandboxd/pool/volume_rw_test.go index c0030381f..8d2713038 100644 --- a/sandboxd/pool/volume_rw_test.go +++ b/sandboxd/pool/volume_rw_test.go @@ -1,6 +1,7 @@ package pool import ( + "context" "errors" "os" "path/filepath" @@ -449,6 +450,9 @@ func TestQuiesceFailureSyncsOnceAndKeepsMarkers(t *testing.T) { if got := eng.syncCount(); got != 1 { t.Errorf("guest syncs=%d, want exactly one fallback for the whole quiesce", got) } + if got := eng.unmountCount(); got != 2 { + t.Errorf("unmounts=%d, want both tried past a busy mount", got) + } if !volumeDirty(first) || !volumeDirty(second) { t.Error("failed unmounts cleared a dirty marker") } @@ -462,6 +466,53 @@ func TestQuiesceFailureSyncsOnceAndKeepsMarkers(t *testing.T) { } } +func TestReleaseSkipsTheRemainingUnmountsOfAWedgedGuest(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + m, base, sb, paths := claimTwoWritableVolumes(t) + eng := &stallingUnmountEngine{fakeEngine: base, firstStall: 2 * engine.VolumeMutationTimeout} + m.eng = eng + start := time.Now() + + if err := m.Release(t.Context(), sb.ID, Cred{Token: sb.Token}); err != nil { + t.Fatalf("Release: %v", err) + } + + if len(eng.deadlines) != 1 { + t.Errorf("unmount calls %d, want 1: a wedged guest skips the rest", len(eng.deadlines)) + } + if wall := time.Since(start); wall != engine.VolumeMutationTimeout { + t.Errorf("release waited %v, want %v", wall, engine.VolumeMutationTimeout) + } + if got := base.syncCount(); got != 1 { + t.Errorf("guest syncs %d, want the fallback once", got) + } + for _, path := range paths { + if !volumeDirty(path) { + t.Errorf("marker of %s cleared without a clean unmount", path) + } + } + }) +} + +func TestReleaseGivesASlowFlushItsOwnMutationBudget(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + m, base, sb, _ := claimTwoWritableVolumes(t) + stall := engine.VolumeMutationTimeout * 2 / 3 + eng := &stallingUnmountEngine{fakeEngine: base, firstStall: stall} + m.eng = eng + start := time.Now() + + if err := m.Release(t.Context(), sb.ID, Cred{Token: sb.Token}); err != nil { + t.Fatalf("Release: %v", err) + } + + want := []time.Time{start.Add(engine.VolumeMutationTimeout), start.Add(stall + engine.VolumeMutationTimeout)} + if !slices.EqualFunc(eng.deadlines, want, time.Time.Equal) { + t.Errorf("unmount deadlines %v, want %v: each volume gets its own budget", eng.deadlines, want) + } + }) +} + func TestPendingRemovalHoldsVolumesUntilDrained(t *testing.T) { path := writeVolumeImage(t, "scratch.img", "data") eng := newFakeEngine() @@ -541,22 +592,6 @@ func TestRestartHoldsVolumesOfAStaleVMUntilItIsRemoved(t *testing.T) { } } -func TestQuiesceBudgetScalesWithMounts(t *testing.T) { - for _, tt := range []struct { - mounts int - want time.Duration - }{ - {1, 2 * engine.VolumeCallTimeout}, - {2, 3 * engine.VolumeCallTimeout}, - {4, volumeQuiesceMax}, - {8, volumeQuiesceMax}, - } { - if got := quiesceBudget(tt.mounts); got != tt.want { - t.Errorf("quiesceBudget(%d)=%s, want %s", tt.mounts, got, tt.want) - } - } -} - func TestVolumeDirtyFailsClosed(t *testing.T) { image := filepath.Join(t.TempDir(), "scratch.img") if err := os.WriteFile(image, []byte("data"), 0o600); err != nil { @@ -642,6 +677,43 @@ func TestWritableDiscoveryAndUsageEvent(t *testing.T) { t.Fatal("claim usage event not found") } +type stallingUnmountEngine struct { + *fakeEngine + firstStall time.Duration + deadlines []time.Time +} + +func (e *stallingUnmountEngine) UnmountVolume(ctx context.Context, sock, mount string) (bool, error) { + deadline, _ := ctx.Deadline() + e.deadlines = append(e.deadlines, deadline) + if len(e.deadlines) == 1 { + select { + case <-time.After(e.firstStall): + case <-ctx.Done(): + return false, errors.New("read silkd frame: use of closed network connection") + } + } + return e.fakeEngine.UnmountVolume(ctx, sock, mount) +} + +func claimTwoWritableVolumes(t *testing.T) (*Manager, *fakeEngine, *types.Sandbox, []string) { + t.Helper() + paths := []string{writeVolumeImage(t, "first.img", "first"), writeVolumeImage(t, "second.img", "second")} + base := newFakeEngine() + m := newVolumeManager(t, base, []config.VolumeSpec{ + {Name: "first", Path: paths[0], Writable: true}, + {Name: "second", Path: paths[1], Writable: true}, + }) + sb, err := m.ClaimProvision(t.Context(), testKey, ClaimOptions{Volumes: []types.Volume{ + {Name: "first", Mode: types.VolumeModeRW}, + {Name: "second", Mode: types.VolumeModeRW}, + }}) + if err != nil { + t.Fatalf("ClaimProvision: %v", err) + } + return m, base, sb, paths +} + func newVolumeManagerAt(t *testing.T, eng *fakeEngine, dataDir string, volumes []config.VolumeSpec) *Manager { t.Helper() cfg := &config.Config{DataDir: dataDir, Volumes: volumes} diff --git a/sandboxd/store/dir/dir.go b/sandboxd/store/dir/dir.go index ec4f1f82a..1250f7296 100644 --- a/sandboxd/store/dir/dir.go +++ b/sandboxd/store/dir/dir.go @@ -168,28 +168,29 @@ func (d *Store) Delete(_ context.Context, id string) error { } func (d *Store) SweepStaging() error { - // ReadDir + suffix, not Glob: the root path may hold glob metacharacters. - if err := utils.RemoveDirEntries(d.root, func(name string) bool { return strings.HasSuffix(name, ".tmp") }); err != nil { - return err - } - return d.sweepRecords() + return d.sweep() } func (d *Store) SweepGenerations(context.Context) error { - return d.sweepRecords() + return d.sweep() } -func (d *Store) sweepRecords() error { +func (d *Store) sweep() error { entries, err := os.ReadDir(d.root) if err != nil { return err } for _, e := range entries { - if !e.IsDir() || !d.idRe.MatchString(e.Name()) { - continue + var sweepErr error + switch { + // ReadDir + suffix, not Glob: the root path may hold glob metacharacters. + case strings.HasSuffix(e.Name(), ".tmp"): + sweepErr = removeAgedEntry(d.root, e) + case e.IsDir() && d.idRe.MatchString(e.Name()): + sweepErr = d.sweepGenerations(e.Name()) } - if err := d.sweepGenerations(e.Name()); err != nil { - return err + if sweepErr != nil { + return sweepErr } } return nil @@ -265,11 +266,11 @@ func (d *Store) publish(_ context.Context, staging, id, digest string) error { if err := touchCurrentGeneration(final, now); err != nil { return err } - tmp := filepath.Join(final, store.MetaFile+".tmp") - if err := os.WriteFile(tmp, metaRaw, 0o600); err != nil { //nolint:gosec // fixed path under our root - return err + tmp, openErr := os.OpenFile(filepath.Join(final, store.MetaFile+".tmp"), os.O_WRONLY|os.O_CREATE|os.O_TRUNC, 0o600) //nolint:gosec // fixed path under our root + if openErr != nil { + return openErr } - if err := os.Rename(tmp, filepath.Join(final, store.MetaFile)); err != nil { + if err := utils.ReplaceFileSync(tmp, filepath.Join(final, store.MetaFile), metaRaw); err != nil { return err } _ = d.sweepGenerations(id) // later publishes and startups retry a failed sweep @@ -432,6 +433,13 @@ func hashChunk(source *digestSource, chunk int, buffer []byte) error { } offset := int64(chunk) * store.DigestChunkSize size := min(store.DigestChunkSize, source.digest.Size-offset) + if sum, hole := store.HoleSum(f, offset, size); hole { + source.digest.Chunks[chunk] = sum + if err := f.Close(); err != nil { + return fmt.Errorf("close export entry %s: %w", source.digest.Path, err) + } + return nil + } hash := sha256.New() n, copyErr := io.CopyBuffer(hash, io.NewSectionReader(f, offset, size), buffer) closeErr := f.Close() diff --git a/sandboxd/store/dir/dir_test.go b/sandboxd/store/dir/dir_test.go index e8e68f4ca..ecc8033f9 100644 --- a/sandboxd/store/dir/dir_test.go +++ b/sandboxd/store/dir/dir_test.go @@ -5,10 +5,13 @@ import ( "crypto/sha256" "errors" "fmt" + "io" "io/fs" "maps" "os" "path/filepath" + "runtime" + "slices" "testing" "time" @@ -194,6 +197,77 @@ func TestDigestWorkersStoreChunksByOffset(t *testing.T) { } } +func TestHoleyExportDigestEqualsTheReadEverythingDigest(t *testing.T) { + root := t.TempDir() + dataTail := writeSparse(t, filepath.Join(root, "disk.img"), 3*store.DigestChunkSize+7, map[int64][]byte{ + store.DigestChunkSize + 5: []byte("data"), + 2 * store.DigestChunkSize: make([]byte, 4096), + 3*store.DigestChunkSize + 6: []byte("z"), + }) + writeSparse(t, filepath.Join(root, "mem"), 2*store.DigestChunkSize+5, map[int64][]byte{0: []byte("head")}) + f, err := os.Open(dataTail) + if err != nil { + t.Fatalf("open: %v", err) + } + defer func() { _ = f.Close() }() + if _, hole := store.HoleSum(f, store.DigestChunkSize, store.DigestChunkSize); hole { + t.Fatal("a chunk holding data reads as a hole") + } + if _, hole := store.HoleSum(f, 0, store.DigestChunkSize); !hole { + t.Fatal("an unwritten chunk does not read as a hole; the filesystem lacks SEEK_DATA") + } + if _, hole := store.HoleSum(f, 2*store.DigestChunkSize, store.DigestChunkSize); hole { + t.Fatal("a chunk of written zeros reads as a hole") + } + + got, err := hashExport(t.Context(), root) + if err != nil { + t.Fatalf("hashExport: %v", err) + } + if want := readEverythingDigest(t, root); got != want { + t.Errorf("hole-skipping digest %q, want the read-everything digest %q", got, want) + } +} + +func TestHoleSkippingDigestAB(t *testing.T) { + if os.Getenv("SANDBOX_DIGEST_AB") == "" { + t.Skip("set SANDBOX_DIGEST_AB=1 to run the 10 GiB interleaved A/B") + } + root := t.TempDir() + writes := map[int64][]byte{} + for i := range int64(4) { + block := make([]byte, 50<<20) + for j := range block { + block[j] = byte(j*7 + int(i)) + } + writes[i*(10<<30)/4] = block + } + writeSparse(t, filepath.Join(root, "disk.img"), 10<<30, writes) + want := readEverythingDigest(t, root) + arms := map[string]func() string{ + "read": func() string { return readEverythingDigest(t, root) }, + "seek": func() string { got, _ := hashExport(t.Context(), root); return got }, + } + times := map[string][]time.Duration{} + for round := range 8 { + order := []string{"read", "seek"} + if round%2 == 1 { + order = []string{"seek", "read"} + } + for _, arm := range order { + start := time.Now() + if got := arms[arm](); got != want { + t.Fatalf("%s arm digest %q, want %q", arm, got, want) + } + times[arm] = append(times[arm], time.Since(start)) + } + } + for _, arm := range []string{"read", "seek"} { + slices.Sort(times[arm]) + t.Logf("%s p50 %v over %v", arm, times[arm][len(times[arm])/2], times[arm]) + } +} + func TestPublishDigestedFailureAndFreshRetry(t *testing.T) { const id = "ck_00000000000000aa" root := t.TempDir() @@ -491,6 +565,66 @@ func TestSweepReclaimsOnlyAgedUncommittedGenerations(t *testing.T) { } } +func TestSweepStagingSparesAPeersFreshStaging(t *testing.T) { + root := t.TempDir() + peer, err := New(root, store.CheckpointIDRe) + if err != nil { + t.Fatalf("new peer: %v", err) + } + restarted, err := New(root, store.CheckpointIDRe) + if err != nil { + t.Fatalf("new restarted node: %v", err) + } + staging, err := peer.Stage("ck_00000000000000aa") + if err != nil { + t.Fatalf("stage: %v", err) + } + seedRecord(t, staging, "in-progress") + + if sweepErr := restarted.SweepStaging(); sweepErr != nil { + t.Fatalf("sweep: %v", sweepErr) + } + if _, statErr := os.Stat(filepath.Join(staging, store.MetaFile)); statErr != nil { + t.Fatalf("a peer's in-progress staging was swept: %v", statErr) + } + + backdate(t, staging) + if sweepErr := restarted.SweepStaging(); sweepErr != nil { + t.Fatalf("sweep aged: %v", sweepErr) + } + if _, statErr := os.Stat(staging); !errors.Is(statErr, fs.ErrNotExist) { + t.Errorf("aged staging survived: %v", statErr) + } +} + +func TestSweepGenerationsRemovesOnlyAgedStaging(t *testing.T) { + st, err := New(t.TempDir(), store.CheckpointIDRe) + if err != nil { + t.Fatalf("New: %v", err) + } + aged, err := st.Stage("ck_00000000000000aa") + if err != nil { + t.Fatalf("stage: %v", err) + } + seedRecord(t, aged, "interrupted") + backdate(t, aged) + fresh, err := st.Stage("ck_00000000000000bb") + if err != nil { + t.Fatalf("stage: %v", err) + } + + if err := st.SweepGenerations(t.Context()); err != nil { + t.Fatalf("SweepGenerations: %v", err) + } + + if _, err := os.Stat(aged); !errors.Is(err, fs.ErrNotExist) { + t.Errorf("staging aged past the grace survived the periodic sweep: %v", err) + } + if _, err := os.Stat(fresh); err != nil { + t.Errorf("fresh staging was swept: %v", err) + } +} + func TestSweepGenerationsToleratesConcurrentDelete(t *testing.T) { const id = "ck_00000000000000aa" st, err := New(t.TempDir(), store.CheckpointIDRe) @@ -610,6 +744,58 @@ func BenchmarkMetas(b *testing.B) { } } +func writeSparse(t testing.TB, path string, size int64, writes map[int64][]byte) string { + t.Helper() + f, err := os.Create(path) + if err != nil { + t.Fatalf("create %s: %v", path, err) + } + defer func() { _ = f.Close() }() + if err = f.Truncate(size); err != nil { + t.Fatalf("size %s: %v", path, err) + } + for off, data := range writes { + if _, err = f.WriteAt(data, off); err != nil { + t.Fatalf("write %s: %v", path, err) + } + } + return path +} + +func readEverythingDigest(t testing.TB, root string) string { + t.Helper() + sources, err := collectDigestSources(root) + if err != nil { + t.Fatalf("collect: %v", err) + } + readAll := func(source *digestSource, chunk int, buffer []byte) error { + f, openErr := os.Open(source.path) + if openErr != nil { + return openErr + } + defer func() { _ = f.Close() }() + offset := int64(chunk) * store.DigestChunkSize + hash := sha256.New() + if _, copyErr := io.CopyBuffer(hash, io.NewSectionReader(f, offset, min(store.DigestChunkSize, source.digest.Size-offset)), buffer); copyErr != nil { + return copyErr + } + copy(source.digest.Chunks[chunk][:], hash.Sum(nil)) + return nil + } + if err = hashChunks(t.Context(), sources, min(runtime.GOMAXPROCS(0), digestWorkerLimit), readAll); err != nil { + t.Fatalf("read every chunk: %v", err) + } + files := make([]store.DigestFile, len(sources)) + for i := range sources { + files[i] = sources[i].digest + } + digest, err := store.AssembleDigest(files) + if err != nil { + t.Fatalf("assemble: %v", err) + } + return digest +} + func metaJSON(metaID string) string { return `{"id":"` + metaID + `"}` } diff --git a/sandboxd/store/s3/s3.go b/sandboxd/store/s3/s3.go index 3babe69f1..e6da7be08 100644 --- a/sandboxd/store/s3/s3.go +++ b/sandboxd/store/s3/s3.go @@ -41,6 +41,7 @@ import ( const ( digestMetadataKey = "content-digest" + cacheDir = "cache" manifestsDir = "manifests" manifestExt = ".json" chunksDir = "chunks" @@ -148,7 +149,7 @@ func (s *Store) Fetch(ctx context.Context, id string) (string, []byte, string, e if err != nil { return "", nil, "", err } - gen := filepath.Join(s.staging, "cache", id, store.ExportGenHash(meta)) + gen := filepath.Join(s.staging, cacheDir, id, store.ExportGenHash(meta)) export := filepath.Join(gen, store.ExportDir) if _, statErr := os.Stat(export); statErr == nil { now := time.Now() @@ -246,7 +247,7 @@ func (s *Store) SetLabels(ctx context.Context, id string, labels []byte) error { } func (s *Store) Delete(ctx context.Context, id string) error { - _ = os.RemoveAll(filepath.Join(s.staging, "cache", id)) + _ = os.RemoveAll(filepath.Join(s.staging, cacheDir, id)) objs, err := s.list(ctx, s.manifestPrefix(id)) if err != nil { return err @@ -268,7 +269,31 @@ func (s *Store) Delete(ctx context.Context, id string) error { return s.deleteKeys(ctx, []string{s.key(id, store.MetaFile)}) } -func (s *Store) SweepGenerations(context.Context) error { return nil } +func (s *Store) SweepGenerations(ctx context.Context) error { + cache := filepath.Join(s.staging, cacheDir) + entries, err := os.ReadDir(cache) + if errors.Is(err, fs.ErrNotExist) { + return nil + } + if err != nil { + return err + } + for _, e := range entries { + dir := filepath.Join(cache, e.Name()) + if !s.idRe.MatchString(e.Name()) || usedWithinGrace(dir) { + continue + } + switch _, _, err := s.readMeta(ctx, e.Name()); { + case errors.Is(err, store.ErrNotFound): + if rmErr := os.RemoveAll(dir); rmErr != nil { + return rmErr + } + case err != nil: + return err + } + } + return nil +} // SweepStaging clears local staging residue at startup. func (s *Store) SweepStaging() error { @@ -715,6 +740,17 @@ func pruneIdleGenerations(dir, keep string) { } } +func usedWithinGrace(dir string) bool { + gens, err := os.ReadDir(dir) + if err != nil { + return true + } + return slices.ContainsFunc(gens, func(gen fs.DirEntry) bool { + info, err := gen.Info() + return err != nil || time.Since(info.ModTime()) < store.GenerationGrace + }) +} + func getChunkBuf() []byte { return *chunkBufs.Get().(*[]byte) } func putChunkBuf(buf []byte) { diff --git a/sandboxd/store/s3/s3_test.go b/sandboxd/store/s3/s3_test.go index 6a0e7b1ef..e1fea1f8a 100644 --- a/sandboxd/store/s3/s3_test.go +++ b/sandboxd/store/s3/s3_test.go @@ -725,6 +725,80 @@ func TestSweepGenerationsLeavesTheBucketAlone(t *testing.T) { } } +func TestSweepGenerationsDropsTheCacheOfARecordAPeerDeleted(t *testing.T) { + const deleted, live = "ck_00000000000000a9", "ck_00000000000000aa" + fake := &fakeS3{objects: map[string][]byte{}} + owner, fetcher := newTestStore(t, fake), newTestStore(t, fake) + for _, id := range []string{deleted, live} { + publishRecord(t, owner, id, []byte(`{"id":"`+id+`"}`), id) + if _, _, _, err := fetcher.Fetch(t.Context(), id); err != nil { + t.Fatalf("fetch %s: %v", id, err) + } + ageCache(t, filepath.Join(fetcher.staging, cacheDir, id)) + } + if err := owner.Delete(t.Context(), deleted); err != nil { + t.Fatalf("Delete: %v", err) + } + + if err := fetcher.SweepGenerations(t.Context()); err != nil { + t.Fatalf("SweepGenerations: %v", err) + } + + if _, err := os.Stat(filepath.Join(fetcher.staging, cacheDir, deleted)); !errors.Is(err, os.ErrNotExist) { + t.Errorf("the cache of a record a peer deleted survived the sweep: %v", err) + } + if _, err := os.Stat(filepath.Join(fetcher.staging, cacheDir, live)); err != nil { + t.Errorf("the cache of a live record was swept: %v", err) + } +} + +func TestSweepGenerationsDropsACacheACanceledFetchInstalledAfterDelete(t *testing.T) { + const id = "ck_00000000000000ab" + fake := &fakeS3{objects: map[string][]byte{}, exportGate: make(chan struct{}), exportEntered: make(chan struct{}, 1), gateOn: "/chunks/"} + st := newTestStore(t, fake) + release := sync.OnceFunc(func() { close(fake.exportGate) }) + t.Cleanup(release) + publishRecord(t, st, id, []byte(`{"id":"`+id+`"}`), "payload") + ctx, cancel := context.WithCancel(t.Context()) + fetched := make(chan error, 1) + go func() { + _, _, _, err := st.Fetch(ctx, id) + fetched <- err + }() + <-fake.exportEntered + cancel() + if err := <-fetched; !errors.Is(err, context.Canceled) { + t.Fatalf("Fetch = %v, want its cancellation", err) + } + if err := st.Delete(t.Context(), id); err != nil { + t.Fatalf("Delete: %v", err) + } + release() + cached := filepath.Join(st.staging, cacheDir, id) + for deadline := time.Now().Add(5 * time.Second); ; time.Sleep(5 * time.Millisecond) { + if entries, _ := os.ReadDir(cached); len(entries) > 0 { + break + } + if time.Now().After(deadline) { + t.Fatal("the detached download never installed its cache") + } + } + + if err := st.SweepGenerations(t.Context()); err != nil { + t.Fatalf("SweepGenerations: %v", err) + } + if _, err := os.Stat(cached); err != nil { + t.Fatalf("a cache used within the grace was swept: %v", err) + } + ageCache(t, cached) + if err := st.SweepGenerations(t.Context()); err != nil { + t.Fatalf("SweepGenerations: %v", err) + } + if _, err := os.Stat(cached); !errors.Is(err, os.ErrNotExist) { + t.Errorf("the cache a canceled fetch installed after the delete survived the sweep: %v", err) + } +} + func TestDeleteRemovesTheLineageMarker(t *testing.T) { const lineage = "sbx_00000000000000a5" fake := &fakeS3{objects: map[string][]byte{}} @@ -839,6 +913,20 @@ func TestFetchFileKeepsHoles(t *testing.T) { } } +func ageCache(t *testing.T, dir string) { + t.Helper() + entries, err := os.ReadDir(dir) + if err != nil { + t.Fatalf("read cache %s: %v", dir, err) + } + old := time.Now().Add(-2 * store.GenerationGrace) + for _, e := range entries { + if err := os.Chtimes(filepath.Join(dir, e.Name()), old, old); err != nil { + t.Fatalf("age cache: %v", err) + } + } +} + func publishRecord(t *testing.T, st *Store, id string, meta []byte, content string) { t.Helper() staging := stageRecord(t, st, id, meta, content) diff --git a/sandboxd/store/sparse.go b/sandboxd/store/sparse.go index 0f6326538..42fa81c17 100644 --- a/sandboxd/store/sparse.go +++ b/sandboxd/store/sparse.go @@ -89,6 +89,15 @@ func WriteChunk(f *os.File, size, index int64, wantSHA256 string, r io.Reader) e return nil } +// HoleSum returns the sum of size zero bytes when f holds no data in [offset, offset+size), false when it does or the seek cannot tell. +func HoleSum(f *os.File, offset, size int64) ([sha256.Size]byte, bool) { + start, end, err := dataExtent(f, offset) + if err != nil || end > start && start < offset+size { + return [sha256.Size]byte{}, false + } + return chunkZeroSum(size), true +} + func dataExtent(f *os.File, pos int64) (start, end int64, err error) { start, err = f.Seek(pos, unix.SEEK_DATA) if errors.Is(err, unix.ENXIO) { diff --git a/sandboxd/store/store.go b/sandboxd/store/store.go index 57c70bacb..f1763ff7f 100644 --- a/sandboxd/store/store.go +++ b/sandboxd/store/store.go @@ -81,7 +81,7 @@ type Store interface { Delete(ctx context.Context, id string) error // SweepStaging removes abandoned staging left by a crash mid-publish. SweepStaging() error - // SweepGenerations runs generation retention without disturbing staging or fetch caches. + // SweepGenerations runs generation retention, sparing staging and fetch caches used within GenerationGrace. SweepGenerations(ctx context.Context) error }