amadan
The replicator's fleet keeps its store private, names what it skips, refuses what it cannot read, and meters the platform's databases under identities the ledger accepts
From the first review of this branch (P2 findings): - MeteredFleet no longer exposes the litestream store, whose own RegisterDB takes an unmetered database. - A database-like file no route claims refuses the store unless the caller names it as seen; named ones are listed. - A unit tenant's state directory with a part that cannot be read refuses the store: discovery passes over it in silence. - The registry and console databases carry fixed app and host names that the durability ledger's attribution accepts. - The store is given the fleet's seven-day snapshot retention. Each has a test and was mutation-checked. Open items are in DISPOSITIONS-metered-replicator.md. Nothing calls this code. Nothing attests.
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
5 files changed,
+236
−25
docs/superpowers/specs/DISPOSITIONS-metered-replicator.md+7 −7internal/ops/litestream_metered.go+59 −4internal/ops/litestream_metered_test.go+103 −6internal/ops/litestream_owners.go+66 −7internal/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 cd4bdce25..24e7c9995 100644 |
| --- a/docs/superpowers/specs/DISPOSITIONS-metered-replicator.md |
| +++ b/docs/superpowers/specs/DISPOSITIONS-metered-replicator.md |
| @@ -4,21 +4,21 @@ 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`. The two P1 rows are fixed; the rest are open. |
| +Reports: `CODEX-FINDINGS-metered-replicator-review1.md`, `OPUS-REVIEW-metered-replicator-review1.md`. The two P1 rows and five P2 rows are fixed or partly fixed, as each says; the rest are open. None of the fixes has been read by a reviewer. |
| | 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. | 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. | |
| -| Astra P2, Opus P2-1: "refuses anything it cannot meter" is true only while the store is built; the returned store still exposes an ordinary registration path | Accepted. Open. | |
| -| Astra P2: platform ownership records cannot bind to the real ledger as written | Accepted. Open. | |
| -| Astra P2: the replacement drops the seven-day recovery window the external replicator keeps | Accepted. Open. | |
| +| Astra P2, Opus P3-5: a discovery error, or an unreadable or vanished database, becomes successful partial coverage | Fixed for unreadable directories: the constructor walks each unit tenant's state directory as discovery does and refuses the store if any part cannot be read (`TestMeteredFleetStoreRefusesAStateDirectoryItCannotRead`). A database that vanishes after the store is built is still open. | |
| +| Astra P2: unknown databases are dropped before the refusal check | Fixed. A database-like file no route claims refuses the store unless the caller names it as seen; named ones are listed on the fleet (`Skipped`). The registry is recognised by the path it is open at. Comparing against the external replicator's own list is still open. `TestMeteredFleetStoreRefusesAnUnclaimedDatabaseNobodyNamed`. | |
| +| Astra P2, Opus P2-1: "refuses anything it cannot meter" is true only while the store is built; the returned store still exposes an ordinary registration path | Fixed at the fleet: `MeteredFleet` keeps the litestream store private and offers only `Open`, `Close`, `Current` and `Databases`. `lease.NewMeteredStore` still returns the bare store to its one caller. | |
| +| Astra P2: platform ownership records cannot bind to the real ledger as written | Fixed. The registry and the console database carry fixed identities that the ledger's attribution accepts for the platform's purpose (`TestPlatformDatabasesHaveIdentitiesTheLedgerAccepts`). | |
| +| Astra P2: the replacement drops the seven-day recovery window the external replicator keeps | Fixed in the setting: the store is given the fleet's snapshot interval and retention. That litestream then keeps a week of snapshots has not been exercised. | |
| | 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 (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. The recorder's check removed; the witness's answer ignored. |
| +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. The platform host blanked; the unclaimed-file refusal off; the unreadable-directory refusal off; the recovery window not set. |
diff --git a/internal/ops/litestream_metered.go b/internal/ops/litestream_metered.go| index 21f50a7e0..c0f7c16ee 100644 |
| --- a/internal/ops/litestream_metered.go |
| +++ b/internal/ops/litestream_metered.go |
| @@ -32,8 +32,19 @@ var ErrOwnershipStale = errors.New("ops: the registry changed since ownership wa |
| // 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 is every database the store replicates, with its owner. |
| Owned []OwnedDB |
| + // Skipped lists the files under the data directory that look like |
| + // databases and that no route claims. They are not replicated. The |
| + // caller named each of them (MeteredFleetStore's accept), so none was |
| + // left out unseen. |
| + Skipped []string |
| + |
| + // store is kept private. A litestream store takes any database through |
| + // its own RegisterDB, with an ordinary unmetered client, and its DBs |
| + // hand out live clients that can be replaced: the constructor's |
| + // refusals would mean nothing to a caller holding it. |
| + store *litestream.Store |
| mu sync.Mutex |
| obs *registry.RouteObservation |
| @@ -67,6 +78,17 @@ func (f *MeteredFleet) Current(ctx context.Context) error { |
| } |
| // Close releases the registry witness. The fleet is stale afterwards. |
| +// Open starts replication. It refuses once ownership is stale. |
| +func (f *MeteredFleet) Open(ctx context.Context) error { |
| + if err := f.Current(ctx); err != nil { |
| + return err |
| + } |
| + return f.store.Open(ctx) |
| +} |
| + |
| +// Databases is how many databases the store replicates. |
| +func (f *MeteredFleet) Databases() int { return len(f.store.DBs()) } |
| + |
| func (f *MeteredFleet) Close() error { |
| f.mu.Lock() |
| defer f.mu.Unlock() |
| @@ -128,7 +150,13 @@ func (r ownedRecorder) Start(operation string, at time.Time) (uint64, error) { |
| // 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) (fleet *MeteredFleet, err error) { |
| +// |
| +// accept names the files discovery skips (database-like files under the |
| +// data directory that no route claims) that the caller has seen and means |
| +// to leave unreplicated. A skipped file that is not in accept refuses the |
| +// store: a database the external replicator still streams, whose route has |
| +// gone, must not drop out of replication because nobody looked. |
| +func MeteredFleetStore(reg *registry.Registry, dataDir, bucket, prefix, region string, accept []string, meter func(OwnedDB) lease.RequestRecorder) (fleet *MeteredFleet, err error) { |
| if meter == nil { |
| return nil, fmt.Errorf("%w: no meter", ErrUnmeterable) |
| } |
| @@ -146,13 +174,37 @@ func MeteredFleetStore(reg *registry.Registry, dataDir, bucket, prefix, region s |
| fleet = nil |
| } |
| }() |
| - owned, unowned, _, err := FleetDBOwners(reg, dataDir) |
| + owned, unowned, skipped, err := FleetDBOwners(reg, dataDir) |
| if err != nil { |
| return nil, err |
| } |
| + // Discovery passes over a directory it cannot read. Here that is a |
| + // directory that may hold a database nobody counted. |
| + for _, r := range obs.Routes { |
| + if r.Backing != registry.BackingUnit || r.DBPath == "" { |
| + continue |
| + } |
| + if bad, unreadable := unreadableUnder(filepath.Clean(r.DBPath)); unreadable { |
| + return nil, fmt.Errorf("%w: cannot read %s under %s's state directory", ErrUnmeterable, bad, r.Host) |
| + } |
| + } |
| if err := fleet.Current(context.Background()); err != nil { |
| return nil, fmt.Errorf("%w (while ownership was being read)", err) |
| } |
| + accepted := map[string]bool{} |
| + for _, p := range accept { |
| + accepted[filepath.Clean(p)] = true |
| + } |
| + var unseen []string |
| + for _, p := range skipped { |
| + if !accepted[filepath.Clean(p)] { |
| + unseen = append(unseen, p) |
| + } |
| + } |
| + if len(unseen) > 0 { |
| + return nil, fmt.Errorf("%w: database files no route claims, not named as accepted: %s", ErrUnmeterable, strings.Join(unseen, ", ")) |
| + } |
| + built.Skipped = skipped |
| if len(unowned) > 0 { |
| return nil, fmt.Errorf("%w: %s", ErrUnmeterable, strings.Join(unowned, ", ")) |
| } |
| @@ -181,6 +233,9 @@ func MeteredFleetStore(reg *registry.Registry, dataDir, bucket, prefix, region s |
| if err != nil { |
| return nil, err |
| } |
| - fleet.Store, fleet.Owned = store, owned |
| + // The fleet's recovery window, not litestream's one-day default: the |
| + // external replicator this replaces keeps a week of snapshots. |
| + store.SnapshotInterval, store.SnapshotRetention = SnapshotInterval, SnapshotRetention |
| + 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 72910213a..12f4af193 100644 |
| --- a/internal/ops/litestream_metered_test.go |
| +++ b/internal/ops/litestream_metered_test.go |
| @@ -11,6 +11,7 @@ import ( |
| "github.com/carlosframework/platform/internal/lease" |
| "github.com/carlosframework/platform/internal/registry" |
| + durability "github.com/carlosframework/platform/internal/usage" |
| ) |
| type nopRecorder struct{} |
| @@ -42,12 +43,12 @@ func TestMeteredFleetStoreRefusesADatabaseItCannotMeter(t *testing.T) { |
| asked = append(asked, o) |
| return nopRecorder{} |
| } |
| - fleet, err := MeteredFleetStore(reg, dir, "replica-bucket", "/metered/v1/", "eu-west-1", meter) |
| + fleet, err := MeteredFleetStore(reg, dir, "replica-bucket", "/metered/v1/", "eu-west-1", nil, meter) |
| if err != nil { |
| t.Fatal(err) |
| } |
| defer fleet.Close() |
| - store, dbs := fleet.Store, fleet.Owned |
| + 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)) |
| } |
| @@ -60,7 +61,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", nil, func(o OwnedDB) lease.RequestRecorder { |
| if o.Path == owned { |
| return nil |
| } |
| @@ -69,7 +70,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, nil); !errors.Is(err, ErrUnmeterable) { |
| t.Fatalf("no meter at all: %v", err) |
| } |
| // A database two accounts claim. |
| @@ -85,7 +86,7 @@ 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", nil, 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) |
| } |
| @@ -145,7 +146,7 @@ func TestMeteredFleetStopsCountingWhenTheRegistryChanges(t *testing.T) { |
| t.Fatal(err) |
| } |
| inner := map[string]*countingRecorder{} |
| - fleet, err := MeteredFleetStore(reg, dir, "replica-bucket", "metered/v1", "eu-west-1", func(o OwnedDB) lease.RequestRecorder { |
| + fleet, err := MeteredFleetStore(reg, dir, "replica-bucket", "metered/v1", "eu-west-1", nil, func(o OwnedDB) lease.RequestRecorder { |
| c := &countingRecorder{} |
| inner[o.Path] = c |
| return c |
| @@ -186,3 +187,99 @@ func TestMeteredFleetStopsCountingWhenTheRegistryChanges(t *testing.T) { |
| fleet.Close() |
| } |
| } |
| + |
| +// The identities the box's own databases are metered under are ones the |
| +// real durability ledger accepts for the platform's purpose. A no-op |
| +// recorder would never notice if they were not. |
| +func TestPlatformDatabasesHaveIdentitiesTheLedgerAccepts(t *testing.T) { |
| + dir := t.TempDir() |
| + reg, err := registry.Open(filepath.Join(dir, "registry.db")) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + owned, unowned, _, err := FleetDBOwners(reg, dir) |
| + if err != nil || len(unowned) != 0 || len(owned) == 0 { |
| + t.Fatalf("owned %+v, unowned %v, err %v", owned, unowned, err) |
| + } |
| + for _, o := range owned { |
| + if !o.Platform { |
| + t.Fatalf("fixture: %+v is not the platform's", o) |
| + } |
| + attr := durability.ReplicationAttribution{Account: o.Account, App: o.App, Host: o.Host, Purpose: durability.PurposePlatform} |
| + if err := attr.Validate(); err != nil { |
| + t.Errorf("%s: the ledger refuses the platform's identity %+v: %v", o.Path, attr, err) |
| + } |
| + } |
| +} |
| + |
| +// A database-like file under the data directory that no route claims is |
| +// not replicated. It may be one the external replicator still streams, for |
| +// a route that has gone. The store is refused until the caller names it as |
| +// seen; then it is listed, not dropped. |
| +func TestMeteredFleetStoreRefusesAnUnclaimedDatabaseNobodyNamed(t *testing.T) { |
| + dir := t.TempDir() |
| + reg, err := registry.Open(filepath.Join(dir, "registry.db")) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + leftover := filepath.Join(dir, "gone.example.com.db") |
| + if err := os.WriteFile(leftover, []byte("x"), 0o644); err != nil { |
| + t.Fatal(err) |
| + } |
| + meter := func(OwnedDB) lease.RequestRecorder { return nopRecorder{} } |
| + fleet, err := MeteredFleetStore(reg, dir, "replica-bucket", "metered/v1", "eu-west-1", nil, meter) |
| + if !errors.Is(err, ErrUnmeterable) || !strings.Contains(err.Error(), leftover) || fleet != nil { |
| + t.Fatalf("an unclaimed database nobody named: fleet %v, err %v; want ErrUnmeterable naming it and no fleet", fleet, err) |
| + } |
| + fleet, err = MeteredFleetStore(reg, dir, "replica-bucket", "metered/v1", "eu-west-1", []string{leftover}, meter) |
| + if err != nil { |
| + t.Fatalf("with the file named as seen: %v", err) |
| + } |
| + defer fleet.Close() |
| + if len(fleet.Skipped) != 1 || fleet.Skipped[0] != leftover { |
| + t.Fatalf("Skipped = %v, want the named file", fleet.Skipped) |
| + } |
| + // The fleet's recovery window, not litestream's default day. |
| + if fleet.store.SnapshotRetention != SnapshotRetention || fleet.store.SnapshotInterval != SnapshotInterval { |
| + t.Fatalf("snapshot policy %s every %s, want %s every %s", fleet.store.SnapshotRetention, fleet.store.SnapshotInterval, SnapshotRetention, SnapshotInterval) |
| + } |
| + if fleet.Databases() != 1 { |
| + t.Fatalf("%d databases, want the registry alone", fleet.Databases()) |
| + } |
| +} |
| + |
| +// Discovery passes over a directory it cannot read. For the replicator |
| +// that is a directory that may hold a database nobody counted, and it |
| +// refuses to start. |
| +func TestMeteredFleetStoreRefusesAStateDirectoryItCannotRead(t *testing.T) { |
| + if os.Geteuid() == 0 { |
| + t.Skip("root reads every directory") |
| + } |
| + dir := t.TempDir() |
| + reg, err := registry.Open(filepath.Join(dir, "registry.db")) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + state := filepath.Join(t.TempDir(), "unit-state") |
| + hidden := filepath.Join(state, "tenants") |
| + if err := os.MkdirAll(hidden, 0o755); err != nil { |
| + t.Fatal(err) |
| + } |
| + if err := os.WriteFile(filepath.Join(state, "main.db"), []byte("x"), 0o644); err != nil { |
| + t.Fatal(err) |
| + } |
| + if err := os.WriteFile(filepath.Join(hidden, "t1.db"), []byte("x"), 0o644); err != nil { |
| + t.Fatal(err) |
| + } |
| + if err := reg.Upsert(registry.Route{Host: "unit.example.com", Account: "bad", App: "unit", DBPath: state, Backing: registry.BackingUnit}); err != nil { |
| + t.Fatal(err) |
| + } |
| + if err := os.Chmod(hidden, 0o000); err != nil { |
| + t.Fatal(err) |
| + } |
| + t.Cleanup(func() { os.Chmod(hidden, 0o755) }) |
| + fleet, err := MeteredFleetStore(reg, dir, "replica-bucket", "metered/v1", "eu-west-1", nil, func(OwnedDB) lease.RequestRecorder { return nopRecorder{} }) |
| + if !errors.Is(err, ErrUnmeterable) || !strings.Contains(err.Error(), hidden) || fleet != nil { |
| + t.Fatalf("a state directory with an unreadable part: fleet %v, err %v; want ErrUnmeterable naming it and no fleet", fleet, err) |
| + } |
| +} |
diff --git a/internal/ops/litestream_owners.go b/internal/ops/litestream_owners.go| index 564a5e64d..11c633695 100644 |
| --- a/internal/ops/litestream_owners.go |
| +++ b/internal/ops/litestream_owners.go |
| @@ -16,11 +16,23 @@ type OwnedDB struct { |
| Path string |
| // Account, App and Host name the route that owns the database. For the |
| // box's own databases (the registry and the console) Account is the |
| - // operations account and Platform is true. |
| + // operations account, Platform is true, and App and Host are the fixed |
| + // names below: the durability ledger's attribution needs a valid app |
| + // and host for the platform's purpose too, and a tenant's are never |
| + // borrowed for it. |
| Account, App, Host string |
| Platform bool |
| } |
| +// The identities the box's own databases are metered under. They are not |
| +// routes: the registry and the console database belong to the platform. |
| +const ( |
| + PlatformRegistryApp = "registry" |
| + PlatformRegistryHost = "registry.platform.internal" |
| + PlatformConsoleApp = "console" |
| + PlatformConsoleHost = "console.platform.internal" |
| +) |
| + |
| // FleetDBOwners is FleetDBs with an owner for each database, for the |
| // platform's own replicator (path B in |
| // docs/superpowers/plans/2026-10-01-durability-attestation-path.md): every |
| @@ -55,9 +67,15 @@ func FleetDBOwners(reg *registry.Registry, dataDir string) (owned []OwnedDB, uno |
| if err != nil { |
| return nil, nil, nil, err |
| } |
| - platform := map[string]bool{ |
| - filepath.Clean(filepath.Join(dataDir, "registry.db")): true, |
| - filepath.Clean(consoleDBPath(dataDir)): true, |
| + // The box's own databases, by what they are: the registry this |
| + // process has open (wherever it lives), the conventional registry |
| + // path, and the console's database. |
| + platform := map[string]OwnedDB{ |
| + filepath.Clean(filepath.Join(dataDir, "registry.db")): {Account: release.OpsAccount, App: PlatformRegistryApp, Host: PlatformRegistryHost, Platform: true}, |
| + filepath.Clean(consoleDBPath(dataDir)): {Account: release.OpsAccount, App: PlatformConsoleApp, Host: PlatformConsoleHost, Platform: true}, |
| + } |
| + if p := reg.Path(); p != "" { |
| + platform[filepath.Clean(p)] = OwnedDB{Account: release.OpsAccount, App: PlatformRegistryApp, Host: PlatformRegistryHost, Platform: true} |
| } |
| // The file each listed path is, by its resolved name. Two listed paths |
| // that are one file are both unowned: one stream must never be written |
| @@ -78,12 +96,13 @@ func FleetDBOwners(reg *registry.Registry, dataDir string) (owned []OwnedDB, uno |
| unowned = append(unowned, p) |
| continue |
| } |
| - if platform[p] { |
| - if !noLinkBeneath(filepath.Clean(dataDir), p) { |
| + if own, ok := platform[p]; ok { |
| + if !noLinkBeneath(filepath.Dir(p), p) { |
| unowned = append(unowned, p) |
| continue |
| } |
| - owned = append(owned, OwnedDB{Path: p, Account: release.OpsAccount, Platform: true}) |
| + own.Path = p |
| + owned = append(owned, own) |
| continue |
| } |
| // Every route that claims the file, exactly or by a directory above |
| @@ -165,3 +184,43 @@ func noLinkBeneath(root, path string) bool { |
| } |
| return true |
| } |
| + |
| +// unreadableUnder walks a unit tenant's state directory the way discovery |
| +// does (treeDBs: the same depth, dot names skipped) and names the first |
| +// directory that cannot be read. Discovery passes such a directory over in |
| +// silence, which is right for a config generator and wrong for a |
| +// replicator that has to say every database was counted: a directory it |
| +// cannot read may hold one. |
| +func unreadableUnder(root string) (string, bool) { |
| + var bad string |
| + var walk func(dir string, depth int) bool |
| + walk = func(dir string, depth int) bool { |
| + if depth > treeDBWalkDepth { |
| + return true |
| + } |
| + entries, err := os.ReadDir(dir) |
| + if err != nil { |
| + bad = dir |
| + return false |
| + } |
| + for _, e := range entries { |
| + if strings.HasPrefix(e.Name(), ".") { |
| + continue |
| + } |
| + p := filepath.Join(dir, e.Name()) |
| + fi, err := os.Stat(p) |
| + if err != nil { |
| + bad = p |
| + return false |
| + } |
| + if fi.IsDir() && !walk(p, depth+1) { |
| + return false |
| + } |
| + } |
| + return true |
| + } |
| + if walk(root, 0) { |
| + return "", false |
| + } |
| + return bad, true |
| +} |
diff --git a/internal/ops/litestream_owners_test.go b/internal/ops/litestream_owners_test.go| index 0cf8bb98a..e9d461daa 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", nil, func(OwnedDB) lease.RequestRecorder { return nopRecorder{} }); !errors.Is(err, ErrUnmeterable) { |
| t.Fatalf("the replicator over a tenant's link: %v, want ErrUnmeterable", err) |
| } |
| } |