amadan

carlos / platform Public

host: registry snapshot pass — current + sidecar + pruned generations under control/

Paul Campbell 89229f23a283c493008e54d0b6d498f2dd3ed863 parent b09e973
2 files changed, +252 −0
  • internal/host/snapshot_loop.go +118 −0
  • internal/host/snapshot_loop_test.go +134 −0
diff --git a/internal/host/snapshot_loop.go b/internal/host/snapshot_loop.go
new file mode 100644
index 000000000..0d199f0ba
--- /dev/null
+++ b/internal/host/snapshot_loop.go
@@ -0,0 +1,118 @@
+package host
+
+import (
+ "context"
+ "encoding/json"
+ "errors"
+ "fmt"
+ "os"
+ "path/filepath"
+ "sort"
+ "time"
+
+ "github.com/carlosframework/platform/internal/registry"
+ "github.com/carlosframework/platform/internal/release"
+)
+
+// The cold-standby spec (2026-08-12 HA posture, Part I rung 1): the current
+// snapshot at a fixed key a runbook can name, a sidecar describing it, and a
+// short tail of prior generations. All under control/ beside control/serial;
+// platform boxes write direct S3, so no store-door ACL is involved.
+const (
+ snapshotKey = "control/registry-snapshot.db"
+ snapshotMetaKey = "control/registry-snapshot.json"
+ snapshotGenPrefix = "control/registry-snapshots/"
+ snapshotKeep = 8
+ snapshotStamp = "20060102T150405Z"
+)
+
+type snapshotMeta struct {
+ V int `json:"v"`
+ Box string `json:"box"`
+ Fingerprint string `json:"fingerprint"`
+ CreatedAt time.Time `json:"created_at"`
+ Bytes int `json:"bytes"`
+}
+
+// registrySnapshotPass snapshots the registry into the bucket when its
+// fingerprint moved since *lastFP. One pass; the loop owns cadence. *lastFP
+// is updated only after every write (including pruning) succeeded, so a
+// partial failure is retried whole next tick.
+func registrySnapshotPass(ctx context.Context, s release.Store, reg *registry.Registry, box string, now time.Time, lastFP *string) error {
+ fp, err := reg.Fingerprint()
+ if err != nil {
+ return fmt.Errorf("snapshot: fingerprint: %w", err)
+ }
+ if fp == *lastFP {
+ return nil
+ }
+
+ dir, err := os.MkdirTemp("", "carlos-registry-snapshot")
+ if err != nil {
+ return err
+ }
+ defer os.RemoveAll(dir)
+ file := filepath.Join(dir, "registry.db")
+ if err := reg.SnapshotTo(file); err != nil {
+ return fmt.Errorf("snapshot: vacuum: %w", err)
+ }
+ body, err := os.ReadFile(file)
+ if err != nil {
+ return err
+ }
+
+ // Generation first (create-only — a stamp collision is a real conflict),
+ // then current + sidecar (replace with the CAS-retry recipe
+ // usage.UpsertEdgeDay uses), then prune.
+ genKey := snapshotGenPrefix + now.UTC().Format(snapshotStamp) + ".db"
+ if _, err := s.Put(ctx, genKey, body, ""); err != nil && !errors.Is(err, release.ErrCASConflict) {
+ return fmt.Errorf("snapshot: put %s: %w", genKey, err)
+ }
+ if err := putReplacing(ctx, s, snapshotKey, body); err != nil {
+ return err
+ }
+ meta := snapshotMeta{V: 1, Box: box, Fingerprint: fp, CreatedAt: now.UTC(), Bytes: len(body)}
+ metaRaw, err := json.Marshal(meta)
+ if err != nil {
+ return err
+ }
+ if err := putReplacing(ctx, s, snapshotMetaKey, metaRaw); err != nil {
+ return err
+ }
+
+ gens, err := s.List(ctx, snapshotGenPrefix)
+ if err != nil {
+ return fmt.Errorf("snapshot: list generations: %w", err)
+ }
+ sort.Strings(gens) // stamps are lexically chronological
+ for len(gens) > snapshotKeep {
+ if err := s.Delete(ctx, gens[0]); err != nil {
+ return fmt.Errorf("snapshot: prune %s: %w", gens[0], err)
+ }
+ gens = gens[1:]
+ }
+
+ *lastFP = fp
+ return nil
+}
+
+// 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.
+func putReplacing(ctx context.Context, s release.Store, key string, body []byte) error {
+ for attempt := 0; attempt < 3; attempt++ {
+ _, etag, err := s.Get(ctx, key)
+ switch {
+ case errors.Is(err, os.ErrNotExist):
+ etag = ""
+ case err != nil:
+ return fmt.Errorf("snapshot: get %s: %w", key, err)
+ }
+ if _, err := s.Put(ctx, key, body, etag); err == nil {
+ return nil
+ } else if !errors.Is(err, release.ErrCASConflict) {
+ return fmt.Errorf("snapshot: put %s: %w", key, err)
+ }
+ }
+ return fmt.Errorf("snapshot: %s: lost CAS race three times", key)
+}
diff --git a/internal/host/snapshot_loop_test.go b/internal/host/snapshot_loop_test.go
new file mode 100644
index 000000000..91764e35e
--- /dev/null
+++ b/internal/host/snapshot_loop_test.go
@@ -0,0 +1,134 @@
+package host
+
+import (
+ "context"
+ "encoding/json"
+ "path/filepath"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/registry"
+ "github.com/carlosframework/platform/internal/release"
+)
+
+func snapTestRegistry(t *testing.T) *registry.Registry {
+ t.Helper()
+ reg, err := registry.Open(filepath.Join(t.TempDir(), "registry.db"))
+ if err != nil {
+ t.Fatal(err)
+ }
+ t.Cleanup(func() { reg.Close() })
+ if err := reg.Upsert(registry.Route{Host: "a.example.com", Kind: "service", Addr: "10.0.0.1:3000", Status: "ready"}); err != nil {
+ t.Fatal(err)
+ }
+ return reg
+}
+
+func TestSnapshotPassWritesCurrentSidecarAndGeneration(t *testing.T) {
+ s := release.NewDirStore(t.TempDir())
+ ctx := context.Background()
+ reg := snapTestRegistry(t)
+ at := time.Date(2026, 8, 12, 3, 0, 0, 0, time.UTC)
+
+ var lastFP string
+ if err := registrySnapshotPass(ctx, s, reg, "box1", at, &lastFP); err != nil {
+ t.Fatal(err)
+ }
+
+ raw, _, err := s.Get(ctx, snapshotKey)
+ if err != nil {
+ t.Fatalf("current snapshot missing: %v", err)
+ }
+ if len(raw) == 0 {
+ t.Fatal("current snapshot is empty")
+ }
+
+ metaRaw, _, err := s.Get(ctx, snapshotMetaKey)
+ if err != nil {
+ t.Fatalf("sidecar missing: %v", err)
+ }
+ var meta snapshotMeta
+ if err := json.Unmarshal(metaRaw, &meta); err != nil {
+ t.Fatal(err)
+ }
+ if meta.V != 1 || meta.Box != "box1" || meta.Fingerprint == "" || meta.Bytes != len(raw) {
+ t.Fatalf("sidecar = %+v (snapshot %d bytes)", meta, len(raw))
+ }
+ if lastFP != meta.Fingerprint {
+ t.Fatalf("lastFP %q not updated to %q", lastFP, meta.Fingerprint)
+ }
+
+ gens, err := s.List(ctx, snapshotGenPrefix)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if len(gens) != 1 || gens[0] != snapshotGenPrefix+"20260812T030000Z.db" {
+ t.Fatalf("generations = %v", gens)
+ }
+}
+
+func TestSnapshotPassSkipsWhenFingerprintUnchanged(t *testing.T) {
+ s := release.NewDirStore(t.TempDir())
+ ctx := context.Background()
+ reg := snapTestRegistry(t)
+ at := time.Date(2026, 8, 12, 3, 0, 0, 0, time.UTC)
+
+ var lastFP string
+ if err := registrySnapshotPass(ctx, s, reg, "box1", at, &lastFP); err != nil {
+ t.Fatal(err)
+ }
+ if err := registrySnapshotPass(ctx, s, reg, "box1", at.Add(5*time.Minute), &lastFP); err != nil {
+ t.Fatal(err)
+ }
+ gens, err := s.List(ctx, snapshotGenPrefix)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if len(gens) != 1 {
+ t.Fatalf("unchanged registry grew generations: %v", gens)
+ }
+
+ // A registry change makes the next pass write again.
+ if err := reg.Upsert(registry.Route{Host: "b.example.com", Kind: "service", Addr: "10.0.0.2:3000", Status: "ready"}); err != nil {
+ t.Fatal(err)
+ }
+ if err := registrySnapshotPass(ctx, s, reg, "box1", at.Add(10*time.Minute), &lastFP); err != nil {
+ t.Fatal(err)
+ }
+ gens, err = s.List(ctx, snapshotGenPrefix)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if len(gens) != 2 {
+ t.Fatalf("changed registry did not write a new generation: %v", gens)
+ }
+}
+
+func TestSnapshotPassPrunesGenerationsBeyondKeep(t *testing.T) {
+ s := release.NewDirStore(t.TempDir())
+ ctx := context.Background()
+ reg := snapTestRegistry(t)
+ at := time.Date(2026, 8, 12, 0, 0, 0, 0, time.UTC)
+
+ var lastFP string
+ for i := 0; i < snapshotKeep+3; i++ {
+ // Change the registry each round so no pass skips.
+ if err := reg.Upsert(registry.Route{Host: "a.example.com", Kind: "service", Addr: "10.0.0.1:3000", Status: "ready", Version: string(rune('a' + i))}); err != nil {
+ t.Fatal(err)
+ }
+ if err := registrySnapshotPass(ctx, s, reg, "box1", at.Add(time.Duration(i)*5*time.Minute), &lastFP); err != nil {
+ t.Fatal(err)
+ }
+ }
+ gens, err := s.List(ctx, snapshotGenPrefix)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if len(gens) != snapshotKeep {
+ t.Fatalf("kept %d generations, want %d: %v", len(gens), snapshotKeep, gens)
+ }
+ // The survivors must be the newest ones: the earliest kept stamp is round 3's.
+ if gens[0] != snapshotGenPrefix+"20260812T001500Z.db" {
+ t.Fatalf("oldest kept generation = %q, want the round-3 stamp", gens[0])
+ }
+}