amadan

carlos / platform Public

ops: one replication owner for every worker (D9)

The box-wide litestream generator and the restore census now share one
predicate, ActivatorReplicated: hibernating exec routes and every worker,
always-on or not. An always-on worker was replicated by both the activator's
lease replicator and the box-wide config; the generator now leaves it out and
the drill covers it. The shrink guard accepts a drop that is explained by the
predicate and carlos ops litestream names each one on stderr; any other drop
is still a shortfall.
Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
claude ce2bfb2a36003de1f7c5598b3bb05dd96c4f43f8 parent 4abca90
6 files changed, +248 −14
  • implementation/census-claims-sidecar-isolation.txt +3 −0
  • internal/ops/hibernate_verify.go +8 −7
  • internal/ops/hibernate_verify_test.go +24 −0
  • internal/ops/litestream.go +55 −5
  • internal/ops/ops.go +15 −2
  • internal/ops/ops_test.go +143 −0
diff --git a/implementation/census-claims-sidecar-isolation.txt b/implementation/census-claims-sidecar-isolation.txt
new file mode 100644
index 000000000..babd959e4
--- /dev/null
+++ b/implementation/census-claims-sidecar-isolation.txt
@@ -0,0 +1,3 @@
+internal/ops/ops.go:runLitestream shrink guard accepts and names a drop because the row is ActivatorReplicated (CoverageShortfallExcept); other drops still refuse
+internal/ops/hibernate_verify.go:HibernatingHosts selects by ActivatorReplicated, so an always-on worker is drilled; reads the whole route through the predicate
+internal/ops/litestream.go:FleetDBsWithSkipped the activator-owned exclusion tests ActivatorReplicated (workers included) instead of Hibernate alone; reads the whole route through the predicate
diff --git a/internal/ops/hibernate_verify.go b/internal/ops/hibernate_verify.go
index 7073f3260..c3d017cc8 100644
--- a/internal/ops/hibernate_verify.go
+++ b/internal/ops/hibernate_verify.go
@@ -40,8 +40,11 @@ type HibernatingRoute struct {
}
// 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.
+// hibernating exec routes and every worker, always-on or not — exactly the
+// set FleetDBs filters out, selected by the SAME predicate (ActivatorReplicated)
+// so the two cannot drift into a gap that belongs to neither pass. The name
+// predates the workers; it is kept so callers and the stored record's
+// vocabulary do not churn.
func HibernatingHosts(reg *registry.Registry) ([]HibernatingRoute, error) {
routes, err := reg.List()
if err != nil {
@@ -51,11 +54,9 @@ func HibernatingHosts(reg *registry.Registry) ([]HibernatingRoute, error) {
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
- // them, so this pass must not claim them.
- if r.Backing == registry.BackingUnit {
- continue
- }
- if r.Hibernate && r.DBPath != "" {
+ // them, and ActivatorReplicated excludes them, so this pass does not
+ // claim them.
+ if ActivatorReplicated(r) {
hosts = append(hosts, HibernatingRoute{Host: r.Host, DBPath: r.DBPath})
}
}
diff --git a/internal/ops/hibernate_verify_test.go b/internal/ops/hibernate_verify_test.go
index d8795965e..3d8656b97 100644
--- a/internal/ops/hibernate_verify_test.go
+++ b/internal/ops/hibernate_verify_test.go
@@ -255,3 +255,27 @@ func TestVerifyHibernatingCatchesAStaleReplica(t *testing.T) {
t.Fatalf("Failures() = %d, want 1", rec.Failures())
}
}
+
+// D9: an always-on worker's replica is the activator's too, so the drill
+// must visit it. A drill that skipped it would leave a database owned by
+// nobody's census.
+func TestHibernatingHostsIncludesAlwaysOnWorker(t *testing.T) {
+ dir := t.TempDir()
+ reg, err := registry.Open(filepath.Join(dir, "registry.db"))
+ if err != nil {
+ t.Fatal(err)
+ }
+ if err := reg.Upsert(registry.Route{Host: "w.example.com", Kind: "sidecar", DBPath: filepath.Join(dir, "w.db")}); err != nil {
+ t.Fatal(err)
+ }
+ if err := reg.Upsert(registry.Route{Host: "a.example.com", DBPath: filepath.Join(dir, "a.db")}); err != nil {
+ t.Fatal(err)
+ }
+ hosts := hostsOf(mustHibernating(t, reg))
+ if !slices.Contains(hosts, "w.example.com") {
+ t.Fatalf("an always-on worker is drilled by nobody: %v", hosts)
+ }
+ if slices.Contains(hosts, "a.example.com") {
+ t.Fatalf("a plain route is drilled: %v", hosts)
+ }
+}
diff --git a/internal/ops/litestream.go b/internal/ops/litestream.go
index de6307f40..47ccddf6e 100644
--- a/internal/ops/litestream.go
+++ b/internal/ops/litestream.go
@@ -189,18 +189,68 @@ func ParseLitestreamDBs(cfg string) []string {
// Growing the set is always fine and needs no ceremony; shrinking it is a
// decision someone has to make on purpose.
func CoverageShortfall(existingConfig string, proposed []string) []string {
+ missing, _ := CoverageShortfallExcept(existingConfig, proposed, nil)
+ return missing
+}
+
+// CoverageShortfallExcept is CoverageShortfall with one reason a database may
+// leave the config on purpose: activatorOwned names the paths whose rows the
+// activator now replicates (ActivatorReplicated), so the box-wide config
+// dropping them hands the file over rather than abandoning it. Those drops
+// are returned separately as expected, so the caller can name each one; every
+// other drop is still a shortfall.
+func CoverageShortfallExcept(existingConfig string, proposed, activatorOwned []string) (missing, expected []string) {
have := map[string]bool{}
for _, p := range proposed {
have[filepath.Clean(p)] = true
}
- var missing []string
+ owned := map[string]bool{}
+ for _, p := range activatorOwned {
+ owned[filepath.Clean(p)] = true
+ }
for _, p := range ParseLitestreamDBs(existingConfig) {
- if !have[filepath.Clean(p)] {
+ switch {
+ case have[filepath.Clean(p)]:
+ case owned[filepath.Clean(p)]:
+ expected = append(expected, p)
+ default:
missing = append(missing, p)
}
}
sort.Strings(missing)
- return missing
+ sort.Strings(expected)
+ return missing, expected
+}
+
+// ActivatorReplicated reports whether the activator's lease replication is
+// this route's only replicator: a hibernating exec route, and any worker
+// (Kind "sidecar") whether it hibernates or stays on, because a worker's wake
+// starts lease replication and always-on supervision keeps it. Unit tenants
+// are the box-wide replicator's (the activator never replicates them), and a
+// route that names no database has nothing to replicate.
+//
+// This one predicate is read by the box-wide generator (FleetDBsWithSkipped,
+// which leaves such a database out) and by the restore census
+// (HibernatingHosts, which drills it), so the two cannot drift into a gap
+// that belongs to neither, or an overlap that gives one WAL two replicators.
+func ActivatorReplicated(rt registry.Route) bool {
+ return rt.Backing != registry.BackingUnit && rt.DBPath != "" && (rt.Hibernate || rt.IsSidecar())
+}
+
+// ActivatorReplicatedPaths lists the database paths of every
+// ActivatorReplicated route, for the shrink guard's expected drops.
+func ActivatorReplicatedPaths(reg *registry.Registry) ([]string, error) {
+ routes, err := reg.List()
+ if err != nil {
+ return nil, err
+ }
+ var paths []string
+ for _, r := range routes {
+ if ActivatorReplicated(r) {
+ paths = append(paths, filepath.Clean(r.DBPath))
+ }
+ }
+ return paths, nil
}
// FleetDBs lists the non-empty *.db files this box-wide replicator owns:
@@ -353,7 +403,7 @@ func FleetDBsWithSkipped(reg *registry.Registry, dataDir string, fenced func(hos
}
continue
}
- if r.Hibernate && r.DBPath != "" {
+ if ActivatorReplicated(r) {
hibernating[filepath.Clean(r.DBPath)] = true
}
}
@@ -436,7 +486,7 @@ func FleetDBsWithSkipped(reg *registry.Registry, dataDir string, fenced func(hos
if err != nil || !fi.Mode().IsRegular() || fi.Size() == 0 {
continue // gone, odd, or an empty placeholder — nothing to replicate
}
- if hibernating[p] {
+ if hibernating[p] { // activator-replicated: see ActivatorReplicated
continue
}
seen[p] = true
diff --git a/internal/ops/ops.go b/internal/ops/ops.go
index 4e09acbca..db54d252e 100644
--- a/internal/ops/ops.go
+++ b/internal/ops/ops.go
@@ -87,7 +87,8 @@ func dataDir(flagVal string) string {
// runLitestream — regenerate the fleet replication config. Idempotent: run it
// after every `carlos add`/`remove` (or on a timer; a no-op re-run is cheap
-// and prints "unchanged"). Databases of hibernating routes are excluded — see
+// and prints "unchanged"). Databases the activator replicates (hibernating
+// routes and every worker, ActivatorReplicated) are excluded — see
// FleetDBs for Eleven's two-replicators lesson.
func runLitestream(args []string) {
fs := flag.NewFlagSet("ops litestream", flag.ExitOnError)
@@ -133,7 +134,19 @@ func runLitestream(args []string) {
// CoverageShortfall. This is what makes a cutover from a hand-maintained
// config to a generated one safe to do unattended.
if prior, err := os.ReadFile(*out); err == nil {
- if missing := CoverageShortfall(string(prior), dbs); len(missing) > 0 && !*allowShrink {
+ owned, err := ActivatorReplicatedPaths(reg)
+ if err != nil {
+ log.Fatalf("litestream: %v", err)
+ }
+ missing, handedOver := CoverageShortfallExcept(string(prior), dbs, owned)
+ if len(handedOver) > 0 {
+ // Expected, not a shortfall: the activator's lease replication has
+ // run for these since the worker's first wake, so only the
+ // duplicate box-wide stanza goes. Named, never silent.
+ fmt.Fprintf(os.Stderr, "litestream: %d database(s) leave %s because the activator's lease replication owns them (one replicator per database):\n %s\n",
+ len(handedOver), *out, strings.Join(handedOver, "\n "))
+ }
+ if len(missing) > 0 && !*allowShrink {
log.Fatalf(`litestream: REFUSING to write %s — it would stop replicating %d database(s) that %s names today:
%s
Nothing has been changed. Either make the registry describe them (a route with
diff --git a/internal/ops/ops_test.go b/internal/ops/ops_test.go
index c16d4c5aa..2904c46fd 100644
--- a/internal/ops/ops_test.go
+++ b/internal/ops/ops_test.go
@@ -2,6 +2,7 @@ package ops
import (
"database/sql"
+ "io"
"os"
"path/filepath"
"slices"
@@ -870,3 +871,145 @@ func TestLayoutJournalPresentIsLstatNotAbsent(t *testing.T) {
t.Fatal("another host was reported held")
}
}
+
+// D9 (sidecar isolation §3.9): an always-on worker is replicated by the
+// activator's lease replicator from its first wake, so the box-wide config
+// must not name its database a second time.
+func TestFleetDBsExcludesAlwaysOnWorker(t *testing.T) {
+ dir := t.TempDir()
+ reg, err := registry.Open(filepath.Join(dir, "registry.db"))
+ if err != nil {
+ t.Fatal(err)
+ }
+ worker := filepath.Join(dir, "w.db")
+ plain := filepath.Join(dir, "a.example.com.db")
+ for _, p := range []string{worker, plain} {
+ if err := os.WriteFile(p, []byte("x"), 0o644); err != nil {
+ t.Fatal(err)
+ }
+ }
+ if err := reg.Upsert(registry.Route{Host: "w.example.com", Kind: "sidecar", Hibernate: false, DBPath: worker}); err != nil {
+ t.Fatal(err)
+ }
+ if err := reg.Upsert(registry.Route{Host: "a.example.com", DBPath: plain}); err != nil {
+ t.Fatal(err)
+ }
+ dbs, err := FleetDBs(reg, dir)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if slices.Contains(dbs, worker) {
+ t.Fatalf("an always-on worker is replicated by both the activator and the box-wide config: %v", dbs)
+ }
+ if !slices.Contains(dbs, plain) {
+ t.Fatalf("an ordinary route lost its replication: %v", dbs)
+ }
+}
+
+func TestActivatorReplicatedPredicate(t *testing.T) {
+ for _, c := range []struct {
+ name string
+ rt registry.Route
+ want bool
+ }{
+ {"hibernating exec", registry.Route{DBPath: "/d", Hibernate: true}, true},
+ {"always-on worker", registry.Route{DBPath: "/d", Kind: "sidecar"}, true},
+ {"hibernating worker", registry.Route{DBPath: "/d", Kind: "sidecar", Hibernate: true}, true},
+ {"plain route", registry.Route{DBPath: "/d"}, false},
+ {"worker without a database", registry.Route{Kind: "sidecar"}, false},
+ {"unit tenant", registry.Route{DBPath: "/d", Hibernate: true, Backing: registry.BackingUnit}, false},
+ {"unit-backed worker", registry.Route{DBPath: "/d", Kind: "sidecar", Backing: registry.BackingUnit}, false},
+ } {
+ if got := ActivatorReplicated(c.rt); got != c.want {
+ t.Errorf("%s: ActivatorReplicated = %v, want %v", c.name, got, c.want)
+ }
+ }
+}
+
+// captureStderr runs fn with os.Stderr redirected and returns what it wrote.
+func captureStderr(t *testing.T, fn func()) string {
+ t.Helper()
+ r, w, err := os.Pipe()
+ if err != nil {
+ t.Fatal(err)
+ }
+ prev := os.Stderr
+ os.Stderr = w
+ done := make(chan string)
+ go func() {
+ b, _ := io.ReadAll(r)
+ done <- string(b)
+ }()
+ func() {
+ defer func() { os.Stderr = prev }()
+ fn()
+ }()
+ _ = w.Close()
+ return <-done
+}
+
+func TestCoverageShortfallAcceptsActivatorReplicatedDrop(t *testing.T) {
+ existing := LitestreamConfig([]string{"/data/a.db", "/data/w.db", "/data/registry.db"}, "b", "p", "eu-west-1")
+ proposed := []string{"/data/a.db", "/data/registry.db"}
+
+ // The library: a drop the activator accounts for is not a shortfall, and
+ // is returned separately so the caller can name it.
+ missing, expected := CoverageShortfallExcept(existing, proposed, []string{"/data/w.db"})
+ if len(missing) != 0 {
+ t.Fatalf("an activator-replicated drop was a shortfall: %v", missing)
+ }
+ if len(expected) != 1 || expected[0] != "/data/w.db" {
+ t.Fatalf("expected drops = %v, want [/data/w.db]", expected)
+ }
+ // Any other reason is still a shortfall, whatever else is excused.
+ missing, expected = CoverageShortfallExcept(existing, []string{"/data/registry.db"}, []string{"/data/w.db"})
+ if len(missing) != 1 || missing[0] != "/data/a.db" || len(expected) != 1 {
+ t.Fatalf("missing = %v, expected = %v, want [/data/a.db] and [/data/w.db]", missing, expected)
+ }
+ // The plain form behaves as before.
+ if missing := CoverageShortfall(existing, proposed); len(missing) != 1 || missing[0] != "/data/w.db" {
+ t.Fatalf("CoverageShortfall = %v, want [/data/w.db]", missing)
+ }
+
+ // The verb: a real registry, a real prior config, no --allow-shrink.
+ dir := t.TempDir()
+ reg, err := registry.Open(filepath.Join(dir, "registry.db"))
+ if err != nil {
+ t.Fatal(err)
+ }
+ a := filepath.Join(dir, "a.example.com.db")
+ w := filepath.Join(dir, "w.db")
+ for _, p := range []string{a, w} {
+ if err := os.WriteFile(p, []byte("x"), 0o644); err != nil {
+ t.Fatal(err)
+ }
+ }
+ if err := reg.Upsert(registry.Route{Host: "a.example.com", DBPath: a}); err != nil {
+ t.Fatal(err)
+ }
+ if err := reg.Upsert(registry.Route{Host: "w.example.com", Kind: "sidecar", DBPath: w}); err != nil {
+ t.Fatal(err)
+ }
+ if err := reg.Close(); err != nil {
+ t.Fatal(err)
+ }
+ out := filepath.Join(dir, "litestream.yml")
+ prior := LitestreamConfig([]string{a, w, filepath.Join(dir, "registry.db")}, "b", "p", "eu-west-1")
+ if err := os.WriteFile(out, []byte(prior), 0o644); err != nil {
+ t.Fatal(err)
+ }
+ stderr := captureStderr(t, func() {
+ runLitestream([]string{"--data", dir, "--bucket", "b", "--prefix", "p", "--region", "eu-west-1",
+ "--out", out, "--litestream-bin", filepath.Join(dir, "no-litestream")})
+ })
+ if !strings.Contains(stderr, w) || !strings.Contains(stderr, "activator") {
+ t.Fatalf("the dropped worker database is not named on stderr:\n%s", stderr)
+ }
+ got, err := os.ReadFile(out)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if strings.Contains(string(got), w) || !strings.Contains(string(got), a) {
+ t.Fatalf("regenerated config wrong:\n%s", got)
+ }
+}