diff --git a/README.md b/README.md index d8223efb..04468e8d 100644 --- a/README.md +++ b/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`, diff --git a/docs/index.md b/docs/index.md index 5d34943c..0c0e15e5 100644 --- a/docs/index.md +++ b/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, diff --git a/docs/security.md b/docs/security.md index d73927e9..6c519f49 100644 --- a/docs/security.md +++ b/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. diff --git a/sandboxd/archives/pg/pg_test.go b/sandboxd/archives/pg/pg_test.go index d07a9de9..5d7ae286 100644 --- a/sandboxd/archives/pg/pg_test.go +++ b/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) diff --git a/sandboxd/metastore/metastore.go b/sandboxd/metastore/metastore.go index fee99631..b8287fbf 100644 --- a/sandboxd/metastore/metastore.go +++ b/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 +} diff --git a/sandboxd/pool/claims.go b/sandboxd/pool/claims.go index 12bcbeb3..5d7111df 100644 --- a/sandboxd/pool/claims.go +++ b/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"` diff --git a/sandboxd/pool/claims_test.go b/sandboxd/pool/claims_test.go index 6f145ea4..0539a537 100644 --- a/sandboxd/pool/claims_test.go +++ b/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 { diff --git a/sandboxd/pool/pool_test.go b/sandboxd/pool/pool_test.go index 10c7c81e..136cd118 100644 --- a/sandboxd/pool/pool_test.go +++ b/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") { diff --git a/sandboxd/pool/setpools.go b/sandboxd/pool/setpools.go index 442a3aac..d4d74357 100644 --- a/sandboxd/pool/setpools.go +++ b/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) diff --git a/sandboxd/pool/telemetry.go b/sandboxd/pool/telemetry.go index d4156736..ada7fd5f 100644 --- a/sandboxd/pool/telemetry.go +++ b/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. diff --git a/sandboxd/store/s3/s3.go b/sandboxd/store/s3/s3.go index 51206e9f..941f93bb 100644 --- a/sandboxd/store/s3/s3.go +++ b/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 { diff --git a/sandboxd/types/types.go b/sandboxd/types/types.go index a7a9983b..4cc8d307 100644 --- a/sandboxd/types/types.go +++ b/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.