| 1 | package host |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "encoding/json" |
| 6 | "errors" |
| 7 | "fmt" |
| 8 | "log" |
| 9 | "math/rand/v2" |
| 10 | "os" |
| 11 | "path/filepath" |
| 12 | "sort" |
| 13 | "time" |
| 14 | |
| 15 | "github.com/carlosframework/platform/internal/registry" |
| 16 | "github.com/carlosframework/platform/internal/release" |
| 17 | ) |
| 18 | |
| 19 | // The cold-standby spec (2026-08-12 HA posture, Part I rung 1): the current |
| 20 | // snapshot at a per-box key a runbook can name (snapshotDBKey), a sidecar |
| 21 | // describing it, and a short tail of prior generations. All under control/ |
| 22 | // beside control/serial; platform boxes write direct S3, so no store-door |
| 23 | // ACL is involved. |
| 24 | const ( |
| 25 | snapshotKeep = 8 |
| 26 | snapshotStamp = "20060102T150405Z" |
| 27 | ) |
| 28 | |
| 29 | // Per-box keys (spec 2026-08-20 Part 1, closes #158): every platform box |
| 30 | // snapshots ITS OWN registry under its own name; boxes never contend. |
| 31 | // The pre-multi-region fixed keys ("control/registry-snapshot.db" etc.) |
| 32 | // are still present in production buckets until both deployments roll — |
| 33 | // they are abandoned, not rewritten; deletion is a rollout step, not code. |
| 34 | func snapshotDBKey(box string) string { return "control/registry-snapshot/" + box + ".db" } |
| 35 | func snapshotMetaKey(box string) string { return "control/registry-snapshot/" + box + ".json" } |
| 36 | func snapshotGenPrefix(box string) string { |
| 37 | return "control/registry-snapshots/" + box + "/" |
| 38 | } |
| 39 | |
| 40 | // snapshotMeta describes the object at snapshotDBKey(box). `Box` is the writing |
| 41 | // box's label (per-instance hashed machine id) — a restore must check it |
| 42 | // names the DEAD box, not the standby, or it is restoring the standby's own |
| 43 | // empty registry. `Fingerprint` carries registry.SnapshotFingerprint (ALL |
| 44 | // columns), not registry.Fingerprint (edge-reload columns only): it is only |
| 45 | // ever compared against itself — the loop seeds lastFP from it and skips an |
| 46 | // unchanged registry — so no reader outside this file depends on which |
| 47 | // digest it is. |
| 48 | type snapshotMeta struct { |
| 49 | V int `json:"v"` |
| 50 | Box string `json:"box"` |
| 51 | Fingerprint string `json:"fingerprint"` |
| 52 | CreatedAt time.Time `json:"created_at"` |
| 53 | Bytes int `json:"bytes"` |
| 54 | } |
| 55 | |
| 56 | // registrySnapshotPass snapshots the registry into the bucket when its |
| 57 | // snapshot fingerprint moved since *lastFP. One pass; the loop owns cadence. |
| 58 | // *lastFP is updated only after every write (including pruning) succeeded, so |
| 59 | // a partial failure is retried whole next tick. |
| 60 | // |
| 61 | // The gate is SnapshotFingerprint, NOT Fingerprint: Fingerprint deliberately |
| 62 | // omits caps, unit, replication, environment, account, canary and |
| 63 | // provisioned, which are precisely the box-scoped columns only this snapshot |
| 64 | // can rebuild — gating on it meant a caps grant or a console unit stamp never |
| 65 | // produced a snapshot at all. |
| 66 | func registrySnapshotPass(ctx context.Context, s release.Store, reg *registry.Registry, box string, now time.Time, lastFP *string) error { |
| 67 | fp, err := reg.SnapshotFingerprint() |
| 68 | if err != nil { |
| 69 | return fmt.Errorf("snapshot: fingerprint: %w", err) |
| 70 | } |
| 71 | if fp == *lastFP { |
| 72 | return nil |
| 73 | } |
| 74 | |
| 75 | dir, err := os.MkdirTemp("", "carlos-registry-snapshot") |
| 76 | if err != nil { |
| 77 | return fmt.Errorf("snapshot: %w", err) |
| 78 | } |
| 79 | defer os.RemoveAll(dir) |
| 80 | file := filepath.Join(dir, "registry.db") |
| 81 | if err := reg.SnapshotTo(file); err != nil { |
| 82 | return fmt.Errorf("snapshot: vacuum: %w", err) |
| 83 | } |
| 84 | body, err := os.ReadFile(file) |
| 85 | if err != nil { |
| 86 | return fmt.Errorf("snapshot: %w", err) |
| 87 | } |
| 88 | |
| 89 | // Generation first (create-only — a same-second re-run or a clock |
| 90 | // step-back can collide on the stamp, in which case Put's ErrCASConflict |
| 91 | // is deliberately swallowed and the retained generation is KEPT rather |
| 92 | // than replaced). Be precise about what that costs: after a clock |
| 93 | // step-back WITH a registry change in between, the retained generation |
| 94 | // holds OLDER content while the current key and its sidecar hold the |
| 95 | // newer — the two are not equivalent, and we are choosing to keep the |
| 96 | // older one under that stamp. What is lost is one intermediate rung of |
| 97 | // the generation tail, never the restore target itself: current + sidecar |
| 98 | // are written unconditionally below, so a restore still gets the newest |
| 99 | // registry. Retrying under a bumped stamp would buy a slightly denser |
| 100 | // tail at the cost of a write loop against a lying clock. |
| 101 | // |
| 102 | // Then current + sidecar (replace with the CAS-retry recipe |
| 103 | // usage.UpsertEdgeDay uses), then prune. |
| 104 | genKey := snapshotGenPrefix(box) + now.UTC().Format(snapshotStamp) + ".db" |
| 105 | if _, err := s.Put(ctx, genKey, body, ""); err != nil && !errors.Is(err, release.ErrCASConflict) { |
| 106 | return fmt.Errorf("snapshot: put %s: %w", genKey, err) |
| 107 | } |
| 108 | // This guard predates per-box keys and is retained for the shared-key |
| 109 | // era and for hand-placed sidecars only (a sidecar copied or edited by |
| 110 | // an operator with a Box field that does not match this box). Under |
| 111 | // per-box keys it reads ONLY this box's own key, and the only writer |
| 112 | // that ever lands there stamps meta.Box = box — so prior.Box != box |
| 113 | // cannot occur through normal operation, INCLUDING the same-name |
| 114 | // standby-restore case the runbook worries about (a rebuilt standby |
| 115 | // reusing the dead box's machine-id before restoring: same name means |
| 116 | // prior.Box already equals box, so this stays silent). That clobber — |
| 117 | // the freshly bootstrapped standby's agent starting and writing its own |
| 118 | // empty registry over the dead box's last-good snapshot before the |
| 119 | // operator restores it — is NOT detected here; the DR runbook's Step 0 |
| 120 | // ordering rule (restore before the agent is allowed to snapshot) owns |
| 121 | // that hazard instead. |
| 122 | if prior := readSnapshotMeta(ctx, s, box); prior != nil && prior.Box != "" && prior.Box != box { |
| 123 | log.Printf("agent: registry snapshot: OVERWRITING a snapshot written by box %q (this box is %q, prior created_at %s) — if this box is a standby that has not been restored yet, STOP and preserve the prior snapshot", |
| 124 | prior.Box, box, prior.CreatedAt.Format(time.RFC3339)) |
| 125 | } |
| 126 | if err := putReplacing(ctx, s, snapshotDBKey(box), body); err != nil { |
| 127 | return err |
| 128 | } |
| 129 | meta := snapshotMeta{V: 1, Box: box, Fingerprint: fp, CreatedAt: now.UTC(), Bytes: len(body)} |
| 130 | metaRaw, err := json.Marshal(meta) |
| 131 | if err != nil { |
| 132 | return fmt.Errorf("snapshot: %w", err) |
| 133 | } |
| 134 | if err := putReplacing(ctx, s, snapshotMetaKey(box), metaRaw); err != nil { |
| 135 | return err |
| 136 | } |
| 137 | |
| 138 | gens, err := s.List(ctx, snapshotGenPrefix(box)) |
| 139 | if err != nil { |
| 140 | return fmt.Errorf("snapshot: list generations: %w", err) |
| 141 | } |
| 142 | sort.Strings(gens) // stamps are lexically chronological |
| 143 | for len(gens) > snapshotKeep { |
| 144 | if err := s.Delete(ctx, gens[0]); err != nil { |
| 145 | return fmt.Errorf("snapshot: prune %s: %w", gens[0], err) |
| 146 | } |
| 147 | gens = gens[1:] |
| 148 | } |
| 149 | |
| 150 | *lastFP = fp |
| 151 | return nil |
| 152 | } |
| 153 | |
| 154 | // seedSnapshotFingerprint reads the sidecar a previous run (or a previous |
| 155 | // process on this same box) left at snapshotMetaKey(box) and returns its |
| 156 | // fingerprint, so a freshly started loop's in-memory lastFP starts where the |
| 157 | // bucket already is rather than empty. Without this, a crash-looping agent |
| 158 | // would re-snapshot an unchanged registry on every restart and burn through |
| 159 | // all snapshotKeep generations with identical copies. Any read or parse |
| 160 | // failure — no sidecar yet, a corrupt object, a store that is down — returns |
| 161 | // "": the loop then snapshots on its first tick, the same behavior a box |
| 162 | // that has never snapshotted before gets. This must never fail the loop. |
| 163 | func seedSnapshotFingerprint(ctx context.Context, s release.Store, box string) string { |
| 164 | meta := readSnapshotMeta(ctx, s, box) |
| 165 | if meta == nil { |
| 166 | return "" |
| 167 | } |
| 168 | return meta.Fingerprint |
| 169 | } |
| 170 | |
| 171 | // readSnapshotMeta returns the sidecar currently in the bucket at this box's |
| 172 | // own key, or nil when there is none, it cannot be read, or it does not |
| 173 | // parse. Advisory only: both callers treat nil as "no prior snapshot" and |
| 174 | // carry on, so a store outage must never fail a pass. |
| 175 | func readSnapshotMeta(ctx context.Context, s release.Store, box string) *snapshotMeta { |
| 176 | raw, _, err := s.Get(ctx, snapshotMetaKey(box)) |
| 177 | if err != nil { |
| 178 | return nil |
| 179 | } |
| 180 | var meta snapshotMeta |
| 181 | if err := json.Unmarshal(raw, &meta); err != nil { |
| 182 | return nil |
| 183 | } |
| 184 | return &meta |
| 185 | } |
| 186 | |
| 187 | // registrySnapshotLoop snapshots the registry to the deployment bucket every |
| 188 | // 5 minutes + jitter when it changed — rung 1 of the HA ladder (cold |
| 189 | // standby restores this file). Platform-box job, retentionLoop's rules: |
| 190 | // skipped on fleet boxes, disabled with a log line when there is no bucket. |
| 191 | // Per-box keys (spec 2026-08-20 Part 1, closes #158): each platform box |
| 192 | // writes under its own name, so N boxes sharing one deployment bucket never |
| 193 | // interleave each other's current/sidecar writes. |
| 194 | func registrySnapshotLoop(reg *registry.Registry) { |
| 195 | ctx := context.Background() |
| 196 | if shouldSkipPublisherElection(os.Getenv) { |
| 197 | log.Printf("agent: registry snapshot loop is a platform-box job — skipped on a fleet box") |
| 198 | return |
| 199 | } |
| 200 | store, err := openDeployStore(ctx) |
| 201 | if err != nil { |
| 202 | log.Printf("agent: registry snapshot loop disabled: %v", err) |
| 203 | return |
| 204 | } |
| 205 | box := metricsBoxLabel(os.Getenv, readMachineID()) |
| 206 | log.Printf("agent: registry snapshot loop armed (5min + jitter, %s)", snapshotDBKey(box)) |
| 207 | lastFP := seedSnapshotFingerprint(ctx, store, box) |
| 208 | delay := time.Minute |
| 209 | for { |
| 210 | time.Sleep(delay) |
| 211 | delay = 5*time.Minute + rand.N(time.Minute) |
| 212 | if err := registrySnapshotPass(ctx, store, reg, box, time.Now(), &lastFP); err != nil { |
| 213 | log.Printf("agent: registry snapshot pass: %v", err) |
| 214 | } |
| 215 | } |
| 216 | } |
| 217 | |
| 218 | // putReplacing writes key regardless of prior state: create-only when |
| 219 | // absent, etag-replace when present, re-reading on a lost race. Three |
| 220 | // attempts, like usage.UpsertEdgeDay. |
| 221 | func putReplacing(ctx context.Context, s release.Store, key string, body []byte) error { |
| 222 | for attempt := 0; attempt < 3; attempt++ { |
| 223 | _, etag, err := s.Get(ctx, key) |
| 224 | switch { |
| 225 | case errors.Is(err, os.ErrNotExist): |
| 226 | etag = "" |
| 227 | case err != nil: |
| 228 | return fmt.Errorf("snapshot: get %s: %w", key, err) |
| 229 | } |
| 230 | if _, err := s.Put(ctx, key, body, etag); err == nil { |
| 231 | return nil |
| 232 | } else if !errors.Is(err, release.ErrCASConflict) { |
| 233 | return fmt.Errorf("snapshot: put %s: %w", key, err) |
| 234 | } |
| 235 | } |
| 236 | return fmt.Errorf("snapshot: %s: lost CAS race three times", key) |
| 237 | } |
| 238 | |
amadan