feat(pool): lifecycle auto-archive for idle sandboxes

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, timeout time.Duration) (time.Duration, error) {
+	start := time.Now()
+	deadline := start.Add(timeout)
+	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, timeout, info.Hibernated, info.Archived, info.Claimed)
+		}
+		time.Sleep(500 * time.Millisecond)
+	}
+}

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,7 +57,7 @@ func (s PoolSpec) ValidateLimits() error {
 	if s.IdleHibernateSeconds < 0 {
 		return fmt.Errorf("idle_hibernate_seconds must not be negative")
 	}
-	return nil
+	return validateArchiveWindow(s.IdleHibernateSeconds, s.ArchiveAfterSeconds, s.ArchiveDeleteAfterSeconds)
 }
 
 // StoreConfig selects a checkpoint backend.
@@ -111,6 +120,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 +233,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":
@@ -279,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
+}

sandboxd/pool/archive.go

@@ -0,0 +1,206 @@
+package pool
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"time"
+
+	"github.com/projecteru2/core/log"
+
+	"github.com/cocoonstack/sandbox/sandboxd/store"
+	"github.com/cocoonstack/sandbox/sandboxd/types"
+)
+
+// 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
+	}
+	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 !benignSweepErr(err):
+				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()
+		m.deleteOrphanArchiveCk(ctx, ck.ID)
+		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()
+		m.deleteOrphanArchiveCk(ctx, ck.ID)
+		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
+}
+
+// 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)
+	}
+}