review: test fix, style, cuts, docs

docs/deploy.md

@@ -586,9 +586,11 @@ the tenant set.
 - **Boot:** a node applies the cell's row over the `pools` section of
   `config.json`. Until the cell's first `PUT`, nodes run their config pools,
   and a `pools.json` left from before the node joined the cell is ignored with
-  a warning; `PUT` its set once to share it. Each node still writes the set it
-  applied to its `pools.json`, which it serves when the row cannot be read at
-  boot.
+  a warning; `PUT` its set once to share it. A node with `meta_store` needs the
+  database at boot, since both shared sets open it there. Each node still
+  writes the set it applied to its `pools.json`, which it serves when the row
+  read fails after the connection succeeds, and which it keeps if it leaves
+  the cell.
 - **Connection:** the pool set has its own query pool, idle between writes,
   and its own `LISTEN sandboxd_pool_sets` session, so each node holds two
   LISTEN sessions; count them in the database's `max_connections`.

sandboxd/config/config.go

@@ -306,12 +306,9 @@ func (c *Config) egressFingerprint() egressFingerprint {
 	fp := egressFingerprint{
 		Classes:  slices.SortedFunc(slices.Values(c.EgressClasses), func(a, b EgressClass) int { return strings.Compare(a.Name, b.Name) }),
 		Internal: slices.Sorted(slices.Values(c.EgressInternalAllow)),
+		Secrets:  slices.SortedFunc(slices.Values(c.Secrets), func(a, b egress.SecretSpec) int { return strings.Compare(a.Name, b.Name) }),
 		Pools:    map[string]PoolSpec{},
 	}
-	for _, s := range c.Secrets {
-		fp.Secrets = append(fp.Secrets, egress.SecretSpec{Name: s.Name, Header: s.Header, ValueEnv: s.ValueEnv})
-	}
-	slices.SortFunc(fp.Secrets, func(a, b egress.SecretSpec) int { return strings.Compare(a.Name, b.Name) })
 	if u := c.EgressUpstream; u != nil {
 		fp.Upstream = &EgressUpstreamConfig{ClaimEnv: u.ClaimEnv, Allow: slices.Sorted(slices.Values(u.Allow))}
 	}

sandboxd/metastore/metastore.go

@@ -48,10 +48,28 @@ func Open(ctx context.Context, dsn, table, schema string) (*pgxpool.Pool, *pgx.C
 // Listen calls synced on each listener connection and notified on each notify, reconnecting with backoff until ctx ends.
 func Listen(ctx context.Context, cfg *pgx.ConnConfig, channel string, synced func(), notified func(payload string)) {
 	logger := log.WithFunc("metastore.Listen")
+	listenOnce := func() error {
+		conn, err := pgx.ConnectConfig(ctx, cfg)
+		if err != nil {
+			return err
+		}
+		defer func() { _ = conn.Close(context.WithoutCancel(ctx)) }()
+		if _, err = conn.Exec(ctx, "LISTEN "+channel); err != nil {
+			return err
+		}
+		synced()
+		for {
+			n, err := conn.WaitForNotification(ctx)
+			if err != nil {
+				return err
+			}
+			notified(n.Payload)
+		}
+	}
 	backoff := listenRetryFirst
 	for {
 		started := time.Now()
-		err := listenOnce(ctx, cfg, channel, synced, notified)
+		err := listenOnce()
 		if ctx.Err() != nil {
 			return
 		}
@@ -68,25 +86,6 @@ func Listen(ctx context.Context, cfg *pgx.ConnConfig, channel string, synced fun
 	}
 }
 
-func listenOnce(ctx context.Context, cfg *pgx.ConnConfig, channel string, synced func(), notified func(payload string)) error {
-	conn, err := pgx.ConnectConfig(ctx, cfg.Copy())
-	if err != nil {
-		return err
-	}
-	defer func() { _ = conn.Close(context.WithoutCancel(ctx)) }()
-	if _, err = conn.Exec(ctx, "LISTEN "+channel); err != nil {
-		return err
-	}
-	synced()
-	for {
-		n, err := conn.WaitForNotification(ctx)
-		if err != nil {
-			return err
-		}
-		notified(n.Payload)
-	}
-}
-
 func ensureTable(ctx context.Context, pool *pgxpool.Pool, table, schema string) error {
 	qctx, cancel := context.WithTimeout(ctx, QueryTimeout)
 	defer cancel()

sandboxd/pool/pool.go

@@ -693,7 +693,7 @@ func (m *Manager) activePool(key types.PoolKey) (*pool, bool) {
 
 func (m *Manager) validate(key types.PoolKey) error {
 	if err := key.Validate(); err != nil {
-		return fmt.Errorf("%w: %v", ErrBadKey, err)
+		return fmt.Errorf("%w: %w", ErrBadKey, err)
 	}
 	if key.Net == types.NetEgress && !m.egress {
 		return ErrNoEgress

sandboxd/poolset/pg/pg_test.go

@@ -3,6 +3,7 @@ package pg
 import (
 	"context"
 	"errors"
+	"strings"
 	"testing"
 	"time"
 
@@ -45,7 +46,8 @@ func TestStoreVersionsTheCellsSet(t *testing.T) {
 
 func TestWatchFiresOnConnectAndOnItsCellsWrites(t *testing.T) {
 	dsn := metastoretest.PGSchema(t)
-	a, b, other := openIn(t, dsn, "c1"), openIn(t, dsn, "c1"), openIn(t, dsn, "c2")
+	_, cell, _ := strings.CutLast(dsn, "=")
+	a, b, other := openIn(t, dsn, cell), openIn(t, dsn, cell), openIn(t, dsn, cell+"-other")
 	ctx, cancel := context.WithCancel(t.Context())
 	fired := make(chan struct{}, 8)
 	done := make(chan struct{})
@@ -56,10 +58,10 @@ func TestWatchFiresOnConnectAndOnItsCellsWrites(t *testing.T) {
 	defer func() { cancel(); <-done }()
 	waitFire(t, fired, "the listener connect")
 	if _, err := other.Store(t.Context(), poolset.Set{}); err != nil {
-		t.Fatalf("store on c2: %v", err)
+		t.Fatalf("store on the other cell: %v", err)
 	}
 	if _, err := b.Store(t.Context(), poolset.Set{}); err != nil {
-		t.Fatalf("store on c1: %v", err)
+		t.Fatalf("store on the cell: %v", err)
 	}
 	waitFire(t, fired, "a write by another node of the cell")
 	select {

sandboxd/server/server.go

@@ -645,12 +645,9 @@ func (s *Server) handleCheckpointProbe(w http.ResponseWriter, r *http.Request) {
 
 func (s *Server) handleListCheckpoints(w http.ResponseWriter, r *http.Request) {
 	ckpts, err := s.mgr.Checkpoints(r.Context(), tenantFrom(r.Context()))
-	if err != nil {
-		log.WithFunc("server.handleListCheckpoints").Error(r.Context(), err, "list checkpoints")
-		writeErr(w, http.StatusInternalServerError, "list checkpoints failed")
-		return
-	}
-	writeJSON(w, http.StatusOK, types.CheckpointListResponse{Checkpoints: ckpts})
+	writeResult(w, r, "list checkpoints", "", "list checkpoints failed", err, func() {
+		writeJSON(w, http.StatusOK, types.CheckpointListResponse{Checkpoints: ckpts})
+	})
 }
 
 func (s *Server) handleDeleteCheckpoint(w http.ResponseWriter, r *http.Request) {
@@ -706,14 +703,7 @@ func (s *Server) handlePutPools(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 	err := s.mgr.SetPools(r.Context(), req.Pools)
-	switch {
-	case writePoolErr(w, err):
-	case err != nil:
-		log.WithFunc("server.handlePutPools").Error(r.Context(), err, "set pools")
-		writeErr(w, http.StatusInternalServerError, "set pools failed")
-	default:
-		s.handleInfo(w, r)
-	}
+	writeResult(w, r, "set pools", "", "set pools failed", err, func() { s.handleInfo(w, r) })
 }
 
 func (s *Server) handleReload(w http.ResponseWriter, r *http.Request) {

sandboxd/store/dir/dir.go

@@ -1,6 +1,4 @@
-// Package dir is the directory record backend: plain files under a
-// root whose filesystem is the operator's choice — local disk keeps
-// records node-local, a shared FUSE mount makes them cluster-wide.
+// Package dir keeps records as plain files under a root: local disk keeps them node-local, a shared FUSE mount makes them cluster-wide.
 package dir
 
 import (
@@ -94,7 +92,7 @@ func (d *Store) Fetch(ctx context.Context, id string) (string, []byte, string, e
 
 func (d *Store) ReadMeta(_ context.Context, id string) ([]byte, error) {
 	raw, err := os.ReadFile(filepath.Join(d.root, id, store.MetaFile)) //nolint:gosec // id pinned by the instance idRe before any call
-	if os.IsNotExist(err) {
+	if errors.Is(err, fs.ErrNotExist) {
 		return nil, store.ErrNotFound
 	}
 	return raw, err
@@ -260,30 +258,20 @@ func (d *Store) publish(_ context.Context, staging, id, digest string) error {
 // sweepGenerations reclaims only entries whose install or supersession time is outside the grace.
 func (d *Store) sweepGenerations(id string) (err error) {
 	final := filepath.Join(d.root, id)
+	var keep []string
 	metaFile, err := os.Open(filepath.Join(final, store.MetaFile)) //nolint:gosec // id pinned by the instance idRe
-	if errors.Is(err, fs.ErrNotExist) {
-		entries, readErr := os.ReadDir(final)
-		if errors.Is(readErr, fs.ErrNotExist) {
-			return nil
-		}
+	switch {
+	case errors.Is(err, fs.ErrNotExist):
+	case err != nil:
+		return err
+	default:
+		defer func() { err = errors.Join(err, metaFile.Close()) }()
+		// the open file pins one inode across a concurrent meta rename.
+		meta, readErr := io.ReadAll(metaFile)
 		if readErr != nil {
 			return readErr
 		}
-		for _, e := range entries {
-			if removeErr := removeAgedEntry(final, e); removeErr != nil {
-				return removeErr
-			}
-		}
-		return nil
-	}
-	if err != nil {
-		return err
-	}
-	defer func() { err = errors.Join(err, metaFile.Close()) }()
-	// the open file pins one inode across a concurrent meta rename.
-	meta, err := io.ReadAll(metaFile)
-	if err != nil {
-		return err
+		keep = []string{store.MetaFile, store.LabelsFile, store.ExportGen(meta), digestName(meta)}
 	}
 	entries, err := os.ReadDir(final)
 	if errors.Is(err, fs.ErrNotExist) {
@@ -292,10 +280,8 @@ func (d *Store) sweepGenerations(id string) (err error) {
 	if err != nil {
 		return err
 	}
-	current := store.ExportGen(meta)
-	currentDigest := digestName(meta)
 	for _, e := range entries {
-		if name := e.Name(); name == store.MetaFile || name == store.LabelsFile || name == current || name == currentDigest {
+		if slices.Contains(keep, e.Name()) {
 			continue
 		}
 		if removeErr := removeAgedEntry(final, e); removeErr != nil {
@@ -378,15 +364,8 @@ func hashExport(ctx context.Context, root string) (string, error) {
 	if err := ctx.Err(); err != nil {
 		return "", err
 	}
-	totalChunks := 0
-	for i := range sources {
-		totalChunks += len(sources[i].digest.Chunks)
-	}
-	if totalChunks > 0 {
-		workers := min(totalChunks, runtime.GOMAXPROCS(0), digestWorkerLimit)
-		if err := hashChunks(ctx, sources, workers, hashChunk); err != nil {
-			return "", err
-		}
+	if err := hashChunks(ctx, sources, min(runtime.GOMAXPROCS(0), digestWorkerLimit), hashChunk); err != nil {
+		return "", err
 	}
 	files := make([]store.DigestFile, len(sources))
 	for i := range sources {

sandboxd/store/dir/dir_test.go

@@ -5,6 +5,7 @@ import (
 	"crypto/sha256"
 	"errors"
 	"fmt"
+	"io/fs"
 	"maps"
 	"os"
 	"path/filepath"
@@ -277,7 +278,7 @@ func TestSweepGenerationsPairsCurrentAndSupersededDigests(t *testing.T) {
 		t.Fatalf("SweepGenerations: %v", err)
 	}
 	for _, path := range []string{firstGen, firstSidecar} {
-		if _, statErr := os.Stat(path); !os.IsNotExist(statErr) {
+		if _, statErr := os.Stat(path); !errors.Is(statErr, fs.ErrNotExist) {
 			t.Errorf("superseded entry %s survived sweep: %v", filepath.Base(path), statErr)
 		}
 	}
@@ -322,7 +323,7 @@ func TestPublishRetriesAfterExpiredInstall(t *testing.T) {
 	if sweepErr := st.SweepGenerations(); sweepErr != nil {
 		t.Fatalf("sweep expired generation: %v", sweepErr)
 	}
-	if _, statErr := os.Stat(gen); !os.IsNotExist(statErr) {
+	if _, statErr := os.Stat(gen); !errors.Is(statErr, fs.ErrNotExist) {
 		t.Fatalf("expired generation survived sweep: %v", statErr)
 	}
 
@@ -357,7 +358,7 @@ func TestPublishSweepsGenerationsBySupersessionAge(t *testing.T) {
 	setAge(t, gen1, store.GenerationGrace+time.Minute)
 	setAge(t, gen2, store.GenerationGrace/2)
 	mustPublish(t, st, id, "fourth")
-	if _, statErr := os.Stat(gen1); !os.IsNotExist(statErr) {
+	if _, statErr := os.Stat(gen1); !errors.Is(statErr, fs.ErrNotExist) {
 		t.Fatalf("generation past its supersession grace survived publish: %v", statErr)
 	}
 	if _, statErr := os.Stat(gen2); statErr != nil {
@@ -461,7 +462,7 @@ func TestSweepReclaimsOnlyAgedUncommittedGenerations(t *testing.T) {
 	if sweepErr := sweeper.SweepStaging(); sweepErr != nil {
 		t.Fatalf("sweep old: %v", sweepErr)
 	}
-	if _, statErr := os.Stat(orphan); !os.IsNotExist(statErr) {
+	if _, statErr := os.Stat(orphan); !errors.Is(statErr, fs.ErrNotExist) {
 		t.Fatalf("aged uncommitted generation survived: %v", statErr)
 	}
 	if _, statErr := os.Stat(filepath.Join(root, orphanID)); statErr != nil {
@@ -599,14 +600,7 @@ func BenchmarkMetas(b *testing.B) {
 	}
 	for i := range 2000 {
 		id := fmt.Sprintf("ck_%016x", i)
-		staging, stageErr := st.Stage(id)
-		if stageErr != nil {
-			b.Fatalf("Stage: %v", stageErr)
-		}
-		seedRecord(b, staging, id)
-		if err = st.Publish(b.Context(), staging, id); err != nil {
-			b.Fatalf("Publish: %v", err)
-		}
+		mustPublish(b, st, id, id)
 	}
 	b.ReportAllocs()
 	for b.Loop() {
@@ -645,7 +639,7 @@ func installUncommitted(t *testing.T, st *Store, root, id, metaID string) (meta
 	return meta, gen
 }
 
-func mustPublish(t *testing.T, st *Store, id, metaID string) {
+func mustPublish(t testing.TB, st *Store, id, metaID string) {
 	t.Helper()
 	staging, err := st.Stage(id)
 	if err != nil {

sandboxd/tenants/pg/pg.go

@@ -41,9 +41,8 @@ const (
 var schema string
 
 type entry struct {
-	rec     *config.TenantRecord
-	sum     [sha256.Size]byte
-	present bool
+	rec *config.TenantRecord
+	sum [sha256.Size]byte
 }
 
 var _ tenants.Source = (*Source)(nil)
@@ -99,7 +98,7 @@ func (s *Source) Resolve(ctx context.Context, sum [sha256.Size]byte) (*config.Te
 	start := s.seq
 	s.mu.RUnlock()
 	switch {
-	case e.present && e.sum == sum:
+	case e.rec != nil && e.sum == sum:
 		return e.rec, nil
 	case negative && time.Now().Before(until):
 		return nil, tenants.ErrUnknown
@@ -137,7 +136,7 @@ func (s *Source) Peek(name string) (config.TenantRecord, tenants.Presence) {
 	case !ok:
 		s.fetch(name)
 		return config.TenantRecord{}, tenants.Unknown
-	case e.present:
+	case e.rec != nil:
 		return *e.rec, tenants.Present
 	default:
 		return config.TenantRecord{}, tenants.Absent
@@ -150,7 +149,7 @@ func (s *Source) Lookup(ctx context.Context, name string) (config.TenantRecord,
 	start := s.seq
 	s.mu.RUnlock()
 	if cached {
-		if e.present {
+		if e.rec != nil {
 			return *e.rec, true, nil
 		}
 		return config.TenantRecord{}, false, nil
@@ -358,7 +357,7 @@ func (s *Source) cache(r *config.TenantRecord, sum [sha256.Size]byte) {
 	if len(s.byName) >= maxCached {
 		s.clearCache()
 	}
-	s.byName[r.Name] = entry{rec: r, sum: sum, present: true}
+	s.byName[r.Name] = entry{rec: r, sum: sum}
 	s.byToken[sum] = r.Name
 }
 
@@ -372,7 +371,7 @@ func (s *Source) evict(key string) {
 		s.floor = s.seq
 		return
 	}
-	if e := s.byName[key]; e.present {
+	if e := s.byName[key]; e.rec != nil {
 		delete(s.byToken, e.sum)
 	}
 	delete(s.byName, key)

sandboxd/tenants/pg/pg_test.go

@@ -93,7 +93,7 @@ func TestARoleWithoutCreateServesAPreCreatedTable(t *testing.T) {
 	if err != nil {
 		t.Fatalf("connect: %v", err)
 	}
-	defer func() { _ = admin.Close(context.Background()) }()
+	defer func() { _ = admin.Close(t.Context()) }()
 	for _, sql := range []string{
 		"CREATE ROLE " + role + " LOGIN PASSWORD 'p'",
 		"GRANT USAGE ON SCHEMA " + schema + " TO " + role,