diff --git a/docs/deploy.md b/docs/deploy.md index 3356eb33..c31f1496 100644 --- a/docs/deploy.md +++ b/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`. diff --git a/sandboxd/config/config.go b/sandboxd/config/config.go index 538ea5ac..d5ed3bed 100644 --- a/sandboxd/config/config.go +++ b/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))} } diff --git a/sandboxd/metastore/metastore.go b/sandboxd/metastore/metastore.go index 0be1f1db..6697a883 100644 --- a/sandboxd/metastore/metastore.go +++ b/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() diff --git a/sandboxd/pool/pool.go b/sandboxd/pool/pool.go index b79e9e3c..ae36997a 100644 --- a/sandboxd/pool/pool.go +++ b/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 diff --git a/sandboxd/poolset/pg/pg_test.go b/sandboxd/poolset/pg/pg_test.go index 412c2564..baa61e28 100644 --- a/sandboxd/poolset/pg/pg_test.go +++ b/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 { diff --git a/sandboxd/server/server.go b/sandboxd/server/server.go index cd609a50..6b923dde 100644 --- a/sandboxd/server/server.go +++ b/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) { diff --git a/sandboxd/store/dir/dir.go b/sandboxd/store/dir/dir.go index c4da653b..9e389146 100644 --- a/sandboxd/store/dir/dir.go +++ b/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 { diff --git a/sandboxd/store/dir/dir_test.go b/sandboxd/store/dir/dir_test.go index 7638058d..16aa9680 100644 --- a/sandboxd/store/dir/dir_test.go +++ b/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 { diff --git a/sandboxd/tenants/pg/pg.go b/sandboxd/tenants/pg/pg.go index c42022bc..613bf288 100644 --- a/sandboxd/tenants/pg/pg.go +++ b/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) diff --git a/sandboxd/tenants/pg/pg_test.go b/sandboxd/tenants/pg/pg_test.go index 3bff4610..28ab7bcf 100644 --- a/sandboxd/tenants/pg/pg_test.go +++ b/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,