pool, store: dead warm VMs, shared staging, cache reclaim, volume unmount budget

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` | `<data_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 `<prefix>chunks/<lineage>/`, with one `<prefix>manifests/<id>/<generation>.json` per published generation naming each file's size and present chunks, and `<prefix><id>/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 `<prefix>lineages/<lineage>/<id>` 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 `<prefix>chunks/<lineage>/`, with one `<prefix>manifests/<id>/<generation>.json` per published generation naming each file's size and present chunks, and `<prefix><id>/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 `<prefix>lineages/<lineage>/<id>` 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

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.
 

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) {

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())

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
 	}

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
 }

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)
 	}

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

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)

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)

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"

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())
 			}
 		})
 	}

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