From e301bfb794b4305e53f73dcdca03cd6e6a694caa Mon Sep 17 00:00:00 2001 From: CMGS Date: Thu, 9 Jul 2026 10:43:33 +0800 Subject: [PATCH 1/3] feat(pool): lifecycle auto-archive for idle sandboxes Add an archived state after hibernation: a hibernated claim idle past archive_after_seconds is checkpointed to the store and its local VM is dropped; the next access wakes it from the store keeping the same id/token/tenant and guest state, and archive_delete_after_seconds bounds retention. The archive image is hidden from the checkpoint API and cannot be listed or deleted while its claim is live. A persist failure on any transition (archive, wake, release, reap) rolls back so a durable claim never diverges from the store. Archived count is surfaced on /info, /metrics, and the SDK NodeInfo. --- e2e/cmd/lifecycle/main.go | 111 ++++++++++ sandboxd/config/config.go | 30 +++ sandboxd/pool/archive.go | 205 +++++++++++++++++ sandboxd/pool/archive_test.go | 394 +++++++++++++++++++++++++++++++++ sandboxd/pool/checkpoint.go | 49 +++- sandboxd/pool/claim.go | 114 ++++++++-- sandboxd/pool/hibernate.go | 9 +- sandboxd/pool/idle_test.go | 2 +- sandboxd/pool/pool.go | 65 +++++- sandboxd/pool/pool_test.go | 12 + sandboxd/pool/telemetry.go | 42 ++-- sandboxd/server/metrics.go | 6 + sandboxd/server/server.go | 4 +- sandboxd/server/server_test.go | 2 + sandboxd/types/bench_test.go | 35 +++ sandboxd/types/types.go | 22 +- scripts/archive-e2e.sh | 74 +++++++ sdk/go/info.go | 1 + 18 files changed, 1114 insertions(+), 63 deletions(-) create mode 100644 e2e/cmd/lifecycle/main.go create mode 100644 sandboxd/pool/archive.go create mode 100644 sandboxd/pool/archive_test.go create mode 100644 sandboxd/types/bench_test.go create mode 100644 scripts/archive-e2e.sh diff --git a/e2e/cmd/lifecycle/main.go b/e2e/cmd/lifecycle/main.go new file mode 100644 index 0000000..f08435d --- /dev/null +++ b/e2e/cmd/lifecycle/main.go @@ -0,0 +1,111 @@ +// lifecycle exercises sandboxd's auto-archive: a claim idles into hibernation, +// then to a store checkpoint with its local VM dropped, and the next access +// wakes it from the store with the same id and guest state intact. The node +// must run a pool with idle_hibernate_seconds + archive_after_seconds set. +package main + +import ( + "bytes" + "context" + "flag" + "fmt" + "os" + "time" + + sandbox "github.com/cocoonstack/sandbox/sdk/go" +) + +const markerPath = "/root/archive-marker" + +func main() { + var ( + addr = flag.String("addr", "127.0.0.1:7777", "sandboxd address") + token = flag.String("token", "", "node api token") + template = flag.String("template", "rt:24.04", "template ref") + netShape = flag.String("net", "none", "network lane: none|egress") + wait = flag.Duration("wait", 90*time.Second, "max wait for the archive sweep") + ) + flag.Parse() + if err := run(*addr, *token, *template, *netShape, *wait); err != nil { + fmt.Fprintln(os.Stderr, "lifecycle:", err) + os.Exit(1) + } +} + +func run(addr, token, template, netShape string, wait time.Duration) error { + ctx := context.Background() + var copts []sandbox.ClientOption + if token != "" { + copts = append(copts, sandbox.WithAPIToken(token)) + } + client, err := sandbox.Connect(addr, copts...) + if err != nil { + return err + } + + sb, err := client.New(ctx, template, sandbox.WithNetwork(sandbox.NetShape(netShape))) + if err != nil { + return fmt.Errorf("claim: %w", err) + } + id := sb.ID + marker := fmt.Appendf(nil, "archive-marker-%d", time.Now().UnixNano()) + if err = sb.WriteFile(ctx, markerPath, marker, nil); err != nil { + return fmt.Errorf("write marker: %w", err) + } + fmt.Printf("claimed id=%s, marker written\n", id) + + elapsed, err := waitArchived(ctx, client, 1, wait) + if err != nil { + return err + } + fmt.Printf("idle-hibernated then archived in %.1fs (local VM dropped)\n", elapsed.Seconds()) + + // The first access on an archived id is the cold wake: fetch the + // checkpoint, provision a fresh local VM, keep the id/token. + wakeStart := time.Now() + got, err := sb.ReadFile(ctx, markerPath) + if err != nil { + return fmt.Errorf("read marker after archive (wake): %w", err) + } + if !bytes.Equal(got, marker) { + return fmt.Errorf("guest state lost across archive/wake: got %q, want %q", got, marker) + } + if sb.ID != id { + return fmt.Errorf("id changed across archive/wake: %s -> %s", id, sb.ID) + } + fmt.Printf("woke id=%s in %.0fms, guest state intact\n", sb.ID, time.Since(wakeStart).Seconds()*1000) + + info, err := client.Info(ctx) + if err != nil { + return err + } + if info.Archived != 0 || info.Claimed != 1 { + return fmt.Errorf("post-wake info archived=%d claimed=%d, want 0/1", info.Archived, info.Claimed) + } + if err = sb.Close(); err != nil { + return fmt.Errorf("release: %w", err) + } + fmt.Println("PASS") + return nil +} + +// waitArchived polls the node until archived reaches n, returning how long the +// reaper's idle→hibernate→archive ladder took. +func waitArchived(ctx context.Context, client *sandbox.Client, n int, max time.Duration) (time.Duration, error) { + start := time.Now() + deadline := start.Add(max) + for { + info, err := client.Info(ctx) + if err != nil { + return 0, err + } + if info.Archived == n { + return time.Since(start), nil + } + if time.Now().After(deadline) { + return 0, fmt.Errorf("archived never reached %d within %s (hibernated=%d archived=%d claimed=%d)", + n, max, info.Hibernated, info.Archived, info.Claimed) + } + time.Sleep(500 * time.Millisecond) + } +} diff --git a/sandboxd/config/config.go b/sandboxd/config/config.go index 7be62c2..108ce36 100644 --- a/sandboxd/config/config.go +++ b/sandboxd/config/config.go @@ -34,6 +34,15 @@ type PoolSpec struct { // and the snapshot, so callers with their own idle logic must not // pay twice. IdleHibernateSeconds int `json:"idle_hibernate_seconds,omitempty"` + + // ArchiveAfterSeconds, when >0, checkpoints a hibernated claim to the + // store after that many idle seconds and drops its local VM; a later call + // restores it. Requires IdleHibernateSeconds>0 and must exceed it. + ArchiveAfterSeconds int `json:"archive_after_seconds,omitempty"` + + // ArchiveDeleteAfterSeconds, when >0, purges an archived claim's store + // checkpoint that many seconds after archiving. Zero keeps it forever. + ArchiveDeleteAfterSeconds int `json:"archive_delete_after_seconds,omitempty"` } // ValidateLimits checks the warm/watermark/idle bounds — shared by the @@ -48,6 +57,18 @@ func (s PoolSpec) ValidateLimits() error { if s.IdleHibernateSeconds < 0 { return fmt.Errorf("idle_hibernate_seconds must not be negative") } + return validateArchiveWindow(s.IdleHibernateSeconds, s.ArchiveAfterSeconds, s.ArchiveDeleteAfterSeconds) +} + +// validateArchiveWindow checks the archive thresholds shared by PoolSpec and +// the node Config: non-negative, and archive_after must sit past idle_hibernate. +func validateArchiveWindow(idle, after, del int) error { + if after < 0 || del < 0 { + return fmt.Errorf("archive seconds must not be negative") + } + if after > 0 && (idle <= 0 || after <= idle) { + return fmt.Errorf("archive_after_seconds %d requires idle_hibernate_seconds>0 and a larger value", after) + } return nil } @@ -111,6 +132,12 @@ type Config struct { // pooled keys. Zero disables. IdleHibernateSeconds int `json:"idle_hibernate_seconds,omitempty"` + // ArchiveAfterSeconds / ArchiveDeleteAfterSeconds are the archive policy + // for unpooled keys; per-pool settings override them. Same rules as + // PoolSpec (archive_after must exceed idle_hibernate). + ArchiveAfterSeconds int `json:"archive_after_seconds,omitempty"` + ArchiveDeleteAfterSeconds int `json:"archive_delete_after_seconds,omitempty"` + // PreviewListen, when set, starts a preview HTTP server on that address // serving guest ports under signed URLs. PreviewSecret (cluster-shared) // signs the tokens; PreviewAdvertise is the base a browser/proxy reaches @@ -218,6 +245,9 @@ func (c *Config) validate() error { if c.IdleHibernateSeconds < 0 { return fmt.Errorf("idle_hibernate_seconds must not be negative, got %d", c.IdleHibernateSeconds) } + if err := validateArchiveWindow(c.IdleHibernateSeconds, c.ArchiveAfterSeconds, c.ArchiveDeleteAfterSeconds); err != nil { + return err + } if cs := c.CheckpointStore; cs != nil { switch cs.Kind { case "", "dir": diff --git a/sandboxd/pool/archive.go b/sandboxd/pool/archive.go new file mode 100644 index 0000000..e1d5bec --- /dev/null +++ b/sandboxd/pool/archive.go @@ -0,0 +1,205 @@ +package pool + +import ( + "context" + "errors" + "fmt" + "time" + + "github.com/projecteru2/core/log" + + "github.com/cocoonstack/sandbox/sandboxd/store" + "github.com/cocoonstack/sandbox/sandboxd/types" +) + +// archiveAfterFor / archiveDeleteFor / archiveEnabledFor resolve a key's +// archive thresholds — the pool's when pooled, the node default otherwise. +// Callers hold m.mu (they read m.pools). +func (m *Manager) archiveAfterFor(key types.PoolKey) time.Duration { + if p, ok := m.pools[key]; ok { + return p.archiveAfter + } + return m.archiveAfterDefault +} + +func (m *Manager) archiveDeleteFor(key types.PoolKey) time.Duration { + if p, ok := m.pools[key]; ok { + return p.archiveDelete + } + return m.archiveDeleteDefault +} + +func (m *Manager) archiveEnabledFor(key types.PoolKey) bool { + return m.archiveAfterFor(key) > 0 +} + +// archiveOnce checkpoints hibernated claims idle past their archive threshold +// to the store and drops their local VM; the next access restores them. +func (m *Manager) archiveOnce(ctx context.Context) { + if !m.archiveEnabled { + return + } + if !m.archiveSweep.CompareAndSwap(false, true) { + return // the previous sweep's archives are still draining + } + now := time.Now() + var victims []*types.Sandbox + m.mu.Lock() + for _, sb := range m.claimed { + if sb.HibernateSnap == "" || sb.ArchiveCk != "" { + continue + } + if _, busy := m.archiving[sb.ID]; busy { + continue // a reap-triggered archive is already exporting this one + } + after := m.archiveAfterFor(sb.Key) + if after <= 0 || now.Sub(sb.LastSeen()) < after { + continue + } + m.archiving[sb.ID] = struct{}{} + victims = append(victims, sb) + } + m.mu.Unlock() + if len(victims) == 0 { + m.archiveSweep.Store(false) + return + } + go func() { + defer m.archiveSweep.Store(false) + logger := log.WithFunc("pool.archiveOnce") + m.runBounded(ctx, len(victims), func(ctx context.Context, i int) { + sb := victims[i] + switch err := m.archive(ctx, sb); { + case err == nil: + logger.Infof(ctx, "archived %s", sb.ID) + case !errors.Is(err, ErrUnknownSandbox) && !errors.Is(err, errWokeMeanwhile): + logger.Errorf(ctx, err, "archive %s", sb.ID) + } + }).Wait() + }() +} + +// archive publishes a hibernated claim's wake image to the store, then swaps +// its compute backing from local VM to store checkpoint (ArchiveCk), preserving +// id/token/tenant so a later access restores it (wakeArchived). +func (m *Manager) archive(ctx context.Context, sb *types.Sandbox) error { + // Must finish even if the sweep ctx is canceled, or the record diverges from the store. + ctx = context.WithoutCancel(ctx) + defer func() { + m.mu.Lock() + delete(m.archiving, sb.ID) + m.mu.Unlock() + }() + m.mu.Lock() + ok := m.claimed[sb.ID] == sb && sb.HibernateSnap != "" && sb.ArchiveCk == "" + m.mu.Unlock() + if !ok { + return errWokeMeanwhile + } + // A hibernated source is copied from its wake image, so no VM starts. + ck, err := m.publishCheckpoint(ctx, sb, "archive", sb.Tenant) + if err != nil { + return err + } + // Re-validate under Transition+m.mu: a wake since the export cleared HibernateSnap. + sb.Transition.Lock() + m.mu.Lock() + if m.claimed[sb.ID] != sb || sb.HibernateSnap == "" || sb.ArchiveCk != "" { + m.mu.Unlock() + sb.Transition.Unlock() + if delErr := m.ckpts.Delete(ctx, ck.ID); delErr != nil { + log.WithFunc("pool.archive").Warnf(ctx, "delete orphaned archive ck %s: %v", ck.ID, delErr) + } + return errWokeMeanwhile + } + snap, vmName, sock, prevDeadline := sb.HibernateSnap, sb.VMName, sb.VsockSocket, sb.Deadline + sb.ArchiveCk = ck.ID + sb.HibernateSnap, sb.VsockSocket, sb.VMName = "", "", "" + if d := m.archiveDeleteFor(sb.Key); d > 0 { + sb.Deadline = time.Now().Add(d) // retention window, not the claim lease + } else { + sb.Deadline = time.Time{} // keep forever: no retention (reap skips a zero deadline) + } + saveErr := m.store.save(m.claimed) + if saveErr != nil { + // Roll back so memory matches the still-durable hibernated record; drop the orphan ck. + sb.ArchiveCk = "" + sb.HibernateSnap, sb.VsockSocket, sb.VMName = snap, sock, vmName + sb.Deadline = prevDeadline + m.mu.Unlock() + sb.Transition.Unlock() + if delErr := m.ckpts.Delete(ctx, ck.ID); delErr != nil { + log.WithFunc("pool.archive").Warnf(ctx, "delete orphaned archive ck %s: %v", ck.ID, delErr) + } + return fmt.Errorf("archive %s: persist claims: %w", sb.ID, saveErr) + } + m.mu.Unlock() + sb.Transition.Unlock() + // Committed: the store ck is authoritative now; reclaim the local footprint. + if rmErr := m.eng.Remove(ctx, vmName); rmErr != nil { + log.WithFunc("pool.archive").Warnf(ctx, "remove archived VM %s: %v", vmName, rmErr) + } + m.dropSnap(ctx, snap) + m.counters.archives.Add(1) + m.recordUsage(ctx, usageEvent{Event: "archive", ID: sb.ID, Reference: ck.ID, Tenant: sb.Tenant}) + return nil +} + +// wakeArchived restores an archived claim from its store checkpoint into a +// fresh local VM, keeping id/token/tenant; the caller holds the Transition lock. +func (m *Manager) wakeArchived(ctx context.Context, sb *types.Sandbox) (string, error) { + ctx = context.WithoutCancel(ctx) + ck := sb.ArchiveCk + dir, _, release, err := m.ckpts.Fetch(ctx, ck) + if errors.Is(err, store.ErrNotFound) { + return "", ErrUnknownSandbox // record disagrees with the store + } + if err != nil { + return "", fmt.Errorf("wake %s: fetch archive: %w", sb.ID, err) + } + defer release() + built, err := m.provision(ctx, sb.Key, dir) + if err != nil { + return "", fmt.Errorf("wake %s: %w", sb.ID, err) + } + live, saved := m.commitWake(ctx, sb, built.VMName, built.VsockSocket) + if !live || !saved { + m.destroy(ctx, built.VMName) // released, or persist rolled back to archived + if !live { + return "", ErrUnknownSandbox + } + return "", fmt.Errorf("wake %s: persist claims", sb.ID) + } + // Consumed into the live VM; drop the store copy. + if delErr := m.ckpts.Delete(ctx, ck); delErr != nil { + log.WithFunc("pool.wakeArchived").Warnf(ctx, "delete consumed archive ck %s: %v", ck, delErr) + } + m.counters.unarchives.Add(1) + m.recordUsage(ctx, usageEvent{Event: "unarchive", ID: sb.ID, Reference: ck, Tenant: sb.Tenant}) + return built.VsockSocket, nil +} + +// commitWake swaps an archived record back to a live VM if still claimed +// (Release/reap skip the Transition lock). A failed persist rolls the record +// back to archived so memory matches disk and the ck stays pinned. +func (m *Manager) commitWake(ctx context.Context, sb *types.Sandbox, vmName, sock string) (live, saved bool) { + m.mu.Lock() + live = m.claimed[sb.ID] == sb + var saveErr error + if live { + ck, deadline := sb.ArchiveCk, sb.Deadline + sb.VMName, sb.VsockSocket, sb.ArchiveCk = vmName, sock, "" + sb.Deadline = time.Now().Add(clampTTL(0)) // a woken sandbox is a fresh lease + if saveErr = m.store.save(m.claimed); saveErr != nil { + sb.VMName, sb.VsockSocket, sb.ArchiveCk, sb.Deadline = "", "", ck, deadline + } + } + m.mu.Unlock() + if saveErr != nil { + log.WithFunc("pool.commitWake").Warnf(ctx, "persist claims: %v", saveErr) + } + if live && saveErr == nil { + sb.Touch() // restart the idle→hibernate→archive clock + } + return live, saveErr == nil +} diff --git a/sandboxd/pool/archive_test.go b/sandboxd/pool/archive_test.go new file mode 100644 index 0000000..52bec6d --- /dev/null +++ b/sandboxd/pool/archive_test.go @@ -0,0 +1,394 @@ +package pool + +import ( + "errors" + "path/filepath" + "testing" + "time" + + "github.com/cocoonstack/sandbox/sandboxd/config" + "github.com/cocoonstack/sandbox/sandboxd/types" +) + +// breakStore points the claim store at a path whose parent is missing, so the +// next save fails — the fault used to exercise the persist-failure paths. +func breakStore(t *testing.T, m *Manager) { + t.Helper() + m.store.path = filepath.Join(t.TempDir(), "gone", "claims.json") +} + +// archivePool is the pool shape the archive tests share: idle→hibernate at 1s, +// archive at 2s, a long retention window (overridden per test). +func archivePool(delete int) config.PoolSpec { + return config.PoolSpec{ + PoolKey: testKey, + Warm: 1, + IdleHibernateSeconds: 1, + ArchiveAfterSeconds: 2, + ArchiveDeleteAfterSeconds: delete, + } +} + +// mustArchive drives a fresh claim to the archived state deterministically +// (hibernate then archive directly), bypassing the sweep thresholds so a test +// can start from a known archived record. +func mustArchive(t *testing.T, m *Manager, sb *types.Sandbox) { + t.Helper() + if err := m.Hibernate(t.Context(), sb.ID, sb.Token); err != nil { + t.Fatalf("hibernate: %v", err) + } + if err := m.archive(t.Context(), sb); err != nil { + t.Fatalf("archive: %v", err) + } +} + +func ckExists(t *testing.T, m *Manager, ck string) bool { + t.Helper() + _, _, release, err := m.ckpts.Fetch(t.Context(), ck) + if err != nil { + return false + } + release() + return true +} + +// TestArchiveLifecycleRoundTrip exercises the real sweep wiring: +// claim→idle-hibernate→archive→wake, and asserts the id/token survive the +// store round-trip while the local VM and its wake image are reclaimed. +func TestArchiveLifecycleRoundTrip(t *testing.T) { + eng := newFakeEngine() + m := newTestManager(t, eng, archivePool(3600)) + sb := mustClaim(t, m, testKey) + id, token, origVM := sb.ID, sb.Token, sb.VMName + + backdate(m, sb, 5*time.Second) + m.idleOnce(t.Context()) + waitFor(t, func() bool { return hibernated(m) == 1 }) + + m.archiveOnce(t.Context()) + waitFor(t, func() bool { return m.ArchivedCount() == 1 }) + + m.mu.Lock() + ck, vm, snap := sb.ArchiveCk, sb.VMName, sb.HibernateSnap + m.mu.Unlock() + if ck == "" || vm != "" || snap != "" { + t.Fatalf("archived record ck=%q vm=%q snap=%q, want ck set + vm/snap cleared", ck, vm, snap) + } + if hibernated(m) != 0 { + t.Error("archived claim still counted as hibernated") + } + if !ckExists(t, m, ck) { + t.Fatal("archive did not publish the checkpoint") + } + waitFor(t, func() bool { return eng.removed(origVM) }) + + sock, err := m.WakeAgentSocket(t.Context(), id, token) + if err != nil { + t.Fatalf("wake archived: %v", err) + } + if sock == "" { + t.Error("wake returned an empty socket") + } + m.mu.Lock() + wokeCk, wokeVM := sb.ArchiveCk, sb.VMName + m.mu.Unlock() + if wokeCk != "" || wokeVM == "" { + t.Errorf("woke record ck=%q vm=%q, want ck cleared + vm set", wokeCk, wokeVM) + } + if m.ArchivedCount() != 0 { + t.Error("woke claim still counted as archived") + } + if ckExists(t, m, ck) { + t.Error("wake left the consumed checkpoint in the store (double storage billing)") + } +} + +// TestArchiveTTLSweepSparesLiveCheckpoint guards the fix where the checkpoint +// TTL sweep deleted the store image backing a live archived claim, bricking it. +func TestArchiveTTLSweepSparesLiveCheckpoint(t *testing.T) { + eng := newFakeEngine() + m := newTestManager(t, eng, archivePool(3600)) + m.ckptTTL = time.Nanosecond // every checkpoint is instantly age-expired + sb := mustClaim(t, m, testKey) + id, token := sb.ID, sb.Token + mustArchive(t, m, sb) + ck := sb.ArchiveCk + + m.sweepExpiredCheckpoints(t.Context()) + if !ckExists(t, m, ck) { + t.Fatal("TTL sweep deleted the checkpoint backing a live archived sandbox") + } + if _, err := m.WakeAgentSocket(t.Context(), id, token); err != nil { + t.Fatalf("wake after sweep: %v", err) + } +} + +// TestArchiveKeepsForeverWhenDeleteZero guards the fix where a zero +// archive-delete window purged archives immediately instead of keeping them. +func TestArchiveKeepsForeverWhenDeleteZero(t *testing.T) { + eng := newFakeEngine() + m := newTestManager(t, eng, archivePool(0)) + sb := mustClaim(t, m, testKey) + id, token := sb.ID, sb.Token + mustArchive(t, m, sb) + ck := sb.ArchiveCk + + m.mu.Lock() + zero := sb.Deadline.IsZero() + m.mu.Unlock() + if !zero { + t.Fatalf("archive with delete=0 set Deadline %v, want zero (keep forever)", sb.Deadline) + } + + m.reapOnce(t.Context()) + if m.ArchivedCount() != 1 || !ckExists(t, m, ck) { + t.Fatal("reap purged a keep-forever archived sandbox") + } + if _, err := m.WakeAgentSocket(t.Context(), id, token); err != nil { + t.Fatalf("wake keep-forever archive: %v", err) + } +} + +// TestArchiveRetentionPurge covers the reapPurge path: an archive past its +// retention deadline drops from claims, deletes its checkpoint, and releases +// its tenant slot. +func TestArchiveRetentionPurge(t *testing.T) { + eng := newFakeEngine() + m := newTestManager(t, eng, archivePool(3600)) + sb, err := m.ClaimProvision(t.Context(), testKey, time.Hour, "acme") + if err != nil { + t.Fatalf("claim: %v", err) + } + mustArchive(t, m, sb) + ck := sb.ArchiveCk + + m.mu.Lock() + sb.Deadline = time.Now().Add(-time.Second) // retention window elapsed + m.mu.Unlock() + + m.reapOnce(t.Context()) + waitFor(t, func() bool { return m.ArchivedCount() == 0 }) + waitFor(t, func() bool { return !ckExists(t, m, ck) }) + + m.mu.Lock() + live := m.tenantLive["acme"] + m.mu.Unlock() + if live != 0 { + t.Errorf("tenant acme has %d live after purge, want 0", live) + } + if got := m.Counters().ArchiveDeletes; got == 0 { + t.Error("purge did not record an archive_delete") + } +} + +// TestReleaseArchivedDeletesCheckpoint guards the fix where releasing an +// archived claim orphaned its store checkpoint and called Remove on an empty +// VM name. +func TestReleaseArchivedDeletesCheckpoint(t *testing.T) { + eng := newFakeEngine() + m := newTestManager(t, eng, archivePool(3600)) + sb := mustClaim(t, m, testKey) + id, token := sb.ID, sb.Token + mustArchive(t, m, sb) + ck := sb.ArchiveCk + removesBefore := len(eng.removedNames()) + + if err := m.Release(t.Context(), id, token); err != nil { + t.Fatalf("release archived: %v", err) + } + if ckExists(t, m, ck) { + t.Error("release orphaned the archive checkpoint in the store") + } + if got := len(eng.removedNames()); got != removesBefore { + t.Errorf("release ran Remove %d extra times on a VM-less archived claim", got-removesBefore) + } + if _, ok := m.claim(id, token); ok { + t.Error("released claim still present") + } +} + +// TestIdleOnceSkipsArchived guards the fix where the idle sweep tried to +// hibernate an archived (VM-less) claim, corrupting its record. +func TestIdleOnceSkipsArchived(t *testing.T) { + eng := newFakeEngine() + m := newTestManager(t, eng, archivePool(3600)) + sb := mustClaim(t, m, testKey) + mustArchive(t, m, sb) + hibBefore := eng.hibernateCount() + backdate(m, sb, time.Hour) + + m.idleOnce(t.Context()) + waitFor(t, func() bool { return !m.idleSweep.Load() }) + if got := eng.hibernateCount(); got != hibBefore { + t.Errorf("idle sweep hibernated an archived claim: %d→%d", hibBefore, got) + } + if m.ArchivedCount() != 1 { + t.Error("idle sweep disturbed the archived claim") + } +} + +// TestDeleteCheckpointCannotBrickArchive guards the fix where the archive +// image was listable and deletable through the ordinary checkpoint API, +// letting a tenant permanently strand its own archived sandbox. +func TestDeleteCheckpointCannotBrickArchive(t *testing.T) { + eng := newFakeEngine() + m := newTestManager(t, eng, archivePool(3600)) + sb := mustClaim(t, m, testKey) + id, token := sb.ID, sb.Token + mustArchive(t, m, sb) + ck := sb.ArchiveCk + + // The archive image is lifecycle-internal — it must not surface as a user + // checkpoint, or be deletable as one. + if ckpts, err := m.Checkpoints(t.Context(), ""); err != nil || len(ckpts) != 0 { + t.Fatalf("Checkpoints listed %d records, want the archive image hidden (%v)", len(ckpts), err) + } + if err := m.DeleteCheckpoint(t.Context(), ck, ""); !errors.Is(err, ErrUnknownCheckpoint) { + t.Fatalf("DeleteCheckpoint(archive ck) = %v, want ErrUnknownCheckpoint", err) + } + if !ckExists(t, m, ck) { + t.Fatal("DeleteCheckpoint removed the archive image backing a live sandbox") + } + if _, err := m.WakeAgentSocket(t.Context(), id, token); err != nil { + t.Fatalf("wake after delete attempt: %v", err) + } +} + +// TestArchivePersistFailureRollsBack guards the durability fix where archive +// destroyed the local VM+snapshot before its transition reached disk: a failed +// persist must roll the record back to hibernated and keep the local backing. +func TestArchivePersistFailureRollsBack(t *testing.T) { + eng := newFakeEngine() + m := newTestManager(t, eng, archivePool(3600)) + sb := mustClaim(t, m, testKey) + if err := m.Hibernate(t.Context(), sb.ID, sb.Token); err != nil { + t.Fatalf("hibernate: %v", err) + } + snap, vm := sb.HibernateSnap, sb.VMName + breakStore(t, m) + + if err := m.archive(t.Context(), sb); err == nil { + t.Fatal("archive succeeded despite a persist failure") + } + m.mu.Lock() + ck, gotSnap, gotVM := sb.ArchiveCk, sb.HibernateSnap, sb.VMName + m.mu.Unlock() + if ck != "" || gotSnap != snap || gotVM != vm { + t.Errorf("not rolled back: ArchiveCk=%q snap=%q vm=%q, want hibernated %q/%q", ck, gotSnap, gotVM, snap, vm) + } + if eng.removed(vm) { + t.Error("archive destroyed the VM after a failed persist") + } + if ckpts, err := m.Checkpoints(t.Context(), ""); err != nil || len(ckpts) != 0 { + t.Errorf("orphaned archive ck not cleaned up: %d records (%v)", len(ckpts), err) + } +} + +// pinnedHidden reports whether the store holds exactly one checkpoint and it is +// hidden from listings — i.e. a live claim's ArchiveCk still pins it. +func pinnedHidden(t *testing.T, m *Manager, ck string) bool { + t.Helper() + ckpts, err := m.Checkpoints(t.Context(), "") + return err == nil && len(ckpts) == 0 && ckExists(t, m, ck) +} + +// TestReleaseArchivedRollsBackOnPersistFailure guards the durability fix where +// a release whose removal did not persist must roll back — the claim survives +// and its ck stays pinned, so a restart still wakes it. +func TestReleaseArchivedRollsBackOnPersistFailure(t *testing.T) { + eng := newFakeEngine() + m := newTestManager(t, eng, archivePool(3600)) + sb := mustClaim(t, m, testKey) + id, token := sb.ID, sb.Token + mustArchive(t, m, sb) + ck := sb.ArchiveCk + breakStore(t, m) + + if err := m.Release(t.Context(), id, token); err == nil { + t.Fatal("release succeeded despite a persist failure") + } + if _, ok := m.claim(id, token); !ok { + t.Error("release dropped the claim despite a failed persist") + } + if !pinnedHidden(t, m, ck) { + t.Error("kept ck is not pinned after a failed release") + } +} + +// TestWakeArchivedRollsBackOnPersistFailure guards the durability fix where a +// wake whose cleared-ArchiveCk record did not persist must roll back to +// archived, keeping the ck pinned. +func TestWakeArchivedRollsBackOnPersistFailure(t *testing.T) { + eng := newFakeEngine() + m := newTestManager(t, eng, archivePool(3600)) + sb := mustClaim(t, m, testKey) + id, token := sb.ID, sb.Token + mustArchive(t, m, sb) + ck := sb.ArchiveCk + breakStore(t, m) + + if _, err := m.WakeAgentSocket(t.Context(), id, token); err == nil { + t.Fatal("wake succeeded despite a persist failure") + } + m.mu.Lock() + archived := sb.ArchiveCk == ck && sb.VMName == "" + m.mu.Unlock() + if !archived { + t.Error("wake did not roll the record back to archived on a failed persist") + } + if !pinnedHidden(t, m, ck) { + t.Error("kept ck is not pinned after a failed wake") + } +} + +// TestReapPurgeRollsBackOnPersistFailure guards the same rollback on the reaper: +// a retention purge that does not persist keeps the claim and its pinned ck. +func TestReapPurgeRollsBackOnPersistFailure(t *testing.T) { + eng := newFakeEngine() + m := newTestManager(t, eng, archivePool(3600)) + sb := mustClaim(t, m, testKey) + id := sb.ID + mustArchive(t, m, sb) + ck := sb.ArchiveCk + m.mu.Lock() + sb.Deadline = time.Now().Add(-time.Second) // retention elapsed + m.mu.Unlock() + breakStore(t, m) + + m.reapOnce(t.Context()) + m.mu.Lock() + _, present := m.claimed[id] + m.mu.Unlock() + if !present { + t.Error("reap dropped the archived claim despite a failed persist") + } + if !pinnedHidden(t, m, ck) { + t.Error("kept ck is not pinned after a failed reap purge") + } +} + +// TestArchiveOnceSkipsInFlight guards the in-flight dedup: a sandbox already +// being archived (marked by a racing reap) must not be exported twice. +func TestArchiveOnceSkipsInFlight(t *testing.T) { + eng := newFakeEngine() + m := newTestManager(t, eng, archivePool(3600)) + sb := mustClaim(t, m, testKey) + if err := m.Hibernate(t.Context(), sb.ID, sb.Token); err != nil { + t.Fatalf("hibernate: %v", err) + } + backdate(m, sb, time.Hour) + + m.mu.Lock() + m.archiving[sb.ID] = struct{}{} // a reap-triggered archive is already exporting it + m.mu.Unlock() + exportsBefore := eng.exportCount() + + m.archiveOnce(t.Context()) + waitFor(t, func() bool { return !m.archiveSweep.Load() }) + if got := eng.exportCount(); got != exportsBefore { + t.Errorf("archiveOnce double-exported an in-flight sandbox: %d→%d", exportsBefore, got) + } + if m.ArchivedCount() != 0 { + t.Error("archiveOnce archived a sandbox already being archived elsewhere") + } +} diff --git a/sandboxd/pool/checkpoint.go b/sandboxd/pool/checkpoint.go index 5c52541..efde74b 100644 --- a/sandboxd/pool/checkpoint.go +++ b/sandboxd/pool/checkpoint.go @@ -37,7 +37,19 @@ func (m *Manager) Checkpoint(ctx context.Context, id, token, name, tenant string } // See Hibernate: a started capture must finish even if the caller hangs up. ctx = context.WithoutCancel(ctx) + ckpt, err := m.publishCheckpoint(ctx, sb, name, tenant) + if err != nil { + return types.Checkpoint{}, err + } + m.counters.checkpoints.Add(1) + m.recordUsage(ctx, usageEvent{Event: "checkpoint", ID: sb.ID, VMName: sb.VMName, Reference: ckpt.ID}) + return ckpt, nil +} +// publishCheckpoint stages the sandbox's exported state, writes the meta, and +// publishes it to the store. Shared by Checkpoint and archive; a hibernated +// source exports its wake image directly (no VM start — refill.sourceSnap). +func (m *Manager) publishCheckpoint(ctx context.Context, sb *types.Sandbox, name, tenant string) (types.Checkpoint, error) { ckpt := types.Checkpoint{ ID: store.CheckpointID(randHex(8)), Name: name, @@ -52,7 +64,7 @@ func (m *Manager) Checkpoint(ctx context.Context, id, token, name, tenant string } defer func() { _ = os.RemoveAll(staging) }() if err = m.exportSource(ctx, sb, filepath.Join(staging, store.ExportDir)); err != nil { - return types.Checkpoint{}, fmt.Errorf("checkpoint %s: %w", id, err) + return types.Checkpoint{}, fmt.Errorf("checkpoint %s: %w", sb.ID, err) } meta, err := json.Marshal(ckpt) if err != nil { @@ -64,8 +76,6 @@ func (m *Manager) Checkpoint(ctx context.Context, id, token, name, tenant string if err := m.ckpts.Publish(ctx, staging, ckpt.ID); err != nil { return types.Checkpoint{}, fmt.Errorf("commit checkpoint: %w", err) } - m.counters.checkpoints.Add(1) - m.recordUsage(ctx, usageEvent{Event: "checkpoint", ID: sb.ID, VMName: sb.VMName, Reference: ckpt.ID}) return ckpt, nil } @@ -109,23 +119,44 @@ func (m *Manager) ClaimCheckpoint(ctx context.Context, ckptID string, ttl time.D // Checkpoints lists the store's checkpoints, newest first — on a shared // backend (a FUSE mount, a bucket), that is the cluster's set, not one // node's. A non-empty tenant filters to that tenant's records; empty (root) -// lists everything. +// lists everything. Checkpoints backing a live archived claim are hidden: +// they are lifecycle-internal wake images, not user checkpoints. func (m *Manager) Checkpoints(ctx context.Context, tenant string) ([]types.Checkpoint, error) { metas, err := m.ckpts.Metas(ctx) if err != nil { return nil, fmt.Errorf("list checkpoints: %w", err) } + pinned := m.pinnedArchiveCks() ckpts := make([]types.Checkpoint, 0, len(metas)) for _, raw := range metas { var ckpt types.Checkpoint - if err := json.Unmarshal(raw, &ckpt); err == nil && tenantOwns(tenant, ckpt.Tenant) { - ckpts = append(ckpts, ckpt) + if err := json.Unmarshal(raw, &ckpt); err != nil || !tenantOwns(tenant, ckpt.Tenant) { + continue + } + if _, archived := pinned[ckpt.ID]; archived { + continue } + ckpts = append(ckpts, ckpt) } slices.SortFunc(ckpts, func(a, b types.Checkpoint) int { return b.CreatedAt.Compare(a.CreatedAt) }) return ckpts, nil } +// pinnedArchiveCks is the set of checkpoint ids backing a live archived claim: +// its wake image, which the listing hides and delete/TTL must spare (deleting +// one would strand its sandbox). +func (m *Manager) pinnedArchiveCks() map[string]struct{} { + m.mu.Lock() + defer m.mu.Unlock() + pinned := make(map[string]struct{}) + for _, sb := range m.claimed { + if sb.ArchiveCk != "" { + pinned[sb.ArchiveCk] = struct{}{} + } + } + return pinned +} + // DeleteCheckpoint removes a checkpoint's snapshot and record. A tenant may // delete only its own records — anything else answers ErrUnknownCheckpoint, // never a hint that the id exists; root (empty tenant) deletes anything. @@ -137,6 +168,9 @@ func (m *Manager) DeleteCheckpoint(ctx context.Context, ckptID, tenant string) e if !tenantOwns(tenant, ckpt.Tenant) { return ErrUnknownCheckpoint } + if _, pinned := m.pinnedArchiveCks()[ckptID]; pinned { + return ErrUnknownCheckpoint // backs a live archived sandbox, not a deletable checkpoint + } if err := m.ckpts.Delete(ctx, ckptID); err != nil { return fmt.Errorf("delete checkpoint: %w", err) } @@ -152,6 +186,9 @@ func (m *Manager) sweepExpiredCheckpoints(ctx context.Context) { } defer m.ckptSweeping.Store(false) logger := log.WithFunc("pool.sweepExpiredCheckpoints") + // Checkpoints already hides archive images backing a live claim, so the + // TTL never reaches one; their retention is the claim's own Deadline + // (reapPurge). An orphaned archive ck carries no live reference and ages out. ckpts, err := m.Checkpoints(ctx, "") if err != nil { logger.Error(ctx, err, "list for retention") diff --git a/sandboxd/pool/claim.go b/sandboxd/pool/claim.go index aa3eab7..3315bc2 100644 --- a/sandboxd/pool/claim.go +++ b/sandboxd/pool/claim.go @@ -3,6 +3,7 @@ package pool import ( "context" "crypto/subtle" + "errors" "fmt" "net" "time" @@ -12,6 +13,16 @@ import ( "github.com/cocoonstack/sandbox/sandboxd/types" ) +// reapAction is what reapOnce does with an expired claim, so the lifecycle +// reaper preserves state (archive) instead of always destroying. +type reapAction int + +const ( + reapDestroy reapAction = iota // engine teardown + drop snapshot + reapArchive // hibernated + archive-enabled: archive, keep the claim + reapPurge // archived past retention: delete the store checkpoint +) + // ClaimWarm transfers ownership of a warm sandbox without provisioning; // ErrNoWarm means the pool is empty (the caller may redirect or provision). // tenant attributes the claim; empty means the operator (root). @@ -86,17 +97,29 @@ func (m *Manager) Release(ctx context.Context, id, token string) error { } delete(m.claimed, id) m.tenantDelta(sb.Tenant, -1) - snap := sb.HibernateSnap + snap, ck, vmName := sb.HibernateSnap, sb.ArchiveCk, sb.VMName saveErr := m.store.save(m.claimed) + if saveErr != nil { + m.claimed[id] = sb // roll back so memory matches the still-durable claim; ck stays pinned + m.tenantDelta(sb.Tenant, 1) + } m.mu.Unlock() if saveErr != nil { log.WithFunc("pool.Release").Errorf(ctx, saveErr, "persist release of %s", id) + return fmt.Errorf("release %s: %w", id, saveErr) + } + // Cleanup must survive the caller hanging up; the claim is already dropped. + ctx = context.WithoutCancel(ctx) + if ck != "" { + m.purgeArchiveCk(ctx, id, ck, sb.Tenant) // archived: no local VM + } + var err error + if vmName != "" { + err = m.eng.Remove(ctx, vmName) } - // The claim is already dropped; removal must survive the caller hanging up. - err := m.eng.Remove(context.WithoutCancel(ctx), sb.VMName) m.dropSnap(ctx, snap) m.counters.releases.Add(1) - m.recordUsage(ctx, usageEvent{Event: "release", ID: id, VMName: sb.VMName}) + m.recordUsage(ctx, usageEvent{Event: "release", ID: id, VMName: vmName}) return err } @@ -123,7 +146,7 @@ func (m *Manager) PreviewDial(ctx context.Context, id string, port uint16) (net. if !ok { return nil, ErrUnknownSandbox } - m.touch(sb) // a live preview stream is data-plane activity + sb.Touch() // a live preview stream is data-plane activity // Preview bypasses the relay's audit tap (it dials the engine directly), // so record the access here — the only data-plane entry that would // otherwise leave no audit trace. @@ -205,7 +228,7 @@ func (m *Manager) finalizeBatch(ctx context.Context, sbs []*types.Sandbox, ttl t now := time.Now() for _, sb := range sbs { stampIdentity(sb, clampTTL(ttl)) - sb.LastActivity = now + sb.TouchAt(now) } m.mu.Lock() if quotaErr := m.quotaErr(len(sbs), sbs[0].Tenant); quotaErr != nil { @@ -242,44 +265,89 @@ func (m *Manager) finalizeBatch(ctx context.Context, sbs []*types.Sandbox, ttl t func (m *Manager) reapOnce(ctx context.Context) { now := time.Now() type victim struct { - id, vmName, snap string + action reapAction + id, vmName, snap, ck, tenant string + sb *types.Sandbox } m.mu.Lock() var expired []victim for id, sb := range m.claimed { - if now.After(sb.Deadline) { - expired = append(expired, victim{id: id, vmName: sb.VMName, snap: sb.HibernateSnap}) + // A zero deadline means no expiry (an archived claim kept forever). + if sb.Deadline.IsZero() || !now.After(sb.Deadline) { + continue + } + switch { + case sb.ArchiveCk != "": + expired = append(expired, victim{action: reapPurge, id: id, ck: sb.ArchiveCk, tenant: sb.Tenant, sb: sb}) + delete(m.claimed, id) + m.tenantDelta(sb.Tenant, -1) + case sb.HibernateSnap != "" && m.archiveEnabledFor(sb.Key): + // End-of-lease hibernated + archive-enabled: archive instead of + // destroy, kept in m.claimed for archive() to re-validate and move. + if _, busy := m.archiving[id]; busy { + continue // an archive is already exporting this sandbox + } + m.archiving[id] = struct{}{} + expired = append(expired, victim{action: reapArchive, id: id, sb: sb}) + default: + expired = append(expired, victim{action: reapDestroy, id: id, vmName: sb.VMName, snap: sb.HibernateSnap, tenant: sb.Tenant, sb: sb}) delete(m.claimed, id) m.tenantDelta(sb.Tenant, -1) } } var saveErr error if len(expired) > 0 { - saveErr = m.store.save(m.claimed) + if saveErr = m.store.save(m.claimed); saveErr != nil { + for _, v := range expired { + if v.action == reapArchive { + delete(m.archiving, v.id) // never removed from m.claimed + } else { + m.claimed[v.id] = v.sb // roll back so memory matches the still-durable claim + m.tenantDelta(v.tenant, 1) + } + } + } } m.mu.Unlock() logger := log.WithFunc("pool.reapOnce") if saveErr != nil { - logger.Error(ctx, saveErr, "persist reap") + logger.Error(ctx, saveErr, "persist reap; rolled back") + return } - // Destroys are engine subprocesses (worst case minutes on a hung stop): - // fan them out bounded so a big batch never stalls the ticker loop. + // Engine subprocesses (worst case minutes on a hung stop): fan out bounded + // so a big batch never stalls the ticker loop. m.runBounded(ctx, len(expired), func(ctx context.Context, i int) { v := expired[i] - m.destroy(ctx, v.vmName) - m.dropSnap(ctx, v.snap) - m.counters.reaps.Add(1) - m.recordUsage(ctx, usageEvent{Event: "reap", ID: v.id, VMName: v.vmName}) - logger.Infof(ctx, "reaped expired sandbox %s (%s)", v.id, v.vmName) + switch v.action { + case reapPurge: + m.purgeArchiveCk(ctx, v.id, v.ck, v.tenant) + logger.Infof(ctx, "purged archived sandbox %s", v.id) + case reapArchive: + switch err := m.archive(ctx, v.sb); { + case err == nil: + logger.Infof(ctx, "archived expired sandbox %s", v.id) + case !errors.Is(err, ErrUnknownSandbox) && !errors.Is(err, errWokeMeanwhile): + logger.Errorf(ctx, err, "archive expired %s", v.id) + } + default: + m.destroy(ctx, v.vmName) + m.dropSnap(ctx, v.snap) + m.counters.reaps.Add(1) + m.recordUsage(ctx, usageEvent{Event: "reap", ID: v.id, VMName: v.vmName}) + logger.Infof(ctx, "reaped expired sandbox %s (%s)", v.id, v.vmName) + } }) } -// touch records data-plane activity for the idle policy. -func (m *Manager) touch(sb *types.Sandbox) { - m.mu.Lock() - sb.LastActivity = time.Now() - m.mu.Unlock() +// purgeArchiveCk deletes an archived claim's store checkpoint and records the +// retention billing event; shared by Release and reapOnce's purge. +func (m *Manager) purgeArchiveCk(ctx context.Context, id, ck, tenant string) { + if err := m.ckpts.Delete(ctx, ck); err != nil { + log.WithFunc("pool.purgeArchiveCk").Warnf(ctx, "delete archive ck %s: %v", ck, err) + } + m.counters.archiveDeletes.Add(1) + m.recordUsage(ctx, usageEvent{Event: "archive_delete", ID: id, Reference: ck, Tenant: tenant}) } func (m *Manager) claim(id, token string) (*types.Sandbox, bool) { diff --git a/sandboxd/pool/hibernate.go b/sandboxd/pool/hibernate.go index 51d51a0..c1b7824 100644 --- a/sandboxd/pool/hibernate.go +++ b/sandboxd/pool/hibernate.go @@ -34,7 +34,7 @@ func (m *Manager) WakeAgentSocket(ctx context.Context, id, token string) (string if !ok { return "", ErrUnknownSandbox } - m.touch(sb) + sb.Touch() return m.wakeResolved(ctx, sb) } @@ -66,6 +66,9 @@ func (m *Manager) hibernateLocked(ctx context.Context, sb *types.Sandbox) error func (m *Manager) wakeResolved(ctx context.Context, sb *types.Sandbox) (string, error) { sb.Transition.Lock() defer sb.Transition.Unlock() + if sb.ArchiveCk != "" { + return m.wakeArchived(ctx, sb) + } if sb.HibernateSnap == "" { return sb.VsockSocket, nil } @@ -113,7 +116,7 @@ func (m *Manager) idleOnce(ctx context.Context) { if p, pooled := m.pools[sb.Key]; pooled { idle = p.idle // pooled keys never take the node default } - if idle <= 0 || sb.HibernateSnap != "" || now.Sub(sb.LastActivity) < idle { + if idle <= 0 || sb.HibernateSnap != "" || sb.ArchiveCk != "" || now.Sub(sb.LastSeen()) < idle { continue } victims = append(victims, victim{sb.ID, sb.Token}) @@ -152,7 +155,7 @@ func (m *Manager) idleHibernate(ctx context.Context, id, token string, sweepStar sb.Transition.Lock() defer sb.Transition.Unlock() m.mu.Lock() - woke := sb.LastActivity.After(sweepStart) || sb.HibernateSnap != "" + woke := sb.LastSeen().After(sweepStart) || sb.HibernateSnap != "" m.mu.Unlock() if woke { return errWokeMeanwhile diff --git a/sandboxd/pool/idle_test.go b/sandboxd/pool/idle_test.go index 4771432..24c2d3d 100644 --- a/sandboxd/pool/idle_test.go +++ b/sandboxd/pool/idle_test.go @@ -74,7 +74,7 @@ func TestActivityStampsBlockIdleSweep(t *testing.T) { func backdate(m *Manager, sb *types.Sandbox, by time.Duration) { m.mu.Lock() - sb.LastActivity = time.Now().Add(-by) + sb.TouchAt(time.Now().Add(-by)) m.mu.Unlock() } diff --git a/sandboxd/pool/pool.go b/sandboxd/pool/pool.go index da18a67..fbe1081 100644 --- a/sandboxd/pool/pool.go +++ b/sandboxd/pool/pool.go @@ -88,6 +88,7 @@ type SandboxSummary struct { Key types.PoolKey `json:"key"` Deadline time.Time `json:"deadline"` Hibernated bool `json:"hibernated"` + Archived bool `json:"archived,omitempty"` FromCheckpoint string `json:"from_checkpoint,omitempty"` } @@ -106,12 +107,14 @@ type pool struct { // floor and warmMax bound the demand-adaptive target (watermark.go); // rate/lead/lastArrival are its EWMA inputs, guarded by the manager // mutex like everything else here. - floor int - warmMax int - idle time.Duration - rate float64 - lead time.Duration - lastArrival time.Time + floor int + warmMax int + idle time.Duration + archiveAfter time.Duration + archiveDelete time.Duration + rate float64 + lead time.Duration + lastArrival time.Time goldenDir string building bool @@ -124,6 +127,8 @@ func (p *pool) applySpec(spec config.PoolSpec) { p.floor = spec.Warm p.warmMax = spec.WarmMax p.idle = time.Duration(spec.IdleHibernateSeconds) * time.Second + p.archiveAfter = time.Duration(spec.ArchiveAfterSeconds) * time.Second + p.archiveDelete = time.Duration(spec.ArchiveDeleteAfterSeconds) * time.Second } // Manager owns the node's pools, claims, and their persistence. @@ -140,6 +145,17 @@ type Manager struct { idleEnabled bool idleSweep atomic.Bool + // archive*Default are the archive thresholds for unpooled keys; pooled keys + // carry theirs on the pool struct. archiveEnabled is set when any pool or + // the node default has archiving on; archiveSweep guards the sweep loop. + archiveAfterDefault time.Duration + archiveDeleteDefault time.Duration + archiveEnabled bool + archiveSweep atomic.Bool + // archiving holds ids with an archive() export in flight, so the reap tick + // and archive sweep don't both re-export the same sandbox. Guarded by m.mu. + archiving map[string]struct{} + // maxClaims caps live claims node-wide (0 = unlimited); tenantMax holds // every configured tenant's cap (0 = unlimited) and doubles as the set of // known tenants; tenantLive counts live claims per tenant so admission @@ -190,6 +206,7 @@ func NewManager(ctx context.Context, cfg *config.Config, eng Engine) (*Manager, pools: make(map[types.PoolKey]*pool, len(cfg.Pools)), claimed: map[string]*types.Sandbox{}, tenantLive: map[string]int{}, + archiving: map[string]struct{}{}, refillSem: make(chan struct{}, maxConcurrentRefills), } if err := os.MkdirAll(m.goldensDir(), 0o750); err != nil { @@ -234,6 +251,9 @@ func NewManager(ctx context.Context, cfg *config.Config, eng Engine) (*Manager, } m.idleDefault = time.Duration(cfg.IdleHibernateSeconds) * time.Second m.idleEnabled = m.idleDefault > 0 + m.archiveAfterDefault = time.Duration(cfg.ArchiveAfterSeconds) * time.Second + m.archiveDeleteDefault = time.Duration(cfg.ArchiveDeleteAfterSeconds) * time.Second + m.archiveEnabled = m.archiveAfterDefault > 0 for _, spec := range cfg.Pools { p := &pool{key: spec.PoolKey} p.applySpec(spec) @@ -241,6 +261,9 @@ func NewManager(ctx context.Context, cfg *config.Config, eng Engine) (*Manager, if spec.IdleHibernateSeconds > 0 { m.idleEnabled = true } + if spec.ArchiveAfterSeconds > 0 { + m.archiveEnabled = true + } } return m, nil } @@ -270,6 +293,7 @@ func (m *Manager) Run(ctx context.Context) { case <-reap.C: m.reapOnce(ctx) m.idleOnce(ctx) + m.archiveOnce(ctx) case <-ckptSweep: m.sweepExpiredCheckpoints(ctx) } @@ -301,6 +325,13 @@ func (m *Manager) Reconcile(ctx context.Context) error { for id, sb := range claims { rec, ok := live[sb.VMName] switch { + case sb.ArchiveCk != "": + // Archived: no local VM by design; the store ck is the durable + // state. Adopt the stub so the id/token survive restart and the + // first exec wakes it (wakeArchived). + m.claimed[id] = sb + m.tenantDelta(sb.Tenant, 1) + continue case ok && rec.State == vmStateRunning: sb.VsockSocket = rec.VsockSocket // Running with the flag set = a wake crashed between restore and @@ -373,7 +404,7 @@ func (m *Manager) Reconcile(ctx context.Context) error { } now := time.Now() for _, sb := range m.claimed { - sb.LastActivity = now + sb.TouchAt(now) } logger.Infof(ctx, "adopted %d claims, %d VMs live", len(m.claimed), len(live)) return saveErr @@ -440,10 +471,14 @@ func (m *Manager) SetPools(ctx context.Context, specs []config.PoolSpec) error { // Recompute rather than latch: removing every idle pool turns the // sweep off again. m.idleEnabled = m.idleDefault > 0 + m.archiveEnabled = m.archiveAfterDefault > 0 for _, p := range m.pools { if p.idle > 0 { m.idleEnabled = true } + if p.archiveAfter > 0 { + m.archiveEnabled = true + } } m.mu.Unlock() @@ -486,6 +521,20 @@ func (m *Manager) Info() ([]PoolInfo, int, int) { return infos, len(m.claimed), hibernated } +// ArchivedCount returns how many live claims are archived to the store (no +// local VM); separate from Info's gauges to keep its arity — and its callers. +func (m *Manager) ArchivedCount() int { + m.mu.Lock() + defer m.mu.Unlock() + n := 0 + for _, sb := range m.claimed { + if sb.ArchiveCk != "" { + n++ + } + } + return n +} + // Sandboxes lists the live claims, for the operator index. func (m *Manager) Sandboxes() []SandboxSummary { m.mu.Lock() @@ -494,7 +543,7 @@ func (m *Manager) Sandboxes() []SandboxSummary { for _, sb := range m.claimed { out = append(out, SandboxSummary{ ID: sb.ID, Key: sb.Key, Deadline: sb.Deadline, - Hibernated: sb.HibernateSnap != "", FromCheckpoint: sb.FromCheckpoint, + Hibernated: sb.HibernateSnap != "", Archived: sb.ArchiveCk != "", FromCheckpoint: sb.FromCheckpoint, }) } slices.SortFunc(out, func(a, b SandboxSummary) int { return strings.Compare(a.ID, b.ID) }) diff --git a/sandboxd/pool/pool_test.go b/sandboxd/pool/pool_test.go index b157145..2d340ec 100644 --- a/sandboxd/pool/pool_test.go +++ b/sandboxd/pool/pool_test.go @@ -666,6 +666,18 @@ func (f *fakeEngine) cloneCount() int { return len(f.clones) } +func (f *fakeEngine) hibernateCount() int { + f.mu.Lock() + defer f.mu.Unlock() + return len(f.hibernates) +} + +func (f *fakeEngine) exportCount() int { + f.mu.Lock() + defer f.mu.Unlock() + return len(f.exports) +} + func (f *fakeEngine) coldCount() int { f.mu.Lock() defer f.mu.Unlock() diff --git a/sandboxd/pool/telemetry.go b/sandboxd/pool/telemetry.go index 22a2014..0f64b86 100644 --- a/sandboxd/pool/telemetry.go +++ b/sandboxd/pool/telemetry.go @@ -28,6 +28,10 @@ type Counters struct { Releases uint64 Reaps uint64 + Archives uint64 + Unarchives uint64 + ArchiveDeletes uint64 + // Nanosecond totals paired with the counters above: avg latency for // dashboards without histogram machinery. ClaimNanos uint64 @@ -36,11 +40,12 @@ type Counters struct { // counters is the live atomic set behind Counters. type counters struct { - claimsWarm, claimsClone, claimsCold atomic.Uint64 - wakes, hibernates atomic.Uint64 - forks, checkpoints, promotes atomic.Uint64 - releases, reaps atomic.Uint64 - claimNanos, wakeNanos atomic.Uint64 + claimsWarm, claimsClone, claimsCold atomic.Uint64 + wakes, hibernates atomic.Uint64 + forks, checkpoints, promotes atomic.Uint64 + releases, reaps atomic.Uint64 + archives, unarchives, archiveDeletes atomic.Uint64 + claimNanos, wakeNanos atomic.Uint64 } // auditFrame is the addressing slice of a request frame worth auditing. @@ -60,18 +65,21 @@ type auditFrame struct { func (m *Manager) Counters() Counters { c := &m.counters return Counters{ - ClaimsWarm: c.claimsWarm.Load(), - ClaimsClone: c.claimsClone.Load(), - ClaimsCold: c.claimsCold.Load(), - Wakes: c.wakes.Load(), - Hibernates: c.hibernates.Load(), - Forks: c.forks.Load(), - Checkpoints: c.checkpoints.Load(), - Promotes: c.promotes.Load(), - Releases: c.releases.Load(), - Reaps: c.reaps.Load(), - ClaimNanos: c.claimNanos.Load(), - WakeNanos: c.wakeNanos.Load(), + ClaimsWarm: c.claimsWarm.Load(), + ClaimsClone: c.claimsClone.Load(), + ClaimsCold: c.claimsCold.Load(), + Wakes: c.wakes.Load(), + Hibernates: c.hibernates.Load(), + Forks: c.forks.Load(), + Checkpoints: c.checkpoints.Load(), + Promotes: c.promotes.Load(), + Releases: c.releases.Load(), + Reaps: c.reaps.Load(), + Archives: c.archives.Load(), + Unarchives: c.unarchives.Load(), + ArchiveDeletes: c.archiveDeletes.Load(), + ClaimNanos: c.claimNanos.Load(), + WakeNanos: c.wakeNanos.Load(), } } diff --git a/sandboxd/server/metrics.go b/sandboxd/server/metrics.go index 2fb5c53..c4cd0c6 100644 --- a/sandboxd/server/metrics.go +++ b/sandboxd/server/metrics.go @@ -19,6 +19,7 @@ type SandboxListResponse struct { // its count, so dashboards derive averages without histogram machinery. func (s *Server) handleMetrics(w http.ResponseWriter, _ *http.Request) { pools, claimed, hibernated := s.mgr.Info() + archived := s.mgr.ArchivedCount() c := s.mgr.Counters() w.Header().Set("Content-Type", "text/plain; version=0.0.4") @@ -29,6 +30,8 @@ func (s *Server) handleMetrics(w http.ResponseWriter, _ *http.Request) { _, _ = fmt.Fprintf(w, "sandboxd_claimed %d\n", claimed) metric("hibernated", "gauge", "claims currently hibernated") _, _ = fmt.Fprintf(w, "sandboxd_hibernated %d\n", hibernated) + metric("archived", "gauge", "claims archived to the checkpoint store") + _, _ = fmt.Fprintf(w, "sandboxd_archived %d\n", archived) if tenants := s.mgr.TenantClaims(); len(tenants) > 0 { metric("tenant_claims", "gauge", "live claims per configured tenant") @@ -65,6 +68,9 @@ func (s *Server) handleMetrics(w http.ResponseWriter, _ *http.Request) { {"promotes_total", "templates promoted", c.Promotes}, {"releases_total", "claims released", c.Releases}, {"reaps_total", "claims reaped at deadline", c.Reaps}, + {"archives_total", "claims archived to the store", c.Archives}, + {"unarchives_total", "archived claims restored", c.Unarchives}, + {"archive_deletes_total", "archived checkpoints purged at retention", c.ArchiveDeletes}, } { metric(row.name, "counter", row.help) _, _ = fmt.Fprintf(w, "sandboxd_%s %d\n", row.name, row.value) diff --git a/sandboxd/server/server.go b/sandboxd/server/server.go index 24c192a..930c30e 100644 --- a/sandboxd/server/server.go +++ b/sandboxd/server/server.go @@ -77,6 +77,7 @@ type Manager interface { WakeAgentSocket(ctx context.Context, id, token string) (string, error) SetPools(ctx context.Context, pools []config.PoolSpec) error Info() ([]pool.PoolInfo, int, int) + ArchivedCount() int } // Dialer opens the hybrid-vsock connection to a VM's silkd. @@ -98,6 +99,7 @@ type InfoResponse struct { Pools []pool.PoolInfo `json:"pools"` Claimed int `json:"claimed"` Hibernated int `json:"hibernated"` + Archived int `json:"archived"` Peers []string `json:"peers,omitempty"` } @@ -425,7 +427,7 @@ func (s *Server) handlePeers(w http.ResponseWriter, _ *http.Request) { func (s *Server) handleInfo(w http.ResponseWriter, _ *http.Request) { pools, claimed, hibernated := s.mgr.Info() - resp := InfoResponse{Pools: pools, Claimed: claimed, Hibernated: hibernated} + resp := InfoResponse{Pools: pools, Claimed: claimed, Hibernated: hibernated, Archived: s.mgr.ArchivedCount()} if s.placer != nil { resp.Peers = s.placer.PeerAddrs() } diff --git a/sandboxd/server/server_test.go b/sandboxd/server/server_test.go index e001264..12fa24f 100644 --- a/sandboxd/server/server_test.go +++ b/sandboxd/server/server_test.go @@ -950,6 +950,8 @@ func (f *fakeManager) Info() ([]pool.PoolInfo, int, int) { return f.infoPools, 0, 0 } +func (f *fakeManager) ArchivedCount() int { return 0 } + type fakeDialer struct { dial func(ctx context.Context, sock string) (net.Conn, error) } diff --git a/sandboxd/types/bench_test.go b/sandboxd/types/bench_test.go new file mode 100644 index 0000000..c29fc7a --- /dev/null +++ b/sandboxd/types/bench_test.go @@ -0,0 +1,35 @@ +package types + +import ( + "sync" + "testing" + "time" +) + +// BenchmarkSandboxTouch measures the relay hot-path last-activity stamp; the +// lock-free atomic store must stay flat under -cpu parallelism (no contention). +func BenchmarkSandboxTouch(b *testing.B) { + sb := &Sandbox{} + b.RunParallel(func(pb *testing.PB) { + for pb.Next() { + sb.Touch() + } + }) +} + +// BenchmarkSandboxTouchMutex is the pre-E1a baseline (mutex-guarded time.Time) +// for contrast: it degrades with parallelism where the atomic stays flat. +func BenchmarkSandboxTouchMutex(b *testing.B) { + var ( + mu sync.Mutex + t time.Time + ) + b.RunParallel(func(pb *testing.PB) { + for pb.Next() { + mu.Lock() + t = time.Now() + mu.Unlock() + } + }) + _ = t +} diff --git a/sandboxd/types/types.go b/sandboxd/types/types.go index 7198383..a912e49 100644 --- a/sandboxd/types/types.go +++ b/sandboxd/types/types.go @@ -9,6 +9,7 @@ import ( "fmt" "regexp" "sync" + "sync/atomic" "time" ) @@ -121,21 +122,34 @@ type Sandbox struct { // HibernateSnap names the memory snapshot while the VM is hibernated; // empty means running. HibernateSnap string `json:"hibernate_snap,omitempty"` + // ArchiveCk names the store checkpoint holding this sandbox's state while + // archived; empty means live or hibernated. While set, VMName/VsockSocket/ + // HibernateSnap are empty (no local VM) and Deadline is the retention deadline. + ArchiveCk string `json:"archive_ck,omitempty"` // FromCheckpoint names the checkpoint this sandbox branched from, for // lineage; empty for pool and template claims. FromCheckpoint string `json:"from_checkpoint,omitempty"` - // LastActivity is the last data-plane connection, for the idle policy; - // runtime-only and guarded by the manager mutex. A restart resets it to - // adoption time. - LastActivity time.Time `json:"-"` + // lastActivity is unix-nanos of the last data-plane connection, for the + // idle policy; lock-free, stamped on the relay hot path. Runtime-only, a + // restart resets it to adoption time. + lastActivity atomic.Int64 // Transition serializes hibernate/wake so concurrent wakes collapse onto // one restore. Lock it before (never under) the manager mutex. Transition sync.Mutex `json:"-"` } +// Touch stamps last data-plane activity; lock-free, called on the relay hot path. +func (s *Sandbox) Touch() { s.lastActivity.Store(time.Now().UnixNano()) } + +// TouchAt stamps last-activity at a caller-supplied instant (batch claim/adoption, tests). +func (s *Sandbox) TouchAt(t time.Time) { s.lastActivity.Store(t.UnixNano()) } + +// LastSeen returns the last data-plane activity time. +func (s *Sandbox) LastSeen() time.Time { return time.Unix(0, s.lastActivity.Load()) } + // Checkpoint is the record of a captured sandbox state: claims cloned from // it branch off the exact captured moment. Node-local, like a template. type Checkpoint struct { diff --git a/scripts/archive-e2e.sh b/scripts/archive-e2e.sh new file mode 100644 index 0000000..9258a22 --- /dev/null +++ b/scripts/archive-e2e.sh @@ -0,0 +1,74 @@ +#!/usr/bin/env bash +# Bare-metal acceptance for M7-E lifecycle auto-archive: a claim idles into +# hibernation, then to a store checkpoint (its local VM dropped), and the next +# access wakes it from the store with the same id and guest state intact. +# Runs a dedicated sandboxd on port 7778 so it never collides with +# sandboxd-e2e.sh (7777). +set -euo pipefail + +TEMPLATE=${TEMPLATE:-rt:24.04} +ADDR=${ADDR:-127.0.0.1:7778} +TOKEN=${TOKEN:-e2e} +REPO=$(cd "$(dirname "$0")/.." && pwd) + +DATA=$(mktemp -d /tmp/archive-e2e.XXXXXX) +DAEMON_PID="" + +cleanup() { + status=$? + if [[ $status -ne 0 && -f "$DATA/daemon.log" ]]; then + echo "== daemon log tail" + tail -30 "$DATA/daemon.log" + fi + [[ -n $DAEMON_PID ]] && kill "$DAEMON_PID" 2>/dev/null || true + wait 2>/dev/null || true + cocoon vm list --format json 2>/dev/null | + jq -r '.[] | select(.config.name | startswith("sbx-")) | .config.name' | + while read -r vm; do + cocoon vm stop --force "$vm" >/dev/null 2>&1 || true + cocoon vm rm --force "$vm" >/dev/null 2>&1 || true + done || true + rm -rf "$DATA" + exit "$status" +} +trap cleanup EXIT + +api() { curl -sf -H "Authorization: Bearer $TOKEN" "http://$ADDR/v1/$1"; } + +echo "== build" +if [[ -n ${SANDBOXD_BIN:-} && -n ${LIFECYCLE_BIN:-} ]]; then + cp "$SANDBOXD_BIN" "$DATA/sandboxd" && cp "$LIFECYCLE_BIN" "$DATA/lifecycle" +else + (cd "$REPO/sandboxd" && GOWORK=off go build -o "$DATA/sandboxd" .) + (cd "$REPO/e2e" && GOWORK=off go build -o "$DATA/lifecycle" ./cmd/lifecycle) +fi + +cat >"$DATA/config.json" <>"$DATA/daemon.log" 2>&1 & +DAEMON_PID=$! +for _ in $(seq 1 20); do curl -sf "http://$ADDR/healthz" >/dev/null && break; sleep 0.5; done + +echo "== wait for golden + warm pool" +for i in $(seq 1 120); do + if api info | jq -e '(.pools|length)>0 and all(.pools[]; .golden and .warm>=.target)' >/dev/null 2>&1; then break; fi + [[ $i == 120 ]] && { echo "pools never ready"; api info | jq . || true; exit 1; } + sleep 1 +done +api info | jq '{pools:(.pools|length),claimed,hibernated,archived}' + +echo "== archive lifecycle: claim -> idle-hibernate -> archive -> wake" +"$DATA/lifecycle" -addr "$ADDR" -token "$TOKEN" -template "$TEMPLATE" -net none -wait 90s + +echo "== post-run node state" +api info | jq '{claimed,hibernated,archived}' +echo "PASS" diff --git a/sdk/go/info.go b/sdk/go/info.go index 3826b81..4d2b6c7 100644 --- a/sdk/go/info.go +++ b/sdk/go/info.go @@ -15,6 +15,7 @@ type NodeInfo struct { Pools []PoolStatus `json:"pools"` Claimed int `json:"claimed"` Hibernated int `json:"hibernated"` + Archived int `json:"archived"` Peers []string `json:"peers,omitempty"` } From 25ffcadfb5053a9beb2b0429ac0b5e217e002b29 Mon Sep 17 00:00:00 2001 From: CMGS Date: Thu, 9 Jul 2026 12:30:07 +0800 Subject: [PATCH 2/3] review(pool): dedup abort-ck delete + benign-sweep-error predicate Post-review /simplify pass: - archive.go: extract deleteOrphanArchiveCk (two identical abort-path ck deletes) and compress the archive*For doc comment. - pool.go: extract benignSweepErr - the released/woke-mid-sweep error set was inlined in idleOnce/archiveOnce/reapOnce. - e2e/lifecycle: rename waitArchived's max param (shadowed the builtin). --- e2e/cmd/lifecycle/main.go | 6 +++--- sandboxd/pool/archive.go | 21 +++++++++++---------- sandboxd/pool/claim.go | 3 +-- sandboxd/pool/hibernate.go | 3 +-- sandboxd/pool/pool.go | 15 +++++++++++---- 5 files changed, 27 insertions(+), 21 deletions(-) diff --git a/e2e/cmd/lifecycle/main.go b/e2e/cmd/lifecycle/main.go index f08435d..a7dcabb 100644 --- a/e2e/cmd/lifecycle/main.go +++ b/e2e/cmd/lifecycle/main.go @@ -91,9 +91,9 @@ func run(addr, token, template, netShape string, wait time.Duration) error { // waitArchived polls the node until archived reaches n, returning how long the // reaper's idle→hibernate→archive ladder took. -func waitArchived(ctx context.Context, client *sandbox.Client, n int, max time.Duration) (time.Duration, error) { +func waitArchived(ctx context.Context, client *sandbox.Client, n int, timeout time.Duration) (time.Duration, error) { start := time.Now() - deadline := start.Add(max) + deadline := start.Add(timeout) for { info, err := client.Info(ctx) if err != nil { @@ -104,7 +104,7 @@ func waitArchived(ctx context.Context, client *sandbox.Client, n int, max time.D } if time.Now().After(deadline) { return 0, fmt.Errorf("archived never reached %d within %s (hibernated=%d archived=%d claimed=%d)", - n, max, info.Hibernated, info.Archived, info.Claimed) + n, timeout, info.Hibernated, info.Archived, info.Claimed) } time.Sleep(500 * time.Millisecond) } diff --git a/sandboxd/pool/archive.go b/sandboxd/pool/archive.go index e1d5bec..34295eb 100644 --- a/sandboxd/pool/archive.go +++ b/sandboxd/pool/archive.go @@ -12,9 +12,7 @@ import ( "github.com/cocoonstack/sandbox/sandboxd/types" ) -// archiveAfterFor / archiveDeleteFor / archiveEnabledFor resolve a key's -// archive thresholds — the pool's when pooled, the node default otherwise. -// Callers hold m.mu (they read m.pools). +// archive*For resolve a key's thresholds (pool's, else node default); callers hold m.mu. func (m *Manager) archiveAfterFor(key types.PoolKey) time.Duration { if p, ok := m.pools[key]; ok { return p.archiveAfter @@ -72,7 +70,7 @@ func (m *Manager) archiveOnce(ctx context.Context) { switch err := m.archive(ctx, sb); { case err == nil: logger.Infof(ctx, "archived %s", sb.ID) - case !errors.Is(err, ErrUnknownSandbox) && !errors.Is(err, errWokeMeanwhile): + case !benignSweepErr(err): logger.Errorf(ctx, err, "archive %s", sb.ID) } }).Wait() @@ -107,9 +105,7 @@ func (m *Manager) archive(ctx context.Context, sb *types.Sandbox) error { if m.claimed[sb.ID] != sb || sb.HibernateSnap == "" || sb.ArchiveCk != "" { m.mu.Unlock() sb.Transition.Unlock() - if delErr := m.ckpts.Delete(ctx, ck.ID); delErr != nil { - log.WithFunc("pool.archive").Warnf(ctx, "delete orphaned archive ck %s: %v", ck.ID, delErr) - } + m.deleteOrphanArchiveCk(ctx, ck.ID) return errWokeMeanwhile } snap, vmName, sock, prevDeadline := sb.HibernateSnap, sb.VMName, sb.VsockSocket, sb.Deadline @@ -128,9 +124,7 @@ func (m *Manager) archive(ctx context.Context, sb *types.Sandbox) error { sb.Deadline = prevDeadline m.mu.Unlock() sb.Transition.Unlock() - if delErr := m.ckpts.Delete(ctx, ck.ID); delErr != nil { - log.WithFunc("pool.archive").Warnf(ctx, "delete orphaned archive ck %s: %v", ck.ID, delErr) - } + m.deleteOrphanArchiveCk(ctx, ck.ID) return fmt.Errorf("archive %s: persist claims: %w", sb.ID, saveErr) } m.mu.Unlock() @@ -203,3 +197,10 @@ func (m *Manager) commitWake(ctx context.Context, sb *types.Sandbox, vmName, soc } return live, saveErr == nil } + +// deleteOrphanArchiveCk drops the published ck when archive() aborts pre-commit. +func (m *Manager) deleteOrphanArchiveCk(ctx context.Context, ckID string) { + if err := m.ckpts.Delete(ctx, ckID); err != nil { + log.WithFunc("pool.archive").Warnf(ctx, "delete orphaned archive ck %s: %v", ckID, err) + } +} diff --git a/sandboxd/pool/claim.go b/sandboxd/pool/claim.go index 3315bc2..9e82249 100644 --- a/sandboxd/pool/claim.go +++ b/sandboxd/pool/claim.go @@ -3,7 +3,6 @@ package pool import ( "context" "crypto/subtle" - "errors" "fmt" "net" "time" @@ -327,7 +326,7 @@ func (m *Manager) reapOnce(ctx context.Context) { switch err := m.archive(ctx, v.sb); { case err == nil: logger.Infof(ctx, "archived expired sandbox %s", v.id) - case !errors.Is(err, ErrUnknownSandbox) && !errors.Is(err, errWokeMeanwhile): + case !benignSweepErr(err): logger.Errorf(ctx, err, "archive expired %s", v.id) } default: diff --git a/sandboxd/pool/hibernate.go b/sandboxd/pool/hibernate.go index c1b7824..e3bdd6f 100644 --- a/sandboxd/pool/hibernate.go +++ b/sandboxd/pool/hibernate.go @@ -2,7 +2,6 @@ package pool import ( "context" - "errors" "fmt" "strings" "time" @@ -137,7 +136,7 @@ func (m *Manager) idleOnce(ctx context.Context) { switch err := m.idleHibernate(ctx, v.id, v.token, now); { case err == nil: logger.Infof(ctx, "idle-hibernated %s", v.id) - case !errors.Is(err, ErrUnknownSandbox) && !errors.Is(err, errWokeMeanwhile): + case !benignSweepErr(err): logger.Errorf(ctx, err, "idle-hibernate %s", v.id) } }).Wait() diff --git a/sandboxd/pool/pool.go b/sandboxd/pool/pool.go index fbe1081..9232839 100644 --- a/sandboxd/pool/pool.go +++ b/sandboxd/pool/pool.go @@ -82,7 +82,7 @@ type Engine interface { DialGuestPort(ctx context.Context, vsockSocket string, port uint16) (net.Conn, error) } -// SandboxSummary is the ops view of one live claim — no tokens. +// SandboxSummary is the ops view of one live claim — no tokens. type SandboxSummary struct { ID string `json:"id"` Key types.PoolKey `json:"key"` @@ -385,7 +385,7 @@ func (m *Manager) Reconcile(ctx context.Context) error { // Snapshot sweep, symmetric to the VM sweep: a hibernate snapshot no // adopted claim references is an orphan (a crash between `vm hibernate` // and the journal commit), and fork/golden-build snapshots are transient - // by construction — none can span a restart. A list failure only skips + // by construction — none can span a restart. A list failure only skips // the sweep: GC must not brick startup. if snaps, listErr := m.eng.SnapshotList(ctx); listErr != nil { logger.Warnf(ctx, "snapshot sweep skipped: %v", listErr) @@ -522,7 +522,7 @@ func (m *Manager) Info() ([]PoolInfo, int, int) { } // ArchivedCount returns how many live claims are archived to the store (no -// local VM); separate from Info's gauges to keep its arity — and its callers. +// local VM); separate from Info's gauges to keep its arity — and its callers. func (m *Manager) ArchivedCount() int { m.mu.Lock() defer m.mu.Unlock() @@ -607,13 +607,20 @@ func dirExists(path string) bool { return err == nil && fi.IsDir() } -// tenantOwns is THE tenancy predicate — every read/delete/overwrite scope +// tenantOwns is THE tenancy predicate — every read/delete/overwrite scope // check goes through it: root (empty tenant) owns everything, a tenant only // records stamped with its own name. func tenantOwns(tenant, owner string) bool { return tenant == "" || tenant == owner } +// benignSweepErr reports whether err is the expected outcome of a housekeeping +// sweep racing the data plane (victim released, or woke mid-sweep), so it is +// not worth logging as a failure. +func benignSweepErr(err error) bool { + return errors.Is(err, ErrUnknownSandbox) || errors.Is(err, errWokeMeanwhile) +} + func vmName(key types.PoolKey) string { return vmPrefix + key.Hash() + "-" + randHex(3) } From cfd7efd044827f15e23345b3dc7abe3f259ead9a Mon Sep 17 00:00:00 2001 From: CMGS Date: Thu, 9 Jul 2026 12:33:23 +0800 Subject: [PATCH 3/3] review(config): move validateArchiveWindow below the type cluster Layout-only: the shared validation helper was wedged between the PoolSpec and StoreConfig type declarations, breaking the config-type cluster. Move it to the bottom with the other validate* funcs (standalone funcs below the type+method blocks). --- sandboxd/config/config.go | 24 ++++++++++++------------ 1 file changed, 12 insertions(+), 12 deletions(-) diff --git a/sandboxd/config/config.go b/sandboxd/config/config.go index 108ce36..92202b9 100644 --- a/sandboxd/config/config.go +++ b/sandboxd/config/config.go @@ -60,18 +60,6 @@ func (s PoolSpec) ValidateLimits() error { return validateArchiveWindow(s.IdleHibernateSeconds, s.ArchiveAfterSeconds, s.ArchiveDeleteAfterSeconds) } -// validateArchiveWindow checks the archive thresholds shared by PoolSpec and -// the node Config: non-negative, and archive_after must sit past idle_hibernate. -func validateArchiveWindow(idle, after, del int) error { - if after < 0 || del < 0 { - return fmt.Errorf("archive seconds must not be negative") - } - if after > 0 && (idle <= 0 || after <= idle) { - return fmt.Errorf("archive_after_seconds %d requires idle_hibernate_seconds>0 and a larger value", after) - } - return nil -} - // StoreConfig selects a checkpoint backend. type StoreConfig struct { Kind string `json:"kind"` @@ -309,3 +297,15 @@ func (c *Config) validateTenants() error { } return nil } + +// validateArchiveWindow checks the archive thresholds shared by PoolSpec and +// the node Config: non-negative, and archive_after must sit past idle_hibernate. +func validateArchiveWindow(idle, after, del int) error { + if after < 0 || del < 0 { + return fmt.Errorf("archive seconds must not be negative") + } + if after > 0 && (idle <= 0 || after <= idle) { + return fmt.Errorf("archive_after_seconds %d requires idle_hibernate_seconds>0 and a larger value", after) + } + return nil +}