carlos / platform Public

Merge pull request #214 from carlosframework/hibernate-freshness

hibernate-verify: the same currency question as the box-wide drill
Paul Campbell c2466bf64ecac698f4b0012cd18cd758c994ffd3 parent dd04065parent d3dbcd1
6 files changed, +233 −70
  • internal/lease/check.go +43 −4
  • internal/lease/check_test.go +33 −2
  • internal/ops/freshness.go +18 −32
  • internal/ops/hibernate_verify.go +45 −12
  • internal/ops/hibernate_verify_test.go +89 −19
  • internal/ops/ops.go +5 −1
diff --git a/internal/lease/check.go b/internal/lease/check.go
index 9902489f0..8fdac6469 100644
--- a/internal/lease/check.go
+++ b/internal/lease/check.go
@@ -3,8 +3,10 @@ package lease
import (
"context"
"database/sql"
+ "encoding/binary"
"errors"
"fmt"
+ "io"
"os"
"path/filepath"
@@ -74,7 +76,7 @@ func CheckDB(path string) error {
// `carlos ops restore-verify` has always restored databases that its own
// replicator was actively streaming, and litestream restores a committed
// position rather than a live tail.
-func VerifyReplica(ctx context.Context, bucket, prefix, region, dir string) error {
+func VerifyReplica(ctx context.Context, bucket, prefix, region, dir string) (uint32, error) {
out := filepath.Join(dir, "verify-"+filepath.Base(prefix)+".db")
// A retry must not trip over a previous attempt's file: litestream refuses
// to restore over an existing output path.
@@ -91,10 +93,47 @@ func VerifyReplica(ctx context.Context, bucket, prefix, region, dir string) erro
restored, err := restoreForVerify(ctx, out, bucket, prefix, region)
if err != nil {
- return fmt.Errorf("restore: %w", err)
+ return 0, fmt.Errorf("restore: %w", err)
}
if !restored {
- return ErrNoReplica
+ return 0, ErrNoReplica
}
- return CheckDB(out)
+ if err := CheckDB(out); err != nil {
+ return 0, err
+ }
+ // The counter is read BEFORE the deferred sweep removes the copy: it is
+ // what lets the caller ask the second question, whether this replica is
+ // still tracking its source (see ops.replicaFreshness).
+ return ChangeCounter(out)
+}
+
+// ChangeCounter reads the SQLite file-change counter: bytes 24-27 of the
+// database header, big-endian. SQLite increments it on every write transaction
+// that modifies the main database file, which makes it the cheapest honest
+// answer to "is this replica still tracking its source?" — compare the number
+// on the live database with the number on a restored copy.
+//
+// It lives here, beside CheckDB and Restore, because both drills need it: the
+// box-wide one (internal/ops) and the hibernation one, which restores through
+// VerifyReplica.
+//
+// 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: it can only make a stale replica look current (a missed alarm),
+// never a current replica look stale (a false one).
+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
}
diff --git a/internal/lease/check_test.go b/internal/lease/check_test.go
index 851b320b3..65828388d 100644
--- a/internal/lease/check_test.go
+++ b/internal/lease/check_test.go
@@ -160,7 +160,7 @@ func TestVerifyReplicaCleansUpEvenOnFailure(t *testing.T) {
}
t.Cleanup(func() { restoreForVerify = prev })
- if err := VerifyReplica(context.Background(), "bucket", "hib/x.example.com", "", dir); err == nil {
+ if _, err := VerifyReplica(context.Background(), "bucket", "hib/x.example.com", "", dir); err == nil {
t.Fatal("VerifyReplica accepted a corrupt restore")
}
left, err := os.ReadDir(dir)
@@ -185,8 +185,39 @@ func TestVerifyReplicaReportsNoReplica(t *testing.T) {
}
t.Cleanup(func() { restoreForVerify = prev })
- err := VerifyReplica(context.Background(), "bucket", "hib/new.example.com", "", t.TempDir())
+ _, err := VerifyReplica(context.Background(), "bucket", "hib/new.example.com", "", t.TempDir())
if !errors.Is(err, ErrNoReplica) {
t.Fatalf("VerifyReplica on a never-replicated prefix = %v, want ErrNoReplica", err)
}
}
+
+// VerifyReplica hands back the restored copy's change counter, which is what
+// lets the hibernation drill ask the second question — is this replica still
+// TRACKING its source? — after the throwaway copy has been swept.
+func TestVerifyReplicaReturnsTheRestoredChangeCounter(t *testing.T) {
+ dir := t.TempDir()
+ src := filepath.Join(t.TempDir(), "source.db")
+ seedDB(t, src)
+ want, err := ChangeCounter(src)
+ if err != nil {
+ t.Fatal(err)
+ }
+
+ prev := restoreForVerify
+ restoreForVerify = func(_ context.Context, out, _, _, _ string) (bool, error) {
+ body, err := os.ReadFile(src)
+ if err != nil {
+ return false, err
+ }
+ return true, os.WriteFile(out, body, 0o600)
+ }
+ t.Cleanup(func() { restoreForVerify = prev })
+
+ got, err := VerifyReplica(context.Background(), "bucket", "hib/x.example.com", "", dir)
+ if err != nil {
+ t.Fatalf("VerifyReplica: %v", err)
+ }
+ if got != want {
+ t.Fatalf("restored change counter = %d, want %d", got, want)
+ }
+}
diff --git a/internal/ops/freshness.go b/internal/ops/freshness.go
index 4dec0c4b2..48847a959 100644
--- a/internal/ops/freshness.go
+++ b/internal/ops/freshness.go
@@ -25,11 +25,11 @@ package ops
// process just read.
import (
- "encoding/binary"
"fmt"
- "io"
"os"
"time"
+
+ "github.com/carlosframework/platform/internal/lease"
)
// staleGrace is how long a source must have been QUIET before a replica that
@@ -74,37 +74,27 @@ func (f freshness) String() string {
}
}
-// 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
-}
+// ChangeCounter is lease.ChangeCounter — one definition of the freshness
+// signal, shared by both drills. Kept exported here because this package's
+// tests and callers are its documented door.
+func ChangeCounter(path string) (uint32, error) { return lease.ChangeCounter(path) }
// 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) {
+ restoredCounter, err := ChangeCounter(restoredPath)
+ if err != nil {
+ return freshnessUnknown, fmt.Sprintf("restored header unreadable: %v", err)
+ }
+ return replicaFreshnessCounter(srcPath, restoredCounter, srcModTime)
+}
+
+// replicaFreshnessCounter is replicaFreshness for a caller that already holds
+// the restored copy's counter and no longer has the file — which is the
+// hibernation drill, where lease.VerifyReplica reads the counter and then
+// sweeps its throwaway copy in the same call.
+func replicaFreshnessCounter(srcPath string, restoredCounter uint32, 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"
@@ -112,10 +102,6 @@ func replicaFreshness(srcPath, restoredPath string, srcModTime time.Time) (fresh
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, ""
}
diff --git a/internal/ops/hibernate_verify.go b/internal/ops/hibernate_verify.go
index 985adfbcd..a9256c29e 100644
--- a/internal/ops/hibernate_verify.go
+++ b/internal/ops/hibernate_verify.go
@@ -30,15 +30,24 @@ import (
"github.com/carlosframework/platform/internal/registry"
)
+// HibernatingRoute is one instance this drill owns: the host names its replica
+// prefix, and DBPath is the live database the replica is supposed to be
+// tracking — carried because "restores clean" and "is current" are two
+// different questions and the second one needs the source.
+type HibernatingRoute struct {
+ Host string
+ DBPath string
+}
+
// HibernatingHosts lists the routes whose replication the activator owns —
// exactly the set FleetDBs filters out, derived with the SAME conditions so
// the two cannot drift into a gap that belongs to neither pass.
-func HibernatingHosts(reg *registry.Registry) ([]string, error) {
+func HibernatingHosts(reg *registry.Registry) ([]HibernatingRoute, error) {
routes, err := reg.List()
if err != nil {
return nil, err
}
- var hosts []string
+ var hosts []HibernatingRoute
for _, r := range routes {
// BackingUnit tenants are the box-wide replicator's job (their wake is
// systemctl and the activator never replicates them) — FleetDBs adds
@@ -47,10 +56,10 @@ func HibernatingHosts(reg *registry.Registry) ([]string, error) {
continue
}
if r.Hibernate && r.DBPath != "" {
- hosts = append(hosts, r.Host)
+ hosts = append(hosts, HibernatingRoute{Host: r.Host, DBPath: r.DBPath})
}
}
- sort.Strings(hosts)
+ sort.Slice(hosts, func(i, j int) bool { return hosts[i].Host < hosts[j].Host })
return hosts, nil
}
@@ -69,16 +78,26 @@ var verifyOne = lease.VerifyReplica
// set LeaseTTL. Each verification is a full restore: in-region S3→EC2
// transfer is free, but the GETs and the restore CPU are not, and an
// unbounded nightly drill over a growing fleet is a bill that grows with it.
-func VerifyHibernating(ctx context.Context, cfg *activator.Config, hosts []string, max int, now time.Time, tmpDir string) map[string]dbverify.Verdict {
+func VerifyHibernating(ctx context.Context, cfg *activator.Config, hosts []HibernatingRoute, max int, now time.Time, tmpDir string) map[string]dbverify.Verdict {
out := map[string]dbverify.Verdict{}
- for i, host := range hosts {
+ for i, rt := range hosts {
if max > 0 && i >= max {
break
}
+ host := rt.Host
prefix := activator.HibernatePrefix(cfg.Prefix, host)
- err := verifyOne(ctx, cfg.Bucket, prefix, cfg.Region, tmpDir)
+ counter, err := verifyOne(ctx, cfg.Bucket, prefix, cfg.Region, tmpDir)
switch {
case err == nil:
+ // Restores clean — now the second question: is it CURRENT? A
+ // hibernating instance's replica has a live owner (the activator's
+ // embedded replicator, under the lease), and an owner that has
+ // stopped writing leaves a replica that still restores perfectly.
+ if f, detail := hibernateFreshness(rt.DBPath, counter); f == freshnessStale || f == freshnessNoSource {
+ out[host] = dbverify.Verdict{VerifiedAt: now.UTC(), Detail: detail}
+ fmt.Printf("STALE %s: %s\n", host, detail)
+ continue
+ }
out[host] = dbverify.Verdict{OK: true, VerifiedAt: now.UTC()}
fmt.Printf("ok %s\n", host)
case errors.Is(err, lease.ErrNoReplica):
@@ -105,26 +124,40 @@ func VerifyHibernating(ctx context.Context, cfg *activator.Config, hosts []strin
return out
}
+// hibernateFreshness asks whether this replica is still tracking its source.
+//
+// A route with no recorded DBPath yields "ok": there is nothing to compare
+// against, and mistaking "we do not know where the source lives" for "the
+// source is gone" would raise a false alarm on a healthy instance.
+// HibernatingHosts only yields routes with a DBPath, so this is a guard rather
+// than a path production takes.
+func hibernateFreshness(dbPath string, restoredCounter uint32) (freshness, string) {
+ if dbPath == "" {
+ return freshnessOK, ""
+ }
+ return replicaFreshnessCounter(dbPath, restoredCounter, sourceModTime(dbPath))
+}
+
// OldestFirst orders hosts by how long ago they were last verified, most
// stale first, so a capped run works through the fleet instead of drilling
// the same alphabetical prefix every night. Hosts with no verdict at all sort
// first — never verified is staler than any timestamp.
-func OldestFirst(hosts []string, prior *dbverify.Record) []string {
+func OldestFirst(hosts []HibernatingRoute, prior *dbverify.Record) []HibernatingRoute {
stamp := map[string]time.Time{}
if prior != nil {
for h, v := range prior.Hosts {
stamp[h] = v.VerifiedAt
}
}
- ordered := append([]string(nil), hosts...)
+ ordered := append([]HibernatingRoute(nil), hosts...)
sort.SliceStable(ordered, func(i, j int) bool {
- a, aok := stamp[ordered[i]]
- b, bok := stamp[ordered[j]]
+ a, aok := stamp[ordered[i].Host]
+ b, bok := stamp[ordered[j].Host]
if aok != bok {
return !aok // never-verified first
}
if a.Equal(b) {
- return ordered[i] < ordered[j] // stable, and deterministic in tests
+ return ordered[i].Host < ordered[j].Host // stable, deterministic in tests
}
return a.Before(b)
})
diff --git a/internal/ops/hibernate_verify_test.go b/internal/ops/hibernate_verify_test.go
index 1c88d2bb3..bce8dc846 100644
--- a/internal/ops/hibernate_verify_test.go
+++ b/internal/ops/hibernate_verify_test.go
@@ -55,10 +55,7 @@ func TestHibernatingHostsIsTheComplementOfFleetDBs(t *testing.T) {
t.Fatal(err)
}
- hosts, err := HibernatingHosts(reg)
- if err != nil {
- t.Fatal(err)
- }
+ hosts := hostsOf(mustHibernating(t, reg))
if !slices.Contains(hosts, "b.example.com") {
t.Fatalf("the hibernating exec route is drilled by nobody: %v", hosts)
}
@@ -82,8 +79,37 @@ func TestHibernatingHostsIsTheComplementOfFleetDBs(t *testing.T) {
}
}
+// routes builds the drill's input from bare host names. The DBPath is left
+// empty on purpose: these tests drive the RESTORE outcomes, and an absent
+// source is reported as its own verdict, which TestVerifyHibernatingStale
+// covers explicitly.
+func routes(hosts ...string) []HibernatingRoute {
+ out := make([]HibernatingRoute, 0, len(hosts))
+ for _, h := range hosts {
+ out = append(out, HibernatingRoute{Host: h})
+ }
+ return out
+}
+
+func hostsOf(rs []HibernatingRoute) []string {
+ out := make([]string, 0, len(rs))
+ for _, r := range rs {
+ out = append(out, r.Host)
+ }
+ return out
+}
+
+func mustHibernating(t *testing.T, reg *registry.Registry) []HibernatingRoute {
+ t.Helper()
+ got, err := HibernatingHosts(reg)
+ if err != nil {
+ t.Fatal(err)
+ }
+ return got
+}
+
// swapVerifyOne points the restore seam at a stub.
-func swapVerifyOne(t *testing.T, fn func(ctx context.Context, bucket, prefix, region, dir string) error) {
+func swapVerifyOne(t *testing.T, fn func(ctx context.Context, bucket, prefix, region, dir string) (uint32, error)) {
t.Helper()
prev := verifyOne
verifyOne = fn
@@ -94,21 +120,21 @@ func swapVerifyOne(t *testing.T, fn func(ctx context.Context, bucket, prefix, re
// "unverified", not a corruption alarm, and a never-woken instance is neither.
func TestVerifyHibernatingVerdicts(t *testing.T) {
now := at("2026-08-15T09:00:00Z")
- swapVerifyOne(t, func(_ context.Context, _, prefix, _, _ string) error {
+ swapVerifyOne(t, func(_ context.Context, _, prefix, _, _ string) (uint32, error) {
switch {
case strings.Contains(prefix, "good"):
- return nil
+ return 1, nil
case strings.Contains(prefix, "damaged"):
- return errors.New("integrity_check: page 4 corrupt")
+ return 0, errors.New("integrity_check: page 4 corrupt")
case strings.Contains(prefix, "new"):
- return lease.ErrNoReplica
+ return 0, lease.ErrNoReplica
}
- return errors.New("unexpected host")
+ return 0, errors.New("unexpected host")
})
cfg := &activator.Config{Bucket: "instances", Prefix: "hib", Region: "eu-west-1"}
got := VerifyHibernating(context.Background(), cfg,
- []string{"good.example.com", "damaged.example.com", "new.example.com"}, 0, now, t.TempDir())
+ routes("good.example.com", "damaged.example.com", "new.example.com"), 0, now, t.TempDir())
if v := got["good.example.com"]; !v.OK || v.Unverified {
t.Fatalf("clean replica verdict = %+v", v)
@@ -134,12 +160,12 @@ func TestVerifyHibernatingVerdicts(t *testing.T) {
// every drill reports "no replica" over perfectly good backups.
func TestVerifyHibernatingUsesTheActivatorsPrefix(t *testing.T) {
var seen string
- swapVerifyOne(t, func(_ context.Context, _, prefix, _, _ string) error {
+ swapVerifyOne(t, func(_ context.Context, _, prefix, _, _ string) (uint32, error) {
seen = prefix
- return nil
+ return 1, nil
})
cfg := &activator.Config{Bucket: "instances", Prefix: "hib"}
- VerifyHibernating(context.Background(), cfg, []string{"x.example.com"}, 0, time.Now(), t.TempDir())
+ VerifyHibernating(context.Background(), cfg, routes("x.example.com"), 0, time.Now(), t.TempDir())
want := activator.HibernatePrefix("hib", "x.example.com")
if seen != want {
@@ -151,13 +177,13 @@ func TestVerifyHibernatingUsesTheActivatorsPrefix(t *testing.T) {
// via Merge, which is what stops a cap from reading as full coverage.
func TestVerifyHibernatingHonoursTheCap(t *testing.T) {
calls := 0
- swapVerifyOne(t, func(context.Context, string, string, string, string) error {
+ swapVerifyOne(t, func(context.Context, string, string, string, string) (uint32, error) {
calls++
- return nil
+ return 1, nil
})
cfg := &activator.Config{Bucket: "instances"}
got := VerifyHibernating(context.Background(), cfg,
- []string{"a.example.com", "b.example.com", "c.example.com"}, 2, time.Now(), t.TempDir())
+ routes("a.example.com", "b.example.com", "c.example.com"), 2, time.Now(), t.TempDir())
if calls != 2 {
t.Fatalf("restored %d replicas under -max 2", calls)
@@ -174,14 +200,58 @@ func TestOldestFirstPutsNeverVerifiedThenStalest(t *testing.T) {
"recent.example.com": {OK: true, VerifiedAt: at("2026-08-14T00:00:00Z")},
"stale.example.com": {OK: true, VerifiedAt: at("2026-08-01T00:00:00Z")},
}}
- got := OldestFirst([]string{"recent.example.com", "stale.example.com", "never.example.com"}, prior)
+ got := hostsOf(OldestFirst(routes("recent.example.com", "stale.example.com", "never.example.com"), prior))
want := []string{"never.example.com", "stale.example.com", "recent.example.com"}
if !slices.Equal(got, want) {
t.Fatalf("OldestFirst = %v, want %v", got, want)
}
// With no prior record at all, the order must still be deterministic.
- if got := OldestFirst([]string{"b.example.com", "a.example.com"}, nil); !slices.Equal(got, []string{"a.example.com", "b.example.com"}) {
+ if got := hostsOf(OldestFirst(routes("b.example.com", "a.example.com"), nil)); !slices.Equal(got, []string{"a.example.com", "b.example.com"}) {
t.Fatalf("OldestFirst with no prior record = %v", got)
}
}
+
+// A hibernating instance's replica has a live owner — the activator's embedded
+// replicator, under the lease — and an owner that has stopped writing leaves a
+// replica that still restores perfectly. Same blind spot as the box-wide
+// drill, same signal: the replica's change counter stopped moving while the
+// database it mirrors did not.
+func TestVerifyHibernatingCatchesAStaleReplica(t *testing.T) {
+ dir := t.TempDir()
+ live := filepath.Join(dir, "live.db")
+ writeCounterDB(t, live, 30)
+ // Quiet for longer than the grace window: unambiguously "replication
+ // stopped", not "still catching up".
+ old := time.Now().Add(-2 * time.Hour)
+ if err := os.Chtimes(live, old, old); err != nil {
+ t.Fatal(err)
+ }
+
+ // The replica restores clean, but at a change counter from long ago.
+ swapVerifyOne(t, func(context.Context, string, string, string, string) (uint32, error) {
+ return 2, nil
+ })
+
+ now := at("2026-08-15T09:00:00Z")
+ got := VerifyHibernating(context.Background(), &activator.Config{Bucket: "instances"},
+ []HibernatingRoute{{Host: "frozen.example.com", DBPath: live}}, 0, now, t.TempDir())
+
+ v := got["frozen.example.com"]
+ if v.OK {
+ t.Fatalf("a replica frozen while its instance kept writing was reported OK: %+v", v)
+ }
+ if v.Unverified {
+ t.Fatalf("a stale replica must be a FAILURE, not merely unverified: %+v", v)
+ }
+ if !strings.Contains(v.Detail, "replication has stopped") {
+ t.Fatalf("verdict does not say what is wrong: %q", v.Detail)
+ }
+
+ // And it must reach the published record as a failure, so the operator
+ // sees it without a session on the box.
+ rec := dbverify.Merge(nil, "box-a", now, []string{"frozen.example.com"}, got)
+ if rec.Failures() != 1 {
+ t.Fatalf("Failures() = %d, want 1", rec.Failures())
+ }
+}
diff --git a/internal/ops/ops.go b/internal/ops/ops.go
index 5a1cefc30..678679cd9 100644
--- a/internal/ops/ops.go
+++ b/internal/ops/ops.go
@@ -267,7 +267,11 @@ func runHibernateVerify(args []string) {
now := time.Now()
results := VerifyHibernating(ctx, cfg, OldestFirst(hosts, prior), *max, now, tmp)
- rec := dbverify.Merge(prior, boxLabel(), now, hosts, results)
+ fleet := make([]string, 0, len(hosts))
+ for _, h := range hosts {
+ fleet = append(fleet, h.Host)
+ }
+ rec := dbverify.Merge(prior, boxLabel(), now, fleet, results)
// Publication is best-effort and LOUD when it fails: the drill's value is
// that someone can read the result without a session on this box, so a