amadan
The registry keeps an ownership generation that only ownership moves
The replicator's witness was SQLite's data_version, which moves on every commit, and the registry is one of the databases the replicator writes its own bookkeeping into: the fleet went stale within minutes of opening (second review, Opus P1, Astra P2). A counter moved by triggers on the routes table now moves when a route is added, removed, or changes host, account, app, database path, backing or hibernation to a different value, and for nothing else. RouteObservation.OwnershipUnchanged reads it on the pinned connection and also answers changed when the registry's path names another file. ObserveRoutes' cleanup dereferenced the named result after the return had cleared it, so any failure after the connection opened was a panic; it now closes what it built.
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
3 files changed,
+336
−2
internal/registry/observe.go+81 −2internal/registry/observe_ownership_test.go+203 −0internal/registry/registry.go+52 −0
diff --git a/internal/registry/observe.go b/internal/registry/observe.go| index 43eb59bb7..268e8e6ae 100644 |
| --- a/internal/registry/observe.go |
| +++ b/internal/registry/observe.go |
| @@ -3,7 +3,9 @@ package registry |
| import ( |
| "context" |
| "database/sql" |
| + "errors" |
| "fmt" |
| + "os" |
| ) |
| // RouteObservation is one read of every route together with a mutation |
| @@ -21,16 +23,35 @@ import ( |
| // commit anywhere — this process's own pool included — is "another |
| // connection" to it. Close releases the connection; an observation that |
| // is never closed leaks one SQLite handle. |
| +// |
| +// A second, narrower witness rides on the same read: the ownership |
| +// generation (migrateRouteOwnership), which moves only when a route is |
| +// added, removed, or changes who owns its database or which replicator |
| +// streams it. OwnershipUnchanged asks that one. |
| type RouteObservation struct { |
| Routes []Route |
| db *sql.DB |
| conn *sql.Conn |
| version int64 |
| + |
| + path string |
| + file os.FileInfo |
| + generation int64 |
| } |
| // ObserveRoutes reads every route over a fresh pinned connection and arms |
| // the witness on it. The caller owns the observation and must Close it. |
| func (r *Registry) ObserveRoutes(ctx context.Context) (obs *RouteObservation, err error) { |
| + // The file the path names, looked at before the connection opens it |
| + // and again once the rows are read, so that a registry put in this |
| + // one's place later can be told from this one. The path naming the |
| + // same file at both ends is taken to mean the connection opened that |
| + // file; a path swapped away and back between the two looks is not |
| + // told apart, and is an operator's doing, not a tenant's. |
| + before, err := os.Stat(r.path) |
| + if err != nil { |
| + return nil, err |
| + } |
| db, err := sql.Open("sqlite", r.path) |
| if err != nil { |
| return nil, err |
| @@ -41,10 +62,15 @@ func (r *Registry) ObserveRoutes(ctx context.Context) (obs *RouteObservation, er |
| db.Close() |
| return nil, err |
| } |
| - obs = &RouteObservation{db: db, conn: conn} |
| + // built, not the named result: "return nil, err" clears the named |
| + // result before the deferred cleanup runs, and the cleanup has to |
| + // close what was built. |
| + built := &RouteObservation{db: db, conn: conn, path: r.path} |
| + obs = built |
| + built.file = before |
| defer func() { |
| if err != nil { |
| - obs.Close() |
| + built.Close() |
| obs = nil |
| } |
| }() |
| @@ -67,6 +93,11 @@ func (r *Registry) ObserveRoutes(ctx context.Context) (obs *RouteObservation, er |
| conn.ExecContext(ctx, "ROLLBACK") |
| return nil, err |
| } |
| + obs.generation, err = ownershipGeneration(ctx, conn) |
| + if err != nil { |
| + conn.ExecContext(ctx, "ROLLBACK") |
| + return nil, err |
| + } |
| obs.Routes, err = listRoutes(ctx, conn) |
| if err != nil { |
| conn.ExecContext(ctx, "ROLLBACK") |
| @@ -75,9 +106,57 @@ func (r *Registry) ObserveRoutes(ctx context.Context) (obs *RouteObservation, er |
| if _, err = conn.ExecContext(ctx, "COMMIT"); err != nil { |
| return nil, err |
| } |
| + after, err := os.Stat(r.path) |
| + if err != nil { |
| + return nil, err |
| + } |
| + if !os.SameFile(before, after) { |
| + return nil, fmt.Errorf("registry: %s was replaced while it was being read", r.path) |
| + } |
| return obs, nil |
| } |
| +// OwnershipUnchanged reports whether the routes still say what they said |
| +// when they were read about which databases there are and who owns each: |
| +// no route added or removed, and none whose host, account, app, database |
| +// path, backing or hibernation has changed, even to a value it has since |
| +// changed back from. Commits that touch nothing else (a status, a due |
| +// time, a certificate date, another table) do not move it, which is the |
| +// difference from Unchanged. |
| +// |
| +// It also answers false when the registry's path no longer names the file |
| +// the rows were read from. The pinned connection would go on reading the |
| +// old file, whose generation never moves again, while every other reader |
| +// has moved to the new one. |
| +// |
| +// As with Unchanged: false with a nil error is a witnessed change, and an |
| +// error is a witness that could not be read, which is never "unchanged". |
| +func (o *RouteObservation) OwnershipUnchanged(ctx context.Context) (bool, error) { |
| + if o.conn == nil || o.file == nil { |
| + return false, errors.New("registry: the observation is closed") |
| + } |
| + now, err := os.Stat(o.path) |
| + if err != nil { |
| + return false, fmt.Errorf("registry: stat %s: %w", o.path, err) |
| + } |
| + if !os.SameFile(o.file, now) { |
| + return false, nil |
| + } |
| + g, err := ownershipGeneration(ctx, o.conn) |
| + if err != nil { |
| + return false, err |
| + } |
| + return g == o.generation, nil |
| +} |
| + |
| +func ownershipGeneration(ctx context.Context, conn *sql.Conn) (int64, error) { |
| + var g int64 |
| + if err := conn.QueryRowContext(ctx, "SELECT generation FROM route_ownership WHERE id = 1").Scan(&g); err != nil { |
| + return 0, fmt.Errorf("read ownership generation: %w", err) |
| + } |
| + return g, nil |
| +} |
| + |
| // Unchanged reports whether no connection has committed to the database |
| // since the rows were read. It may be asked any number of times; asking |
| // is a read and moves nothing. false with a nil error is a witnessed |
diff --git a/internal/registry/observe_ownership_test.go b/internal/registry/observe_ownership_test.go| new file mode 100644 |
| index 000000000..80834b4ad |
| --- /dev/null |
| +++ b/internal/registry/observe_ownership_test.go |
| @@ -0,0 +1,203 @@ |
| +package registry |
| + |
| +import ( |
| + "context" |
| + "os" |
| + "path/filepath" |
| + "testing" |
| +) |
| + |
| +// The ownership witness moves when a route is added, removed, or changes |
| +// who owns its database or which replicator streams it, and for nothing |
| +// else. Every row of the second table is a commit that the whole-database |
| +// witness (Unchanged) reports and that has nothing to do with ownership: |
| +// those are the writes a live box makes all day, and the bookkeeping a |
| +// replicator writes into the registry it replicates. |
| +func TestOwnershipWitnessMovesOnlyWhenOwnershipChanges(t *testing.T) { |
| + const host = "t.example.com" |
| + moves := map[string]func(t *testing.T, w *Registry){ |
| + "the route moves to another account": func(t *testing.T, w *Registry) { |
| + must(t, w.SetAccount(host, "bbbb")) |
| + }, |
| + "the route moves account and back": func(t *testing.T, w *Registry) { |
| + must(t, w.SetAccount(host, "bbbb")) |
| + must(t, w.SetAccount(host, "aaaa")) |
| + }, |
| + "the route is deleted": func(t *testing.T, w *Registry) { |
| + must(t, w.Delete(host)) |
| + }, |
| + "the route is deleted and made again, the same": func(t *testing.T, w *Registry) { |
| + r, err := w.Get(host) |
| + must(t, err) |
| + must(t, w.Delete(host)) |
| + must(t, w.Upsert(*r)) |
| + }, |
| + "another route is added": func(t *testing.T, w *Registry) { |
| + must(t, w.Upsert(Route{Host: "u.example.com", Slug: "u", Kind: "instance", App: "u", Account: "aaaa"})) |
| + }, |
| + "the route starts hibernating": func(t *testing.T, w *Registry) { |
| + must(t, w.SetHibernate(host, true)) |
| + }, |
| + "the route's database moves": func(t *testing.T, w *Registry) { |
| + must(t, w.SetDBPath(host, "/data/t.db", "/data/elsewhere.db")) |
| + }, |
| + "the route becomes unit-backed": func(t *testing.T, w *Registry) { |
| + must(t, w.SetBacking(host, BackingUnit)) |
| + }, |
| + "a deploy that changes the app": func(t *testing.T, w *Registry) { |
| + r, err := w.Get(host) |
| + must(t, err) |
| + r.App = "other" |
| + must(t, w.Upsert(*r)) |
| + }, |
| + "a raw statement from some other build": func(t *testing.T, w *Registry) { |
| + _, err := w.sql.Exec(`UPDATE routes SET account = 'cccc', status = 'ready' WHERE host = ?`, host) |
| + must(t, err) |
| + }, |
| + } |
| + stays := map[string]func(t *testing.T, w *Registry){ |
| + "a worker's next due time": func(t *testing.T, w *Registry) { |
| + must(t, w.SetNextDue(host, 1234)) |
| + }, |
| + "a DNS check": func(t *testing.T, w *Registry) { |
| + must(t, w.SetDNSStatus(host, "ok", 99)) |
| + }, |
| + "a certificate's expiry": func(t *testing.T, w *Registry) { |
| + must(t, w.SetCertExpiresAt(host, 99)) |
| + }, |
| + "a status change": func(t *testing.T, w *Registry) { |
| + must(t, w.SetStatus(host, "starting")) |
| + }, |
| + "a deploy's versions": func(t *testing.T, w *Registry) { |
| + must(t, w.SetVersions(host, "v2", "v1")) |
| + }, |
| + "a deploy that writes the whole row with the same owner": func(t *testing.T, w *Registry) { |
| + r, err := w.Get(host) |
| + must(t, err) |
| + r.Version, r.Status = "v3", "ready" |
| + must(t, w.Upsert(*r)) |
| + }, |
| + "a channel change": func(t *testing.T, w *Registry) { |
| + must(t, w.SetChannel(host, "beta")) |
| + }, |
| + "a replicator's bookkeeping in the registry": func(t *testing.T, w *Registry) { |
| + _, err := w.sql.Exec(`CREATE TABLE IF NOT EXISTS _litestream_seq (id INTEGER PRIMARY KEY, seq INTEGER); INSERT INTO _litestream_seq (id, seq) VALUES (1, 1) ON CONFLICT (id) DO UPDATE SET seq = seq + 1`) |
| + must(t, err) |
| + }, |
| + } |
| + run := func(name string, change func(*testing.T, *Registry), wantUnchanged bool) { |
| + t.Run(name, func(t *testing.T) { |
| + path := filepath.Join(t.TempDir(), "registry.db") |
| + reader, err := Open(path) |
| + must(t, err) |
| + writer, err := Open(path) |
| + must(t, err) |
| + must(t, writer.Upsert(Route{Host: host, Slug: "t", Kind: "instance", App: "t", Account: "aaaa", DBPath: "/data/t.db", Version: "v1"})) |
| + ctx := context.Background() |
| + obs, err := reader.ObserveRoutes(ctx) |
| + must(t, err) |
| + defer obs.Close() |
| + if ok, err := obs.OwnershipUnchanged(ctx); err != nil || !ok { |
| + t.Fatalf("before any write: unchanged=%v err=%v", ok, err) |
| + } |
| + change(t, writer) |
| + ok, err := obs.OwnershipUnchanged(ctx) |
| + must(t, err) |
| + if ok != wantUnchanged { |
| + t.Fatalf("OwnershipUnchanged = %v, want %v", ok, wantUnchanged) |
| + } |
| + // Every one of these is a commit, and the whole-database |
| + // witness says so: the two witnesses differ exactly on the |
| + // second table. |
| + if all, err := obs.Unchanged(ctx); err != nil || all { |
| + t.Fatalf("the whole-database witness missed the commit: unchanged=%v err=%v", all, err) |
| + } |
| + // Asking is a read: the answer does not change by being asked. |
| + if again, err := obs.OwnershipUnchanged(ctx); err != nil || again != ok { |
| + t.Fatalf("asked again: %v then %v (err %v)", ok, again, err) |
| + } |
| + }) |
| + } |
| + for name, change := range moves { |
| + run(name, change, false) |
| + } |
| + for name, change := range stays { |
| + run(name, change, true) |
| + } |
| +} |
| + |
| +// A registry file put in the old one's place leaves the pinned connection |
| +// reading the old file, whose generation never moves again. The witness |
| +// says changed, because nobody else is reading that file any more. |
| +func TestOwnershipWitnessSeesTheRegistryReplaced(t *testing.T) { |
| + dir := t.TempDir() |
| + path := filepath.Join(dir, "registry.db") |
| + reg, err := Open(path) |
| + must(t, err) |
| + must(t, reg.Upsert(Route{Host: "t.example.com", Slug: "t", Kind: "instance", App: "t", Account: "aaaa"})) |
| + other, err := Open(filepath.Join(dir, "other.db")) |
| + must(t, err) |
| + must(t, other.Upsert(Route{Host: "t.example.com", Slug: "t", Kind: "instance", App: "t", Account: "bbbb"})) |
| + must(t, other.Close()) |
| + ctx := context.Background() |
| + obs, err := reg.ObserveRoutes(ctx) |
| + must(t, err) |
| + defer obs.Close() |
| + must(t, os.Rename(filepath.Join(dir, "other.db"), path)) |
| + if ok, err := obs.OwnershipUnchanged(ctx); err != nil || ok { |
| + t.Fatalf("a replaced registry: unchanged=%v err=%v, want a witnessed change", ok, err) |
| + } |
| + // And one that has gone altogether is not proof of anything. |
| + must(t, os.Remove(path)) |
| + if ok, err := obs.OwnershipUnchanged(ctx); err == nil || ok { |
| + t.Fatalf("a missing registry: unchanged=%v err=%v, want an error", ok, err) |
| + } |
| +} |
| + |
| +// A registry written by a build from before the generation existed gains |
| +// the counter and its triggers on open, and a registry that has lost its |
| +// counter row cannot be vouched for. |
| +func TestOwnershipWitnessOnAnOlderRegistry(t *testing.T) { |
| + path := filepath.Join(t.TempDir(), "registry.db") |
| + old, err := Open(path) |
| + must(t, err) |
| + must(t, old.Upsert(Route{Host: "t.example.com", Slug: "t", Kind: "instance", App: "t", Account: "aaaa"})) |
| + for _, stmt := range []string{ |
| + `DROP TRIGGER routes_ownership_insert`, `DROP TRIGGER routes_ownership_delete`, |
| + `DROP TRIGGER routes_ownership_update`, `DROP TABLE route_ownership`, |
| + } { |
| + _, err := old.sql.Exec(stmt) |
| + must(t, err) |
| + } |
| + must(t, old.Close()) |
| + reg, err := Open(path) |
| + must(t, err) |
| + ctx := context.Background() |
| + obs, err := reg.ObserveRoutes(ctx) |
| + must(t, err) |
| + defer obs.Close() |
| + must(t, reg.SetAccount("t.example.com", "bbbb")) |
| + if ok, err := obs.OwnershipUnchanged(ctx); err != nil || ok { |
| + t.Fatalf("after migration a change of account: unchanged=%v err=%v", ok, err) |
| + } |
| + _, err = reg.sql.Exec(`DELETE FROM route_ownership`) |
| + must(t, err) |
| + if ok, err := obs.OwnershipUnchanged(ctx); err == nil || ok { |
| + t.Fatalf("a registry with no counter: unchanged=%v err=%v, want an error", ok, err) |
| + } |
| + if _, err := reg.ObserveRoutes(ctx); err == nil { |
| + t.Fatal("an observation was taken of a registry with no ownership counter") |
| + } |
| + // A closed observation vouches for nothing. |
| + must(t, obs.Close()) |
| + if ok, err := obs.OwnershipUnchanged(ctx); err == nil || ok { |
| + t.Fatalf("a closed observation: unchanged=%v err=%v", ok, err) |
| + } |
| +} |
| + |
| +func must(t *testing.T, err error) { |
| + t.Helper() |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| +} |
diff --git a/internal/registry/registry.go b/internal/registry/registry.go| index 7dfc050cd..6a2e37781 100644 |
| --- a/internal/registry/registry.go |
| +++ b/internal/registry/registry.go |
| @@ -476,6 +476,9 @@ CREATE TABLE IF NOT EXISTS routes ( |
| return err |
| } |
| } |
| + if err := r.migrateRouteOwnership(); err != nil { |
| + return err |
| + } |
| // route_aliases is a derived table (a later reconcile pass owns every |
| // row, materializing domains.Claim records into it) — a plain CREATE |
| // TABLE IF NOT EXISTS, no legacy shape to carry forward. |
| @@ -556,6 +559,55 @@ CREATE INDEX IF NOT EXISTS timers_due ON timers (next_at) WHERE next_at > 0;`); |
| return r.migrateFleets() |
| } |
| +// routeOwnershipColumns are the columns of a route that decide which |
| +// databases the box replicates and who pays for each one: the route's name, |
| +// its account and app, where its database is, and whether the box-wide |
| +// replicator or the activator streams it (backing, hibernate). The |
| +// ownership generation moves when one of these changes and for nothing |
| +// else in the row. Adding a column that discovery or ownership reads |
| +// (internal/ops FleetDBsWithSkipped, FleetDBOwners) means adding it here. |
| +var routeOwnershipColumns = []string{"host", "account", "app", "db_path", "backing", "hibernate"} |
| + |
| +// migrateRouteOwnership creates the ownership generation: one counter that |
| +// moves, inside the writer's own transaction, whenever a route is added, |
| +// removed, or changes one of routeOwnershipColumns to a different value. |
| +// |
| +// It exists because PRAGMA data_version (RouteObservation.Unchanged) moves |
| +// on every commit to the registry, and most commits have nothing to do |
| +// with who owns what: a worker's next due time, a certificate's expiry, a |
| +// DNS check, a wake or a sleep's status, and the bookkeeping a replicator |
| +// writes into the very database it replicates. A reader that must stop |
| +// when ownership changes, and must not stop otherwise, needs a witness |
| +// that only ownership moves. |
| +// |
| +// The counter is moved by triggers, so every writer moves it whatever |
| +// build it is and whichever statement it uses, and a change that is made |
| +// and then undone still moves it twice. Triggers are created on every |
| +// open (IF NOT EXISTS), after the routes table and its columns exist. |
| +func (r *Registry) migrateRouteOwnership() error { |
| + changed := make([]string, 0, len(routeOwnershipColumns)) |
| + for _, c := range routeOwnershipColumns { |
| + changed = append(changed, "OLD."+c+" IS NOT NEW."+c) |
| + } |
| + const bump = `UPDATE route_ownership SET generation = generation + 1 WHERE id = 1;` |
| + for _, stmt := range []string{ |
| + `CREATE TABLE IF NOT EXISTS route_ownership ( |
| + id INTEGER PRIMARY KEY CHECK (id = 1), |
| + generation INTEGER NOT NULL |
| +)`, |
| + `INSERT OR IGNORE INTO route_ownership (id, generation) VALUES (1, 1)`, |
| + `CREATE TRIGGER IF NOT EXISTS routes_ownership_insert AFTER INSERT ON routes BEGIN ` + bump + ` END`, |
| + `CREATE TRIGGER IF NOT EXISTS routes_ownership_delete AFTER DELETE ON routes BEGIN ` + bump + ` END`, |
| + `CREATE TRIGGER IF NOT EXISTS routes_ownership_update AFTER UPDATE OF ` + strings.Join(routeOwnershipColumns, ", ") + ` ON routes |
| +WHEN ` + strings.Join(changed, " OR ") + ` BEGIN ` + bump + ` END`, |
| + } { |
| + if _, err := r.sql.Exec(stmt); err != nil { |
| + return err |
| + } |
| + } |
| + return nil |
| +} |
| + |
| // migrateAccounts brings the accounts table to its multi-account shape: |
| // |
| // id INTEGER PRIMARY KEY (supplied by the caller, NOT autoincrement) |