From 3a77e8c5b1c296ca86971f69e8d51b3d052132a4 Mon Sep 17 00:00:00 2001 From: CMGS Date: Mon, 13 Jul 2026 12:48:33 +0800 Subject: [PATCH 1/3] feat(config): persist API-applied pools and detect cluster config drift pools.json makes PUT /v1/pools survive restart (rebuild from operator intent, not the config seed); the mesh gossips a cluster-invariant config digest that warns and raises a gauge on divergence; SetPoolsCluster fans a set across peers; the gossip epoch persists so a backwards clock cannot regress placement. Durable writes use a unique-temp fsync helper with monotonic per-writer guards so concurrent persists can neither lose an update nor race a shared temp into a failed rename. --- docs/cluster.md | 57 ++++++++++++- docs/deploy.md | 2 +- sandboxd/config/config.go | 38 +++++++++ sandboxd/config/config_test.go | 28 +++++++ sandboxd/main.go | 9 +- sandboxd/mesh/epoch.go | 28 +++++++ sandboxd/mesh/mesh.go | 107 +++++++++++++++++++++--- sandboxd/mesh/mesh_test.go | 2 +- sandboxd/mesh/state_test.go | 90 ++++++++++++++++++++ sandboxd/pool/pool.go | 19 +++++ sandboxd/pool/poolstore.go | 143 ++++++++++++++++++++++++++++++++ sandboxd/pool/poolstore_test.go | 101 ++++++++++++++++++++++ sandboxd/pool/setpools.go | 9 +- sandboxd/server/metrics.go | 5 ++ sandboxd/server/server.go | 1 + sandboxd/server/server_test.go | 1 + sandboxd/utils/utils.go | 51 ++++++++++++ sdk/go/info.go | 17 ++-- sdk/go/pools.go | 85 +++++++++++++++++++ sdk/go/pools_test.go | 111 +++++++++++++++++++++++++ 20 files changed, 881 insertions(+), 23 deletions(-) create mode 100644 sandboxd/mesh/epoch.go create mode 100644 sandboxd/mesh/state_test.go create mode 100644 sandboxd/pool/poolstore.go create mode 100644 sandboxd/pool/poolstore_test.go create mode 100644 sdk/go/pools.go create mode 100644 sdk/go/pools_test.go diff --git a/docs/cluster.md b/docs/cluster.md index db0a56a..0e9926e 100644 --- a/docs/cluster.md +++ b/docs/cluster.md @@ -123,10 +123,65 @@ handle `Sandbox.Promote` returns is still owner-bound — `template.New` and `template.Delete` dial the owner directly, no gossip involved — and is the race-free choice immediately after a promote. +## State ownership + +Each kind of state has exactly one source of truth, and a restart rebuilds +fully from it: + +| state | source of truth | survives restart | +|---|---|---| +| operator config (`tenants`, `secrets`, egress policies, `bridge`/`network`, `mesh`, `preview_secret`, `egress_ca`) | `config.json` (human/deploy-tool owned) | re-read at boot | +| API-applied pool targets (`PUT /v1/pools`) | `/pools.json` (machine owned) | yes | +| claims | the claims journal + `Reconcile` | yes | +| placement hints (warm counts, template sets) | gossip | rebuilt | + +Pools are managed API-first. The first time a node takes `PUT /v1/pools`, it +writes the applied set to `/pools.json` and from then on **that file +seeds the pools at boot**, overriding the `pools` section of `config.json` (a +loud log line notes this; if the two disagree because someone edited +`config.json` afterward, a second warning names it). The file's presence is the +ownership marker — delete `pools.json` to return the node to config-owned pools. +Egress stays config-owned regardless: the API rejects egress specs, so +`pools.json` never carries them, and egress policies re-merge from `config.json` +by pool key at boot. + +Cluster-wide pool changes are a client-side fan-out, not a gossiped desired +state (pools are legitimately heterogeneous per node): `Client.SetPoolsCluster` +PUTs the set to the entry node and every peer and returns a per-node result; +retrying the failed nodes is the whole protocol, because the apply is an +idempotent declarative replace. A non-nil error means peer discovery itself +failed and the fan-out reached only the entry node — an incomplete apply to +retry, kept distinct from a genuine single-node cluster (nil error). It fits a +homogeneous cluster — a spec set that names an egress-lane pool is refused on a +node without an attachment, so nodes that differ in capacity or attachment take +a per-node `Client.SetPools`. + +### Cluster-invariant config + +Some config must match on every node or the cluster fails in confusing ways: + +| config | what breaks on mismatch | +|---|---| +| `api_token`, `tenants` | the SDK replays the authorizing token across a redirect; a peer missing that tenant/token answers 401 | +| `preview_secret` | a preview URL signed on one node fails verification on another | +| `mesh.cluster_key` | nodes cannot join / decrypt gossip at all | +| `egress_ca` cluster root | a guest checkpointed/redirected across nodes trusts the root; a divergent root fails interception | + +Each node gossips a digest of these (HMAC-keyed by `cluster_key` when set; +otherwise a token-free digest of tenant names + the CA root, so nothing +brute-forceable rides cleartext gossip). A mismatch logs a warning at the moment +the divergent node appears — not at the first unlucky redirect — and raises the +`sandboxd_config_digest_mismatch` gauge. It is warn-only: a rolling credential +rotation is a legitimate transient mismatch, so a divergence never partitions +the mesh. + ## Cluster checklist - memberlist port (e.g. 7946) open node-to-node, TCP **and** UDP - `advertise_addr` set to a routable address on every node (never loopback, never a wildcard) -- same `api_token` and `tenants` set everywhere +- same `api_token`, `tenants`, `preview_secret`, and `egress_ca` root everywhere + (a mismatch warns and shows in `sandboxd_config_digest_mismatch`) - `cluster_key` set if the gossip network is not otherwise trusted +- pool changes via `Client.SetPoolsCluster` (or per-node `SetPools`); the applied + set persists to `pools.json` and survives restart diff --git a/docs/deploy.md b/docs/deploy.md index 54a6b93..05ac088 100644 --- a/docs/deploy.md +++ b/docs/deploy.md @@ -66,7 +66,7 @@ sandboxd reads one JSON file (`-config`, default | `archive_after_seconds` | 0 (off) | tier below hibernation: a hibernated claim idle this long is checkpointed to the store and its local VM dropped, freeing the node entirely; the next call restores it transparently (a checkpoint restore's latency). Requires `idle_hibernate_seconds > 0` and must exceed it. Node-wide for unpooled keys; per-pool overrides for that pool | | `archive_delete_after_seconds` | 0 (keep) | purge an archived claim's store checkpoint this long after it was archived, reclaiming storage; the claim is then gone for good. Same node-wide/per-pool split | | `mesh` | unset | join a cluster ([Clusters](cluster.md)); unset = single node | -| `pools[]` | — | warm pools. `warm` defaults to 4; `net` is `none` or `egress`; `size` is a tier, below. Retune online without a restart via [`PUT /v1/pools`](sandboxd-api.md#put-v1pools) — omitted pools drain | +| `pools[]` | — | warm pools. `warm` defaults to 4; `net` is `none` or `egress`; `size` is a tier, below. Retune online without a restart via [`PUT /v1/pools`](sandboxd-api.md#put-v1pools) — omitted pools drain. This is the **first-boot seed**: once a node takes a `PUT /v1/pools`, the applied set persists to `/pools.json` and overrides this section on every later boot (a startup log notes it); delete `pools.json` to return to config-owned pools. Egress stays config-owned either way. See [state ownership](cluster.md#state-ownership) | Size tiers (free-form CPU/memory is deliberately not accepted — it would fragment the warm pools): diff --git a/sandboxd/config/config.go b/sandboxd/config/config.go index 507765f..e072d6d 100644 --- a/sandboxd/config/config.go +++ b/sandboxd/config/config.go @@ -2,9 +2,15 @@ package config import ( + "crypto/hmac" + "crypto/sha256" + "encoding/base64" + "encoding/hex" + "encoding/json" "fmt" "os" "slices" + "strings" "github.com/cocoonstack/sandbox/sandboxd/egress" "github.com/cocoonstack/sandbox/sandboxd/store/s3" @@ -204,6 +210,38 @@ func (c *Config) HasEgress() bool { return c.Bridge != "" || c.Network != "" } +// ClusterDigest fingerprints the config every mesh node must share, so a +// divergent node is caught at join instead of at a later 401 or failed +// cross-node interception: api_token, the tenant name→token set, preview_secret, +// and the egress cluster root. With a cluster_key set (gossip is encrypted) it +// HMACs the full set including token material; without one gossip may be +// cleartext, so it covers only non-secret identity (tenant names + CA root), +// keeping a weak token off the wire. +func (c *Config) ClusterDigest(caFingerprint string) string { + names := make([]string, len(c.Tenants)) + for i, t := range c.Tenants { + names[i] = t.Name + } + slices.Sort(names) + if c.Mesh != nil && c.Mesh.ClusterKey != "" { + if key, err := base64.StdEncoding.DecodeString(c.Mesh.ClusterKey); err == nil { + type auth struct{ Name, Token string } + tenants := make([]auth, len(c.Tenants)) + for i, t := range c.Tenants { + tenants[i] = auth{t.Name, t.Token} + } + slices.SortFunc(tenants, func(a, b auth) int { return strings.Compare(a.Name, b.Name) }) + raw, _ := json.Marshal([]any{c.APIToken, c.PreviewSecret, caFingerprint, tenants}) + mac := hmac.New(sha256.New, key) + mac.Write(raw) + return hex.EncodeToString(mac.Sum(nil)) + } + } + raw, _ := json.Marshal([]any{caFingerprint, names}) + sum := sha256.Sum256(raw) + return hex.EncodeToString(sum[:]) +} + // guardsEgressLane: any tenant policy counts (claims mint egress keys without a // pool), a none-lane pool policy does not (it locks no tap). func (c *Config) guardsEgressLane() bool { diff --git a/sandboxd/config/config_test.go b/sandboxd/config/config_test.go index c6833b0..e775f48 100644 --- a/sandboxd/config/config_test.go +++ b/sandboxd/config/config_test.go @@ -1,6 +1,7 @@ package config import ( + "encoding/base64" "os" "path/filepath" "strings" @@ -9,6 +10,33 @@ import ( "github.com/cocoonstack/sandbox/sandboxd/types" ) +func TestClusterDigest(t *testing.T) { + base := &Config{APIToken: "tok", PreviewSecret: "ps", Tenants: []TenantSpec{{Name: "acme", Token: "t1"}}} + d := base.ClusterDigest("ca-fp") + if base.ClusterDigest("ca-fp") != d { + t.Fatal("digest is not stable for identical config") + } + if (&Config{APIToken: "tok", PreviewSecret: "ps", Tenants: []TenantSpec{{Name: "beta", Token: "t1"}}}).ClusterDigest("ca-fp") == d { + t.Error("a tenant-name change is not reflected") + } + if base.ClusterDigest("other-fp") == d { + t.Error("an egress CA root change is not reflected") + } + // Without a cluster_key gossip may be cleartext, so token material must be + // excluded: an api_token change alone must NOT alter the on-wire digest. + if (&Config{APIToken: "other", PreviewSecret: "ps", Tenants: base.Tenants}).ClusterDigest("ca-fp") != d { + t.Error("api_token leaked into the keyless digest") + } + // With a cluster_key the digest is an HMAC over token material, so an + // api_token change IS covered. + key := base64.StdEncoding.EncodeToString(make([]byte, 32)) + keyed := &Config{APIToken: "tok", PreviewSecret: "ps", Tenants: base.Tenants, Mesh: &MeshConfig{ClusterKey: key}} + keyedDiff := &Config{APIToken: "other", PreviewSecret: "ps", Tenants: base.Tenants, Mesh: &MeshConfig{ClusterKey: key}} + if keyed.ClusterDigest("ca-fp") == keyedDiff.ClusterDigest("ca-fp") { + t.Error("with a cluster_key the api_token must be covered by the digest") + } +} + func TestLoadAppliesDefaults(t *testing.T) { path := writeConfig(t, `{"pools":[{"template":"rt:24.04","net":"none","size":"small"}]}`) cfg, err := Load(path) diff --git a/sandboxd/main.go b/sandboxd/main.go index 91614e3..2b765d2 100644 --- a/sandboxd/main.go +++ b/sandboxd/main.go @@ -82,7 +82,7 @@ func main() { var placer server.Placer if cfg.Mesh != nil { - msh, err := startMesh(cfg) + msh, err := startMesh(cfg, mgr) if err != nil { logger.Fatalf(ctx, err, "start mesh") } @@ -141,7 +141,7 @@ func main() { logger.Info(ctx, "sandboxd stopped; VMs stay alive for the next reconcile") } -func startMesh(cfg *config.Config) (*mesh.Mesh, error) { +func startMesh(cfg *config.Config, mgr *pool.Manager) (*mesh.Mesh, error) { mc := cfg.Mesh mlCfg := memberlist.DefaultLANConfig() host, portStr, err := net.SplitHostPort(mc.Bind) @@ -167,10 +167,13 @@ func startMesh(cfg *config.Config) (*mesh.Mesh, error) { return nil, fmt.Errorf("mesh cluster key: %w", err) } } - msh, err := mesh.New(mlCfg, nodeID, cfg.AdvertiseAddr, key) + msh, err := mesh.New(mlCfg, nodeID, cfg.AdvertiseAddr, key, cfg.DataDir) if err != nil { return nil, err } + // Publish the cluster-invariant config digest before Join, so the first + // gossip a peer sees already carries it. + msh.SetSelfDigest(cfg.ClusterDigest(mgr.EgressCAFingerprint())) if err := msh.Join(mc.Join); err != nil { _ = msh.Shutdown() return nil, err diff --git a/sandboxd/mesh/epoch.go b/sandboxd/mesh/epoch.go new file mode 100644 index 0000000..3d09519 --- /dev/null +++ b/sandboxd/mesh/epoch.go @@ -0,0 +1,28 @@ +package mesh + +import ( + "os" + "strconv" + "strings" + + "github.com/cocoonstack/sandbox/sandboxd/utils" +) + +// loadEpoch reads the persisted gossip epoch, or 0 when the file is absent or +// unreadable — the wall-clock seed then wins, which is the pre-persistence +// behavior. +func loadEpoch(path string) uint64 { + raw, err := os.ReadFile(path) //nolint:gosec // node-local data-dir path + if err != nil { + return 0 + } + n, err := strconv.ParseUint(strings.TrimSpace(string(raw)), 10, 64) + if err != nil { + return 0 + } + return n +} + +func storeEpoch(path string, epoch uint64) error { + return utils.WriteFileSync(path, []byte(strconv.FormatUint(epoch, 10)), 0o600) +} diff --git a/sandboxd/mesh/mesh.go b/sandboxd/mesh/mesh.go index 473cacd..709f408 100644 --- a/sandboxd/mesh/mesh.go +++ b/sandboxd/mesh/mesh.go @@ -7,15 +7,18 @@ package mesh import ( + "context" "encoding/json" "fmt" "maps" "math/rand/v2" + "path/filepath" "slices" "sync" "time" "github.com/hashicorp/memberlist" + "github.com/projecteru2/core/log" ) // leaveTimeout bounds the graceful-leave broadcast on shutdown so a wedged @@ -30,33 +33,44 @@ type NodeState struct { Epoch uint64 `json:"epoch"` Pools map[string]int `json:"pools"` // PoolKey hash → warm count Templates []string `json:"templates,omitempty"` // promoted-template key hashes on disk + Digest string `json:"digest,omitempty"` // cluster-invariant config digest } // Mesh is the node's view of the cluster and its own gossiped state. type Mesh struct { - ml *memberlist.Memberlist + ml *memberlist.Memberlist + epochPath string mu sync.Mutex self NodeState view map[string]NodeState // node_id → latest known state (includes self) + + epochMu sync.Mutex + epochWritten uint64 // highest epoch on disk; guarded by epochMu } // New starts a mesh member listening per cfg. selfAddr is the data-plane // address peers should dial for a redirect; secretKey (16/24/32 bytes) enables -// gossip encryption when non-empty. -func New(cfg *memberlist.Config, nodeID, selfAddr string, secretKey []byte) (*Mesh, error) { +// gossip encryption when non-empty; dataDir holds the persisted epoch. +func New(cfg *memberlist.Config, nodeID, selfAddr string, secretKey []byte, dataDir string) (*Mesh, error) { + epochPath := filepath.Join(dataDir, "mesh-epoch") + // Seed from the persisted epoch, floored at wall-clock: a restart under a + // backwards clock would otherwise regress below peers' last-seen epoch, so + // its fresh state is silently rejected. Persist the seed at once. + epoch := max(loadEpoch(epochPath), uint64(time.Now().UnixNano())) //nolint:gosec // UnixNano is positive for current times m := &Mesh{ + epochPath: epochPath, self: NodeState{ NodeID: nodeID, Addr: selfAddr, - // Seed the epoch from wall-clock so a restarted node's fresh counts - // aren't rejected by peers still holding its pre-restart (higher) - // epoch; Epoch++ keeps intra-process monotonicity above that base. - Epoch: uint64(time.Now().UnixNano()), //nolint:gosec // UnixNano is positive for current times - Pools: map[string]int{}, + Epoch: epoch, + Pools: map[string]int{}, }, view: map[string]NodeState{}, } + if err := m.persistEpoch(epoch); err != nil { + return nil, fmt.Errorf("persist mesh epoch: %w", err) + } m.view[nodeID] = m.self cfg.Name = nodeID @@ -90,14 +104,48 @@ func (m *Mesh) Join(seeds []string) error { // every second for nothing. func (m *Mesh) UpdateSelf(pools map[string]int, templates []string) { m.mu.Lock() - defer m.mu.Unlock() if maps.Equal(m.self.Pools, pools) && slices.Equal(m.self.Templates, templates) { + m.mu.Unlock() return } m.self.Epoch++ m.self.Pools = pools m.self.Templates = templates m.view[m.self.NodeID] = m.self + epoch := m.self.Epoch + m.mu.Unlock() + // Persist off the manager lock (redirect placement reads it): the bump is + // the new floor a restart must not regress below. + if err := m.persistEpoch(epoch); err != nil { + log.WithFunc("mesh.UpdateSelf").Warnf(context.Background(), "persist epoch: %v", err) + } +} + +// SetSelfDigest records this node's cluster-invariant config digest so peers can +// detect divergence; call it once before Join, before any gossip ships. +func (m *Mesh) SetSelfDigest(digest string) { + m.mu.Lock() + defer m.mu.Unlock() + m.self.Digest = digest + m.view[m.self.NodeID] = m.self +} + +// ConfigMismatches counts peers whose config digest differs from this node's — +// a gauge for alerting on a divergent cluster-invariant config (the intermittent +// 401 / cross-node interception class), recomputed per read from the view. +func (m *Mesh) ConfigMismatches() int { + m.mu.Lock() + defer m.mu.Unlock() + if m.self.Digest == "" { + return 0 + } + n := 0 + for id, st := range m.view { + if id != m.self.NodeID && st.Digest != "" && st.Digest != m.self.Digest { + n++ + } + } + return n } // Candidates returns up to two peer addresses that report warm(keyHash) > 0, @@ -188,6 +236,22 @@ func (m *Mesh) Shutdown() error { return m.ml.Shutdown() } +// persistEpoch durably records the gossip epoch, serialized so a lower value +// from a slower concurrent UpdateSelf can never overwrite a higher one already +// on disk — the persisted floor must only ever advance. +func (m *Mesh) persistEpoch(epoch uint64) error { + m.epochMu.Lock() + defer m.epochMu.Unlock() + if epoch <= m.epochWritten { + return nil + } + if err := storeEpoch(m.epochPath, epoch); err != nil { + return err + } + m.epochWritten = epoch + return nil +} + // forget drops a departed node from the placement view so redirects stop // targeting a dead peer; SWIM detected the death, the view must follow. func (m *Mesh) forget(nodeID string) { @@ -207,10 +271,31 @@ func (m *Mesh) merge(states []NodeState) { if st.NodeID == m.self.NodeID { continue } - if cur, ok := m.view[st.NodeID]; !ok || st.Epoch > cur.Epoch { - m.view[st.NodeID] = st + cur, ok := m.view[st.NodeID] + if ok && st.Epoch <= cur.Epoch { + continue + } + // Warn when a peer's config digest diverges from this node's, on each + // distinct new value (not once per lifetime): a mismatched + // cluster-invariant config (api_token/tenants/preview_secret/egress CA + // root) makes cross-node redirects 401 and interception fail — the + // worst failure mode to debug at first unlucky redirect. Warn-only: + // refusing the merge would turn a routine rolling credential rotation + // into a partition. + if m.self.Digest != "" && st.Digest != "" && st.Digest != m.self.Digest && (!ok || cur.Digest != st.Digest) { + log.WithFunc("mesh.merge").Warnf(context.Background(), + "peer %s config digest %s differs from this node's %s: cluster-invariant config diverges (redirects may 401, interception may fail)", + st.NodeID, short(st.Digest), short(m.self.Digest)) } + m.view[st.NodeID] = st + } +} + +func short(digest string) string { + if len(digest) > 12 { + return digest[:12] } + return digest } var _ memberlist.Delegate = (*delegate)(nil) diff --git a/sandboxd/mesh/mesh_test.go b/sandboxd/mesh/mesh_test.go index 3f18de6..6bae73e 100644 --- a/sandboxd/mesh/mesh_test.go +++ b/sandboxd/mesh/mesh_test.go @@ -160,7 +160,7 @@ func startNode(t *testing.T, host string, port int, id string) *node { cfg.AdvertiseAddr = host cfg.PushPullInterval = 200 * time.Millisecond cfg.Logger = discardLogger() - m, err := New(cfg, id, id+":7777", nil) + m, err := New(cfg, id, id+":7777", nil, t.TempDir()) if err != nil { t.Fatalf("new mesh %s: %v", id, err) } diff --git a/sandboxd/mesh/state_test.go b/sandboxd/mesh/state_test.go new file mode 100644 index 0000000..6f79ce9 --- /dev/null +++ b/sandboxd/mesh/state_test.go @@ -0,0 +1,90 @@ +package mesh + +import ( + "path/filepath" + "sync" + "testing" + + "github.com/hashicorp/memberlist" +) + +func newBoundMesh(t *testing.T, dataDir string) *Mesh { + t.Helper() + cfg := memberlist.DefaultLocalConfig() + cfg.BindPort = 0 + cfg.Logger = discardLogger() + m, err := New(cfg, "self", "self:7777", nil, dataDir) + if err != nil { + t.Fatalf("new mesh: %v", err) + } + t.Cleanup(func() { _ = m.Shutdown() }) + return m +} + +func TestEpochRoundTrip(t *testing.T) { + p := filepath.Join(t.TempDir(), "e") + if loadEpoch(p) != 0 { + t.Error("missing epoch file must read 0") + } + if err := storeEpoch(p, 42); err != nil { + t.Fatalf("store: %v", err) + } + if got := loadEpoch(p); got != 42 { + t.Errorf("loadEpoch = %d, want 42", got) + } +} + +func TestEpochSeededFromPersistedFloor(t *testing.T) { + dir := t.TempDir() + huge := uint64(1) << 62 // above any plausible wall-clock nanos + if err := storeEpoch(filepath.Join(dir, "mesh-epoch"), huge); err != nil { + t.Fatalf("store: %v", err) + } + m := newBoundMesh(t, dir) + if m.self.Epoch != huge { + t.Errorf("epoch seeded %d, want persisted floor %d (a backwards clock must not regress it)", m.self.Epoch, huge) + } +} + +func TestUpdateSelfPersistsEpoch(t *testing.T) { + dir := t.TempDir() + m := newBoundMesh(t, dir) + before := m.self.Epoch + m.UpdateSelf(map[string]int{"k": 1}, nil) + if got := loadEpoch(filepath.Join(dir, "mesh-epoch")); got <= before { + t.Errorf("persisted epoch %d did not advance past the seed %d", got, before) + } +} + +func TestPersistEpochMonotonic(t *testing.T) { + dir := t.TempDir() + m := newBoundMesh(t, dir) + base := m.self.Epoch // New already seeded a wall-clock floor + high := base + 1000 + if err := m.persistEpoch(high); err != nil { + t.Fatalf("persist high: %v", err) + } + // Concurrent lower values must never regress the on-disk floor below high. + var wg sync.WaitGroup + for e := base + 1; e < high; e++ { + wg.Go(func() { + if err := m.persistEpoch(e); err != nil { + t.Errorf("persist %d: %v", e, err) + } + }) + } + wg.Wait() + if got := loadEpoch(filepath.Join(dir, "mesh-epoch")); got != high { + t.Errorf("persisted epoch %d, want the %d floor held against lower concurrent writes", got, high) + } +} + +func TestConfigDigestMismatch(t *testing.T) { + m := newTestMesh(t, "self") + m.SetSelfDigest("self-digest") + m.merge([]NodeState{{NodeID: "peerA", Addr: "a:1", Epoch: 1, Digest: "self-digest"}}) + m.merge([]NodeState{{NodeID: "peerB", Addr: "b:1", Epoch: 1, Digest: "other-digest"}}) + if n := m.ConfigMismatches(); n != 1 { + t.Errorf("ConfigMismatches = %d, want 1 (only peerB diverges)", n) + } +} diff --git a/sandboxd/pool/pool.go b/sandboxd/pool/pool.go index 0596fb7..2aea752 100644 --- a/sandboxd/pool/pool.go +++ b/sandboxd/pool/pool.go @@ -155,6 +155,11 @@ type Manager struct { maxFork int store *claimStore + // poolStore persists the last API-applied pool set; configSeedHash is the + // config.json pools' hash, to warn when a file edit is overridden by it. + poolStore *poolStore + configSeedHash string + // idleDefault is the idle-hibernate threshold for unpooled keys; pooled // keys carry theirs on the pool struct. Zero means disabled. idleDefault time.Duration @@ -251,6 +256,7 @@ func NewManager(ctx context.Context, cfg *config.Config, eng Engine, secrets *eg lockEgress: cfg.Bridge != "", maxFork: maxFork, store: newClaimStore(cfg.DataDir), + poolStore: newPoolStore(cfg.DataDir), pools: make(map[types.PoolKey]*pool, len(cfg.Pools)), claimed: map[string]*types.Sandbox{}, tenantLive: map[string]int{}, @@ -336,12 +342,25 @@ func NewManager(ctx context.Context, cfg *config.Config, eng Engine, secrets *eg } m.egressCA = ca } + m.configSeedHash = poolSeedHash(cfg.Pools) + if err := m.adoptPersistedPools(ctx); err != nil { + return nil, err + } if m.archiveEnabled && m.ckptTTL == 0 { log.WithFunc("pool.NewManager").Warn(ctx, "archive enabled with checkpoint_ttl_hours=0: a checkpoint whose delete fails is not reclaimed") } return m, nil } +// EgressCAFingerprint is the egress cluster root's fingerprint, or "" when this +// node does no interception; it feeds the cluster-invariant config digest. +func (m *Manager) EgressCAFingerprint() string { + if m.egressCA == nil { + return "" + } + return m.egressCA.Fingerprint() +} + func loadEgressCA(cfg *config.EgressCAConfig) (*egress.CA, error) { root, err := os.ReadFile(cfg.RootCert) //nolint:gosec // operator-configured ca path if err != nil { diff --git a/sandboxd/pool/poolstore.go b/sandboxd/pool/poolstore.go new file mode 100644 index 0000000..1e6fbcc --- /dev/null +++ b/sandboxd/pool/poolstore.go @@ -0,0 +1,143 @@ +package pool + +import ( + "context" + "crypto/sha256" + "encoding/hex" + "encoding/json" + "errors" + "fmt" + "io/fs" + "os" + "path/filepath" + "slices" + "strings" + "sync" + "sync/atomic" + + "github.com/projecteru2/core/log" + + "github.com/cocoonstack/sandbox/sandboxd/config" + "github.com/cocoonstack/sandbox/sandboxd/utils" +) + +// poolsFile records the last API-applied pool set (PUT /v1/pools) so a restart +// rebuilds from operator intent, not the config.json seed that only grows +// staler under API-first management. ConfigSeed is the config.json pools' hash +// at write time; a boot that finds it changed warns that the file edit is +// overridden. Egress is never here — SetPools rejects it, so it stays +// config-owned and is re-merged by key at load. +type poolsFile struct { + ConfigSeed string `json:"config_seed"` + Pools []config.PoolSpec `json:"pools"` +} + +type poolStore struct { + path string + + seq atomic.Uint64 // sequence source; bumped under the manager mutex (apply order) + mu sync.Mutex + written uint64 // highest sequence on disk; guarded by mu +} + +func newPoolStore(dataDir string) *poolStore { + return &poolStore{path: filepath.Join(dataDir, "pools.json")} +} + +// load returns the persisted pool file, or nil when the node has never taken a +// PUT /v1/pools (config.json seeds the pools, unchanged behavior). +func (s *poolStore) load() (*poolsFile, error) { + raw, err := os.ReadFile(s.path) + if errors.Is(err, fs.ErrNotExist) { + return nil, nil + } + if err != nil { + return nil, fmt.Errorf("read pools file: %w", err) + } + var pf poolsFile + if err := json.Unmarshal(raw, &pf); err != nil { + return nil, fmt.Errorf("parse pools file: %w", err) + } + return &pf, nil +} + +// commit durably writes the applied set, serialized by s.mu; a seq no newer +// than the last written is a no-op, so two concurrent SetPools calls can never +// let an already-superseded set win the on-disk write and revert the latest. +func (s *poolStore) commit(seq uint64, pf poolsFile) error { + s.mu.Lock() + defer s.mu.Unlock() + if seq <= s.written { + return nil + } + raw, err := json.Marshal(pf) + if err != nil { + return fmt.Errorf("encode pools file: %w", err) + } + if err := utils.WriteFileSync(s.path, raw, 0o600); err != nil { + return err + } + s.written = seq + return nil +} + +// adoptPersistedPools replaces the config-seeded pools with the last API-applied +// set (pools.json) when present, so a restart rebuilds from operator intent. +// Egress stays config-owned (kept in m.poolEgress, re-merged by key); the egress +// CA must already be loaded so golden adoption can check its fingerprint. A +// restored pool that no longer validates (e.g. an egress lane with the bridge +// now removed) fails the boot, matching config-load discipline. +func (m *Manager) adoptPersistedPools(ctx context.Context) error { + pf, err := m.poolStore.load() + if err != nil { + return err + } + if pf == nil { + return nil + } + logger := log.WithFunc("pool.adoptPersistedPools") + if pf.ConfigSeed != m.configSeedHash { + logger.Warnf(ctx, "config.json pools differ from the API-applied set and are overridden; delete %s to return to config-owned pools", m.poolStore.path) + } + clear(m.pools) + m.idleEnabled = m.idleDefault > 0 + m.archiveEnabled = m.archiveAfterDefault > 0 + for _, spec := range pf.Pools { + spec = normalizePoolSpec(spec) + if err := m.validate(spec.PoolKey); err != nil { + return fmt.Errorf("restore pool %q: %w", spec.Template, err) + } + if err := spec.ValidateLimits(); err != nil { + return fmt.Errorf("restore pool %q: %w", spec.Template, err) + } + p := &pool{key: spec.PoolKey} + p.applySpec(spec) + if g := filepath.Join(m.goldensDir(), spec.Hash()); dirExists(g) && m.goldenCAMatches(g, m.poolEgress[spec.PoolKey].Intercepts()) { + p.goldenDir = g + } + m.pools[spec.PoolKey] = p + if spec.IdleHibernateSeconds > 0 { + m.idleEnabled = true + } + if spec.ArchiveAfterSeconds > 0 { + m.archiveEnabled = true + } + } + logger.Infof(ctx, "restored %d API-applied pools from pools.json", len(pf.Pools)) + return nil +} + +// poolSeedHash digests a config pool set's warm-target shape, order-independent; +// egress is excluded (config-owned, re-merged separately) so the hash tracks +// exactly the targets pools.json would override. +func poolSeedHash(specs []config.PoolSpec) string { + shaped := make([]config.PoolSpec, len(specs)) + copy(shaped, specs) + for i := range shaped { + shaped[i].Egress = nil + } + slices.SortFunc(shaped, func(a, b config.PoolSpec) int { return strings.Compare(a.Hash(), b.Hash()) }) + raw, _ := json.Marshal(shaped) + sum := sha256.Sum256(raw) + return hex.EncodeToString(sum[:]) +} diff --git a/sandboxd/pool/poolstore_test.go b/sandboxd/pool/poolstore_test.go new file mode 100644 index 0000000..3c32e8b --- /dev/null +++ b/sandboxd/pool/poolstore_test.go @@ -0,0 +1,101 @@ +package pool + +import ( + "encoding/json" + "os" + "path/filepath" + "sync" + "testing" + + "github.com/cocoonstack/sandbox/sandboxd/config" + "github.com/cocoonstack/sandbox/sandboxd/types" +) + +var ( + seedKey = types.PoolKey{Template: "seed:24.04", Net: types.NetNone, Size: types.SizeSmall} + apiKey = types.PoolKey{Template: "api:24.04", Net: types.NetNone, Size: types.SizeSmall} +) + +func TestPersistedPoolsSurviveRestart(t *testing.T) { + dir := t.TempDir() + m := newTestManagerAt(t, newFakeEngine(), dir, config.PoolSpec{PoolKey: seedKey, Warm: 1}) + if err := m.SetPools(t.Context(), []config.PoolSpec{{PoolKey: apiKey, Warm: 3}}); err != nil { + t.Fatalf("SetPools: %v", err) + } + // Restart: a fresh manager on the same data dir, still config-seeded with the + // old pool, must rebuild from pools.json (the API-applied set), not the seed. + m2 := newTestManagerAt(t, newFakeEngine(), dir, config.PoolSpec{PoolKey: seedKey, Warm: 1}) + m2.mu.Lock() + _, hasSeed := m2.pools[seedKey] + restored, hasAPI := m2.pools[apiKey] + m2.mu.Unlock() + if hasSeed || !hasAPI { + t.Fatalf("restart did not restore API pools: seed=%v api=%v", hasSeed, hasAPI) + } + if restored.floor != 3 { + t.Errorf("restored warm floor = %d, want 3", restored.floor) + } +} + +func TestDeletingPoolsFileRestoresConfigSeed(t *testing.T) { + dir := t.TempDir() + m := newTestManagerAt(t, newFakeEngine(), dir, config.PoolSpec{PoolKey: seedKey, Warm: 1}) + if err := m.SetPools(t.Context(), []config.PoolSpec{{PoolKey: apiKey, Warm: 3}}); err != nil { + t.Fatalf("SetPools: %v", err) + } + if err := os.Remove(filepath.Join(dir, "pools.json")); err != nil { + t.Fatalf("remove pools.json: %v", err) + } + m2 := newTestManagerAt(t, newFakeEngine(), dir, config.PoolSpec{PoolKey: seedKey, Warm: 1}) + m2.mu.Lock() + _, hasSeed := m2.pools[seedKey] + _, hasAPI := m2.pools[apiKey] + m2.mu.Unlock() + if !hasSeed || hasAPI { + t.Fatalf("deleting pools.json did not restore the config seed: seed=%v api=%v", hasSeed, hasAPI) + } +} + +func TestConcurrentSetPoolsPersistLatest(t *testing.T) { + dir := t.TempDir() + m := newTestManagerAt(t, newFakeEngine(), dir, config.PoolSpec{PoolKey: seedKey, Warm: 1}) + var wg sync.WaitGroup + for warm := 1; warm <= 8; warm++ { + wg.Go(func() { + if err := m.SetPools(t.Context(), []config.PoolSpec{{PoolKey: apiKey, Warm: warm}}); err != nil { + t.Errorf("SetPools(warm=%d): %v", warm, err) + } + }) + } + wg.Wait() + // The set that applied last (highest sequence) owns memory; the persisted + // file must match it, never revert to an already-superseded concurrent write. + m.mu.Lock() + want := m.pools[apiKey].floor + m.mu.Unlock() + pf, err := m.poolStore.load() + if err != nil || pf == nil { + t.Fatalf("load pools.json: pf=%v err=%v", pf, err) + } + if len(pf.Pools) != 1 || pf.Pools[0].Warm != want { + t.Fatalf("persisted %+v, want the last-applied warm %d", pf.Pools, want) + } +} + +func TestRestoredEgressPoolNeedsAttachment(t *testing.T) { + dir := t.TempDir() + pf := poolsFile{Pools: []config.PoolSpec{ + {PoolKey: types.PoolKey{Template: "eg:24.04", Net: types.NetEgress, Size: types.SizeSmall}, Warm: 1}, + }} + raw, err := json.Marshal(pf) + if err != nil { + t.Fatalf("marshal: %v", err) + } + if err := os.WriteFile(filepath.Join(dir, "pools.json"), raw, 0o600); err != nil { + t.Fatalf("write pools.json: %v", err) + } + // A node with no bridge/network must refuse to start a restored egress pool. + if _, err := NewManager(t.Context(), &config.Config{DataDir: dir}, newFakeEngine(), testSecrets(t)); err == nil { + t.Error("NewManager accepted a restored egress-lane pool without an attachment") + } +} diff --git a/sandboxd/pool/setpools.go b/sandboxd/pool/setpools.go index f1236e2..5822c34 100644 --- a/sandboxd/pool/setpools.go +++ b/sandboxd/pool/setpools.go @@ -40,6 +40,8 @@ func (m *Manager) SetPools(ctx context.Context, specs []config.PoolSpec) error { var trim []string m.mu.Lock() + // Sequenced under the mutex (apply order) so commit drops a write a later apply superseded. + seq := m.poolStore.seq.Add(1) now := time.Now() for key, p := range m.pools { spec, ok := desired[key] @@ -92,7 +94,12 @@ func (m *Manager) SetPools(ctx context.Context, specs []config.PoolSpec) error { m.destroy(ctx, trim[i]) }).Wait() m.refillOnce(runCtx) - return nil + // Persist the applied set so a restart rebuilds from it, not the config seed. + persisted := make([]config.PoolSpec, 0, len(desired)) + for _, spec := range desired { + persisted = append(persisted, spec) + } + return m.poolStore.commit(seq, poolsFile{ConfigSeed: m.configSeedHash, Pools: persisted}) } // normalizePoolSpec fills the wire defaults, mirroring ClaimRequest.Key(): diff --git a/sandboxd/server/metrics.go b/sandboxd/server/metrics.go index 5ff8cfa..148331d 100644 --- a/sandboxd/server/metrics.go +++ b/sandboxd/server/metrics.go @@ -48,6 +48,11 @@ func (s *Server) handleMetrics(w http.ResponseWriter, _ *http.Request) { _, _ = fmt.Fprintf(w, "sandboxd_pool_target{template=%q,net=%q,size=%q} %d\n", p.Key.Template, p.Key.Net, p.Key.Size, p.Target) } + if s.placer != nil { + metric("config_digest_mismatch", "gauge", "peers whose cluster-invariant config digest differs from this node") + _, _ = fmt.Fprintf(w, "sandboxd_config_digest_mismatch %d\n", s.placer.ConfigMismatches()) + } + metric("claims_total", "counter", "claims served, by provisioning tier") _, _ = fmt.Fprintf(w, "sandboxd_claims_total{tier=\"warm\"} %d\n", c.ClaimsWarm) _, _ = fmt.Fprintf(w, "sandboxd_claims_total{tier=\"clone\"} %d\n", c.ClaimsClone) diff --git a/sandboxd/server/server.go b/sandboxd/server/server.go index 9079b01..8aa48bf 100644 --- a/sandboxd/server/server.go +++ b/sandboxd/server/server.go @@ -90,6 +90,7 @@ type Placer interface { Candidates(keyHash string) []string TemplateOwners(keyHash string) []string PeerAddrs() []string + ConfigMismatches() int } // InfoResponse is the wire reply of GET /v1/info. Peers lists the other nodes' diff --git a/sandboxd/server/server_test.go b/sandboxd/server/server_test.go index 9006b71..328fab5 100644 --- a/sandboxd/server/server_test.go +++ b/sandboxd/server/server_test.go @@ -990,3 +990,4 @@ type fakePlacer struct { func (f *fakePlacer) Candidates(string) []string { return f.addrs } func (f *fakePlacer) TemplateOwners(string) []string { return f.owners } func (f *fakePlacer) PeerAddrs() []string { return f.addrs } +func (f *fakePlacer) ConfigMismatches() int { return 0 } diff --git a/sandboxd/utils/utils.go b/sandboxd/utils/utils.go index 04cebf4..b44b065 100644 --- a/sandboxd/utils/utils.go +++ b/sandboxd/utils/utils.go @@ -7,9 +7,60 @@ import ( "errors" "fmt" "io" + "os" + "path/filepath" "strings" ) +// WriteFileSync durably replaces path with data: a uniquely-named sibling temp +// file is written, fsync'd, renamed over path, and the parent directory fsync'd +// so the rename survives a crash. Machine-owned state that a restart must trust +// (pools.json, the mesh epoch) uses this rather than a bare write+rename. The +// temp name is per-call unique, so concurrent writers to the same path never +// race a shared temp into a lost rename. +func WriteFileSync(path string, data []byte, perm os.FileMode) error { + dir := filepath.Dir(path) + f, err := os.CreateTemp(dir, filepath.Base(path)+".*.tmp") + if err != nil { + return fmt.Errorf("create temp: %w", err) + } + tmp := f.Name() + renamed := false + defer func() { + if !renamed { + _ = os.Remove(tmp) //nolint:gosec // our own just-created temp under a caller-owned data-dir path + } + }() + if err = f.Chmod(perm); err != nil { + _ = f.Close() + return fmt.Errorf("chmod %s: %w", tmp, err) + } + if _, err = f.Write(data); err != nil { + _ = f.Close() + return fmt.Errorf("write %s: %w", tmp, err) + } + if err = f.Sync(); err != nil { + _ = f.Close() + return fmt.Errorf("sync %s: %w", tmp, err) + } + if err = f.Close(); err != nil { + return fmt.Errorf("close %s: %w", tmp, err) + } + if err = os.Rename(tmp, path); err != nil { //nolint:gosec // our own temp, caller-owned data-dir path + return fmt.Errorf("rename %s: %w", path, err) + } + renamed = true + d, err := os.Open(dir) //nolint:gosec // caller-owned data-dir path + if err != nil { + return fmt.Errorf("open dir: %w", err) + } + defer func() { _ = d.Close() }() + if err = d.Sync(); err != nil { + return fmt.Errorf("sync dir: %w", err) + } + return nil +} + // DecodeStrictJSON decodes one JSON value into v, refusing unknown fields, // duplicated keys, and trailing data — for hand-edited operator input, where // a typo must fail instead of silently changing what was configured (an diff --git a/sdk/go/info.go b/sdk/go/info.go index 4d2b6c7..2eb9028 100644 --- a/sdk/go/info.go +++ b/sdk/go/info.go @@ -44,17 +44,24 @@ func (c *Client) Info(ctx context.Context) (*NodeInfo, error) { return &info, nil } -// peers fetches the cluster's node addresses, best-effort. It reads the -// tenant-accessible /v1/peers, not /v1/info (operator-only), so Lookup -// works under a tenant token. +// peers fetches the cluster's node addresses, best-effort — a discovery +// failure yields nil, harmless for Lookup's read scatter. func (c *Client) peers(ctx context.Context) []string { + addrs, _ := c.peersOrErr(ctx) + return addrs +} + +// peersOrErr fetches the cluster's node addresses, surfacing a discovery +// failure. It reads the tenant-accessible /v1/peers, not /v1/info +// (operator-only), so it works under a tenant token. +func (c *Client) peersOrErr(ctx context.Context) ([]string, error) { ctx, cancel := context.WithTimeout(ctx, peersTimeout) defer cancel() reply, err := doJSON[struct { Peers []string `json:"peers"` }](ctx, c, http.MethodGet, c.addr, "/v1/peers", nil, c.apiToken, "peers") if err != nil { - return nil + return nil, err } - return reply.Peers + return reply.Peers, nil } diff --git a/sdk/go/pools.go b/sdk/go/pools.go new file mode 100644 index 0000000..e7fdee3 --- /dev/null +++ b/sdk/go/pools.go @@ -0,0 +1,85 @@ +package sandbox + +import ( + "bytes" + "context" + "fmt" + "net/http" + "sync" +) + +// PoolSpec is a desired warm pool for SetPools. Egress is config-owned on the +// node and rejected by the API, so it has no field here. +type PoolSpec struct { + Template string `json:"template"` + Net NetShape `json:"net,omitempty"` + Size Size `json:"size,omitempty"` + Warm int `json:"warm"` + WarmMax int `json:"warm_max,omitempty"` + IdleHibernateSeconds int `json:"idle_hibernate_seconds,omitempty"` + ArchiveAfterSeconds int `json:"archive_after_seconds,omitempty"` + ArchiveDeleteAfterSeconds int `json:"archive_delete_after_seconds,omitempty"` +} + +// PoolResult is one node's outcome from SetPoolsCluster. +type PoolResult struct { + Addr string + Info *NodeInfo + Err error +} + +type poolUpdate struct { + Pools []PoolSpec `json:"pools"` +} + +// SetPools replaces the entry node's desired warm pools (PUT /v1/pools): a +// declarative full replace, so omitted pools drain. Requires the operator token. +func (c *Client) SetPools(ctx context.Context, pools []PoolSpec) (*NodeInfo, error) { + return c.setPoolsAt(ctx, c.addr, pools) +} + +// SetPoolsCluster applies pools to the entry node and every peer from +// /v1/peers, returning a per-node result. The apply is an idempotent +// declarative replace, so retrying the failed nodes is the whole consistency +// protocol. A non-nil error means peer discovery failed and the fan-out +// reached only the entry node (plus any peers already seen) — an incomplete +// set that a caller must retry, distinct from a genuine single-node cluster +// where the error is nil. It fits a homogeneous cluster; nodes that differ in +// capacity or egress attachment should take a per-node SetPools instead. +func (c *Client) SetPoolsCluster(ctx context.Context, pools []PoolSpec) ([]PoolResult, error) { + peers, peersErr := c.peersOrErr(ctx) + seen := map[string]struct{}{} + var addrs []string + for _, a := range append([]string{c.addr}, peers...) { + if _, dup := seen[a]; dup { + continue + } + seen[a] = struct{}{} + addrs = append(addrs, a) + } + results := make([]PoolResult, len(addrs)) + var wg sync.WaitGroup + for i, addr := range addrs { + wg.Go(func() { + info, err := c.setPoolsAt(ctx, addr, pools) + results[i] = PoolResult{Addr: addr, Info: info, Err: err} + }) + } + wg.Wait() + if peersErr != nil { + return results, fmt.Errorf("discover peers: %w", peersErr) + } + return results, nil +} + +func (c *Client) setPoolsAt(ctx context.Context, addr string, pools []PoolSpec) (*NodeInfo, error) { + body, err := encodeBody("pools", poolUpdate{Pools: pools}) + if err != nil { + return nil, err + } + info, err := doJSON[NodeInfo](ctx, c, http.MethodPut, addr, "/v1/pools", bytes.NewReader(body), c.apiToken, "pools") + if err != nil { + return nil, err + } + return &info, nil +} diff --git a/sdk/go/pools_test.go b/sdk/go/pools_test.go new file mode 100644 index 0000000..b76e0c6 --- /dev/null +++ b/sdk/go/pools_test.go @@ -0,0 +1,111 @@ +package sandbox + +import ( + "encoding/json" + "net/http" + "net/http/httptest" + "strings" + "sync" + "testing" +) + +func TestSetPools(t *testing.T) { + var got poolUpdate + ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodPut || r.URL.Path != "/v1/pools" { + t.Errorf("got %s %s, want PUT /v1/pools", r.Method, r.URL.Path) + } + if auth := r.Header.Get("Authorization"); auth != "Bearer root" { + t.Errorf("auth = %q, want Bearer root", auth) + } + _ = json.NewDecoder(r.Body).Decode(&got) + _ = json.NewEncoder(w).Encode(NodeInfo{Claimed: 2}) + })) + defer ts.Close() + + c := testClient(t, ts, WithAPIToken("root")) + info, err := c.SetPools(t.Context(), []PoolSpec{{Template: "rt:24.04", Net: NetNone, Size: Small, Warm: 3}}) + if err != nil { + t.Fatalf("SetPools: %v", err) + } + if info.Claimed != 2 { + t.Errorf("info = %+v, want Claimed 2", info) + } + if len(got.Pools) != 1 || got.Pools[0].Warm != 3 || got.Pools[0].Template != "rt:24.04" { + t.Errorf("server received %+v", got.Pools) + } +} + +func TestSetPoolsClusterFansOut(t *testing.T) { + var mu sync.Mutex + applied := map[string]int{} + pools := func(w http.ResponseWriter, r *http.Request) { + var body poolUpdate + _ = json.NewDecoder(r.Body).Decode(&body) + mu.Lock() + applied[r.Host] = body.Pools[0].Warm + mu.Unlock() + _ = json.NewEncoder(w).Encode(NodeInfo{}) + } + peer := httptest.NewServer(http.HandlerFunc(pools)) + defer peer.Close() + peerAddr := strings.TrimPrefix(peer.URL, "http://") + + entry := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path == "/v1/peers" { + _ = json.NewEncoder(w).Encode(map[string][]string{"peers": {peerAddr}}) + return + } + pools(w, r) + })) + defer entry.Close() + + c := testClient(t, entry, WithAPIToken("root")) + results, err := c.SetPoolsCluster(t.Context(), []PoolSpec{{Template: "rt:24.04", Net: NetNone, Size: Small, Warm: 5}}) + if err != nil { + t.Fatalf("SetPoolsCluster: %v", err) + } + if len(results) != 2 { + t.Fatalf("results = %d, want 2 (entry + one peer)", len(results)) + } + for _, res := range results { + if res.Err != nil { + t.Errorf("node %s: %v", res.Addr, res.Err) + } + } + mu.Lock() + defer mu.Unlock() + if len(applied) != 2 { + t.Fatalf("applied to %d nodes, want 2: %v", len(applied), applied) + } + for addr, warm := range applied { + if warm != 5 { + t.Errorf("node %s saw warm %d, want 5", addr, warm) + } + } +} + +func TestSetPoolsClusterPeerDiscoveryFails(t *testing.T) { + var applied int + entry := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path == "/v1/peers" { + w.WriteHeader(http.StatusInternalServerError) + return + } + applied++ + _ = json.NewEncoder(w).Encode(NodeInfo{}) + })) + defer entry.Close() + + c := testClient(t, entry, WithAPIToken("root")) + results, err := c.SetPoolsCluster(t.Context(), []PoolSpec{{Template: "rt:24.04", Net: NetNone, Size: Small, Warm: 5}}) + if err == nil { + t.Fatal("SetPoolsCluster returned nil error despite peer-discovery failure") + } + if len(results) != 1 || results[0].Err != nil { + t.Fatalf("results = %+v, want the entry node applied cleanly", results) + } + if applied != 1 { + t.Errorf("applied to %d nodes, want 1 (entry only)", applied) + } +} From 9c450c7b7387a2ed6c014ec0685bb62af59c77a9 Mon Sep 17 00:00:00 2001 From: CMGS Date: Mon, 13 Jul 2026 13:24:43 +0800 Subject: [PATCH 2/3] fix(mesh): persist epoch before gossip and seed strictly above the floor MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A restart under a backwards clock re-published an epoch equal to the one peers already held, and merge's strict `>` then rejected the node's fresh pools/templates/digest. UpdateSelf also published the new epoch into the gossiped view before persisting it, so a crash in that window left disk behind what peers had seen — the same stranding. Seed from max(wall, persisted+1), persist the candidate before publishing (hold the old advertised state on write failure), and reject an above-MaxInt64 persisted value as corrupt so the monotonic counter cannot saturate. --- sandboxd/mesh/epoch.go | 12 ++++++++---- sandboxd/mesh/mesh.go | 33 ++++++++++++++++++++++----------- sandboxd/mesh/mesh_test.go | 6 ++++-- sandboxd/mesh/state_test.go | 37 ++++++++++++++++++++++++++++++++++--- 4 files changed, 68 insertions(+), 20 deletions(-) diff --git a/sandboxd/mesh/epoch.go b/sandboxd/mesh/epoch.go index 3d09519..d0d3fb3 100644 --- a/sandboxd/mesh/epoch.go +++ b/sandboxd/mesh/epoch.go @@ -1,6 +1,7 @@ package mesh import ( + "math" "os" "strconv" "strings" @@ -8,16 +9,19 @@ import ( "github.com/cocoonstack/sandbox/sandboxd/utils" ) -// loadEpoch reads the persisted gossip epoch, or 0 when the file is absent or -// unreadable — the wall-clock seed then wins, which is the pre-persistence -// behavior. +// loadEpoch reads the persisted gossip epoch, or 0 when the file is absent, +// unreadable, or holds an implausible value — the wall-clock seed then wins, +// which is the pre-persistence behavior. The epoch is a UnixNano-derived +// counter that only ever increments by one, so a value above MaxInt64 cannot +// have arisen legitimately; treating it as corrupt keeps a crafted or +// bit-rotted file from seeding a saturated counter that can never advance. func loadEpoch(path string) uint64 { raw, err := os.ReadFile(path) //nolint:gosec // node-local data-dir path if err != nil { return 0 } n, err := strconv.ParseUint(strings.TrimSpace(string(raw)), 10, 64) - if err != nil { + if err != nil || n > math.MaxInt64 { return 0 } return n diff --git a/sandboxd/mesh/mesh.go b/sandboxd/mesh/mesh.go index 709f408..b61c8b0 100644 --- a/sandboxd/mesh/mesh.go +++ b/sandboxd/mesh/mesh.go @@ -54,10 +54,13 @@ type Mesh struct { // gossip encryption when non-empty; dataDir holds the persisted epoch. func New(cfg *memberlist.Config, nodeID, selfAddr string, secretKey []byte, dataDir string) (*Mesh, error) { epochPath := filepath.Join(dataDir, "mesh-epoch") - // Seed from the persisted epoch, floored at wall-clock: a restart under a - // backwards clock would otherwise regress below peers' last-seen epoch, so - // its fresh state is silently rejected. Persist the seed at once. - epoch := max(loadEpoch(epochPath), uint64(time.Now().UnixNano())) //nolint:gosec // UnixNano is positive for current times + // Seed strictly above the persisted floor, then floor at wall-clock. The + // persisted value is the last epoch peers were shown; seeding at it (not + // above) ties the stale copy peers still hold, and merge's `>` rejects the + // tie, so a backwards-clock restart's fresh pools/templates/digest never + // propagate. The +1 keeps the restart one ahead even when the clock + // regressed; loadEpoch caps the floor at MaxInt64 so the +1 cannot wrap. + epoch := max(uint64(time.Now().UnixNano()), loadEpoch(epochPath)+1) //nolint:gosec // UnixNano is positive for current times m := &Mesh{ epochPath: epochPath, self: NodeState{ @@ -108,17 +111,25 @@ func (m *Mesh) UpdateSelf(pools map[string]int, templates []string) { m.mu.Unlock() return } - m.self.Epoch++ - m.self.Pools = pools - m.self.Templates = templates - m.view[m.self.NodeID] = m.self - epoch := m.self.Epoch + epoch := m.self.Epoch + 1 m.mu.Unlock() - // Persist off the manager lock (redirect placement reads it): the bump is - // the new floor a restart must not regress below. + // Persist the candidate epoch before it can gossip: memberlist may ship the + // new self state the instant it is published, so a crash between publish and + // persist would strand peers on an epoch a backwards-clock restart cannot + // climb back above. On write failure keep the old advertised state rather + // than gossip an epoch the disk does not back. if err := m.persistEpoch(epoch); err != nil { log.WithFunc("mesh.UpdateSelf").Warnf(context.Background(), "persist epoch: %v", err) + return } + m.mu.Lock() + if epoch > m.self.Epoch { + m.self.Epoch = epoch + m.self.Pools = pools + m.self.Templates = templates + m.view[m.self.NodeID] = m.self + } + m.mu.Unlock() } // SetSelfDigest records this node's cluster-invariant config digest so peers can diff --git a/sandboxd/mesh/mesh_test.go b/sandboxd/mesh/mesh_test.go index 6bae73e..60e0b7a 100644 --- a/sandboxd/mesh/mesh_test.go +++ b/sandboxd/mesh/mesh_test.go @@ -4,6 +4,7 @@ import ( "fmt" "io" "log" + "path/filepath" "slices" "testing" "time" @@ -140,8 +141,9 @@ func TestTwoNodeClusterGossipsPools(t *testing.T) { func newTestMesh(t *testing.T, id string) *Mesh { t.Helper() return &Mesh{ - self: NodeState{NodeID: id, Addr: id + ":7777", Pools: map[string]int{}}, - view: map[string]NodeState{id: {NodeID: id, Addr: id + ":7777"}}, + epochPath: filepath.Join(t.TempDir(), "mesh-epoch"), + self: NodeState{NodeID: id, Addr: id + ":7777", Pools: map[string]int{}}, + view: map[string]NodeState{id: {NodeID: id, Addr: id + ":7777"}}, } } diff --git a/sandboxd/mesh/state_test.go b/sandboxd/mesh/state_test.go index 6f79ce9..8c741a0 100644 --- a/sandboxd/mesh/state_test.go +++ b/sandboxd/mesh/state_test.go @@ -1,6 +1,7 @@ package mesh import ( + "os" "path/filepath" "sync" "testing" @@ -34,15 +35,45 @@ func TestEpochRoundTrip(t *testing.T) { } } -func TestEpochSeededFromPersistedFloor(t *testing.T) { +func TestLoadEpochRejectsImplausibleValue(t *testing.T) { + p := filepath.Join(t.TempDir(), "e") + // A saturated/corrupt value must not seed a counter that +1 cannot climb + // past and that would tie a peer's stale copy forever. + if err := os.WriteFile(p, []byte("18446744073709551615"), 0o600); err != nil { + t.Fatalf("write: %v", err) + } + if got := loadEpoch(p); got != 0 { + t.Errorf("loadEpoch of a MaxUint64 file = %d, want 0 (treated as corrupt)", got) + } +} + +func TestEpochSeededAbovePersistedFloor(t *testing.T) { dir := t.TempDir() huge := uint64(1) << 62 // above any plausible wall-clock nanos if err := storeEpoch(filepath.Join(dir, "mesh-epoch"), huge); err != nil { t.Fatalf("store: %v", err) } m := newBoundMesh(t, dir) - if m.self.Epoch != huge { - t.Errorf("epoch seeded %d, want persisted floor %d (a backwards clock must not regress it)", m.self.Epoch, huge) + // Strictly above: the persisted value is what peers last saw, so seeding at + // it would tie their stale copy and merge's `>` would reject the restart. + if m.self.Epoch != huge+1 { + t.Errorf("epoch seeded %d, want persisted floor + 1 = %d (a backwards clock must land above peers' last-seen epoch)", m.self.Epoch, huge+1) + } +} + +func TestUpdateSelfPersistFailKeepsOldState(t *testing.T) { + dir := t.TempDir() + m := newBoundMesh(t, dir) + before := m.self.Epoch + // Point the epoch file at a non-existent directory so the durable write + // fails; the candidate epoch must not be published while the disk lags. + m.epochPath = filepath.Join(dir, "gone", "mesh-epoch") + m.UpdateSelf(map[string]int{"k": 1}, nil) + if m.self.Epoch != before { + t.Errorf("self epoch advanced to %d on a failed persist, want %d held", m.self.Epoch, before) + } + if len(m.self.Pools) != 0 { + t.Errorf("self pools published %v despite a failed persist", m.self.Pools) } } From 637636a3a6d7334118faa350638b10da63fe3d6f Mon Sep 17 00:00:00 2001 From: CMGS Date: Mon, 13 Jul 2026 13:38:17 +0800 Subject: [PATCH 3/3] review: tighten PR#33 comments to the one-line-WHY budget --- sandboxd/config/config.go | 10 +++------ sandboxd/config/config_test.go | 5 +---- sandboxd/main.go | 3 +-- sandboxd/mesh/epoch.go | 7 ++----- sandboxd/mesh/mesh.go | 36 ++++++++++++--------------------- sandboxd/mesh/state_test.go | 9 +++------ sandboxd/pool/pool.go | 8 +++----- sandboxd/pool/poolstore.go | 26 +++++++++--------------- sandboxd/pool/poolstore_test.go | 6 ++---- sandboxd/utils/utils.go | 9 +++------ sdk/go/info.go | 8 +++----- sdk/go/pools.go | 12 ++++------- 12 files changed, 47 insertions(+), 92 deletions(-) diff --git a/sandboxd/config/config.go b/sandboxd/config/config.go index e072d6d..911791c 100644 --- a/sandboxd/config/config.go +++ b/sandboxd/config/config.go @@ -210,13 +210,9 @@ func (c *Config) HasEgress() bool { return c.Bridge != "" || c.Network != "" } -// ClusterDigest fingerprints the config every mesh node must share, so a -// divergent node is caught at join instead of at a later 401 or failed -// cross-node interception: api_token, the tenant name→token set, preview_secret, -// and the egress cluster root. With a cluster_key set (gossip is encrypted) it -// HMACs the full set including token material; without one gossip may be -// cleartext, so it covers only non-secret identity (tenant names + CA root), -// keeping a weak token off the wire. +// ClusterDigest fingerprints the must-match config so a divergent node is +// caught early, not at a later 401. With cluster_key it HMACs token material +// too; without one only tenant names + CA root ride the (maybe cleartext) wire. func (c *Config) ClusterDigest(caFingerprint string) string { names := make([]string, len(c.Tenants)) for i, t := range c.Tenants { diff --git a/sandboxd/config/config_test.go b/sandboxd/config/config_test.go index e775f48..613ff11 100644 --- a/sandboxd/config/config_test.go +++ b/sandboxd/config/config_test.go @@ -22,13 +22,10 @@ func TestClusterDigest(t *testing.T) { if base.ClusterDigest("other-fp") == d { t.Error("an egress CA root change is not reflected") } - // Without a cluster_key gossip may be cleartext, so token material must be - // excluded: an api_token change alone must NOT alter the on-wire digest. + // Keyless: token material must stay off the (maybe cleartext) wire. if (&Config{APIToken: "other", PreviewSecret: "ps", Tenants: base.Tenants}).ClusterDigest("ca-fp") != d { t.Error("api_token leaked into the keyless digest") } - // With a cluster_key the digest is an HMAC over token material, so an - // api_token change IS covered. key := base64.StdEncoding.EncodeToString(make([]byte, 32)) keyed := &Config{APIToken: "tok", PreviewSecret: "ps", Tenants: base.Tenants, Mesh: &MeshConfig{ClusterKey: key}} keyedDiff := &Config{APIToken: "other", PreviewSecret: "ps", Tenants: base.Tenants, Mesh: &MeshConfig{ClusterKey: key}} diff --git a/sandboxd/main.go b/sandboxd/main.go index 2b765d2..3904b36 100644 --- a/sandboxd/main.go +++ b/sandboxd/main.go @@ -171,8 +171,7 @@ func startMesh(cfg *config.Config, mgr *pool.Manager) (*mesh.Mesh, error) { if err != nil { return nil, err } - // Publish the cluster-invariant config digest before Join, so the first - // gossip a peer sees already carries it. + // Publish the config digest before Join, so the first gossip carries it. msh.SetSelfDigest(cfg.ClusterDigest(mgr.EgressCAFingerprint())) if err := msh.Join(mc.Join); err != nil { _ = msh.Shutdown() diff --git a/sandboxd/mesh/epoch.go b/sandboxd/mesh/epoch.go index d0d3fb3..f943950 100644 --- a/sandboxd/mesh/epoch.go +++ b/sandboxd/mesh/epoch.go @@ -10,11 +10,8 @@ import ( ) // loadEpoch reads the persisted gossip epoch, or 0 when the file is absent, -// unreadable, or holds an implausible value — the wall-clock seed then wins, -// which is the pre-persistence behavior. The epoch is a UnixNano-derived -// counter that only ever increments by one, so a value above MaxInt64 cannot -// have arisen legitimately; treating it as corrupt keeps a crafted or -// bit-rotted file from seeding a saturated counter that can never advance. +// unreadable, or above MaxInt64 — a UnixNano-derived counter cannot legitimately +// exceed it, so a corrupt value falls back to the wall-clock seed. func loadEpoch(path string) uint64 { raw, err := os.ReadFile(path) //nolint:gosec // node-local data-dir path if err != nil { diff --git a/sandboxd/mesh/mesh.go b/sandboxd/mesh/mesh.go index b61c8b0..d2e6aea 100644 --- a/sandboxd/mesh/mesh.go +++ b/sandboxd/mesh/mesh.go @@ -54,12 +54,9 @@ type Mesh struct { // gossip encryption when non-empty; dataDir holds the persisted epoch. func New(cfg *memberlist.Config, nodeID, selfAddr string, secretKey []byte, dataDir string) (*Mesh, error) { epochPath := filepath.Join(dataDir, "mesh-epoch") - // Seed strictly above the persisted floor, then floor at wall-clock. The - // persisted value is the last epoch peers were shown; seeding at it (not - // above) ties the stale copy peers still hold, and merge's `>` rejects the - // tie, so a backwards-clock restart's fresh pools/templates/digest never - // propagate. The +1 keeps the restart one ahead even when the clock - // regressed; loadEpoch caps the floor at MaxInt64 so the +1 cannot wrap. + // Seed strictly above the persisted floor (the last epoch peers saw): + // seeding at it ties their stale copy and merge's `>` rejects the restart's + // fresh state. loadEpoch caps the floor so the +1 cannot wrap. epoch := max(uint64(time.Now().UnixNano()), loadEpoch(epochPath)+1) //nolint:gosec // UnixNano is positive for current times m := &Mesh{ epochPath: epochPath, @@ -113,11 +110,9 @@ func (m *Mesh) UpdateSelf(pools map[string]int, templates []string) { } epoch := m.self.Epoch + 1 m.mu.Unlock() - // Persist the candidate epoch before it can gossip: memberlist may ship the - // new self state the instant it is published, so a crash between publish and - // persist would strand peers on an epoch a backwards-clock restart cannot - // climb back above. On write failure keep the old advertised state rather - // than gossip an epoch the disk does not back. + // Persist the candidate before publishing it: memberlist gossips self the + // instant it enters the view, so a crash before the write would strand peers + // on an epoch a backwards-clock restart can't beat. Hold old state on failure. if err := m.persistEpoch(epoch); err != nil { log.WithFunc("mesh.UpdateSelf").Warnf(context.Background(), "persist epoch: %v", err) return @@ -142,8 +137,7 @@ func (m *Mesh) SetSelfDigest(digest string) { } // ConfigMismatches counts peers whose config digest differs from this node's — -// a gauge for alerting on a divergent cluster-invariant config (the intermittent -// 401 / cross-node interception class), recomputed per read from the view. +// the gauge for alerting on a divergent cluster, recomputed per read. func (m *Mesh) ConfigMismatches() int { m.mu.Lock() defer m.mu.Unlock() @@ -247,9 +241,8 @@ func (m *Mesh) Shutdown() error { return m.ml.Shutdown() } -// persistEpoch durably records the gossip epoch, serialized so a lower value -// from a slower concurrent UpdateSelf can never overwrite a higher one already -// on disk — the persisted floor must only ever advance. +// persistEpoch durably records the epoch, serialized so a slower concurrent +// UpdateSelf's lower value cannot overwrite a higher one already on disk. func (m *Mesh) persistEpoch(epoch uint64) error { m.epochMu.Lock() defer m.epochMu.Unlock() @@ -286,13 +279,10 @@ func (m *Mesh) merge(states []NodeState) { if ok && st.Epoch <= cur.Epoch { continue } - // Warn when a peer's config digest diverges from this node's, on each - // distinct new value (not once per lifetime): a mismatched - // cluster-invariant config (api_token/tenants/preview_secret/egress CA - // root) makes cross-node redirects 401 and interception fail — the - // worst failure mode to debug at first unlucky redirect. Warn-only: - // refusing the merge would turn a routine rolling credential rotation - // into a partition. + // Warn on each distinct divergent digest (not once per lifetime): a + // mismatched api_token/tenants/preview_secret/CA root 401s cross-node + // redirects and fails interception. Warn-only — refusing would partition + // a rolling credential rotation. if m.self.Digest != "" && st.Digest != "" && st.Digest != m.self.Digest && (!ok || cur.Digest != st.Digest) { log.WithFunc("mesh.merge").Warnf(context.Background(), "peer %s config digest %s differs from this node's %s: cluster-invariant config diverges (redirects may 401, interception may fail)", diff --git a/sandboxd/mesh/state_test.go b/sandboxd/mesh/state_test.go index 8c741a0..309694e 100644 --- a/sandboxd/mesh/state_test.go +++ b/sandboxd/mesh/state_test.go @@ -37,8 +37,7 @@ func TestEpochRoundTrip(t *testing.T) { func TestLoadEpochRejectsImplausibleValue(t *testing.T) { p := filepath.Join(t.TempDir(), "e") - // A saturated/corrupt value must not seed a counter that +1 cannot climb - // past and that would tie a peer's stale copy forever. + // A saturated value must not seed a counter that +1 can't climb past. if err := os.WriteFile(p, []byte("18446744073709551615"), 0o600); err != nil { t.Fatalf("write: %v", err) } @@ -54,8 +53,7 @@ func TestEpochSeededAbovePersistedFloor(t *testing.T) { t.Fatalf("store: %v", err) } m := newBoundMesh(t, dir) - // Strictly above: the persisted value is what peers last saw, so seeding at - // it would tie their stale copy and merge's `>` would reject the restart. + // Strictly above: seeding at the persisted value would tie peers' stale copy. if m.self.Epoch != huge+1 { t.Errorf("epoch seeded %d, want persisted floor + 1 = %d (a backwards clock must land above peers' last-seen epoch)", m.self.Epoch, huge+1) } @@ -65,8 +63,7 @@ func TestUpdateSelfPersistFailKeepsOldState(t *testing.T) { dir := t.TempDir() m := newBoundMesh(t, dir) before := m.self.Epoch - // Point the epoch file at a non-existent directory so the durable write - // fails; the candidate epoch must not be published while the disk lags. + // A non-existent dir makes the durable write fail: the epoch must not publish. m.epochPath = filepath.Join(dir, "gone", "mesh-epoch") m.UpdateSelf(map[string]int{"k": 1}, nil) if m.self.Epoch != before { diff --git a/sandboxd/pool/pool.go b/sandboxd/pool/pool.go index 2aea752..aa8e67f 100644 --- a/sandboxd/pool/pool.go +++ b/sandboxd/pool/pool.go @@ -155,10 +155,8 @@ type Manager struct { maxFork int store *claimStore - // poolStore persists the last API-applied pool set; configSeedHash is the - // config.json pools' hash, to warn when a file edit is overridden by it. poolStore *poolStore - configSeedHash string + configSeedHash string // config pools' hash, to warn when a file edit is overridden // idleDefault is the idle-hibernate threshold for unpooled keys; pooled // keys carry theirs on the pool struct. Zero means disabled. @@ -352,8 +350,8 @@ func NewManager(ctx context.Context, cfg *config.Config, eng Engine, secrets *eg return m, nil } -// EgressCAFingerprint is the egress cluster root's fingerprint, or "" when this -// node does no interception; it feeds the cluster-invariant config digest. +// EgressCAFingerprint is the egress cluster root's fingerprint, or "" when the +// node does no interception. func (m *Manager) EgressCAFingerprint() string { if m.egressCA == nil { return "" diff --git a/sandboxd/pool/poolstore.go b/sandboxd/pool/poolstore.go index 1e6fbcc..490af3f 100644 --- a/sandboxd/pool/poolstore.go +++ b/sandboxd/pool/poolstore.go @@ -21,12 +21,9 @@ import ( "github.com/cocoonstack/sandbox/sandboxd/utils" ) -// poolsFile records the last API-applied pool set (PUT /v1/pools) so a restart -// rebuilds from operator intent, not the config.json seed that only grows -// staler under API-first management. ConfigSeed is the config.json pools' hash -// at write time; a boot that finds it changed warns that the file edit is -// overridden. Egress is never here — SetPools rejects it, so it stays -// config-owned and is re-merged by key at load. +// poolsFile records the last API-applied pool set so a restart rebuilds from +// operator intent, not the config.json seed. ConfigSeed is the config pools' +// hash at write time; a boot finding it changed warns the edit is overridden. type poolsFile struct { ConfigSeed string `json:"config_seed"` Pools []config.PoolSpec `json:"pools"` @@ -62,8 +59,7 @@ func (s *poolStore) load() (*poolsFile, error) { } // commit durably writes the applied set, serialized by s.mu; a seq no newer -// than the last written is a no-op, so two concurrent SetPools calls can never -// let an already-superseded set win the on-disk write and revert the latest. +// than the last written is a no-op, so a superseded concurrent write can't win. func (s *poolStore) commit(seq uint64, pf poolsFile) error { s.mu.Lock() defer s.mu.Unlock() @@ -81,12 +77,9 @@ func (s *poolStore) commit(seq uint64, pf poolsFile) error { return nil } -// adoptPersistedPools replaces the config-seeded pools with the last API-applied -// set (pools.json) when present, so a restart rebuilds from operator intent. -// Egress stays config-owned (kept in m.poolEgress, re-merged by key); the egress -// CA must already be loaded so golden adoption can check its fingerprint. A -// restored pool that no longer validates (e.g. an egress lane with the bridge -// now removed) fails the boot, matching config-load discipline. +// adoptPersistedPools rebuilds the pools from pools.json (the last API-applied +// set) over the config seed when present. Egress stays config-owned; a restored +// pool that no longer validates fails the boot, matching config-load discipline. func (m *Manager) adoptPersistedPools(ctx context.Context) error { pf, err := m.poolStore.load() if err != nil { @@ -127,9 +120,8 @@ func (m *Manager) adoptPersistedPools(ctx context.Context) error { return nil } -// poolSeedHash digests a config pool set's warm-target shape, order-independent; -// egress is excluded (config-owned, re-merged separately) so the hash tracks -// exactly the targets pools.json would override. +// poolSeedHash digests a pool set's warm-target shape, order-independent and +// egress-excluded, so it tracks exactly the targets pools.json would override. func poolSeedHash(specs []config.PoolSpec) string { shaped := make([]config.PoolSpec, len(specs)) copy(shaped, specs) diff --git a/sandboxd/pool/poolstore_test.go b/sandboxd/pool/poolstore_test.go index 3c32e8b..c0e6a10 100644 --- a/sandboxd/pool/poolstore_test.go +++ b/sandboxd/pool/poolstore_test.go @@ -22,8 +22,7 @@ func TestPersistedPoolsSurviveRestart(t *testing.T) { if err := m.SetPools(t.Context(), []config.PoolSpec{{PoolKey: apiKey, Warm: 3}}); err != nil { t.Fatalf("SetPools: %v", err) } - // Restart: a fresh manager on the same data dir, still config-seeded with the - // old pool, must rebuild from pools.json (the API-applied set), not the seed. + // Restart on the same data dir, still config-seeded: must rebuild from pools.json. m2 := newTestManagerAt(t, newFakeEngine(), dir, config.PoolSpec{PoolKey: seedKey, Warm: 1}) m2.mu.Lock() _, hasSeed := m2.pools[seedKey] @@ -68,8 +67,7 @@ func TestConcurrentSetPoolsPersistLatest(t *testing.T) { }) } wg.Wait() - // The set that applied last (highest sequence) owns memory; the persisted - // file must match it, never revert to an already-superseded concurrent write. + // The last-applied set owns memory; the persisted file must match it. m.mu.Lock() want := m.pools[apiKey].floor m.mu.Unlock() diff --git a/sandboxd/utils/utils.go b/sandboxd/utils/utils.go index b44b065..df1a0a6 100644 --- a/sandboxd/utils/utils.go +++ b/sandboxd/utils/utils.go @@ -12,12 +12,9 @@ import ( "strings" ) -// WriteFileSync durably replaces path with data: a uniquely-named sibling temp -// file is written, fsync'd, renamed over path, and the parent directory fsync'd -// so the rename survives a crash. Machine-owned state that a restart must trust -// (pools.json, the mesh epoch) uses this rather than a bare write+rename. The -// temp name is per-call unique, so concurrent writers to the same path never -// race a shared temp into a lost rename. +// WriteFileSync durably replaces path with data (temp + fsync + rename + dir +// fsync) so the rename survives a crash. The temp name is per-call unique, so +// concurrent writers to the same path can't race a shared temp into a lost rename. func WriteFileSync(path string, data []byte, perm os.FileMode) error { dir := filepath.Dir(path) f, err := os.CreateTemp(dir, filepath.Base(path)+".*.tmp") diff --git a/sdk/go/info.go b/sdk/go/info.go index 2eb9028..149a16e 100644 --- a/sdk/go/info.go +++ b/sdk/go/info.go @@ -44,16 +44,14 @@ func (c *Client) Info(ctx context.Context) (*NodeInfo, error) { return &info, nil } -// peers fetches the cluster's node addresses, best-effort — a discovery -// failure yields nil, harmless for Lookup's read scatter. +// peers fetches the cluster's node addresses, best-effort (nil on failure). func (c *Client) peers(ctx context.Context) []string { addrs, _ := c.peersOrErr(ctx) return addrs } -// peersOrErr fetches the cluster's node addresses, surfacing a discovery -// failure. It reads the tenant-accessible /v1/peers, not /v1/info -// (operator-only), so it works under a tenant token. +// peersOrErr fetches the node addresses, surfacing a discovery failure. It reads +// /v1/peers (tenant-accessible), so it works under a tenant token. func (c *Client) peersOrErr(ctx context.Context) ([]string, error) { ctx, cancel := context.WithTimeout(ctx, peersTimeout) defer cancel() diff --git a/sdk/go/pools.go b/sdk/go/pools.go index e7fdee3..9395ca5 100644 --- a/sdk/go/pools.go +++ b/sdk/go/pools.go @@ -38,14 +38,10 @@ func (c *Client) SetPools(ctx context.Context, pools []PoolSpec) (*NodeInfo, err return c.setPoolsAt(ctx, c.addr, pools) } -// SetPoolsCluster applies pools to the entry node and every peer from -// /v1/peers, returning a per-node result. The apply is an idempotent -// declarative replace, so retrying the failed nodes is the whole consistency -// protocol. A non-nil error means peer discovery failed and the fan-out -// reached only the entry node (plus any peers already seen) — an incomplete -// set that a caller must retry, distinct from a genuine single-node cluster -// where the error is nil. It fits a homogeneous cluster; nodes that differ in -// capacity or egress attachment should take a per-node SetPools instead. +// SetPoolsCluster applies pools to the entry node and every peer, returning a +// per-node result; retrying failed nodes is the whole protocol (idempotent +// replace). A non-nil error means peer discovery failed and only the entry node +// was reached — an incomplete apply to retry, not a single-node cluster (nil). func (c *Client) SetPoolsCluster(ctx context.Context, pools []PoolSpec) ([]PoolResult, error) { peers, peersErr := c.peersOrErr(ctx) seen := map[string]struct{}{}