review: audit after the node-loss, reader and S3 chunk merges

README.md

@@ -42,8 +42,10 @@ performance) — source in
   host-side credential injection and per-claim upstream proxies, runtime
   tenants, config reload (SIGHUP), usage + audit journals, /metrics, reap +
   restart reconcile with VMM cold restart, memberlist mesh with redirect
-  placement, and an optional shared PostgreSQL meta store for tenants,
-  pools and the cross-node adoption of archived claims
+  placement, a checkpoint store on a shared directory or S3 (sparse,
+  content-addressed chunks), and an optional shared PostgreSQL meta store
+  for tenants, pools, node leases and the cross-node adoption of archived
+  claims and of running claims a pool replicates (`replicate_seconds`)
 - `sdk/go/` — Go SDK: `Connect/New/Lookup`, `Exec/Run`, files,
   `Push/Pull`, sessions, `Find/Replace`, `Watch`, git verbs, `OpenPty`,
   `Renew/Fork/Hibernate/Promote/Checkpoint`, claim env, `DialPort/ProxyPort/PreviewURL`,

docs/index.md

@@ -21,7 +21,8 @@ vsock-only I/O (hardened default); `net=egress` attaches a bridge/CNI NIC.
 - [Deploying sandboxd](deploy.md) — single node, configuration reference,
   images, running as a service
 - [Clusters](cluster.md) — joining a mesh, querying members, how redirect
-  placement works, relocating handles
+  placement works, relocating handles, and surviving node loss (archived and
+  replicated claims)
 - [sandboxd HTTP API](sandboxd-api.md) — every endpoint: claim/release,
   renew, hibernate, fork, promote, checkpoints, preview, the guest-port
   relay, claim env, live volume changes, runtime tenants, config reload,

docs/security.md

@@ -64,10 +64,10 @@ all of them.
    packet in that shape can reach the local L2 segment.
 4. **Set `mesh.cluster_key` unless the gossip network is itself trusted**,
    and open the memberlist port node-to-node only.
-5. **Treat the checkpoint store like backups.** A checkpoint embeds full
-   guest memory — including any secret the workload held at capture.
-   Restrict store access; use S3 server-side encryption or an encrypted
-   mount where that matters.
+5. **Treat the checkpoint store like backups.** A checkpoint, an archive and
+   a replica each embed full guest memory, including any secret the workload
+   held at capture. Restrict store access; use S3 server-side encryption or
+   an encrypted mount where that matters.
 6. **No custom NAT64 prefix routed on a sandboxd host** — the SSRF guard
    cannot see through an operator-specific translator prefix
    ([egress](egress.md#deployment-constraints)).
@@ -149,6 +149,8 @@ Facts to plan around, stated so the boundary is honest:
 - **Rate limiting is capacity-based only** (`max_claims`, per-tenant
   caps answer 429). The API assumes callers inside the trust boundary;
   front it with your own limiter if semi-trusted automation can reach it.
-- **A dead node's sandboxes die with it** — memory state is node-local by
-  design ([clusters](cluster.md)); durability is the checkpoint store's
-  job, not the node's.
+- **A dead node's running sandboxes die with it** unless their pool
+  replicates them (`replicate_seconds`): another node then restores the last
+  replica, losing what ran after it. Archived claims are adopted from the
+  checkpoint store ([clusters](cluster.md#archived-claims-across-node-loss));
+  durability is the checkpoint store's job, not the node's.

sandboxd/archives/pg/pg_test.go

@@ -2,6 +2,7 @@ package pg
 
 import (
 	"errors"
+	"sync"
 	"testing"
 	"time"
 
@@ -123,6 +124,25 @@ func TestLeasesAreScopedToTheirCell(t *testing.T) {
 	}
 }
 
+func TestNodesStartingTogetherCreateTheTablesOnce(t *testing.T) {
+	dsn := metastoretest.PGSchema(t)
+	var wg sync.WaitGroup
+	errs := make([]error, 8)
+	for i := range errs {
+		wg.Go(func() {
+			r, err := Open(t.Context(), dsn, "c1")
+			if err == nil {
+				r.Close()
+			}
+			errs[i] = err
+		})
+	}
+	wg.Wait()
+	if err := errors.Join(errs...); err != nil {
+		t.Fatalf("a node starting beside others failed: %v", err)
+	}
+}
+
 func open(t *testing.T, dsn, cell string) *Registry {
 	t.Helper()
 	r, err := Open(t.Context(), dsn, cell)

sandboxd/metastore/metastore.go

@@ -90,15 +90,22 @@ func Listen(ctx context.Context, cfg *pgx.ConnConfig, channel string, synced fun
 func EnsureTable(ctx context.Context, pool *pgxpool.Pool, table, schema string) error {
 	qctx, cancel := context.WithTimeout(ctx, QueryTimeout)
 	defer cancel()
-	var exists bool
-	if err := pool.QueryRow(qctx, `SELECT to_regclass($1) IS NOT NULL`, table).Scan(&exists); err != nil {
-		return fmt.Errorf("find table %s: %w", table, err)
-	}
-	if exists {
-		return nil
+	if exists, err := tableExists(qctx, pool, table); err != nil || exists {
+		return err
 	}
 	if _, err := pool.Exec(qctx, schema); err != nil {
+		if exists, _ := tableExists(qctx, pool, table); exists {
+			return nil
+		}
 		return fmt.Errorf("create table %s: %w", table, err)
 	}
 	return nil
 }
+
+func tableExists(ctx context.Context, pool *pgxpool.Pool, table string) (bool, error) {
+	var exists bool
+	if err := pool.QueryRow(ctx, `SELECT to_regclass($1) IS NOT NULL`, table).Scan(&exists); err != nil {
+		return false, fmt.Errorf("find table %s: %w", table, err)
+	}
+	return exists, nil
+}

sandboxd/pool/claims.go

@@ -54,7 +54,7 @@ type claimDTO struct {
 	TAP            string             `json:"tap,omitempty"`
 	HibernateSnap  string             `json:"hibernate_snap,omitempty"`
 	PendingSnap    string             `json:"pending_snap,omitempty"`
-	ReaderSafe     bool               `json:"reader_safe,omitempty"`
+	ReaderSafe     bool               `json:"reader_safe,omitzero"`
 	ArchiveCk      string             `json:"archive_ck,omitempty"`
 	ArchiveEpoch   int64              `json:"archive_epoch,omitzero"`
 	ArchivePending string             `json:"archive_pending,omitempty"`

sandboxd/pool/claims_test.go

@@ -9,6 +9,7 @@ import (
 	"reflect"
 	"runtime"
 	"slices"
+	"strings"
 	"sync"
 	"syscall"
 	"testing"
@@ -121,6 +122,18 @@ func TestClaimStoreSnapshotDetachesVolumes(t *testing.T) {
 	}
 }
 
+func TestClaimRowOmitsUnsetFlags(t *testing.T) {
+	raw, err := encodeRow(new(dtoOf(&types.Sandbox{ID: "sb_1", VMName: "vm"})))
+	if err != nil {
+		t.Fatalf("encode: %v", err)
+	}
+	for _, flag := range []string{"reader_safe", "no_egress"} {
+		if strings.Contains(string(raw), flag) {
+			t.Errorf("row %s carries the unset flag %s", raw, flag)
+		}
+	}
+}
+
 func TestStoreLoadMissingFile(t *testing.T) {
 	got, err := newClaimStore(t.TempDir(), false).load()
 	if err != nil {

sandboxd/pool/pool_test.go

@@ -624,6 +624,7 @@ func TestSetPoolsRejectsConfigOwnedFields(t *testing.T) {
 		{"storage", config.PoolSpec{PoolKey: testKey, Warm: 1, Storage: "40G"}},
 		{"replicate_seconds", config.PoolSpec{PoolKey: testKey, Warm: 1, ReplicateSeconds: 60}},
 		{"reader", config.PoolSpec{PoolKey: testKey, Warm: 1, Reader: true}},
+		{"egress_upstream_env", config.PoolSpec{PoolKey: testKey, Warm: 1, EgressUpstreamEnv: "UPSTREAM"}},
 	} {
 		err := m.SetPools(t.Context(), []config.PoolSpec{tt.spec})
 		if !errors.Is(err, ErrBadKey) || !strings.Contains(err.Error(), tt.field+" is set in the config file") {

sandboxd/pool/setpools.go

@@ -40,27 +40,22 @@ func (m *Manager) desiredPools(specs []config.PoolSpec) (map[types.PoolKey]confi
 		if err := spec.ValidateLimits(); err != nil {
 			return nil, fmt.Errorf("%w: %w", ErrBadCount, err)
 		}
-		// egress is config-owned; accepting it here would silently drop it
-		if spec.Egress != nil {
-			return nil, fmt.Errorf("%w: pool %q: egress is set in the config file, not via the API", ErrBadKey, spec.Template)
-		}
-		if spec.Warmup != nil {
-			return nil, fmt.Errorf("%w: pool %q: warmup is set in the config file, not via the API", ErrBadKey, spec.Template)
-		}
-		if spec.CaptureTrim {
-			return nil, fmt.Errorf("%w: pool %q: capture_trim is set in the config file, not via the API", ErrBadKey, spec.Template)
-		}
-		if spec.Reader {
-			return nil, fmt.Errorf("%w: pool %q: reader is set in the config file, not via the API", ErrBadKey, spec.Template)
-		}
-		if spec.Storage != "" {
-			return nil, fmt.Errorf("%w: pool %q: storage is set in the config file, not via the API", ErrBadKey, spec.Template)
-		}
-		if spec.ReplicateSeconds != 0 {
-			return nil, fmt.Errorf("%w: pool %q: replicate_seconds is set in the config file, not via the API", ErrBadKey, spec.Template)
-		}
-		if spec.EgressUpstreamEnv != "" {
-			return nil, fmt.Errorf("%w: pool %q: egress_upstream_env is set in the config file, not via the API", ErrBadKey, spec.Template)
+		// config-owned fields; accepting one here would silently drop it
+		for _, f := range []struct {
+			name string
+			set  bool
+		}{
+			{"egress", spec.Egress != nil},
+			{"warmup", spec.Warmup != nil},
+			{"capture_trim", spec.CaptureTrim},
+			{"reader", spec.Reader},
+			{"storage", spec.Storage != ""},
+			{"replicate_seconds", spec.ReplicateSeconds != 0},
+			{"egress_upstream_env", spec.EgressUpstreamEnv != ""},
+		} {
+			if f.set {
+				return nil, fmt.Errorf("%w: pool %q: %s is set in the config file, not via the API", ErrBadKey, spec.Template, f.name)
+			}
 		}
 		if _, ok := desired[spec.PoolKey]; ok {
 			return nil, fmt.Errorf("%w: duplicate pool %q", ErrBadKey, spec.Template)

sandboxd/pool/telemetry.go

@@ -14,8 +14,12 @@ import (
 	"github.com/cocoonstack/sandbox/sandboxd/egress"
 )
 
-// AuditLineCap bounds how much of a relayed request frame the audit tap captures and records.
-const AuditLineCap = 4096
+const (
+	// AuditLineCap bounds how much of a relayed request frame the audit tap captures and records.
+	AuditLineCap = 4096
+
+	egressEvent = "egress"
+)
 
 // Counters is a monotonic snapshot for /metrics; gauges come from Info.
 type Counters struct {
@@ -62,22 +66,18 @@ type requestFrame struct {
 }
 
 type auditFrame struct {
-	Op       string   `json:"op"`
-	Argv     []string `json:"argv,omitempty"`
-	Path     string   `json:"path,omitempty"`
-	Dest     string   `json:"dest,omitempty"`
-	Method   string   `json:"method,omitempty"`
-	From     string   `json:"from,omitempty"`
-	To       string   `json:"to,omitempty"`
-	URL      string   `json:"url,omitempty"`
-	Session  string   `json:"session,omitempty"`
-	Port     uint16   `json:"port,omitzero"`
-	Decision string   `json:"decision,omitempty"`
-	Secret   string   `json:"secret,omitempty"`
-	Upstream string   `json:"upstream,omitempty"`
-	Added    []string `json:"added,omitempty"`
-	Changed  []string `json:"changed,omitempty"`
-	Removed  []string `json:"removed,omitempty"`
+	Op      string   `json:"op"`
+	Argv    []string `json:"argv,omitempty"`
+	Path    string   `json:"path,omitempty"`
+	Dest    string   `json:"dest,omitempty"`
+	From    string   `json:"from,omitempty"`
+	To      string   `json:"to,omitempty"`
+	URL     string   `json:"url,omitempty"`
+	Session string   `json:"session,omitempty"`
+	Port    uint16   `json:"port,omitzero"`
+	Added   []string `json:"added,omitempty"`
+	Changed []string `json:"changed,omitempty"`
+	Removed []string `json:"removed,omitempty"`
 }
 
 type egressFrame struct {
@@ -178,11 +178,11 @@ func (m *Manager) recordEgress(ctx context.Context, id, tenant string, ev egress
 			ID   string    `json:"id"`
 			egressFrame
 		}{time.Now(), id, egressFrame{
-			Op: "egress", Dest: ev.Host, Method: ev.Method, Port: ev.Port, Decision: decision, Secret: ev.Injected, Upstream: ev.Upstream,
+			Op: egressEvent, Dest: ev.Host, Method: ev.Method, Port: ev.Port, Decision: decision, Secret: ev.Injected, Upstream: ev.Upstream,
 			Tenant: tenant, Src: ev.Local, Dst: ev.Remote,
 		}})
 	}
-	m.recordUsage(ctx, usageEvent{Event: "egress", ID: id, Tenant: tenant, Reference: ev.Host, Upstream: ev.Upstream, Secret: ev.Injected})
+	m.recordUsage(ctx, usageEvent{Event: egressEvent, ID: id, Tenant: tenant, Reference: ev.Host, Upstream: ev.Upstream, Secret: ev.Injected})
 }
 
 // recordUsage appends one billing event; failures are logged, never propagated.

sandboxd/store/s3/s3.go

@@ -510,29 +510,34 @@ func (s *Store) referencedChunks(ctx context.Context, id, lineage string) (map[s
 			ids = append(ids, path.Base(*marker.Key))
 		}
 	}
-	refs := &chunkRefs{live: map[string]map[string]struct{}{}}
+	have := map[string]struct{}{}
+	var mu sync.Mutex
 	g, gctx := errgroup.WithContext(ctx)
 	g.SetLimit(metaReadConcurrency)
 	for _, recordID := range ids {
 		g.Go(func() error {
 			man, err := s.currentManifest(gctx, recordID)
-			if errors.Is(err, store.ErrNotFound) {
+			switch {
+			case errors.Is(err, store.ErrNotFound):
 				return nil
-			}
-			if err != nil {
+			case err != nil:
 				return err
+			case man.Lineage != lineage:
+				return nil
+			}
+			mu.Lock()
+			defer mu.Unlock()
+			for _, file := range man.Files {
+				for _, chunk := range file.Chunks {
+					have[chunk.SHA256] = struct{}{}
+				}
 			}
-			refs.add(man)
 			return nil
 		})
 	}
 	if err := g.Wait(); err != nil {
 		return nil, err
 	}
-	have := refs.live[lineage]
-	if have == nil {
-		have = map[string]struct{}{}
-	}
 	return have, nil
 }
 
@@ -659,26 +664,6 @@ func (u *chunkUploader) claim(sha string) bool {
 	return true
 }
 
-type chunkRefs struct {
-	mu   sync.Mutex
-	live map[string]map[string]struct{} // lineage -> chunk sha
-}
-
-func (r *chunkRefs) add(man manifest) {
-	r.mu.Lock()
-	defer r.mu.Unlock()
-	set := r.live[man.Lineage]
-	if set == nil {
-		set = map[string]struct{}{}
-		r.live[man.Lineage] = set
-	}
-	for _, file := range man.Files {
-		for _, chunk := range file.Chunks {
-			set[chunk.SHA256] = struct{}{}
-		}
-	}
-}
-
 func publishFiles(staging string) ([]publishFile, error) {
 	var files []publishFile
 	err := store.WalkExport(filepath.Join(staging, store.ExportDir), func(rel, abs string, _ fs.FileInfo) error {

sandboxd/types/types.go

@@ -571,7 +571,7 @@ type Sandbox struct {
 	HibernateSnap string `json:"hibernate_snap,omitempty"`
 	// PendingSnap is the journaled intent of a hibernate in flight, cleared by its commit.
 	PendingSnap string `json:"pending_snap,omitempty"`
-	ReaderSafe  bool   `json:"reader_safe,omitempty"`
+	ReaderSafe  bool   `json:"reader_safe,omitzero"`
 	// ArchiveCk names the store checkpoint holding an archived sandbox; empty means local.
 	ArchiveCk string `json:"archive_ck,omitempty"`
 	// ArchiveEpoch is the archive registry epoch this node last held for the claim; zero means no row.