amadan

carlos / platform Public

Merge pull request #211 from carlosframework/replica-freshness

restore-verify: prove the replica is CURRENT, not merely restorable
Paul Campbell 0ed1ed1214920769b1dda6acbf92bb097d8d1b06 parent b77856aparent 3acdb8b
3 files changed, +333 −7
  • internal/ops/freshness.go +140 −0
  • internal/ops/litestream.go +36 −4
  • internal/ops/ops_test.go +157 −3
diff --git a/internal/ops/freshness.go b/internal/ops/freshness.go
new file mode 100644
index 000000000..4dec0c4b2
--- /dev/null
+++ b/internal/ops/freshness.go
@@ -0,0 +1,140 @@
+package ops
+
+// Is the replica CURRENT, not merely restorable?
+//
+// The drill's original question — "does this replica restore clean?" — has a
+// blind spot that only shows up in the failure it is meant to catch. A
+// replicator that dies leaves behind a replica that is valid SQLite, restores
+// perfectly, and passes `integrity_check` forever, while the database it
+// mirrors moves on without it. The drill stays green; the backup quietly
+// becomes a snapshot of the day the replicator died.
+//
+// This is not hypothetical. On 2026-08-11 the flagship's litestream exited
+// with "permission denied" writing its shadow WAL into a DynamicUser tenant's
+// state directory (a root drop-in fixed it the same minute). Had nobody been
+// watching the journal, nothing in the platform would have said a word: the
+// next morning's drill would have restored four frozen replicas and reported
+// them clean.
+//
+// The signal used here is SQLite's own: the file-change counter in the
+// database header, which advances on every write transaction that reaches the
+// main database file. Compare the source's counter with the restored copy's,
+// and a replica that has stopped tracking its source shows up as a number that
+// stopped moving. No S3 layout knowledge, no litestream version dependency,
+// no clock skew between machines — the comparison is between two files this
+// process just read.
+
+import (
+ "encoding/binary"
+ "fmt"
+ "io"
+ "os"
+ "time"
+)
+
+// staleGrace is how long a source must have been QUIET before a replica that
+// trails it is called stale. It exists to make false alarms impossible in the
+// ordinary case: a database written seconds before the drill ran is expected
+// to be a transaction or two ahead of its replica, and reporting that as a
+// fault would put the drill permanently red on the busiest — most valuable —
+// databases on the box, which is precisely how a check stops being read.
+//
+// 15 minutes is far longer than any replication lag this fleet has (litestream
+// syncs on change, seconds at most) and far shorter than the gap that matters
+// (a replicator that died).
+const staleGrace = 15 * time.Minute
+
+type freshness int
+
+const (
+ // freshnessOK: the replica is level with its source, or trailing a source
+ // that was written too recently to judge.
+ freshnessOK freshness = iota
+ // freshnessStale: the source has been quiet for longer than staleGrace and
+ // the replica is STILL behind it. Replication is not happening.
+ freshnessStale
+ // freshnessNoSource: the config names a database that is not on disk. The
+ // replica cannot ever advance again, and nothing is watching the gap.
+ freshnessNoSource
+ // freshnessUnknown: one of the two headers could not be read. Say so;
+ // never let "could not tell" read as "fine".
+ freshnessUnknown
+)
+
+func (f freshness) String() string {
+ switch f {
+ case freshnessOK:
+ return "ok"
+ case freshnessStale:
+ return "STALE"
+ case freshnessNoSource:
+ return "no-source"
+ default:
+ return "unknown"
+ }
+}
+
+// ChangeCounter reads the SQLite file-change counter: bytes 24-27 of the
+// database header, big-endian (see the SQLite file-format spec, "Database
+// Header"). SQLite increments it on every write transaction that modifies the
+// main database file.
+//
+// In WAL mode the counter in the main file only advances at a checkpoint, so
+// a source with a hot WAL reads LOWER than reality. That skew is deliberately
+// harmless here: it can only make a stale replica look current (a missed
+// alarm), never make a current replica look stale (a false one). Given a
+// choice, a durability check should under-claim.
+func ChangeCounter(path string) (uint32, error) {
+ f, err := os.Open(path)
+ if err != nil {
+ return 0, err
+ }
+ defer f.Close()
+
+ var header [28]byte
+ if _, err := io.ReadFull(f, header[:]); err != nil {
+ return 0, fmt.Errorf("read header: %w", err)
+ }
+ if string(header[:15]) != "SQLite format 3" {
+ return 0, fmt.Errorf("not a SQLite database")
+ }
+ return binary.BigEndian.Uint32(header[24:28]), nil
+}
+
+// replicaFreshness compares a restored replica against the live database it
+// mirrors. srcModTime is the source's last-modified time, passed in so the
+// caller stats once and tests can drive the grace window.
+func replicaFreshness(srcPath, restoredPath string, srcModTime time.Time) (freshness, string) {
+ srcCounter, err := ChangeCounter(srcPath)
+ if os.IsNotExist(err) {
+ return freshnessNoSource, "source database is not on disk — this replica can never advance again"
+ }
+ if err != nil {
+ return freshnessUnknown, fmt.Sprintf("source header unreadable: %v", err)
+ }
+ restoredCounter, err := ChangeCounter(restoredPath)
+ if err != nil {
+ return freshnessUnknown, fmt.Sprintf("restored header unreadable: %v", err)
+ }
+ if restoredCounter >= srcCounter {
+ return freshnessOK, ""
+ }
+ if quiet := time.Since(srcModTime); quiet <= staleGrace {
+ // Behind, but the source was written within the grace window — this is
+ // ordinary lag, and the next pass will judge it properly.
+ return freshnessOK, ""
+ }
+ return freshnessStale, fmt.Sprintf(
+ "replica is at change %d, the live database at %d, and the database has been quiet since %s — replication has stopped",
+ restoredCounter, srcCounter, srcModTime.UTC().Format(time.RFC3339))
+}
+
+// sourceModTime is the source database's mtime, or the zero time when it
+// cannot be stated (which replicaFreshness will treat through its own paths).
+func sourceModTime(path string) time.Time {
+ fi, err := os.Stat(path)
+ if err != nil {
+ return time.Time{}
+ }
+ return fi.ModTime()
+}
diff --git a/internal/ops/litestream.go b/internal/ops/litestream.go
index 718b42938..a3e7baeeb 100644
--- a/internal/ops/litestream.go
+++ b/internal/ops/litestream.go
@@ -291,7 +291,7 @@ func RestoreVerify(litestreamBin, cfgPath, tmpDir string) error {
return fmt.Errorf("restore-verify: no databases in %s — regenerate it: carlos ops litestream", cfgPath)
}
- failures := 0
+ failures, stale := 0, 0
for _, db := range dbs {
// Sweep this database's copy however the check ends — a failure must
// not leave the wreckage behind for the next restore to be charged for.
@@ -304,18 +304,50 @@ func RestoreVerify(litestreamBin, cfgPath, tmpDir string) error {
time.Sleep(retryDelay)
err = restoreAndCheckFn(litestreamBin, cfgPath, db, tmpDir)
}
+ // Freshness is judged BEFORE the copy is swept, and only once the
+ // restore itself passed: "restores clean but stopped tracking its
+ // source" is a different finding from "does not restore", and
+ // conflating them would hide whichever came second.
+ var fresh freshness
+ var freshDetail string
+ if err == nil {
+ fresh, freshDetail = replicaFreshness(db, out, sourceModTime(db))
+ }
removeRestored(out) // every path, success included
if err != nil {
failures++
fmt.Printf("FAIL %s: %v\n", db, err)
continue
}
- fmt.Printf("ok %s\n", db)
+ switch fresh {
+ case freshnessStale:
+ stale++
+ fmt.Printf("STALE %s: %s\n", db, freshDetail)
+ case freshnessNoSource:
+ stale++
+ fmt.Printf("STALE %s: %s\n", db, freshDetail)
+ case freshnessUnknown:
+ // Never a pass and never an alarm — see the freshness doc.
+ fmt.Printf("ok %s (freshness unknown: %s)\n", db, freshDetail)
+ default:
+ fmt.Printf("ok %s\n", db)
+ }
}
- if failures > 0 {
+ switch {
+ case failures > 0 && stale > 0:
+ return fmt.Errorf("restore-verify: %d of %d replicas failed to restore clean, and %d more restore but have stopped tracking their source",
+ failures, len(dbs), stale)
+ case failures > 0:
return fmt.Errorf("restore-verify: %d of %d replicas failed to restore clean", failures, len(dbs))
+ case stale > 0:
+ // A stale replica restores perfectly and is still a durability
+ // failure: what comes back is the database as it was when replication
+ // stopped. It fails the run for the same reason an unrestorable one
+ // does.
+ return fmt.Errorf("restore-verify: %d of %d replicas restore clean but have STOPPED TRACKING their source — replication is broken for them",
+ stale, len(dbs))
}
- fmt.Printf("restore-verify: all %d replicas restore clean\n", len(dbs))
+ fmt.Printf("restore-verify: all %d replicas restore clean and are current\n", len(dbs))
return nil
}
diff --git a/internal/ops/ops_test.go b/internal/ops/ops_test.go
index 27babd0c8..1e1d059d0 100644
--- a/internal/ops/ops_test.go
+++ b/internal/ops/ops_test.go
@@ -349,8 +349,16 @@ func TestSnapshotPolicySupported(t *testing.T) {
// at the LARGEST SINGLE database instead of their sum.
func TestRestoreVerifyHoldsOneRestoreAtATime(t *testing.T) {
tmp := t.TempDir()
+ srcDir := t.TempDir()
cfgPath := filepath.Join(t.TempDir(), "litestream.yml")
- dbs := []string{"/data/a.db", "/data/b.db", "/data/c.db"}
+ // Real source databases: the freshness check reads their headers, and a
+ // config naming files that do not exist is itself a finding now.
+ var dbs []string
+ for _, name := range []string{"a.db", "b.db", "c.db"} {
+ p := filepath.Join(srcDir, name)
+ writeCounterDB(t, p, 3)
+ dbs = append(dbs, p)
+ }
if err := os.WriteFile(cfgPath, []byte(LitestreamConfig(dbs, "b", "p", "eu-west-1")), 0o644); err != nil {
t.Fatal(err)
}
@@ -358,8 +366,13 @@ func TestRestoreVerifyHoldsOneRestoreAtATime(t *testing.T) {
maxSeen := 0
prev := restoreAndCheckFn
restoreAndCheckFn = func(_, _, dbPath, tmpDir string) error {
- // Stand in for a restore: leave a file behind, like the real one does.
- if err := os.WriteFile(filepath.Join(tmpDir, filepath.Base(dbPath)), []byte("restored"), 0o600); err != nil {
+ // Stand in for a restore of a CURRENT replica: a byte copy of the
+ // source, so its change counter matches and freshness passes.
+ body, err := os.ReadFile(dbPath)
+ if err != nil {
+ return err
+ }
+ if err := os.WriteFile(filepath.Join(tmpDir, filepath.Base(dbPath)), body, 0o600); err != nil {
return err
}
ents, err := os.ReadDir(tmpDir)
@@ -387,3 +400,144 @@ func TestRestoreVerifyHoldsOneRestoreAtATime(t *testing.T) {
t.Fatalf("RestoreVerify left %d file(s) behind", len(left))
}
}
+
+// --- replica freshness ---------------------------------------------------
+//
+// The drill proves a replica RESTORES. It has never proved the replica is
+// CURRENT — and those come apart exactly when it matters. On 2026-08-11 the
+// flagship's litestream died with "permission denied" writing its shadow WAL
+// into a DynamicUser tenant's state dir; had nobody noticed by hand, the daily
+// drill would have gone on passing indefinitely, restoring a replica frozen at
+// the moment of death. A valid backup of last week is not a backup.
+
+func writeCounterDB(t *testing.T, path string, writes int) {
+ t.Helper()
+ db, err := sql.Open("sqlite", path)
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer db.Close()
+ // DELETE journalling: the header change counter advances per commit
+ // without needing a checkpoint, which keeps the fixture honest about what
+ // it is testing.
+ if _, err := db.Exec("PRAGMA journal_mode=DELETE"); err != nil {
+ t.Fatal(err)
+ }
+ if _, err := db.Exec("CREATE TABLE IF NOT EXISTS t (id INTEGER PRIMARY KEY)"); err != nil {
+ t.Fatal(err)
+ }
+ for i := 0; i < writes; i++ {
+ // A fresh rowid each call, so re-seeding an existing file adds
+ // transactions rather than colliding on the primary key.
+ if _, err := db.Exec("INSERT INTO t (id) VALUES (NULL)"); err != nil {
+ t.Fatal(err)
+ }
+ }
+}
+
+func TestChangeCounterAdvancesWithWrites(t *testing.T) {
+ dir := t.TempDir()
+ a := filepath.Join(dir, "a.db")
+ writeCounterDB(t, a, 1)
+ first, err := ChangeCounter(a)
+ if err != nil {
+ t.Fatal(err)
+ }
+ writeCounterDB(t, a, 5)
+ second, err := ChangeCounter(a)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if second <= first {
+ t.Fatalf("change counter did not advance across writes: %d then %d", first, second)
+ }
+
+ junk := filepath.Join(dir, "junk.db")
+ if err := os.WriteFile(junk, []byte("nope"), 0o600); err != nil {
+ t.Fatal(err)
+ }
+ if _, err := ChangeCounter(junk); err == nil {
+ t.Fatal("ChangeCounter accepted a file that is not a SQLite database")
+ }
+}
+
+func TestReplicaFreshness(t *testing.T) {
+ dir := t.TempDir()
+ local := filepath.Join(dir, "local.db")
+ behind := filepath.Join(dir, "behind.db")
+ writeCounterDB(t, behind, 1) // the replica, frozen early
+ writeCounterDB(t, local, 9) // the source, written since
+
+ long := time.Now().Add(-2 * time.Hour)
+ recent := time.Now()
+
+ // Quiet since well before the drill, replica still behind: the replicator
+ // has had every chance to catch up and has not. That is the alarm.
+ if v, detail := replicaFreshness(local, behind, long); v != freshnessStale {
+ t.Fatalf("a replica behind a long-quiet source = %v (%s), want stale", v, detail)
+ }
+ // Written moments ago: the replica being a transaction or two behind is
+ // ordinary lag, not a fault. Never cry wolf here — a drill that goes red
+ // on healthy churn is a drill people stop reading.
+ if v, _ := replicaFreshness(local, behind, recent); v != freshnessOK {
+ t.Fatalf("a replica behind a just-written source = %v, want ok (still catching up)", v)
+ }
+ // Caught up.
+ if v, _ := replicaFreshness(local, local, long); v != freshnessOK {
+ t.Fatalf("an up-to-date replica = %v, want ok", v)
+ }
+ // The source is gone: a config naming a database that no longer exists is
+ // its own finding — the replica can never advance again.
+ if v, _ := replicaFreshness(filepath.Join(dir, "absent.db"), behind, long); v != freshnessNoSource {
+ t.Fatalf("a missing source = %v, want no-source", v)
+ }
+ // Unreadable header: say nothing rather than invent a verdict.
+ junk := filepath.Join(dir, "junk.db")
+ if err := os.WriteFile(junk, []byte("nope"), 0o600); err != nil {
+ t.Fatal(err)
+ }
+ if v, _ := replicaFreshness(junk, behind, long); v != freshnessUnknown {
+ t.Fatalf("an unreadable source = %v, want unknown", v)
+ }
+}
+
+// The whole point, end to end: a replica that restores perfectly but has
+// stopped tracking its source must FAIL the run. Before this, such a replica
+// was reported "ok" every night for as long as the replicator stayed dead.
+func TestRestoreVerifyFailsOnAStaleReplica(t *testing.T) {
+ srcDir, replicaDir := t.TempDir(), t.TempDir()
+ cfgPath := filepath.Join(t.TempDir(), "litestream.yml")
+
+ live := filepath.Join(srcDir, "live.db")
+ writeCounterDB(t, live, 2)
+ // Snapshot the replica HERE — this stands for the moment the replicator
+ // died — then let the source move on without it.
+ frozen, err := os.ReadFile(live)
+ if err != nil {
+ t.Fatal(err)
+ }
+ writeCounterDB(t, live, 20)
+ // And make the source quiet for longer than the grace window, so this is
+ // unambiguously "replication stopped" rather than "still catching up".
+ old := time.Now().Add(-2 * time.Hour)
+ if err := os.Chtimes(live, old, old); err != nil {
+ t.Fatal(err)
+ }
+
+ if err := os.WriteFile(cfgPath, []byte(LitestreamConfig([]string{live}, "b", "p", "eu-west-1")), 0o644); err != nil {
+ t.Fatal(err)
+ }
+ prev := restoreAndCheckFn
+ restoreAndCheckFn = func(_, _, dbPath, tmpDir string) error {
+ return os.WriteFile(filepath.Join(tmpDir, filepath.Base(dbPath)), frozen, 0o600)
+ }
+ t.Cleanup(func() { restoreAndCheckFn = prev })
+
+ err = RestoreVerify("litestream", cfgPath, replicaDir)
+ if err == nil {
+ t.Fatal("a replica frozen at the moment its replicator died passed the drill")
+ }
+ if !strings.Contains(err.Error(), "STOPPED TRACKING") {
+ t.Fatalf("error does not name the failure: %v", err)
+ }
+}