amadan
The replicator stops counting when the registry changes: ownership is read under the registry's witness and every request asks it
From the first review of this branch (both reviewers, P1): who owns a database was read once, when the store was built, and the recorder bound then went on counting to that owner after the route moved to another account, was deleted and made again, or began hibernating. MeteredFleetStore now returns a MeteredFleet. The ownership read is taken under registry.RouteObservation and checked again afterwards. Every recorder a replica client is given asks the witness before it starts a request: after any commit to the registry, no request is started or counted, the owner's ledger is told its day is incomplete, and Current says the store must be rebuilt. Open: one request already past its check is still counted to the owner as read; nothing rebuilds the store yet, because nothing calls it. Nothing calls this code. Nothing attests.
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
4 files changed,
+230
−20
docs/superpowers/specs/DISPOSITIONS-metered-replicator.md+4 −4internal/ops/litestream_metered.go+122 −11internal/ops/litestream_metered_test.go+103 −4internal/ops/litestream_owners_test.go+1 −1
diff --git a/docs/superpowers/specs/DISPOSITIONS-metered-replicator.md b/docs/superpowers/specs/DISPOSITIONS-metered-replicator.md| index 1d211ca1a..cd4bdce25 100644 |
| --- a/docs/superpowers/specs/DISPOSITIONS-metered-replicator.md |
| +++ b/docs/superpowers/specs/DISPOSITIONS-metered-replicator.md |
| @@ -4,11 +4,11 @@ Branch `feat/carloku-metered-replicator-20261001`. Nothing calls this code and n |
| ## Review one (reviewed `ebab1a37`; Astra rejected, Opus rejected) |
| -Reports: `CODEX-FINDINGS-metered-replicator-review1.md`, `OPUS-REVIEW-metered-replicator-review1.md`. One row below is fixed; the rest are open. |
| +Reports: `CODEX-FINDINGS-metered-replicator-review1.md`, `OPUS-REVIEW-metered-replicator-review1.md`. The two P1 rows are fixed; the rest are open. |
| | Finding | Disposition | |
| | --- | --- | |
| -| Both, P1: the owner of a database is resolved once, when its client is built, and never again. A route moved to another account, or an instance deleted and made again, goes on being counted to the first owner. | Accepted. Open. Ownership has to be bound to the registry generation it was read at and re-checked, with the client refused or rebuilt on a change. | |
| +| Both, P1: the owner of a database is resolved once, when its client is built, and never again. A route moved to another account, or an instance deleted and made again, goes on being counted to the first owner. | Fixed in the second commit after `40558c1d`. Ownership is read under the registry's change witness (`registry.RouteObservation`), checked again once the read is done, and every recorder a client is given asks the witness before it starts a request. After ANY commit to the registry no request is started or counted, the owner's ledger is told its day is incomplete, and `MeteredFleet.Current` says the store must be rebuilt. The rule is blunt on purpose: this code cannot tell a change of owner from any other change. Reproduction: `TestMeteredFleetStopsCountingWhenTheRegistryChanges` (the route moves account; starts hibernating; an unrelated route is added). Still open: one request already past its check when the registry changes is counted to the owner as read; and nothing yet rebuilds the store, because nothing calls it. | |
| | Both, P1: ownership and the duplicate-file check compare paths as text. A tenant symlink, or two spellings of one path, bills one tenant for another's database. | Fixed in the commit after `40558c1d`. A listed path is owned only if nothing beneath the claiming route's directory is a link, and no two listed paths resolve to one file; either makes it unowned and the replicator refuses to start. The store compares files by identity (`os.SameFile`). Shadow state moved out of the tenant's directory to `<data dir>/.replicator/<replica key>`. Reproductions: `TestFleetDBOwnersDoesNotFollowATenantsLink`, `TestMeteredStoreRefusesWhatItCannotAttribute`. Still open: the check is made when the store is built, and a tenant can replace a file with a link afterwards; closing that needs the databases opened beneath their root, and the ownership re-check of the first row. | |
| | Astra P2, Opus P3-5: a discovery error, or an unreadable or vanished database, becomes successful partial coverage | Accepted. Open. Must refuse or hold. | |
| | Astra P2: unknown databases are dropped before the refusal check | Accepted. Open. | |
| @@ -18,7 +18,7 @@ Reports: `CODEX-FINDINGS-metered-replicator-review1.md`, `OPUS-REVIEW-metered-re |
| | Opus P2-2: nothing enforces that the replica prefix is not the external binary's lineage | Accepted. Open. | |
| | Opus P2-4: the fleet store drops `skipped` and has no shrink guard | Accepted. Open. | |
| | Both, tests: none shows the right recorder on the right client or a store-driven request being metered; several only exercise a fake | Accepted. Open. | |
| -| Opus P3-1 to P3-4: two non-atomic registry reads; an arbitrary instance for one account and app; prefix uniqueness by exact string; a typed-nil recorder | Accepted. Open. | |
| +| Opus P3-1 (two non-atomic registry reads): closed by the witness, which is taken before the reads and checked after them. Opus P3-2 to P3-4: an arbitrary instance for one account and app; prefix uniqueness by exact string; a typed-nil recorder | Accepted. Open. | |
| | Opus: what is still missing before this could attest, beyond the launch branch's item 1 | Recorded in the report; to be carried into the plan. | |
| -Mutation checks in a scratch copy, each failing its test: the link check off; a file listed twice accepted; the store comparing paths and not files. |
| +Mutation checks in a scratch copy, each failing its test: the link check off; a file listed twice accepted; the store comparing paths and not files. The recorder's check removed; the witness's answer ignored. |
diff --git a/internal/ops/litestream_metered.go b/internal/ops/litestream_metered.go| index 2fc7f4c0f..21f50a7e0 100644 |
| --- a/internal/ops/litestream_metered.go |
| +++ b/internal/ops/litestream_metered.go |
| @@ -1,10 +1,13 @@ |
| package ops |
| import ( |
| + "context" |
| "errors" |
| "fmt" |
| "path/filepath" |
| "strings" |
| + "sync" |
| + "time" |
| "github.com/benbjohnson/litestream" |
| @@ -19,6 +22,81 @@ import ( |
| // through its ledger. |
| var ErrUnmeterable = errors.New("ops: a replicated database has no owner to meter it against") |
| +// ErrOwnershipStale: the registry has changed since this replicator read |
| +// who owns each database. A route may have moved to another account, been |
| +// deleted, been made again, or started hibernating; the meters this store |
| +// was built with may now be the wrong ones. Every replica request is |
| +// refused from then on, and the store has to be built again. |
| +var ErrOwnershipStale = errors.New("ops: the registry changed since ownership was read; the replicator must be rebuilt") |
| + |
| +// MeteredFleet is the platform's replicator for one box, with the witness |
| +// its ownership was read under. Close releases the witness. |
| +type MeteredFleet struct { |
| + Store *litestream.Store |
| + Owned []OwnedDB |
| + |
| + mu sync.Mutex |
| + obs *registry.RouteObservation |
| + stale bool |
| + recorders map[string]lease.RequestRecorder // by database path; what each client was given |
| +} |
| + |
| +// Current reports whether ownership is still as it was read: nil, or |
| +// ErrOwnershipStale once any commit has landed on the registry since. It |
| +// asks the registry's own change witness (registry.RouteObservation), which |
| +// sees a change that was made and undone as well as one that stands. Once |
| +// stale, always stale. A witness that cannot be read is not proof of no |
| +// change, and counts as one. |
| +func (f *MeteredFleet) Current(ctx context.Context) error { |
| + f.mu.Lock() |
| + defer f.mu.Unlock() |
| + if f.stale { |
| + return ErrOwnershipStale |
| + } |
| + if f.obs == nil { |
| + f.stale = true |
| + return ErrOwnershipStale |
| + } |
| + ctx, cancel := context.WithTimeout(ctx, 5*time.Second) |
| + defer cancel() |
| + if ok, err := f.obs.Unchanged(ctx); err != nil || !ok { |
| + f.stale = true |
| + return ErrOwnershipStale |
| + } |
| + return nil |
| +} |
| + |
| +// Close releases the registry witness. The fleet is stale afterwards. |
| +func (f *MeteredFleet) Close() error { |
| + f.mu.Lock() |
| + defer f.mu.Unlock() |
| + f.stale = true |
| + if f.obs == nil { |
| + return nil |
| + } |
| + err := f.obs.Close() |
| + f.obs = nil |
| + return err |
| +} |
| + |
| +// ownedRecorder is the recorder a replica client is actually given: the |
| +// owner's own, behind a check that the owner is still the owner. A request |
| +// that would be made after the registry changed is not started and is not |
| +// counted to anybody. The owner's ledger is told its day is incomplete, |
| +// because a replica that stops being written is a gap, not a zero. |
| +type ownedRecorder struct { |
| + lease.RequestRecorder |
| + fleet *MeteredFleet |
| +} |
| + |
| +func (r ownedRecorder) Start(operation string, at time.Time) (uint64, error) { |
| + if err := r.fleet.Current(context.Background()); err != nil { |
| + r.RequestRecorder.Incomplete(at) |
| + return 0, err |
| + } |
| + return r.RequestRecorder.Start(operation, at) |
| +} |
| + |
| // MeteredFleetStore builds the platform's own replicator for this box: one |
| // litestream store over every database FleetDBs lists, each replica client |
| // metered against the database's owner (path B in |
| @@ -28,36 +106,68 @@ var ErrUnmeterable = errors.New("ops: a replicated database has no owner to mete |
| // box's durability ledger. It is asked once for each database. A nil |
| // recorder refuses the whole store: there is no unmetered fallback. |
| // |
| +// Who owns each database is read once, here, and a route can change hands |
| +// afterwards. So the read is taken under the registry's change witness, |
| +// and every recorder is wrapped (ownedRecorder): the first request after |
| +// ANY commit to the registry is refused, and every one after it. The rule |
| +// is deliberately blunt. This code cannot tell a change that moves an |
| +// owner from one that does not, and counting a write to an account that |
| +// no longer owns the database is the one thing it must never do. The |
| +// caller rebuilds the store when Current says so. One request already |
| +// past its check when the registry changes can still be counted to the |
| +// owner as read; closing that needs the registry's writer and this |
| +// replicator to share a lock, which nothing here does. |
| +// |
| // prefix is the replica lineage this replicator writes. It must not be one |
| // the external binary has written to: the two do not share a format or a |
| // metadata directory, and a restore from a prefix both have touched is not |
| // something anyone has tested. |
| // |
| -// The store is returned unopened, and nothing here attests anything. A box |
| +// The store is returned unopened, inside a MeteredFleet the caller must |
| +// Close, and nothing here attests anything. A box |
| // may say that every replica write was counted only when this is the only |
| // thing on it that holds the replica bucket's credentials, which is a |
| // change to the box and not to this code. |
| -func MeteredFleetStore(reg *registry.Registry, dataDir, bucket, prefix, region string, meter func(OwnedDB) lease.RequestRecorder) (*litestream.Store, []OwnedDB, error) { |
| +func MeteredFleetStore(reg *registry.Registry, dataDir, bucket, prefix, region string, meter func(OwnedDB) lease.RequestRecorder) (fleet *MeteredFleet, err error) { |
| if meter == nil { |
| - return nil, nil, fmt.Errorf("%w: no meter", ErrUnmeterable) |
| + return nil, fmt.Errorf("%w: no meter", ErrUnmeterable) |
| + } |
| + // The witness first, then the ownership read, then the witness again: |
| + // what was read is what the registry held for the whole of the read. |
| + obs, err := reg.ObserveRoutes(context.Background()) |
| + if err != nil { |
| + return nil, err |
| } |
| + built := &MeteredFleet{obs: obs, recorders: map[string]lease.RequestRecorder{}} |
| + fleet = built |
| + defer func() { |
| + if err != nil { |
| + built.Close() |
| + fleet = nil |
| + } |
| + }() |
| owned, unowned, _, err := FleetDBOwners(reg, dataDir) |
| if err != nil { |
| - return nil, nil, err |
| + return nil, err |
| + } |
| + if err := fleet.Current(context.Background()); err != nil { |
| + return nil, fmt.Errorf("%w (while ownership was being read)", err) |
| } |
| if len(unowned) > 0 { |
| - return nil, nil, fmt.Errorf("%w: %s", ErrUnmeterable, strings.Join(unowned, ", ")) |
| + return nil, fmt.Errorf("%w: %s", ErrUnmeterable, strings.Join(unowned, ", ")) |
| } |
| if len(owned) == 0 { |
| - return nil, nil, errors.New("ops: no databases to replicate") |
| + return nil, errors.New("ops: no databases to replicate") |
| } |
| prefix = strings.Trim(prefix, "/") |
| dbs := make([]lease.MeteredDB, 0, len(owned)) |
| for _, o := range owned { |
| - rec := meter(o) |
| - if rec == nil { |
| - return nil, nil, fmt.Errorf("%w: no recorder for %s (%s/%s)", ErrUnmeterable, o.Path, o.Account, o.App) |
| + inner := meter(o) |
| + if inner == nil { |
| + return nil, fmt.Errorf("%w: no recorder for %s (%s/%s)", ErrUnmeterable, o.Path, o.Account, o.App) |
| } |
| + rec := ownedRecorder{RequestRecorder: inner, fleet: fleet} |
| + fleet.recorders[o.Path] = rec |
| key := replicaKey(o.Path) |
| if prefix != "" { |
| key = prefix + "/" + key |
| @@ -69,7 +179,8 @@ func MeteredFleetStore(reg *registry.Registry, dataDir, bucket, prefix, region s |
| } |
| store, err := lease.NewMeteredStore(bucket, region, dbs) |
| if err != nil { |
| - return nil, nil, err |
| + return nil, err |
| } |
| - return store, owned, nil |
| + fleet.Store, fleet.Owned = store, owned |
| + return fleet, nil |
| } |
diff --git a/internal/ops/litestream_metered_test.go b/internal/ops/litestream_metered_test.go| index 305b12248..72910213a 100644 |
| --- a/internal/ops/litestream_metered_test.go |
| +++ b/internal/ops/litestream_metered_test.go |
| @@ -1,6 +1,7 @@ |
| package ops |
| import ( |
| + "context" |
| "errors" |
| "os" |
| "path/filepath" |
| @@ -41,10 +42,12 @@ func TestMeteredFleetStoreRefusesADatabaseItCannotMeter(t *testing.T) { |
| asked = append(asked, o) |
| return nopRecorder{} |
| } |
| - store, dbs, err := MeteredFleetStore(reg, dir, "replica-bucket", "/metered/v1/", "eu-west-1", meter) |
| + fleet, err := MeteredFleetStore(reg, dir, "replica-bucket", "/metered/v1/", "eu-west-1", meter) |
| if err != nil { |
| t.Fatal(err) |
| } |
| + defer fleet.Close() |
| + store, dbs := fleet.Store, fleet.Owned |
| if len(dbs) != 2 || len(asked) != 2 || len(store.DBs()) != 2 { |
| t.Fatalf("store over %d databases, %d owners, meter asked %d times: want the route's database and the registry", len(store.DBs()), len(dbs), len(asked)) |
| } |
| @@ -57,7 +60,7 @@ func TestMeteredFleetStoreRefusesADatabaseItCannotMeter(t *testing.T) { |
| } |
| } |
| // A recorder that cannot be bound: nothing is replicated unmetered. |
| - _, _, err = MeteredFleetStore(reg, dir, "replica-bucket", "metered/v1", "eu-west-1", func(o OwnedDB) lease.RequestRecorder { |
| + _, err = MeteredFleetStore(reg, dir, "replica-bucket", "metered/v1", "eu-west-1", func(o OwnedDB) lease.RequestRecorder { |
| if o.Path == owned { |
| return nil |
| } |
| @@ -66,7 +69,7 @@ func TestMeteredFleetStoreRefusesADatabaseItCannotMeter(t *testing.T) { |
| if !errors.Is(err, ErrUnmeterable) || !strings.Contains(err.Error(), owned) { |
| t.Fatalf("a database with no recorder: %v, want ErrUnmeterable naming it", err) |
| } |
| - if _, _, err := MeteredFleetStore(reg, dir, "replica-bucket", "metered/v1", "eu-west-1", nil); !errors.Is(err, ErrUnmeterable) { |
| + if _, err := MeteredFleetStore(reg, dir, "replica-bucket", "metered/v1", "eu-west-1", nil); !errors.Is(err, ErrUnmeterable) { |
| t.Fatalf("no meter at all: %v", err) |
| } |
| // A database two accounts claim. |
| @@ -82,8 +85,104 @@ func TestMeteredFleetStoreRefusesADatabaseItCannotMeter(t *testing.T) { |
| t.Fatal(err) |
| } |
| } |
| - _, _, err = MeteredFleetStore(reg, dir, "replica-bucket", "metered/v1", "eu-west-1", meter) |
| + _, err = MeteredFleetStore(reg, dir, "replica-bucket", "metered/v1", "eu-west-1", meter) |
| if !errors.Is(err, ErrUnmeterable) || !strings.Contains(err.Error(), shared) { |
| t.Fatalf("a database with no single owner: %v, want ErrUnmeterable naming it", err) |
| } |
| } |
| + |
| +// countingRecorder counts what it is asked to start and remembers being |
| +// told its day is incomplete. |
| +type countingRecorder struct { |
| + started int |
| + incomplete int |
| +} |
| + |
| +func (c *countingRecorder) Start(string, time.Time) (uint64, error) { |
| + c.started++ |
| + return uint64(c.started), nil |
| +} |
| +func (c *countingRecorder) Finish(uint64, lease.RequestOutcome) error { return nil } |
| +func (c *countingRecorder) Abandon(uint64, time.Time) {} |
| +func (c *countingRecorder) Incomplete(time.Time) { c.incomplete++ } |
| +func (c *countingRecorder) Hold(time.Time) func() { return func() {} } |
| + |
| +// Who owns a database is read when the replicator is built. A route can |
| +// change hands afterwards: moved to another account, deleted and made |
| +// again, put to sleep. The recorder a replica client holds must not go on |
| +// counting to the owner it was built for. After any commit to the |
| +// registry, a request is not started and not counted to anybody, the old |
| +// owner's ledger is told its day is incomplete, and the fleet says it must |
| +// be rebuilt. |
| +func TestMeteredFleetStopsCountingWhenTheRegistryChanges(t *testing.T) { |
| + for name, change := range map[string]func(t *testing.T, reg *registry.Registry, db string){ |
| + "the route moves to another account": func(t *testing.T, reg *registry.Registry, db string) { |
| + if err := reg.Upsert(registry.Route{Host: "a.example.com", Account: "bae", App: "one", DBPath: db}); err != nil { |
| + t.Fatal(err) |
| + } |
| + }, |
| + "the route starts hibernating": func(t *testing.T, reg *registry.Registry, db string) { |
| + if err := reg.Upsert(registry.Route{Host: "a.example.com", Account: "bac", App: "one", DBPath: db, Hibernate: true}); err != nil { |
| + t.Fatal(err) |
| + } |
| + }, |
| + "an unrelated route is added": func(t *testing.T, reg *registry.Registry, db string) { |
| + if err := reg.Upsert(registry.Route{Host: "other.example.com", Account: "bae", App: "two"}); err != nil { |
| + t.Fatal(err) |
| + } |
| + }, |
| + } { |
| + dir := t.TempDir() |
| + reg, err := registry.Open(filepath.Join(dir, "registry.db")) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + db := filepath.Join(dir, "a.example.com.db") |
| + if err := os.WriteFile(db, []byte("x"), 0o644); err != nil { |
| + t.Fatal(err) |
| + } |
| + if err := reg.Upsert(registry.Route{Host: "a.example.com", Account: "bac", App: "one", DBPath: db}); err != nil { |
| + t.Fatal(err) |
| + } |
| + inner := map[string]*countingRecorder{} |
| + fleet, err := MeteredFleetStore(reg, dir, "replica-bucket", "metered/v1", "eu-west-1", func(o OwnedDB) lease.RequestRecorder { |
| + c := &countingRecorder{} |
| + inner[o.Path] = c |
| + return c |
| + }) |
| + if err != nil { |
| + t.Fatalf("%s: %v", name, err) |
| + } |
| + rec := fleet.recorders[db] |
| + if rec == nil || inner[db] == nil { |
| + t.Fatalf("%s: fixture: no recorder for the route's database", name) |
| + } |
| + at := time.Date(2026, 10, 1, 10, 0, 0, 0, time.UTC) |
| + if _, err := rec.Start("PutObject", at); err != nil || inner[db].started != 1 { |
| + t.Fatalf("%s: a request before any change: %v, started %d", name, err, inner[db].started) |
| + } |
| + if err := fleet.Current(context.Background()); err != nil { |
| + t.Fatalf("%s: current before any change: %v", name, err) |
| + } |
| + change(t, reg, db) |
| + if _, err := rec.Start("PutObject", at.Add(time.Second)); !errors.Is(err, ErrOwnershipStale) { |
| + t.Fatalf("%s: a request after the change: %v, want ErrOwnershipStale", name, err) |
| + } |
| + if inner[db].started != 1 { |
| + t.Fatalf("%s: a request after the change was counted to the owner as first read (%d started)", name, inner[db].started) |
| + } |
| + if inner[db].incomplete != 1 { |
| + t.Fatalf("%s: the old owner's ledger was not told its day is incomplete", name) |
| + } |
| + if err := fleet.Current(context.Background()); !errors.Is(err, ErrOwnershipStale) { |
| + t.Fatalf("%s: the fleet does not say it must be rebuilt: %v", name, err) |
| + } |
| + // Stale for good: every recorder, the platform's included. |
| + for path, r := range fleet.recorders { |
| + if _, err := r.Start("PutObject", at.Add(2*time.Second)); !errors.Is(err, ErrOwnershipStale) { |
| + t.Fatalf("%s: %s still counts after the change: %v", name, path, err) |
| + } |
| + } |
| + fleet.Close() |
| + } |
| +} |
diff --git a/internal/ops/litestream_owners_test.go b/internal/ops/litestream_owners_test.go| index af3b57bf0..0cf8bb98a 100644 |
| --- a/internal/ops/litestream_owners_test.go |
| +++ b/internal/ops/litestream_owners_test.go |
| @@ -172,7 +172,7 @@ func TestFleetDBOwnersDoesNotFollowATenantsLink(t *testing.T) { |
| if !isUnowned[theirs] { |
| t.Errorf("a file listed twice (once through a link) is not named as unowned: %v", unowned) |
| } |
| - if _, _, err := MeteredFleetStore(reg, dir, "test-bucket", "replicas", "eu-west-1", func(OwnedDB) lease.RequestRecorder { return nopRecorder{} }); !errors.Is(err, ErrUnmeterable) { |
| + if _, err := MeteredFleetStore(reg, dir, "test-bucket", "replicas", "eu-west-1", func(OwnedDB) lease.RequestRecorder { return nopRecorder{} }); !errors.Is(err, ErrUnmeterable) { |
| t.Fatalf("the replicator over a tenant's link: %v, want ErrUnmeterable", err) |
| } |
| } |