amadan
The replicator's fourth build: the third review's bounded findings
A unit route's named file is checked from the top of the tenant's own directory chain, so a link in a directory above it is refused (Opus P1-1). The hold passes to the replacement at Open, after the old fleet's last reading at Close (P2-1), and reaches a margin past the finding (P3-1). A transient witness error is retried before it counts as a change (P3-2); the stale flag is checked again under the admission lock (Astra P1-4); a route's rowid cannot be moved (Astra P2-5). Lineage and shadow state are keyed by owner (P2-4). A unit route with no path, an empty claimed file, a name in another case and a database under any name in a tenant's tree are reported, not passed over (P2-3). What stays open, the swap during litestream's first open above all, is in DISPOSITIONS-metered-replicator.md, review three.
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
8 files changed,
+417
−100
docs/superpowers/specs/DISPOSITIONS-metered-replicator.md+24 −0internal/ops/litestream.go+22 −2internal/ops/litestream_metered.go+98 −36internal/ops/litestream_metered_test.go+75 −48internal/ops/litestream_owners.go+89 −14internal/ops/litestream_owners_test.go+74 −0internal/registry/observe_ownership_test.go+28 −0internal/registry/registry.go+7 −0
CI failed — run details
diff --git a/docs/superpowers/specs/DISPOSITIONS-metered-replicator.md b/docs/superpowers/specs/DISPOSITIONS-metered-replicator.md| index 514ada5a3..07406e6f1 100644 |
| --- a/docs/superpowers/specs/DISPOSITIONS-metered-replicator.md |
| +++ b/docs/superpowers/specs/DISPOSITIONS-metered-replicator.md |
| @@ -74,3 +74,27 @@ Mutation checks in a scratch copy, twenty-five. Twenty-two failed their tests: t |
| Not built, and still needed before this could replace the external binary: the replica credentials out of everything else's reach; a restore path; the prefix-lineage guard; the shrink guard; opening databases beneath a pinned root; a caller that rebuilds the fleet; the registry migration reviewed as a schema change on every box (it adds one table and three triggers, and an older build writing to a migrated registry still moves the counter). |
| +## Review three (reviewed `c8627694`; Opus accepted with fixes, Astra rejected) |
| + |
| +Reports: `OPUS-REVIEW-metered-replicator-review3.md`, `CODEX-FINDINGS-metered-replicator-review3.md`. Both found the request gate closed (a refused attempt reaches neither the network nor the meter, on every client litestream builds), the ownership generation closed for the routes table and every ordinary writer, `Close` closed, and the shadow state private beneath `.replicator`. The findings, and what the fourth build did: |
| + |
| +| Finding | Disposition | Where | Reproduction | |
| +| --- | --- | --- | --- | |
| +| Opus P1-1, Astra P1: a unit route that names one file was checked as a named file: `Dir(p)` reached through `Stat`, nothing above it checked, no owner check. A tenant who renames the directory and puts a link in its place has another tenant's database streamed as its own. | Fixed. A unit route's file is a `tenantFile`: the check climbs from the file's directory while the directory above is the same user's and not a link, requires the directory above the top of that chain to be the platform's, refuses a link anywhere from the top down (the top included), and requires the file to be the chain's user's. An exec route's file stays a `namedFile` (its directory is the activator's). | `litestream_owners.go` `checkOwnedFile` | `TestFleetDBOwnersCheckAUnitRoutesNamedFileLikeATree` | |
| +| Opus P1-2: a swap during litestream's first open pins it to the wrong file for good; `lookAgain` compares the path's current file, not what litestream has open. | Open. Needs litestream to expose the identity of the file it opened, or to open beneath a pinned root; a litestream patch, not made. Stated in `checkOwnedFile`'s comment and here: the second look does not narrow this case. | | | |
| +| Opus P2-1: the hold was passed to the replacement at build, so a fleet built before the old one noticed inherited nothing. | Fixed. `Close` asks the witness one last time before releasing it, and the replacement reads the old fleet's window at `Open`, which already requires the old fleet closed. | `litestream_metered.go` `Close`, `Open` | `TestMeteredFleetReplacementInheritsTheHold` (a case where the old fleet notices only as it closes) | |
| +| Opus P2-2: any route added or removed stops the replicator and holds every day; most such routes have no database. Not stated. | Stated. The generation moves on every insert and delete of a route on purpose: whether a new route claims a database is not known until its files are looked at, and a rebuild is how they are. Narrowing it to rows with a `db_path` or unit backing is a later refinement, with the hold narrowed to the recorders affected; neither is built. On a busy box this means frequent rebuilds, which the caller that nothing yet is must do. | this file | | |
| +| Opus P2-3, Astra P2-6: databases still drop out unseen: a unit route with no `DBPath`; names matched by case; a file with an extension other than `.db`; a claimed file that is missing or empty; a header read that fails. | Fixed in part: a unit route with no path is a problem; names are matched in any case; every other regular file in a tenant's tree is read for SQLite's header, and one that cannot be read is a problem; an explicitly claimed file that is empty is listed as skipped and one that cannot be looked at is a problem. Still open: a claimed file that is missing is passed over (nothing to replicate), an exec route whose path is outside the globs, and there is still no comparison against the external replicator's own inventory before cutover. | `litestream.go`, `litestream_owners.go` | `TestFleetInventoryNamesWhatItCannotReplicate`, `TestStrictTreeWalkReportsWhatDiscoveryPassesOver` | |
| +| Opus P2-4: the replica prefix and shadow state were keyed by path, so a path that changed hands continued the old owner's lineage. | Fixed: both are keyed by the owner's account, then the path. | `litestream_metered.go` `ownedReplicaKey` | the shadow-state and opened-store tests, under the new keys | |
| +| Opus P3-1, Astra P1 (part): the hold's top was the instant the change was found; a request admitted before and recorded after it, across midnight, fell on a day nobody held. | Fixed: the hold reaches a margin past the finding as well as before the last proof. Open, as Astra states: a request held between admission and recording for longer than the margin, which nothing bounds; and an admission that completed before the finding is not tracked to its recording. | `litestream_metered.go` `goStale`, `ownershipMargin` | `TestStaleFleetHoldsEveryDayInDoubt` ("found just before midnight") | |
| +| Opus P3-2: one transient witness error made the fleet stale for good. | Fixed: a reading that fails is tried three times before it is taken for a change. | `litestream_metered.go` `admit` | none (a fake registry that fails once is not easily arranged) | |
| +| Astra P1-4: an admission that was waiting for the lock when the second look made the fleet stale went on to a registry reading and was admitted. | Fixed: the stale flag is checked again under the lock. | `litestream_metered.go` `admit` | none | |
| +| Astra P2-5: an `UPDATE OR REPLACE` that moves a row onto another's rowid deletes that row without its delete trigger and names no watched column. | Fixed: a trigger refuses any change to a route's rowid. | `registry.go` `routes_rowid_immutable` | `TestRoutesRowidIsImmutable` | |
| +| Astra P1-2: `platformOwned(dataDir)` checks the data directory and not its ancestors; a tenant-writable parent could swap the whole directory. | Open, stated. The data directory's ancestors on a box are `/data` and `/`, root's; the check does not establish that. | | | |
| +| Astra P1-3 (rest), Opus (crash): what a process that dies between the change and the rebuild leaves; `Finish` and `Abandon` of an admitted request go to the old owner's recorder; the transition exists only in memory. | Open, stated more exactly: requests admitted before the change and finished before the crash are complete rows on a day nobody held; writing the transition down before the request is sent is the fix, not built. | | | |
| +| Opus P3-3, Astra P1 (part): a bind mount from the same filesystem has the same device number. | Open, stated: only an operator or systemd can make one. | | | |
| +| Astra P2-7: the shutdown tests pass with `store.Close` removed; the opened test does not compare attempts received with attempts recorded; the second look is not exercised through the watch. | Open; the tests are not strengthened in this build. | | | |
| +| Opus: comments claimed more than the code (the link rule, "narrows the window", the margin). | Corrected where the code changed; the rest stated above as open. | | | |
| + |
| +The fourth build is not read by a reviewer. The branch stays unwired and unattested, and item 1 of OUTSTANDING.md stays a launch blocker: nothing on any box can yet prove its replication writes. |
| + |
diff --git a/internal/ops/litestream.go b/internal/ops/litestream.go| index 5fa1dc8a2..7ddd07979 100644 |
| --- a/internal/ops/litestream.go |
| +++ b/internal/ops/litestream.go |
| @@ -22,7 +22,9 @@ package ops |
| import ( |
| "bytes" |
| + "errors" |
| "fmt" |
| + "io/fs" |
| "os" |
| "os/exec" |
| "path/filepath" |
| @@ -358,11 +360,15 @@ func fleetDBsFromRoutes(routes []registry.Route, registryPath, dataDir string, t |
| // Unit tenants' data lives outside dataDir (their systemd |
| // StateDirectory), so the glob below can never find them on its |
| // own — DBPath, when the operator set it, is the only way this |
| - // replicator learns where they are. |
| + // replicator learns where they are. For the platform's own |
| + // replicator a unit route with none is a place a database |
| + // could be that nobody can look at. |
| if r.DBPath != "" { |
| if _, seen := unitPaths[filepath.Clean(r.DBPath)]; !seen { |
| unitPaths[filepath.Clean(r.DBPath)] = r.Host |
| } |
| + } else if tree != nil { |
| + problems = append(problems, r.Host+": a unit route with no database path") |
| } |
| continue |
| } |
| @@ -463,7 +469,21 @@ func fleetDBsFromRoutes(routes []registry.Route, registryPath, dataDir string, t |
| } |
| fi, err := os.Stat(p) |
| if err != nil || !fi.Mode().IsRegular() || fi.Size() == 0 { |
| - continue // gone, odd, or an empty placeholder — nothing to replicate |
| + // Gone, odd, or an empty placeholder: nothing to replicate. For |
| + // the platform's own replicator an explicitly claimed file in |
| + // that state is said, not passed over (an empty placeholder |
| + // becomes a database without anyone being told). |
| + if tree != nil && claimed[p] { |
| + switch { |
| + case err != nil && !errors.Is(err, fs.ErrNotExist): |
| + problems = append(problems, p+": cannot be looked at") |
| + case err == nil && !fi.Mode().IsRegular(): |
| + problems = append(problems, p+": not a regular file") |
| + case err == nil: |
| + skipped = append(skipped, p) |
| + } |
| + } |
| + continue |
| } |
| if hibernating[p] { |
| continue |
diff --git a/internal/ops/litestream_metered.go b/internal/ops/litestream_metered.go| index c6afca93a..cf41924ed 100644 |
| --- a/internal/ops/litestream_metered.go |
| +++ b/internal/ops/litestream_metered.go |
| @@ -49,13 +49,25 @@ var ( |
| inventoryWatchInterval = 30 * time.Second |
| ) |
| -// ownershipMargin is how far before the last proven instant a stale |
| -// fleet's hold reaches. A request is admitted by the gate and then |
| -// recorded and sent; the recording has a budget of milliseconds. The |
| -// margin is far wider than that, so the hold covers every request that |
| -// was admitted before a change and sent after it. |
| +// ownershipMargin is how far on either side of the doubtful span a stale |
| +// fleet's hold reaches: before the last proven instant, and after the |
| +// instant the change was found. A request is admitted by the gate and |
| +// then recorded and sent; between its admission and its recording it can |
| +// be scheduled late, and a recording made after the finding falls on the |
| +// day of its own instant. The margin is far wider than the meter's budget, |
| +// so the hold covers a request admitted before a change and recorded after |
| +// the finding, across a midnight included. What it does not cover is a |
| +// request held between admission and recording for longer than the |
| +// margin, which nothing here bounds. |
| const ownershipMargin = time.Minute |
| +// witnessAttempts and witnessRetryDelay: how a reading of the ownership |
| +// witness that fails is tried again before it is taken for a change. |
| +var ( |
| + witnessAttempts = 3 |
| + witnessRetryDelay = 200 * time.Millisecond |
| +) |
| + |
| // MeteredFleetOptions says what the platform's replicator is built over. |
| type MeteredFleetOptions struct { |
| Registry *registry.Registry |
| @@ -82,10 +94,11 @@ type MeteredFleetOptions struct { |
| // it: when ownership goes stale it tells every recorder it was given. |
| Meter func(OwnedDB) lease.RequestRecorder |
| // Replaces is the fleet this one is built to take over from, if any. |
| - // If that fleet went stale, every recorder of this one is told the |
| - // same days are incomplete: the new owner of a database that changed |
| - // hands is one of them, and the old fleet never had its recorder. The |
| - // old fleet must be closed before this one is opened. |
| + // The old fleet must be closed before this one is opened, and closing |
| + // it asks the witness one last time. At Open, if that fleet went |
| + // stale, every recorder of this one is told the same days are |
| + // incomplete: the new owner of a database that changed hands is one |
| + // of them, and the old fleet never had its recorder. |
| Replaces *MeteredFleet |
| // SyncInterval is litestream's replica sync tick. Zero keeps its |
| // default. |
| @@ -121,10 +134,11 @@ type MeteredFleet struct { |
| store *litestream.Store |
| obs *registry.RouteObservation |
| - stale atomic.Bool |
| - drift atomic.Bool |
| - closed atomic.Bool |
| - proven atomic.Int64 // the last instant ownership was shown unchanged, in Unix nanoseconds |
| + stale atomic.Bool |
| + drift atomic.Bool |
| + closed atomic.Bool |
| + inherited atomic.Bool |
| + proven atomic.Int64 // the last instant ownership was shown unchanged, in Unix nanoseconds |
| staleMu sync.Mutex |
| staleFrom, staleAt time.Time |
| @@ -166,15 +180,32 @@ func (f *MeteredFleet) admit(ctx context.Context) error { |
| } |
| f.mu.Lock() |
| defer f.mu.Unlock() |
| - if f.obs == nil || f.closed.Load() { |
| + // Again, under the lock: the second look at the files (lookAgain) makes |
| + // the fleet stale without the lock, and an admission that was waiting |
| + // for it must not go on to a reading of the registry that says nothing |
| + // about the files. |
| + if f.obs == nil || f.closed.Load() || f.stale.Load() { |
| return ErrOwnershipStale |
| } |
| asked := time.Now() |
| // Not the request's context: a request cancelled while the witness is |
| // being read must not be mistaken for a registry that cannot be read. |
| - wctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) |
| - defer cancel() |
| - if ok, err := f.obs.OwnershipUnchanged(wctx); err != nil || !ok { |
| + // A reading that fails is tried again a few times before it is taken |
| + // for a change: a busy registry is ordinary, and one error must not |
| + // stop the box's replication for good. One that keeps failing is not |
| + // proof of no change, and counts as one. |
| + var ok bool |
| + var err error |
| + for attempt := 0; attempt < witnessAttempts; attempt++ { |
| + wctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) |
| + ok, err = f.obs.OwnershipUnchanged(wctx) |
| + cancel() |
| + if err == nil { |
| + break |
| + } |
| + time.Sleep(witnessRetryDelay) |
| + } |
| + if err != nil || !ok { |
| f.goStale(time.Now()) |
| return ErrOwnershipStale |
| } |
| @@ -189,20 +220,22 @@ func (f *MeteredFleet) admit(ctx context.Context) error { |
| // the watch, with no request in sight. |
| // |
| // The days in doubt run from a margin before the last instant ownership |
| -// was shown unchanged, to now. A request sent before the registry changed |
| -// was sent for the owner of that moment and is counted to them. The ones |
| -// in doubt were admitted before the change and sent after it; they were |
| -// admitted no earlier than a moment before the change, which is later than |
| -// the last proof less the margin. |
| +// was shown unchanged to a margin after the instant the change was found. |
| +// A request sent before the registry changed was sent for the owner of |
| +// that moment and is counted to them. The ones in doubt were admitted |
| +// before the change and recorded after it; they were admitted no earlier |
| +// than a moment before the change, which is later than the last proof |
| +// less the margin, and recorded no later than a moment after the finding |
| +// (ownershipMargin). |
| func (f *MeteredFleet) goStale(at time.Time) { |
| if !f.stale.CompareAndSwap(false, true) { |
| return |
| } |
| - from := time.Unix(0, f.proven.Load()).Add(-ownershipMargin) |
| + from, to := time.Unix(0, f.proven.Load()).Add(-ownershipMargin), at.Add(ownershipMargin) |
| f.staleMu.Lock() |
| - f.staleFrom, f.staleAt = from, at |
| + f.staleFrom, f.staleAt = from, to |
| f.staleMu.Unlock() |
| - holdDays(f.recorders, from, at) |
| + holdDays(f.recorders, from, to) |
| } |
| // staleWindow is the span a stale fleet held, for the fleet that replaces |
| @@ -244,6 +277,17 @@ func (f *MeteredFleet) Open(ctx context.Context) error { |
| if f.replaces != nil && !f.replaces.closed.Load() { |
| return errors.New("ops: the replicator this one replaces has not been closed") |
| } |
| + // The fleet this one replaces has closed, and closing asks the witness |
| + // one last time, so whatever it had to say about its last days it has |
| + // said by now. That is told to this fleet's recorders here, once: a |
| + // database that changed hands has a new owner, whose recorder the old |
| + // fleet never had. Read at build it would be missed whenever the new |
| + // fleet is built before the old one notices. |
| + if f.replaces != nil && f.inherited.CompareAndSwap(false, true) { |
| + if from, at, stale := f.replaces.staleWindow(); stale { |
| + holdDays(f.recorders, from, at) |
| + } |
| + } |
| if err := f.admit(ctx); err != nil { |
| return err |
| } |
| @@ -374,14 +418,26 @@ func (f *MeteredFleet) Close() error { |
| first = store.Close(ctx) |
| cancel() |
| } |
| - f.closed.Store(true) |
| f.stopOnce.Do(func() { close(f.stop) }) |
| f.watching.Wait() |
| f.mu.Lock() |
| defer f.mu.Unlock() |
| if f.obs == nil { |
| + f.closed.Store(true) |
| return first |
| } |
| + // One last reading before the witness is released: a change that |
| + // landed after the last admission, and that no watch tick saw, is |
| + // still this fleet's to hold, and the fleet that replaces it reads the |
| + // hold at Open. |
| + if !f.stale.Load() { |
| + wctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) |
| + if ok, err := f.obs.OwnershipUnchanged(wctx); err != nil || !ok { |
| + f.goStale(time.Now()) |
| + } |
| + cancel() |
| + } |
| + f.closed.Store(true) |
| err := f.obs.Close() |
| f.obs = nil |
| if first != nil { |
| @@ -476,9 +532,14 @@ func MeteredFleetStore(o MeteredFleetOptions) (fleet *MeteredFleet, err error) { |
| if err := built.admit(context.Background()); err != nil { |
| return nil, fmt.Errorf("%w (while ownership was being read)", err) |
| } |
| + // A replica lineage and its shadow state belong to an owner, not to a |
| + // path: a path that changes hands (an account moved, an instance |
| + // deleted and made again under its name) starts a new lineage, so the |
| + // old owner's data is never restored from under the new one and the |
| + // old owner's retention and compaction are never billed to the new. |
| keys := make([]string, 0, len(inv.Owned)) |
| for _, owned := range inv.Owned { |
| - keys = append(keys, replicaKey(owned.Path)) |
| + keys = append(keys, ownedReplicaKey(owned)) |
| } |
| metas, err := prepareShadowState(built.dataDir, keys) |
| if err != nil { |
| @@ -492,7 +553,7 @@ func MeteredFleetStore(o MeteredFleetOptions) (fleet *MeteredFleet, err error) { |
| return nil, fmt.Errorf("%w: no recorder for %s (%s/%s)", ErrUnmeterable, owned.Path, owned.Account, owned.App) |
| } |
| built.recorders = append(built.recorders, rec) |
| - key := replicaKey(owned.Path) |
| + key := ownedReplicaKey(owned) |
| replica := key |
| if prefix != "" { |
| replica = prefix + "/" + key |
| @@ -508,17 +569,18 @@ func MeteredFleetStore(o MeteredFleetOptions) (fleet *MeteredFleet, err error) { |
| store.SnapshotInterval, store.SnapshotRetention = SnapshotInterval, SnapshotRetention |
| built.store, built.Owned, built.Skipped = store, inv.Owned, inv.Skipped |
| built.facts, built.inventory = inv.facts, inv.render() |
| - // The fleet this one replaces went stale over some span of days. A |
| - // database that changed hands has a new owner, whose recorder the old |
| - // fleet never had. |
| - if o.Replaces != nil { |
| - if from, at, stale := o.Replaces.staleWindow(); stale { |
| - holdDays(built.recorders, from, at) |
| - } |
| - } |
| + // What the fleet this one replaces held is passed on at Open, once the |
| + // old fleet has closed and said its last. |
| return built, nil |
| } |
| +// ownedReplicaKey is the replica lineage of a database for its owner: |
| +// the account, then the path. The platform's own databases are keyed |
| +// under the operations account like any other. |
| +func ownedReplicaKey(o OwnedDB) string { |
| + return o.Account + "/" + replicaKey(o.Path) |
| +} |
| + |
| // prepareShadowState makes the directories litestream keeps each |
| // database's shadow state in, under <data dir>/.replicator, and answers |
| // with each key's directory. |
diff --git a/internal/ops/litestream_metered_test.go b/internal/ops/litestream_metered_test.go| index 80f76e3bc..3f0c79564 100644 |
| --- a/internal/ops/litestream_metered_test.go |
| +++ b/internal/ops/litestream_metered_test.go |
| @@ -519,55 +519,79 @@ func TestStaleFleetHoldsEveryDayInDoubt(t *testing.T) { |
| // has a new owner, whose recorder the old fleet never had. And it does not |
| // open while the fleet it replaces is still open. |
| func TestMeteredFleetReplacementInheritsTheHold(t *testing.T) { |
| + // noticed: the old fleet saw the change itself before it was closed. |
| + // Not noticed: the new fleet is built straight after the change, before |
| + // the old one's watch ran, and the old fleet's last reading at Close is |
| + // what finds it. |
| + for name, noticed := range map[string]bool{"the old fleet noticed before it closed": true, "the old fleet noticed only as it closed": false} { |
| + t.Run(name, func(t *testing.T) { |
| + newFakeBucket(t) |
| + dir := platformDir(t) |
| + reg := openRegistry(t, filepath.Join(dir, "registry.db")) |
| + db := writeFile(t, filepath.Join(dir, "a.example.com.db")) |
| + upsert(t, reg, registry.Route{Host: "a.example.com", Account: "bac", App: "one", DBPath: db}) |
| + o := fleetOptions(reg, dir) |
| + old, err := MeteredFleetStore(o) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + defer old.Close() |
| + if err := reg.SetAccount("a.example.com", "bae"); err != nil { |
| + t.Fatal(err) |
| + } |
| + if noticed { |
| + if err := old.Current(context.Background()); !errors.Is(err, ErrOwnershipStale) { |
| + t.Fatalf("the old fleet after the change: %v", err) |
| + } |
| + } |
| + recs := &recorders{} |
| + o.Meter, o.Replaces = recs.meter, old |
| + next, err := MeteredFleetStore(o) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + defer next.Close() |
| + var owner OwnedDB |
| + for _, od := range next.Owned { |
| + if od.Path == db { |
| + owner = od |
| + } |
| + } |
| + if owner.Account != "bae" { |
| + t.Fatalf("the new fleet's owner of the database: %+v", owner) |
| + } |
| + if err := next.Open(context.Background()); err == nil || !strings.Contains(err.Error(), "not been closed") { |
| + t.Fatalf("opening beside the fleet it replaces: %v", err) |
| + } |
| + for path, rec := range recs.all() { |
| + if _, _, inc := rec.counts(); inc != 0 { |
| + t.Fatalf("%s: told before the old fleet had closed and said its last", path) |
| + } |
| + } |
| + if err := old.Close(); err != nil { |
| + t.Fatal(err) |
| + } |
| + if err := old.Current(context.Background()); !errors.Is(err, ErrOwnershipStale) { |
| + t.Fatalf("the old fleet after closing: %v", err) |
| + } |
| + if err := next.Open(context.Background()); err != nil { |
| + t.Fatal(err) |
| + } |
| + for path, rec := range recs.all() { |
| + if _, _, inc := rec.counts(); inc == 0 { |
| + t.Errorf("%s: the new fleet's recorder was not told the days the old fleet held", path) |
| + } |
| + } |
| + if err := next.Close(); err != nil { |
| + t.Fatal(err) |
| + } |
| + }) |
| + } |
| + // A fleet that replaces one that was current inherits nothing. |
| newFakeBucket(t) |
| dir := platformDir(t) |
| reg := openRegistry(t, filepath.Join(dir, "registry.db")) |
| - db := writeFile(t, filepath.Join(dir, "a.example.com.db")) |
| - upsert(t, reg, registry.Route{Host: "a.example.com", Account: "bac", App: "one", DBPath: db}) |
| o := fleetOptions(reg, dir) |
| - old, err := MeteredFleetStore(o) |
| - if err != nil { |
| - t.Fatal(err) |
| - } |
| - defer old.Close() |
| - if err := reg.SetAccount("a.example.com", "bae"); err != nil { |
| - t.Fatal(err) |
| - } |
| - if err := old.Current(context.Background()); !errors.Is(err, ErrOwnershipStale) { |
| - t.Fatalf("the old fleet after the change: %v", err) |
| - } |
| - recs := &recorders{} |
| - o.Meter, o.Replaces = recs.meter, old |
| - next, err := MeteredFleetStore(o) |
| - if err != nil { |
| - t.Fatal(err) |
| - } |
| - defer next.Close() |
| - var owner OwnedDB |
| - for _, od := range next.Owned { |
| - if od.Path == db { |
| - owner = od |
| - } |
| - } |
| - if owner.Account != "bae" { |
| - t.Fatalf("the new fleet's owner of the database: %+v", owner) |
| - } |
| - for path, rec := range recs.all() { |
| - if _, _, inc := rec.counts(); inc == 0 { |
| - t.Errorf("%s: the new fleet's recorder was not told the days the old fleet held", path) |
| - } |
| - } |
| - if err := next.Open(context.Background()); err == nil || !strings.Contains(err.Error(), "not been closed") { |
| - t.Fatalf("opening beside the fleet it replaces: %v", err) |
| - } |
| - // A fleet that replaces one that was current inherits nothing. |
| - if err := old.Close(); err != nil { |
| - t.Fatal(err) |
| - } |
| - if err := next.Close(); err != nil { |
| - t.Fatal(err) |
| - } |
| - o.Replaces = nil |
| clean, err := MeteredFleetStore(o) |
| if err != nil { |
| t.Fatal(err) |
| @@ -582,6 +606,9 @@ func TestMeteredFleetReplacementInheritsTheHold(t *testing.T) { |
| t.Fatal(err) |
| } |
| defer after.Close() |
| + if err := after.Open(context.Background()); err != nil { |
| + t.Fatal(err) |
| + } |
| for path, rec := range recs2.all() { |
| if _, _, inc := rec.counts(); inc != 0 { |
| t.Errorf("%s: told its day is incomplete after a clean handover", path) |
| @@ -741,7 +768,7 @@ func TestMeteredFleetKeepsShadowStateOutOfTenantsReach(t *testing.T) { |
| private(t, at) |
| } |
| } |
| - if got := filepath.Join(root, replicaKey(db)); func() bool { _, err := os.Stat(got); return err != nil }() { |
| + if got := filepath.Join(root, "bad", replicaKey(db)); func() bool { _, err := os.Stat(got); return err != nil }() { |
| t.Fatalf("no shadow directory for the tenant's database at %s", got) |
| } |
| }) |
| @@ -760,7 +787,7 @@ func TestMeteredFleetKeepsShadowStateOutOfTenantsReach(t *testing.T) { |
| if err := os.MkdirAll(filepath.Join(dir, ".replicator"), 0o700); err != nil { |
| t.Fatal(err) |
| } |
| - first := strings.Split(replicaKey(filepath.Join(dir, "registry.db")), string(filepath.Separator))[0] |
| + first := "bab" // the operations account, which owns the registry |
| if err := os.Mkdir(filepath.Join(dir, ".replicator", first), 0o777); err != nil { |
| t.Fatal(err) |
| } |
| @@ -944,7 +971,7 @@ func TestMeteredFleetOpenedReplicatesCountsAndStops(t *testing.T) { |
| eventually(t, "the counts to settle", func() bool { return bucket.inFlight.Load() == 0 }) |
| // The shadow state is still the platform's. |
| root := filepath.Join(dir, ".replicator") |
| - for at := filepath.Join(root, replicaKey(db)); ; at = filepath.Dir(at) { |
| + for at := filepath.Join(root, "bac", replicaKey(db)); ; at = filepath.Dir(at) { |
| fi, err := os.Lstat(at) |
| if err != nil { |
| t.Fatal(err) |
diff --git a/internal/ops/litestream_owners.go b/internal/ops/litestream_owners.go| index 2b0381b8b..f57e32f1d 100644 |
| --- a/internal/ops/litestream_owners.go |
| +++ b/internal/ops/litestream_owners.go |
| @@ -170,10 +170,17 @@ func fleetInventory(routes []registry.Route, registryPath, dataDir string) (Flee |
| continue |
| } |
| // The claim is on a directory the tenant can write in, or on the |
| - // file itself. |
| + // file itself. A unit tenant's file is checked as its tree would |
| + // be, from the file's own directory: the tenant can write there, |
| + // so the directory above must be the platform's and the file the |
| + // tenant's own. An exec route's file is checked as a named file: |
| + // its directory is the activator's. |
| root, kind := filepath.Clean(owner.DBPath), tenantTree |
| if root == p { |
| root, kind = filepath.Dir(p), namedFile |
| + if owner.Backing == registry.BackingUnit { |
| + kind = tenantFile |
| + } |
| } |
| facts, why := checkOwnedFile(root, p, kind) |
| if why != "" { |
| @@ -223,9 +230,16 @@ const ( |
| // could swap its whole tree for a link), and the file must belong to |
| // the user the tree belongs to. |
| tenantTree |
| - // namedFile: a route names the file itself. Nothing is asked of its |
| - // directory here: an exec instance's directory may be the tenant's. |
| + // namedFile: an exec route names the file itself. Nothing is asked of |
| + // its directory here: an exec instance's directory is the activator's, |
| + // and the activator, not a tenant, puts files in it. |
| namedFile |
| + // tenantFile: a unit route names the file itself. The file's whole |
| + // chain of directories, up to where the tenant's ownership stops, must |
| + // be the tenant's and contain no link, and the directory above that |
| + // must be the platform's: the tenant can write anywhere in its own |
| + // chain, and a link anywhere in it leads wherever the tenant likes. |
| + tenantFile |
| ) |
| // checkOwnedFile decides whether path, claimed from root, is a file its |
| @@ -289,6 +303,48 @@ func checkOwnedFile(root, path string, kind claimKind) (fileFacts, string) { |
| return fileFacts{}, why |
| } |
| } |
| + case tenantFile: |
| + // The root is not the file's directory but the top of the tenant's |
| + // own chain: climb from the file's directory while the directory |
| + // above is the same user's and not a link. What is above that must |
| + // be the platform's, and nothing from the top of the chain down may |
| + // be a link, the top included (an operator's link is allowed for a |
| + // tree claim's root, which the route names; this root is inferred). |
| + owner, lerr := os.Lstat(path) |
| + if lerr != nil || !owner.Mode().IsRegular() { |
| + return fileFacts{}, "it is not a regular file" |
| + } |
| + ownerFacts, ok := factsOf(owner) |
| + if !ok { |
| + return fileFacts{}, "file ownership cannot be read on this platform" |
| + } |
| + for { |
| + parent := filepath.Dir(root) |
| + if parent == root { |
| + return fileFacts{}, "its directory chain reaches the root of the filesystem without leaving its owner" |
| + } |
| + pfi, perr := os.Lstat(parent) |
| + if perr != nil { |
| + return fileFacts{}, "a directory above it cannot be looked at" |
| + } |
| + pf, ok := factsOf(pfi) |
| + if !ok || pfi.Mode()&os.ModeSymlink != 0 || !pfi.IsDir() || pf.UID != ownerFacts.UID { |
| + break |
| + } |
| + root = parent |
| + } |
| + if top, terr := os.Lstat(root); terr != nil || top.Mode()&os.ModeSymlink != 0 || !top.IsDir() { |
| + return fileFacts{}, "the top of its directory chain is a link or cannot be looked at" |
| + } |
| + if why := platformOwned(filepath.Dir(root)); why != "" { |
| + return fileFacts{}, why |
| + } |
| + if rootInfo, err = os.Stat(root); err != nil { |
| + return fileFacts{}, "the directory it is claimed from cannot be looked at" |
| + } |
| + if rootFacts, ok = factsOf(rootInfo); !ok { |
| + return fileFacts{}, "file ownership cannot be read on this platform" |
| + } |
| } |
| rel, err := filepath.Rel(root, path) |
| if err != nil || rel == "." || rel == ".." || strings.HasPrefix(rel, ".."+string(filepath.Separator)) { |
| @@ -321,7 +377,7 @@ func checkOwnedFile(root, path string, kind claimKind) (fileFacts, string) { |
| if facts.Nlink != 1 { |
| return fileFacts{}, "it has more than one name (a hard link)" |
| } |
| - if kind == tenantTree && facts.UID != rootFacts.UID { |
| + if (kind == tenantTree || kind == tenantFile) && facts.UID != rootFacts.UID { |
| return fileFacts{}, "it does not belong to the user its tenant's directory belongs to" |
| } |
| for _, suffix := range []string{"-wal", "-shm", "-journal"} { |
| @@ -426,8 +482,20 @@ func strictTreeDBs(root string) (dbs, skipped, problems []string) { |
| case !fi.Mode().IsRegular() || fi.Size() == 0: |
| case strings.HasSuffix(name, ".db") && !hidden && !strings.HasPrefix(name, "."): |
| dbs = append(dbs, p) |
| - case sqliteName(name) || (filepath.Ext(name) == "" && sqliteHeader(p)): |
| + case sqliteName(name): |
| skipped = append(skipped, p) |
| + case strings.HasSuffix(strings.ToLower(name), "-wal") || strings.HasSuffix(strings.ToLower(name), "-shm") || strings.HasSuffix(strings.ToLower(name), "-journal"): |
| + // A database's own sidecar, which travels with it. |
| + default: |
| + // Any other regular file is read for SQLite's header: a |
| + // database under some other name is a database. One |
| + // that cannot be read is a place a database could be. |
| + switch isDB, rerr := sqliteHeader(p); { |
| + case rerr != nil: |
| + problems = append(problems, p+": cannot be read") |
| + case isDB: |
| + skipped = append(skipped, p) |
| + } |
| } |
| } |
| } |
| @@ -435,26 +503,33 @@ func strictTreeDBs(root string) (dbs, skipped, problems []string) { |
| return dbs, skipped, problems |
| } |
| -// sqliteName reports whether a file name is one SQLite databases go by. |
| +// sqliteName reports whether a file name is one SQLite databases go by, |
| +// in any case. |
| func sqliteName(name string) bool { |
| - switch filepath.Ext(name) { |
| + switch strings.ToLower(filepath.Ext(name)) { |
| case ".db", ".sqlite", ".sqlite3", ".db3": |
| return true |
| } |
| return false |
| } |
| -// sqliteHeader reports whether the file begins with SQLite's header. It is |
| -// asked only of files with no extension. |
| -func sqliteHeader(path string) bool { |
| +// sqliteHeader reports whether the file begins with SQLite's header. A |
| +// file too short to hold one is not a database; a file that cannot be |
| +// opened or read is an error, which the caller treats as a place it could |
| +// not look. |
| +func sqliteHeader(path string) (bool, error) { |
| f, err := os.Open(path) |
| if err != nil { |
| - return false |
| + return false, err |
| } |
| defer f.Close() |
| var head [16]byte |
| - if _, err := io.ReadFull(f, head[:]); err != nil { |
| - return false |
| + n, err := io.ReadFull(f, head[:]) |
| + if err == io.EOF || err == io.ErrUnexpectedEOF { |
| + return false, nil |
| + } |
| + if err != nil { |
| + return false, err |
| } |
| - return string(head[:]) == "SQLite format 3\x00" |
| + return n == len(head) && string(head[:]) == "SQLite format 3\x00", nil |
| } |
diff --git a/internal/ops/litestream_owners_test.go b/internal/ops/litestream_owners_test.go| index 4748c3cd8..78989b2bf 100644 |
| --- a/internal/ops/litestream_owners_test.go |
| +++ b/internal/ops/litestream_owners_test.go |
| @@ -401,3 +401,77 @@ func TestStrictTreeWalkReportsWhatDiscoveryPassesOver(t *testing.T) { |
| } |
| } |
| } |
| + |
| +// A unit route that names one file is checked as the tenant's tree would |
| +// be, from the file's own directory: a link in the directory above the |
| +// file, which the tenant can make, leads to a database nobody listed, and |
| +// must not be owned. |
| +func TestFleetDBOwnersCheckAUnitRoutesNamedFileLikeATree(t *testing.T) { |
| + dir := platformDir(t) |
| + reg := openRegistry(t, filepath.Join(dir, "registry.db")) |
| + state := filepath.Join(platformDir(t), "unit-a") |
| + own := writeFile(t, filepath.Join(state, "data", "data.db")) |
| + upsert(t, reg, registry.Route{Host: "a.example.com", Account: "bac", App: "one", DBPath: own, Backing: registry.BackingUnit}) |
| + inv, err := FleetDBOwners(reg, dir) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if o, ok := ownedByPath(inv)[own]; !ok || o.Account != "bac" { |
| + t.Fatalf("the tenant's own named file: %+v (unowned %v, problems %v)", inv.Owned, inv.Unowned, inv.Problems) |
| + } |
| + // The tenant swaps the directory above the file for a link to a |
| + // directory holding a database nobody lists. |
| + elsewhere := writeFile(t, filepath.Join(platformDir(t), "not-listed", "data.db")) |
| + if err := os.Rename(filepath.Join(state, "data"), filepath.Join(state, "data-was")); err != nil { |
| + t.Fatal(err) |
| + } |
| + if err := os.Symlink(filepath.Dir(elsewhere), filepath.Join(state, "data")); err != nil { |
| + t.Fatal(err) |
| + } |
| + inv, err = FleetDBOwners(reg, dir) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if _, ok := ownedByPath(inv)[own]; ok || !named(inv.Unowned, own, "") { |
| + t.Fatalf("a named file reached through a link above it: owned %+v, unowned %v", inv.Owned, inv.Unowned) |
| + } |
| +} |
| + |
| +// A unit route with no database path, and a claimed file that is missing, |
| +// empty or not a regular file, are said to the replicator, not passed over. |
| +func TestFleetInventoryNamesWhatItCannotReplicate(t *testing.T) { |
| + dir := platformDir(t) |
| + reg := openRegistry(t, filepath.Join(dir, "registry.db")) |
| + missing := filepath.Join(platformDir(t), "gone", "app.db") |
| + empty := filepath.Join(platformDir(t), "empty", "app.db") |
| + if err := os.MkdirAll(filepath.Dir(empty), 0o755); err != nil { |
| + t.Fatal(err) |
| + } |
| + if err := os.WriteFile(empty, nil, 0o644); err != nil { |
| + t.Fatal(err) |
| + } |
| + upsert(t, reg, |
| + registry.Route{Host: "nopath.example.com", Account: "bac", App: "one", Backing: registry.BackingUnit}, |
| + registry.Route{Host: "missing.example.com", Account: "bac", App: "two", DBPath: missing, Backing: registry.BackingUnit}, |
| + registry.Route{Host: "empty.example.com", Account: "bac", App: "three", DBPath: empty, Backing: registry.BackingUnit}, |
| + ) |
| + inv, err := FleetDBOwners(reg, dir) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if !named(inv.Problems, "nopath.example.com", "no database path") { |
| + t.Errorf("a unit route with no path is not a problem: %v", inv.Problems) |
| + } |
| + if !named(inv.Skipped, empty, "") { |
| + t.Errorf("an empty claimed file is not listed as skipped: %v", inv.Skipped) |
| + } |
| + // A missing file is nothing to replicate and nothing to look at; a |
| + // file that appears later is the second look's to report. |
| + if named(inv.Problems, missing, "") || named(inv.Skipped, missing, "") { |
| + t.Errorf("a missing claimed file is reported: problems %v skipped %v", inv.Problems, inv.Skipped) |
| + } |
| + // The config generator's own discovery is unchanged by any of this. |
| + if dbs, err := FleetDBs(reg, dir); err != nil || len(dbs) != 1 { |
| + t.Fatalf("FleetDBs: %v %v", dbs, err) |
| + } |
| +} |
diff --git a/internal/registry/observe_ownership_test.go b/internal/registry/observe_ownership_test.go| index 80834b4ad..e5cd65a33 100644 |
| --- a/internal/registry/observe_ownership_test.go |
| +++ b/internal/registry/observe_ownership_test.go |
| @@ -4,6 +4,7 @@ import ( |
| "context" |
| "os" |
| "path/filepath" |
| + "strings" |
| "testing" |
| ) |
| @@ -201,3 +202,30 @@ func must(t *testing.T, err error) { |
| t.Fatal(err) |
| } |
| } |
| + |
| +// A route's rowid never changes: an UPDATE OR REPLACE that moved a row onto |
| +// another's rowid would delete that other row without the delete trigger |
| +// when recursive triggers are off, and the ownership generation would not |
| +// move. The database refuses the move. |
| +func TestRoutesRowidIsImmutable(t *testing.T) { |
| + path := filepath.Join(t.TempDir(), "registry.db") |
| + w, err := Open(path) |
| + must(t, err) |
| + must(t, w.Upsert(Route{Host: "a.example.com", Slug: "a", Kind: "instance", App: "a", Account: "aaaa", DBPath: "/data/a.db"})) |
| + must(t, w.Upsert(Route{Host: "b.example.com", Slug: "b", Kind: "instance", App: "b", Account: "bbbb", DBPath: "/data/b.db"})) |
| + obs, err := w.ObserveRoutes(context.Background()) |
| + must(t, err) |
| + defer obs.Close() |
| + _, err = w.sql.Exec(`UPDATE OR REPLACE routes SET rowid = (SELECT rowid FROM routes WHERE host = 'b.example.com') WHERE host = 'a.example.com'`) |
| + if err == nil || !strings.Contains(err.Error(), "routes_rowid_immutable") { |
| + t.Fatalf("a rowid move: %v, want it refused", err) |
| + } |
| + routes, err := w.List() |
| + must(t, err) |
| + if len(routes) != 2 { |
| + t.Fatalf("%d routes after the refused move, want 2", len(routes)) |
| + } |
| + if ok, err := obs.OwnershipUnchanged(context.Background()); err != nil || !ok { |
| + t.Fatalf("a refused move moved the generation: %v %v", ok, err) |
| + } |
| +} |
diff --git a/internal/registry/registry.go b/internal/registry/registry.go| index 6a2e37781..673de6221 100644 |
| --- a/internal/registry/registry.go |
| +++ b/internal/registry/registry.go |
| @@ -600,6 +600,13 @@ func (r *Registry) migrateRouteOwnership() error { |
| `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`, |
| + // A row's rowid never changes. An UPDATE OR REPLACE that moved a |
| + // row onto another's rowid would delete that other row without |
| + // firing its delete trigger when recursive triggers are off, and |
| + // name none of the columns above: a route gone with the counter |
| + // unmoved. No writer of this database has a reason to move a |
| + // rowid, so it is refused outright. |
| + `CREATE TRIGGER IF NOT EXISTS routes_rowid_immutable BEFORE UPDATE ON routes WHEN OLD.rowid <> NEW.rowid BEGIN SELECT RAISE(ABORT, 'routes_rowid_immutable'); END`, |
| } { |
| if _, err := r.sql.Exec(stmt); err != nil { |
| return err |