| index 0d199f0ba..486ca01df 100644 |
| --- a/internal/host/snapshot_loop.go |
| +++ b/internal/host/snapshot_loop.go |
| @@ -5,6 +5,8 @@ import ( |
| "encoding/json" |
| "errors" |
| "fmt" |
| + "log" |
| + "math/rand/v2" |
| "os" |
| "path/filepath" |
| "sort" |
| @@ -49,7 +51,7 @@ func registrySnapshotPass(ctx context.Context, s release.Store, reg *registry.Re |
| |
| dir, err := os.MkdirTemp("", "carlos-registry-snapshot") |
| if err != nil { |
| - return err |
| + return fmt.Errorf("snapshot: %w", err) |
| } |
| defer os.RemoveAll(dir) |
| file := filepath.Join(dir, "registry.db") |
| @@ -58,10 +60,13 @@ func registrySnapshotPass(ctx context.Context, s release.Store, reg *registry.Re |
| } |
| body, err := os.ReadFile(file) |
| if err != nil { |
| - return err |
| + return fmt.Errorf("snapshot: %w", err) |
| } |
| |
| - // Generation first (create-only — a stamp collision is a real conflict), |
| + // Generation first (create-only — a same-second re-run or a clock |
| + // step-back can collide on the stamp, in which case Put's ErrCASConflict |
| + // is deliberately swallowed: the existing generation for that second is |
| + // as good as the one we'd have written, so there is nothing to retry), |
| // then current + sidecar (replace with the CAS-retry recipe |
| // usage.UpsertEdgeDay uses), then prune. |
| genKey := snapshotGenPrefix + now.UTC().Format(snapshotStamp) + ".db" |
| @@ -74,7 +79,7 @@ func registrySnapshotPass(ctx context.Context, s release.Store, reg *registry.Re |
| meta := snapshotMeta{V: 1, Box: box, Fingerprint: fp, CreatedAt: now.UTC(), Bytes: len(body)} |
| metaRaw, err := json.Marshal(meta) |
| if err != nil { |
| - return err |
| + return fmt.Errorf("snapshot: %w", err) |
| } |
| if err := putReplacing(ctx, s, snapshotMetaKey, metaRaw); err != nil { |
| return err |
| @@ -96,6 +101,58 @@ func registrySnapshotPass(ctx context.Context, s release.Store, reg *registry.Re |
| return nil |
| } |
| |
| +// seedSnapshotFingerprint reads the sidecar a previous run (or a previous |
| +// process on this same box) left at snapshotMetaKey and returns its |
| +// fingerprint, so a freshly started loop's in-memory lastFP starts where the |
| +// bucket already is rather than empty. Without this, a crash-looping agent |
| +// would re-snapshot an unchanged registry on every restart and burn through |
| +// all snapshotKeep generations with identical copies. Any read or parse |
| +// failure — no sidecar yet, a corrupt object, a store that is down — returns |
| +// "": the loop then snapshots on its first tick, the same behavior a box |
| +// that has never snapshotted before gets. This must never fail the loop. |
| +func seedSnapshotFingerprint(ctx context.Context, s release.Store) string { |
| + raw, _, err := s.Get(ctx, snapshotMetaKey) |
| + if err != nil { |
| + return "" |
| + } |
| + var meta snapshotMeta |
| + if err := json.Unmarshal(raw, &meta); err != nil { |
| + return "" |
| + } |
| + return meta.Fingerprint |
| +} |
| + |
| +// registrySnapshotLoop snapshots the registry to the deployment bucket every |
| +// 5 minutes + jitter when it changed — rung 1 of the HA ladder (cold |
| +// standby restores this file). Platform-box job, retentionLoop's rules: |
| +// skipped on fleet boxes, disabled with a log line when there is no bucket. |
| +// The fixed control/ keys this writes assume one platform box per deployment |
| +// bucket (true for every deployment today); a second concurrent writer would |
| +// interleave current/sidecar writes from two registries. |
| +func registrySnapshotLoop(reg *registry.Registry) { |
| + ctx := context.Background() |
| + if shouldSkipPublisherElection(os.Getenv) { |
| + log.Printf("agent: registry snapshot loop is a platform-box job — skipped on a fleet box") |
| + return |
| + } |
| + store, err := openDeployStore(ctx) |
| + if err != nil { |
| + log.Printf("agent: registry snapshot loop disabled: %v", err) |
| + return |
| + } |
| + box := metricsBoxLabel(os.Getenv, readMachineID()) |
| + log.Printf("agent: registry snapshot loop armed (5min + jitter, %s)", snapshotKey) |
| + lastFP := seedSnapshotFingerprint(ctx, store) |
| + delay := time.Minute |
| + for { |
| + time.Sleep(delay) |
| + delay = 5*time.Minute + rand.N(time.Minute) |
| + if err := registrySnapshotPass(ctx, store, reg, box, time.Now(), &lastFP); err != nil { |
| + log.Printf("agent: registry snapshot pass: %v", err) |
| + } |
| + } |
| +} |
| + |
| // putReplacing writes key regardless of prior state: create-only when |
| // absent, etag-replace when present, re-reading on a lost race. Three |
| // attempts, like usage.UpsertEdgeDay. |
amadan