3678b940a · carlos/platform

diff --git a/DISPOSITIONS-replicator-pinned-root.md b/DISPOSITIONS-replicator-pinned-root.md

index ffb8a26e6..3f65c2a0d 100644

--- a/DISPOSITIONS-replicator-pinned-root.md

+++ b/DISPOSITIONS-replicator-pinned-root.md

@@ -178,3 +178,24 @@ Opus: accept with fixes, one P1, two P2. Astra: reject, three P1, five P2. Repor

| Astra: the design's settlement formula and the plan's Task 3 | the formula names the clamp to now; Task 3 names `RequiredDatabases(registryPath)` and ownership. |

| Astra: the `GateAnswers` equivalence claim | narrowed as asked: recovery leaves `admitted` without an answer; the named sequence kills the mutant; the spike note says so. |

| Opus: mutants to add | the replay closing on an empty counted set over unsettled rows (killed by the pending-to-row test); closure without the discovery floor (killed by the edges test); a hold never published (killed by the Publish-level test). |

+

+## Review ten (of `f11464ec`: the ninth review answered, Task 4's fourth slice)

+

+Opus: accept with fixes, no P1, three P2. Astra: reject, one P1, five P2, one P3. Reports: `logs/opus-replroot-review10.md`, `logs/astra-replroot-review10.md`. The gate on `f11464ec` passed (exit 0, three cross builds). The answers land as their own commit; plan Task 4's fifth slice (the gate and the fleet) lands after them.

+

+| Finding | Disposition |

+|---|---|

+| Astra 1 (P1), Opus P2-3: an unresolved latch republished from the earliest dirty day, and the enumeration truncated at 400 days | `publish` reads the ledger's earliest day (`MIN(date) FROM days`) under its mutex and enumerates the latch from it; a span past the bound (4000 days, the hold writer's) is an error from `Publish`, never a truncated list, and so is a hold or dirtying interval past it; the Publish-level test publishes a day complete, latches an unresolved discovery days later, and reads the old day incomplete; the floor mutant is killed. |

+| Astra 2, Opus P2-1: a latch replayed twice; two open uncertainties | `Replay` checks closed first, then reads the latch again under `drainMu` and `l.mu` (a cleared latch is a no-op) and takes its instant after the locks; `replayTx` folds into an open uncertainty of the fleet, lowering its bound, rather than inserting a second row; a seam runs a `Discover` between a `Replay`'s first latch read and its locks (one uncertainty), and a direct fold is tested; both mutants are killed. A `CloseFleet` carrying the replay reads the latch under the same locks. |

+| Astra 3: clearing an unresolved latch left days published barred | the replay's transaction dirties, with a fresh sequence, every day the overlay barred (from the latched bound, or from the ledger's earliest day while unresolved) through now; the Publish-level test takes a day complete, unresolved-barred, and complete again. |

+| Astra 4: the handoff compared identities; an inode reused after the listing directory was removed would pass | the strict walk keeps the directories that listed a database open, with their chain to the tree's root (`fleetPins.listedDirs`, `listedTrees`), and the ownership check inspects the file in that retained directory (I1(b)) after requiring the name-resolved chain to be it, directory by directory (litestream opens by name until Part B); the retained chain is what the owned file keeps, the check's own acquisition released; after the pass the listing directories no owned chain reaches are closed (`trimListed`), so the descriptor count stays what `/proc` shows. An inode cannot be reused while the directory is held open, which no fixture can show deterministically; the mutants that never use the listing directory, never compare the name-resolved chain, and never keep the directory open are each killed. The identity-comparison helpers of revision 15 are gone. |

+| Astra 5: unsupported YAML read as an empty set | a flow document, a quoted `dbs` or `path` key, a `dbs` key not at the top level, and a document that mentions `dbs` with no recognised key are refused before anything is read as empty; tested. |

+| Astra 6, Opus P2-2: a fleet closed with an uncertainty nothing will close | `CloseFleet` closes an open uncertainty of the fleet on the settlement's own test (no unsettled row, nothing pending) before it marks the fleet closed; the sequence (empty settlement told while the counter still says one, the counter dropping, the closure) is a test, and the schedule ends at quiescence with no open uncertainty anywhere; the plan's gate contract says the counter is decremented before the ledger is told. |

+| Astra 7, Opus P3-6: closed refusals inconsistent; the journal tests masked missing guards | `Replay` refuses a closed ledger before anything else; the journal tests and the discovery tests assert the ledger's own error; the admission test's refused inputs use fresh tokens. |

+| Opus P3-1: the parser took a replica's `path:` | only a key at the entry's own indent is the database's path; the file-replica fixture in both orders; the mutant is killed. |

+| Opus P3-2: the chain rule's holes | superseded by the retained directory (Astra 4). |

+| Opus P3-3: the schedule could not kill its subjects | the schedule computes its expected effective set from its own record of the chain and the rows' fleet ids, drives admissions, settlements, discoveries and closures on any live fleet, and ends at quiescence; the backfill's and acknowledgement's hold writes are the rows a discovery already wrote under box-wide holds, inspection-only until owner scope, and the test says so. |

+| Opus P3-4: the journal laxer than the model | a repeat `OpenFleet` compares every column; a predecessor takes one successor (tested); `Discover`'s state check is the gate slice's, stated in the plan. |

+| Opus P3-5: `Replay`'s instant before its locks | taken after them. |

+| Astra: the model's recovery acknowledges through the existing chain, not by the first new fleet | recorded for slice 6 (recovery): the journal transaction follows the design, not the model's shortcut. |

+| Astra: survivors | the admission's own day asserted through `Publish` before any seal; the unresolved latch through `Publish`; the schedule's expected set; fresh tokens. `O_NOCTTY` stays inspection-only. |

diff --git a/docs/site/cli.md b/docs/site/cli.md

index 7dbf87da3..07106c948 100644

--- a/docs/site/cli.md

+++ b/docs/site/cli.md

@@ -1443,8 +1443,9 @@ opens the registry without creating or migrating it and prints the platform

replicator's inventory (owned, skipped, unowned, problems, and the required

databases with whether each is reachable and owned) beside the external

replicator's live configuration (`-config`, default `/etc/litestream.yml`;

-block-style YAML, items at any indentation, quoted or bare paths; a

-configuration the reader cannot take is an error, never an empty list) and

+block-style YAML with bare keys, items at any indentation, quoted or

+bare paths; a configuration the reader cannot take, a flow document, a

+quoted key or an indented `dbs`, is an error, never an empty list) and

the list a regeneration would write, every path quoted; the external lists

and the inventory are separate looks, one after the other, and the required

set is read from this process's environment (`CONSOLE_DB`), which the

diff --git a/docs/superpowers/plans/2026-10-03-replicator-authority-plan.md b/docs/superpowers/plans/2026-10-03-replicator-authority-plan.md

index 78dffab06..425103823 100644

--- a/docs/superpowers/plans/2026-10-03-replicator-authority-plan.md

+++ b/docs/superpowers/plans/2026-10-03-replicator-authority-plan.md

@@ -34,9 +34,9 @@ The admission model (`internal/ops/admissionmodel`) is the executable contract;

1. **The tables and the admission's row (built: `internal/usage/durabilityadmissions.go`).** `admissions(token TEXT PRIMARY KEY, fleet_id, account, app, admitted_at, transport_done_at, recorder_started_at, recorder_done_at, settled_at, how)`, `fleets(fleet_id PRIMARY KEY, box, witness_generation, owners, opened_at, state, predecessor, acknowledged_by)`, `uncertainties(id, fleet_id, from_ns, to_ns NULL, reason)`, `inherited(fleet_id, uncertainty_id)`, `holds(day, account, app, uncertainty_id, written_at)`, all `IF NOT EXISTS` beside today's tables. The `DurabilityLedger` gains `Admit(token, fleet, owner, app, at) error` (the token allocated by the caller before the write, `NewAdmissionToken`, so an ambiguous commit is read back by it; one transaction, `synchronous=FULL`; an exact repeat is the write that landed, a repeat with other facts is refused; the gate's budget, `ErrAdmissionNotDurable` past it and the goroutine staying to mark a late commit `unissued`, are slice 5's, the gate's, and are not built here), `TransportDone(token, at, lost bool)` (`lost` is not recorded: transport-lost is never a value of `how`), `RecorderStarted(token, at)`, `RecorderDone(token, at, how)` (`finished` and `abandoned` need the recorder registered), settlement by whichever lands last in its own transaction at the latest of the two completions and now, and `Unissued(token, at)` (an edge from `admitted` alone: refused after any participant activity); every transition refuses an instant before the admission's, is idempotent for an exact repeat, and refuses on a closed ledger. Mirrors: `TestUnchangedFleetHoldsNothing` and `TestThreeMidnightsBothOrders` (the row's settlement in both orders; the interval written at settlement is slice 3's), `TestCommitAfterDiscoveryIsUnissued`, `TestHistoricalTermCountsUnrecordedSettlements`; `TestBudgetExpiryWithoutDiscovery` is slice 5's.

2. **The barrier (built: `DurabilityContent.Barred`, counted in the snapshot's read transaction; `CompletenessFor` refuses while it is positive; box-wide, so the ancestry query waits for owner scope).** The publisher's snapshot (`snapshotOnce`) treats a day incomplete for an owner while a durable unsettled admission of that owner, or of any fleet whose unacknowledged ancestry includes the admission's fleet, is admitted on or before that day (box-wide: any owner); it is a query over `admissions` and `fleets`, writes nothing, and lifts at settlement. Mirrors: the model's barrier invariant and `TestSuccessorBarredByPredecessorAdmission`, `TestLandedAmbiguousCommitBarsBeforeReadBack`.

-3. **Discovery, the uncertainty and the latch (built in the ledger, with review nine's corrections: every hold write and every admission and settlement dirty their days through `touchDay` with a fresh sequence and `publish` enumerates the latched discoveries' days (`latchedDates`), so publication revisits a day a bound reaches or a barrier leaves; `discoveryState.generation` makes a snapshot read again across a replay; a failed discovery latches the bound and counted set it computed or latches unresolved; `Replay` recomputes the bound in its transaction and closes on `maybeClose`'s test; `uncertainties.discovered_at` floors every closure; `SetPendingCounter` takes an `*atomic.Int64`, lock-free by contract. `Discover` with the fixed bound and the counted set, `uncertainties` and box-wide `holds` rows in one transaction, closed at once with nothing counted or pending; the settlement's hook writing `[admitted_at, settled_at]` and closing when no row and no pending registration remains, the fleet's pending count by `SetPendingCounter`; `EmptySettlement`; a failed transaction latched in memory with its bound and counted tokens, barring publication through the snapshot's `Barred` until `Replay`, one transaction; `DurabilityContent.Held` refusing completeness; the fleet's discovery mutex, registration and the latch's refusal of requests are the gate slice).** `Discover(fleet, at, proof, lastCleanLook)`: under the fleet's discovery mutex (new in `MeteredFleet`: registration, the recheck and the start of a discovery serialise on it, the pending set counted), the fixed bound, the counted set, the `uncertainties` row and the holds of `[from, now]` for the owners reached in one transaction, closed at once with nothing pending; a failed write latches the pending discovery in the existing failure latch (extended with the bound, the owners reached and the counted tokens; the publisher's overlay reaches successors through the ancestry query) and the fleet refuses requests until `Replay`, one transaction as the design has it. Mirrors: `TestZeroPendingClosesAtOnceAcrossMidnight`, `TestFailedDiscoveryLatchBarsUntilReplay`, `TestFailedDiscoveryReplayKeepsSettlements`, `TestReplayHoldsTheLowerBoundWhileOpen`, `TestReplayWritesSettledTokenWhileOpen`, `TestLatchOverlayReachesSuccessor`, the lower-bound tests (the in-flight request across a clean look, the proof term, the earlier-of threshold, the fixed cutoff in both orders, the admission instant, the exact bound, the empty terminals excluded).

-4. **The fleet journal, construction backfill and acknowledgement (built: `internal/usage/durabilityfleets.go`: `OpenFleet(fleetID, box, witnessGeneration, owners, predecessor, at)` with construction backfill of the unacknowledged ancestry's effective set; `FleetClosing`; `CloseFleet`, refusing with an unsettled row or a pending registration and carrying a latched replay in its own transaction; `Acknowledge`, root-first and terminal, copying the predecessor's effective set into `inherited` and writing the closed ones' holds; `ReadFleet`, `Ancestry`, `Effective`, `Reach`, `Inherited`; `stale` written inside the discovery's transaction; holds being box-wide on this branch, the backfill and acknowledgement writes are the rows a discovery already wrote, and the owner-scoped writes are these same calls with the owners' rows. Mirrors: the ancestry-and-root-first sequence, the healthy chain, and a recorded random schedule asserting the per-step inheritance invariant.)** `OpenFleet(predecessor, owners, witness)` writes the `open` row and the backfill of every interval its unacknowledged ancestry holds; `stale`, `closing`, `closed` as the design's transactions (closure carries a pending replay and refuses with an unsettled admission); `Acknowledge(successor)` root-first, copying the effective set into `inherited` and writing the holds for the closed ones. The ledger's reach query replaces the model's `reach`. Mirrors: `TestAncestryReachAndRootFirstAcknowledgement`, `TestHealthyChainContributesNothing`, the per-step inheritance invariant as a property test over a recorded schedule.

-5. **The gate and the fleet.** Contract (review nine, Opus P2-2): the ledger is never called with the fleet's discovery mutex held, and the pending counter the ledger reads is the lock-free `*atomic.Int64`, so a settlement under the ledger's mutex never waits on the mutex a gate recheck holds while it writes `unissued`. `MeteredFleet.admit` becomes the proof (unchanged) plus the admission: registration in the pending set under the discovery mutex, the ledger's `Admit`, the recheck, the token handed to the SDK middleware (`lease.RequestGate` returns the token; the meter's `Start` carries it; the finalize step writes `TransportDone`). `lookNow` is the discovery when it finds a change; `closeNow` is `closing` then `closed`. The helper process does not exist yet (Part B): the transport and the recorder run in the agent, so the lifecycle guard is the fleet's own, `transport-lost` is written at `Close` for a transport that never finalised, and `unrecorded` where the meter's `Start` failed. Mirrors: `TestHelperDeathAndRecorderGuard` (as the agent's close), `TestSendNeedsDurableIssuedAdmission` (the middleware refuses without a token), `TestReadBackOnlyAfterTheHandleCloses`, `TestBetweenCommitAndAnswer`.

+3. **Discovery, the uncertainty and the latch (built in the ledger, with review nine's and ten's corrections: `Replay` reads the latch again under the locks and takes its instant after them, and a replay that finds an uncertainty of the fleet open folds into it (one open uncertainty per fleet); publication enumerates an unresolved latch from the ledger's earliest day, capped from the newest day backwards; every hold write and every admission and settlement dirty their days through `touchDay` with a fresh sequence and `publish` enumerates the latched discoveries' days (`latchedDates`), so publication revisits a day a bound reaches or a barrier leaves; `discoveryState.generation` makes a snapshot read again across a replay; a failed discovery latches the bound and counted set it computed or latches unresolved; `Replay` recomputes the bound in its transaction and closes on `maybeClose`'s test; `uncertainties.discovered_at` floors every closure; `SetPendingCounter` takes an `*atomic.Int64`, lock-free by contract. `Discover` with the fixed bound and the counted set, `uncertainties` and box-wide `holds` rows in one transaction, closed at once with nothing counted or pending; the settlement's hook writing `[admitted_at, settled_at]` and closing when no row and no pending registration remains, the fleet's pending count by `SetPendingCounter`; `EmptySettlement`; a failed transaction latched in memory with its bound and counted tokens, barring publication through the snapshot's `Barred` until `Replay`, one transaction; `DurabilityContent.Held` refusing completeness; the fleet's discovery mutex, registration and the latch's refusal of requests are the gate slice).** `Discover(fleet, at, proof, lastCleanLook)`: under the fleet's discovery mutex (new in `MeteredFleet`: registration, the recheck and the start of a discovery serialise on it, the pending set counted), the fixed bound, the counted set, the `uncertainties` row and the holds of `[from, now]` for the owners reached in one transaction, closed at once with nothing pending; a failed write latches the pending discovery in the existing failure latch (extended with the bound, the owners reached and the counted tokens; the publisher's overlay reaches successors through the ancestry query) and the fleet refuses requests until `Replay`, one transaction as the design has it. Mirrors: `TestZeroPendingClosesAtOnceAcrossMidnight`, `TestFailedDiscoveryLatchBarsUntilReplay`, `TestFailedDiscoveryReplayKeepsSettlements`, `TestReplayHoldsTheLowerBoundWhileOpen`, `TestReplayWritesSettledTokenWhileOpen`, `TestLatchOverlayReachesSuccessor`, the lower-bound tests (the in-flight request across a clean look, the proof term, the earlier-of threshold, the fixed cutoff in both orders, the admission instant, the exact bound, the empty terminals excluded).

+4. **The fleet journal, construction backfill and acknowledgement.** `OpenFleet(predecessor, owners, witness)` writes the `open` row and the backfill of every interval its unacknowledged ancestry holds; `stale`, `closing`, `closed` as the design's transactions (closure carries a pending replay and refuses with an unsettled admission); `Acknowledge(successor)` root-first, copying the effective set into `inherited` and writing the holds for the closed ones. The ledger's reach query replaces the model's `reach`. Mirrors: `TestAncestryReachAndRootFirstAcknowledgement`, `TestHealthyChainContributesNothing`, the per-step inheritance invariant as a property test over a recorded schedule.

+5. **The gate and the fleet.** Contract (review nine, Opus P2-2; review ten, Opus P2-2): the ledger is never called with the fleet's discovery mutex held, and the pending counter the ledger reads is the lock-free `*atomic.Int64`, so a settlement under the ledger's mutex never waits on the mutex a gate recheck holds while it writes `unissued`; a registration that resolves decrements the counter before it tells the ledger (`EmptySettlement`, or the row's own transition), so the closure test sees the counter already down, and `CloseFleet` closes an open uncertainty on the same test regardless; the ledger's `Discover` does not check the fleet's journal state (the model returns unless open), which is this slice's to enforce before it calls. `MeteredFleet.admit` becomes the proof (unchanged) plus the admission: registration in the pending set under the discovery mutex, the ledger's `Admit`, the recheck, the token handed to the SDK middleware (`lease.RequestGate` returns the token; the meter's `Start` carries it; the finalize step writes `TransportDone`). `lookNow` is the discovery when it finds a change; `closeNow` is `closing` then `closed`. The helper process does not exist yet (Part B): the transport and the recorder run in the agent, so the lifecycle guard is the fleet's own, `transport-lost` is written at `Close` for a transport that never finalised, and `unrecorded` where the meter's `Start` failed. Mirrors: `TestHelperDeathAndRecorderGuard` (as the agent's close), `TestSendNeedsDurableIssuedAdmission` (the middleware refuses without a token), `TestReadBackOnlyAfterTheHandleCloses`, `TestBetweenCommitAndAnswer`.

6. **The producer fence and recovery.** `admissions.lock` flocked by the agent before its first admission on an open file description every helper will later inherit (today: the agent alone); `OpenDurabilityLedger`'s `recover` extended: with the fence, in one transaction, every unsettled admission `recovered-unknown` with its interval, every fleet not `closed` closed with its interval, every unacknowledged fleet acknowledged root-first by the first fleet built. Mirrors: `TestFenceOutlivesAgent`, `TestFenceHeldTokenSentAfterCrash`, `TestLandedAmbiguousThenCrash`, the kill-at-every-transition tests (a second process reopening the ledger after a kill at each boundary).

7. **Part C2's reader.** The `holds` rows in the published day (`DurabilityContent`), the reader treating a box-wide row as failing every owner and an owner row as failing that owner alone, and the older-reader marker (a box-wide incompleteness marker whenever any hold exists). Box-wide stays the default; decision 10's per-owner fleets are not built.

diff --git a/docs/superpowers/specs/2026-10-03-replicator-pinned-roots-design.md b/docs/superpowers/specs/2026-10-03-replicator-pinned-roots-design.md

index 3437b7015..4a759848f 100644

--- a/docs/superpowers/specs/2026-10-03-replicator-pinned-roots-design.md

+++ b/docs/superpowers/specs/2026-10-03-replicator-pinned-roots-design.md

@@ -1,6 +1,6 @@

# The replicator's authority: retained descriptors, a durable admission, claims by identity

-Date: 2026-10-03. Status: design, revision 15 (the ninth code review: a hold, a latched discovery, an admission and its settlement are publication events, dirtying their days above any snapshot's watermark, and publication enumerates a latched discovery's days as it does the failure latch's; a snapshot reads again when the latched set changes across its read; a failed discovery keeps the bound and counted set its transaction computed, or latches unresolved and bars every day until the replay computes them; the replay closes on the settlement's own test, every unsettled row and the pending count; the discovery instant is stored and floors the closure; the settlement formula names the clamp to now; the handoff from the tree walk to the ownership checks compares the identity of the file and of every directory on the chain that listed it). Revision 14 was (the eighth code review: the recovery paragraph agrees with the transition table, an orphan's missing timestamps become now and a completion recorded before the crash keeps its instant; required coverage stated as ownership, with the deployment condition for the console database). Revision 13 was (the sixth code review: the predicate's compatibility stated as within the pinned roots, with the two safe wider shapes and the kept platform-directory refusal; the walk's pre-pin absent exception; the link-shapes line extended). Revision 12 was (Part A's root rule stated as the plan has it: the platform walk ends where the platform's authority ends, a tenant-owned component's parent, and the chain beneath is retained and rechecked rather than refused). Revision 11 was (the fourth code review: the historical term's filter named `finished` and `abandoned` and so excluded `unrecorded`, a settlement whose transport did send; it now excludes only the empty terminals, `unissued` and `cancelled`; the look's order is stated as what safety needs rather than a sequence; the closing look agrees). Revision 10 was revision 9's correction of the lower bound refined after the third code review: the historical term's threshold is the earlier of the registry proof and the last clean look's start, since a witness change is known only to be after the proof; an empty terminal, unissued or cancelled, never sent and is not a settlement for that term; and every look reads the witness and rechecks every retained chain before it inspects files, so a clean look's start bounds every kind of change. Revision 9 was one mechanics correction to the uncertainty's lower bound, found by the second code review of the admission model: an admission admitted before the last clean look, still in flight across a change after it and settled clean before the change was noticed, was outside the bound; the bound now also takes the admission instant of every admission settled at or after the last clean look's start; the invariant I2 is unchanged and the model's named test and schedules cover it). Revision 8 was the accepted text: revision 7 was accepted with fixes by Astra (no P1; one P2: a negative read-back was not conclusive while the admission writer could still commit, and the pending set had three definitions; one P3: the restore wording made a salt mismatch imply a snapshot), both applied here. Revision 6 was rejected by Astra on one P1: a discovery whose write failed had no rule for the admissions that settled before its retry, so the replayed interval could lose a settlement; and three P2s: a helper dying before its recorder started left an admission nobody could finish; the transition table lacked the abort edges and the pending set's cleanup; the lineage record cited a durability discipline the shadow state does not have; each answered here. Revision 5 was rejected by Astra on two P1s: the admission's insertion had no ordering point against discovery, so a discovery between the proof and the commit counted nothing and a later send fell outside the closed uncertainty; construction did not backfill already-closed intervals for a successor's new owners before acknowledgement; and P2s: the transition table and the helper-death rule disagreed and the recorder participant was synthesised while its worker could still run; proof-only callers of admit would have created admissions with no terminal; a terminal `.` lost the directory requirement; conflicting summaries; each answered here. Revision 4 was rejected by Astra on four P1s: the helper has several connections of its own and needs SQLite's within-process lock bookkeeping between them; the producer fence did not cover a helper that outlives the agent with an admission in hand; an intermediate fleet closed before acknowledging broke transitive protection; a fleet could be marked closed with a discovery uncommitted; and four P2s: a dead helper's admissions had no terminal in the running agent, the sleeping binding persisted a reusable inode, the lineage predicate named header fields that identify nothing, the walk's grammar refused a `.` target the predicate accepts; each answered here. Revision 3 was rejected by Astra on five P1s: two SQLite VFSs in one process cannot coordinate through POSIX locks, since the library's own per-inode bookkeeping is in-process; a recorder that abandons early could settle an admission before its transport ended; an admission-only barrier does not cover a discovery whose uncertainty write failed after every admission settled; links at acknowledgement contradicted acknowledging only after settlement, and an inherited uncertainty vanished at the next successor; the ledger's lock does not prove the old process gone; and four P2s: closure at zero pending and the margin's extra day, sleeping claims with no binding, absolute and `.` spellings in the walk, `(dev, ino)` as a lineage; each answered here. Revision 2 was rejected by Astra on three P1s: the VFS delegated SQLite's Unix file methods, which open the SHM and check moved files by pathname themselves; every settled admission wrote a permanent hold, so ordinary replication never read complete; a late owner-scoped extension did not reach successors that had already acknowledged; and six P2s: orphan admissions never terminated, `..` and chain edges unspecified, `dup` sharing the directory offset, the platform-link predicate not preserved and terminal links unclassified, the uid producer without a lifecycle, one billing owner not one replicator; each answered here. Revision 1 was rejected by Astra on five P1s: SQLite canonicalises a `/proc` pathname before it opens anything and this driver passes no `nofollow`; `os.Root` confines but re-resolves every name, so a swap between a listing and an inspection still moves the answer; a `Start` row is not an admission, since the meter sends without one; the attempt-derived window lost a request completed before the change was noticed and the days between an admission and a finish; `closed_at` was not a handover; plus the directory open before `O_DIRECTORY`, per-owner holds the ledger cannot store, a uid with no producer, I3 against Part D, and tests). Branch `design-replicator-pinned-root`, off `feat/carloku-metered-replicator-20261001` at `e1280811`. It answers the eleven findings the twentieth review deferred (`DISPOSITIONS-metered-replicator.md`, review twenty) and keeps decisions 10 to 13 and approval H (`DECISIONS-FOR-PAUL.md`, section D, on the launch branch) as parameters. Nothing is built, wired or applied.

+Date: 2026-10-03. Status: design, revision 16 (the tenth code review: the directory that listed a database is retained across the handoff to the ownership check and is the directory the file is inspected in, the name-resolved chain being required to be it; one open uncertainty per fleet, a replay reading its latch under the locks and folding into an open uncertainty; a fleet is never closed with an uncertainty nothing will close; a latch that clears dirties the days it barred, and publication enumerates an unresolved latch from the ledger's earliest day with an explicit bound, never a truncated list). Revision 15 was (the ninth code review: a hold, a latched discovery, an admission and its settlement are publication events, dirtying their days above any snapshot's watermark, and publication enumerates a latched discovery's days as it does the failure latch's; a snapshot reads again when the latched set changes across its read; a failed discovery keeps the bound and counted set its transaction computed, or latches unresolved and bars every day until the replay computes them; the replay closes on the settlement's own test, every unsettled row and the pending count; the discovery instant is stored and floors the closure; the settlement formula names the clamp to now; the handoff from the tree walk to the ownership checks compares the identity of the file and of every directory on the chain that listed it). Revision 14 was (the eighth code review: the recovery paragraph agrees with the transition table, an orphan's missing timestamps become now and a completion recorded before the crash keeps its instant; required coverage stated as ownership, with the deployment condition for the console database). Revision 13 was (the sixth code review: the predicate's compatibility stated as within the pinned roots, with the two safe wider shapes and the kept platform-directory refusal; the walk's pre-pin absent exception; the link-shapes line extended). Revision 12 was (Part A's root rule stated as the plan has it: the platform walk ends where the platform's authority ends, a tenant-owned component's parent, and the chain beneath is retained and rechecked rather than refused). Revision 11 was (the fourth code review: the historical term's filter named `finished` and `abandoned` and so excluded `unrecorded`, a settlement whose transport did send; it now excludes only the empty terminals, `unissued` and `cancelled`; the look's order is stated as what safety needs rather than a sequence; the closing look agrees). Revision 10 was revision 9's correction of the lower bound refined after the third code review: the historical term's threshold is the earlier of the registry proof and the last clean look's start, since a witness change is known only to be after the proof; an empty terminal, unissued or cancelled, never sent and is not a settlement for that term; and every look reads the witness and rechecks every retained chain before it inspects files, so a clean look's start bounds every kind of change. Revision 9 was one mechanics correction to the uncertainty's lower bound, found by the second code review of the admission model: an admission admitted before the last clean look, still in flight across a change after it and settled clean before the change was noticed, was outside the bound; the bound now also takes the admission instant of every admission settled at or after the last clean look's start; the invariant I2 is unchanged and the model's named test and schedules cover it). Revision 8 was the accepted text: revision 7 was accepted with fixes by Astra (no P1; one P2: a negative read-back was not conclusive while the admission writer could still commit, and the pending set had three definitions; one P3: the restore wording made a salt mismatch imply a snapshot), both applied here. Revision 6 was rejected by Astra on one P1: a discovery whose write failed had no rule for the admissions that settled before its retry, so the replayed interval could lose a settlement; and three P2s: a helper dying before its recorder started left an admission nobody could finish; the transition table lacked the abort edges and the pending set's cleanup; the lineage record cited a durability discipline the shadow state does not have; each answered here. Revision 5 was rejected by Astra on two P1s: the admission's insertion had no ordering point against discovery, so a discovery between the proof and the commit counted nothing and a later send fell outside the closed uncertainty; construction did not backfill already-closed intervals for a successor's new owners before acknowledgement; and P2s: the transition table and the helper-death rule disagreed and the recorder participant was synthesised while its worker could still run; proof-only callers of admit would have created admissions with no terminal; a terminal `.` lost the directory requirement; conflicting summaries; each answered here. Revision 4 was rejected by Astra on four P1s: the helper has several connections of its own and needs SQLite's within-process lock bookkeeping between them; the producer fence did not cover a helper that outlives the agent with an admission in hand; an intermediate fleet closed before acknowledging broke transitive protection; a fleet could be marked closed with a discovery uncommitted; and four P2s: a dead helper's admissions had no terminal in the running agent, the sleeping binding persisted a reusable inode, the lineage predicate named header fields that identify nothing, the walk's grammar refused a `.` target the predicate accepts; each answered here. Revision 3 was rejected by Astra on five P1s: two SQLite VFSs in one process cannot coordinate through POSIX locks, since the library's own per-inode bookkeeping is in-process; a recorder that abandons early could settle an admission before its transport ended; an admission-only barrier does not cover a discovery whose uncertainty write failed after every admission settled; links at acknowledgement contradicted acknowledging only after settlement, and an inherited uncertainty vanished at the next successor; the ledger's lock does not prove the old process gone; and four P2s: closure at zero pending and the margin's extra day, sleeping claims with no binding, absolute and `.` spellings in the walk, `(dev, ino)` as a lineage; each answered here. Revision 2 was rejected by Astra on three P1s: the VFS delegated SQLite's Unix file methods, which open the SHM and check moved files by pathname themselves; every settled admission wrote a permanent hold, so ordinary replication never read complete; a late owner-scoped extension did not reach successors that had already acknowledged; and six P2s: orphan admissions never terminated, `..` and chain edges unspecified, `dup` sharing the directory offset, the platform-link predicate not preserved and terminal links unclassified, the uid producer without a lifecycle, one billing owner not one replicator; each answered here. Revision 1 was rejected by Astra on five P1s: SQLite canonicalises a `/proc` pathname before it opens anything and this driver passes no `nofollow`; `os.Root` confines but re-resolves every name, so a swap between a listing and an inspection still moves the answer; a `Start` row is not an admission, since the meter sends without one; the attempt-derived window lost a request completed before the change was noticed and the days between an admission and a finish; `closed_at` was not a handover; plus the directory open before `O_DIRECTORY`, per-owner holds the ledger cannot store, a uid with no producer, I3 against Part D, and tests). Branch `design-replicator-pinned-root`, off `feat/carloku-metered-replicator-20261001` at `e1280811`. It answers the eleven findings the twentieth review deferred (`DISPOSITIONS-metered-replicator.md`, review twenty) and keeps decisions 10 to 13 and approval H (`DECISIONS-FOR-PAUL.md`, section D, on the launch branch) as parameters. Nothing is built, wired or applied.

## What stands today

diff --git a/internal/ops/authoritybridge.go b/internal/ops/authoritybridge.go

index 097d9880c..3e5b8dc01 100644

--- a/internal/ops/authoritybridge.go

+++ b/internal/ops/authoritybridge.go

@@ -39,13 +39,16 @@ type fleetPins struct {

pins *authority.Pins

roots map[string]*authority.Root

// listed is the facts of every database the strict tree walk listed,

- // by its spelled path, and listedChain the facts of every directory

- // from the tree's root down to the one that listed it: the identities

- // the ownership checks must find again (the listing-to-inspection

- // rule across the handoff: the file, and the directory that listed

- // it, are the ones inspected, or the file is refused).

+ // by its spelled path, and listedDirs the retained directory that

+ // listed it: the authority the ownership check inspects the file in

+ // (I1(b): the directory that listed is the directory inspected, held

+ // open across the handoff so its identity cannot be reused). The

+ // walk's kept directories and tree acquisitions are the pins' to

+ // close: trimmed to the owned files' chains after the ownership pass,

+ // the rest at Close.

listed map[string]authority.Facts

- listedChain map[string][]authority.Facts

+ listedDirs map[string]*authority.Dir

+ listedTrees []*listedTree

}

func newFleetPins() (*fleetPins, error) {

@@ -53,18 +56,71 @@ func newFleetPins() (*fleetPins, error) {

if err != nil {

return nil, err

}

- p := &fleetPins{slash: slash, roots: map[string]*authority.Root{}, listed: map[string]authority.Facts{}, listedChain: map[string][]authority.Facts{}}

+ p := &fleetPins{slash: slash, roots: map[string]*authority.Root{}, listed: map[string]authority.Facts{}, listedDirs: map[string]*authority.Dir{}}

p.pins = &authority.Pins{Slash: slash, Platform: platformFacts}

return p, nil

}

func (p *fleetPins) Close() {

+ for _, t := range p.listedTrees {

+ t.close()

+ }

+ p.listedTrees = nil

for _, r := range p.roots {

r.Close()

}

p.slash.Close()

}

+// listedTree is one strict walk's retained directories: the tree's

+// acquisition (top and its chain, released by release) and the kept

+// directories beneath it that listed a database.

+type listedTree struct {

+ top *authority.Dir

+ dirs []*authority.Dir // deepest last

+ release func()

+}

+

+func (t *listedTree) close() {

+ for i := len(t.dirs) - 1; i >= 0; i-- {

+ t.dirs[i].Close()

+ }

+ t.dirs = nil

+ if t.release != nil {

+ t.release()

+ t.release = nil

+ }

+}

+

+// trimListed closes every retained listing directory no owned file's

+// chain reaches, and releases a tree none of whose directories is in

+// use: what stays open is what Descriptors counts.

+func (p *fleetPins) trimListed(inUse map[*authority.Dir]bool) {

+ var kept []*listedTree

+ for _, t := range p.listedTrees {

+ var dirs []*authority.Dir

+ for _, d := range t.dirs {

+ if inUse[d] {

+ dirs = append(dirs, d)

+ } else {

+ d.Close()

+ }

+ t.dirs = dirs

+ if len(dirs) == 0 && !inUse[t.top] {

+ t.close()

+ continue

+ }

+ kept = append(kept, t)

+ }

+ p.listedTrees = kept

+ for path, d := range p.listedDirs {

+ if !inUse[d] {

+ delete(p.listedDirs, path)

+ }

+}

+

// comps splits a spelled absolute path into its components beneath the

// pinned slash; a path not beneath it is not something the fleet can look at.

func (p *fleetPins) comps(path string) ([]string, bool) {

@@ -539,35 +595,22 @@ func (p *fleetPins) recheck() error {

return first

}

-// chainFacts is the identity of every directory from top down to d, by

-// the retained parent links; nil when top is not above d.

-func chainFacts(d, top *authority.Dir) []authority.Facts {

- var rev []authority.Facts

+// chainBelow is the retained directories strictly below top (found above

+// d by identity) down to d, nearest top first; false when top is not

+// above d.

+func chainBelow(d, top *authority.Dir) ([]*authority.Dir, bool) {

+ var rev []*authority.Dir

for cur := d; cur != nil; cur = cur.Parent() {

- rev = append(rev, cur.Facts())

- if cur == top {

- out := make([]authority.Facts, len(rev))

- for i, f := range rev {

- out[len(rev)-1-i] = f

+ if cur.Facts().Same(top.Facts()) {

+ out := make([]*authority.Dir, len(rev))

+ for k, x := range rev {

+ out[len(rev)-1-k] = x

}

- return out

+ return out, true

}

+ rev = append(rev, cur)

}

- return nil

-}

-

-// chainMatches reports whether the directories ending at d (d and its

-// retained parents, nearest first) are, by identity, the listed chain's

-// (its last entry being the listing directory).

-func chainMatches(d *authority.Dir, listed []authority.Facts) bool {

- cur := d

- for i := len(listed) - 1; i >= 0; i-- {

- if cur == nil || !cur.Facts().Same(listed[i]) {

- return false

- }

- cur = cur.Parent()

- }

- return true

+ return nil, false

}

// ancestryChanged is a test seam over Root.AncestryChanged.

diff --git a/internal/ops/litestream.go b/internal/ops/litestream.go

index 815b05776..de60822c1 100644

--- a/internal/ops/litestream.go

+++ b/internal/ops/litestream.go

@@ -28,6 +28,7 @@ import (

"os"

"os/exec"

"path/filepath"

+ "regexp"

"sort"

"strconv"

"strings"

@@ -178,12 +179,30 @@ func replicaKey(dbPath string) string {

// pinned by TestConfigRoundTrip) — it is not a YAML parser.

func ParseLitestreamDBs(cfg string) ([]string, error) {

type item struct {

- indent int

+ indent int // the dash's

+ keys int // the indent of the entry's own keys, once known

path string

seen bool

}

+ // Documents this reader does not take are refused before anything is

+ // read as empty: a flow document, a quoted dbs or path key, a dbs key

+ // not at column zero (a nested or indented mapping).

+ for n, raw := range strings.Split(cfg, "\n") {

+ trim := strings.TrimSpace(raw)

+ switch {

+ case trim == "" || strings.HasPrefix(trim, "#"):

+ continue

+ case n == 0 && (strings.HasPrefix(trim, "{") || strings.HasPrefix(trim, "[")):

+ return nil, errors.New("litestream config: a flow document; block style only")

+ case quotedKeyRE.MatchString(trim):

+ return nil, fmt.Errorf("litestream config line %d: a quoted key (%s); bare keys only", n+1, trim)

+ case strings.HasPrefix(trim, "dbs:") && len(raw)-len(strings.TrimLeft(raw, " \t")) > 0 && !strings.HasPrefix(strings.TrimLeft(raw, " \t"), "-"):

+ return nil, fmt.Errorf("litestream config line %d: a dbs key that is not at the top level", n+1)

+ }

var dbs []string

inDBs := false

+ foundDBs := false

var cur *item

finish := func() error {

if cur != nil && !cur.seen {

@@ -207,6 +226,7 @@ func ParseLitestreamDBs(cfg string) ([]string, error) {

if indent == 0 {

if rest, ok := strings.CutPrefix(trim, "dbs:"); ok {

rest = strings.TrimSpace(stripComment(rest))

+ foundDBs = true

switch rest {

case "":

inDBs = true

@@ -233,17 +253,21 @@ func ParseLitestreamDBs(cfg string) ([]string, error) {

if err := finish(); err != nil {

return nil, err

}

- cur = &item{indent: indent}

- trim = strings.TrimLeft(strings.TrimPrefix(trim, "-"), " \t")

- if trim == "" {

+ cur = &item{indent: indent, keys: -1}

+ rest := strings.TrimLeft(strings.TrimPrefix(trim, "-"), " \t")

+ if rest == "" {

continue

}

- indent += 2 // the item's first key on the dash's line

+ indent += len(trim) - len(rest) // the entry's first key on the dash's line

+ trim = rest

}

if cur == nil {

return nil, fmt.Errorf("litestream config line %d: %q is under dbs but not in a list entry", n+1, trim)

}

- if v, ok := strings.CutPrefix(trim, "path:"); ok && indent > cur.indent {

+ if cur.keys < 0 && indent > cur.indent {

+ cur.keys = indent // the entry's own keys are at the first key's indent; deeper lines belong to a nested mapping (a replica's own path:)

+ }

+ if v, ok := strings.CutPrefix(trim, "path:"); ok && indent == cur.keys {

val, err := yamlScalar(strings.TrimSpace(stripComment(v)))

if err != nil {

return nil, fmt.Errorf("litestream config line %d: %v", n+1, err)

@@ -260,9 +284,15 @@ func ParseLitestreamDBs(cfg string) ([]string, error) {

if len(dbs) == 0 && sawPathKey {

return nil, errors.New("litestream config: path keys found outside a dbs list")

}

+ if !foundDBs && strings.Contains(cfg, "dbs") {

+ return nil, errors.New("litestream config: the document mentions dbs but no top-level block-style dbs key was recognised")

+ }

return dbs, nil

}

+// quotedKeyRE is a quoted dbs or path key, which this reader refuses.

+var quotedKeyRE = regexp.MustCompile(`^-?\s*["'](dbs|path)["']\s*:`)

+

// stripComment removes a trailing YAML comment (a # after whitespace).

func stripComment(s string) string {

if i := strings.Index(s, " #"); i >= 0 {

diff --git a/internal/ops/litestream_owners.go b/internal/ops/litestream_owners.go

index c0c460c43..d5d9e4c33 100644

--- a/internal/ops/litestream_owners.go

+++ b/internal/ops/litestream_owners.go

@@ -389,15 +389,6 @@ func fleetInventory(routes []registry.Route, registryPath, dataDir string) (Flee

}

continue

}

- // the directory that listed the file is the one it was checked

- // in, and so is every directory above it to the tree's root (the

- // same file moved into a replacement directory passes the file's

- // identity and not this)

- if lc, ok := pins.listedChain[p]; ok && facts.keep != nil && !chainMatches(facts.keep.dir, lc) {

- unowned(p, "it changed under the look (the directory it was checked in is not the directory that listed it)")

- facts.keep.release()

- continue

- }

inv.Owned = append(inv.Owned, OwnedDB{Path: p, Account: owner.Account, App: owner.App, Host: owner.Host, Platform: owner.Account == release.OpsAccount})

inv.facts[p] = facts

}

@@ -444,6 +435,17 @@ func fleetInventory(routes []registry.Route, registryPath, dataDir string) (Flee

}

inv.Problems = append(inv.Problems, r.Path+": a required database is not owned ("+r.Why+")")

}

+ // the retained listing directories: those on an owned file's chain

+ // stay for the fleet's life, the rest are closed now

+ inUse := map[*authority.Dir]bool{}

+ for _, f := range inv.facts {

+ if f.keep != nil {

+ for d := f.keep.dir; d != nil; d = d.Parent() {

+ inUse[d] = true

+ }

+ pins.trimListed(inUse)

sort.Slice(inv.Owned, func(i, j int) bool { return inv.Owned[i].Path < inv.Owned[j].Path })

sort.Strings(inv.Unowned)

return inv, nil

@@ -779,8 +781,34 @@ func checkOwnedFileOn(pins *fleetPins, root, path string, kind claimKind) (fileF

}

name := comps[len(comps)-1]

at = filepath.Join(at, name)

+ // A file the strict walk listed is inspected in the directory that

+ // listed it, retained since (I1(b)); the chain the name leads to must

+ // be that chain, directory by directory (litestream opens by name

+ // until Part B, so a name that leads elsewhere is refused as well),

+ // and then the name-resolved chain is released and the retained one

+ // kept. The retained directories are the pins' to close.

+ borrowed := false

+ if ld, ok := pins.listedDirs[path]; ok {

+ listedChain, found := chainBelow(ld, dir)

+ if !found || len(listedChain) != len(opened) {

+ return fileFacts{}, "it changed under the look (the directory it was checked in is not the directory that listed it)"

+ }

+ for i := range opened {

+ if !opened[i].Facts().Same(listedChain[i].Facts()) {

+ return fileFacts{}, "it changed under the look (the directory it was checked in is not the directory that listed it)"

+ }

+ for i := len(opened) - 1; i >= 0; i-- {

+ opened[i].Close()

+ }

+ opened = nil

+ cur, borrowed = ld, true

+ }

k, lf, lerr := cur.Lookat(name)

if lerr != nil || k == authority.Absent {

+ if borrowed {

+ return fileFacts{}, "it changed under the look (it is no longer in the directory that listed it)"

+ }

return fileFacts{}, "part of its path cannot be looked at"

}

if k == authority.Link {

@@ -836,6 +864,12 @@ func checkOwnedFileOn(pins *fleetPins, root, path string, kind claimKind) (fileF

}

kept = true

chainDir, chainOpened, chainRelease := cur, opened, release

+ if borrowed {

+ // the claim root's own acquisition is not kept: the retained

+ // listing chain reaches the same root, and the pins own it

+ release()

+ chainRelease = func() {}

+ }

keep := &retainedChain{dir: chainDir, name: name, facts: lf, release: func() {

for i := len(chainOpened) - 1; i >= 0; i-- {

chainOpened[i].Close()

@@ -957,7 +991,25 @@ func strictTreeDBsOn(pins *fleetPins, root string) (dbs, skipped, problems []str

if why != "" {

return nil, nil, []string{root + ": cannot be read (" + why + ")"}

}

- defer release()

+ // the directories that list a database are kept open, with their

+ // chain to the tree's root, and handed to the ownership check as the

+ // authority it inspects the file in; a walk that lists none releases

+ // everything here

+ kept := map[*authority.Dir]bool{}

+ tree := &listedTree{top: top, release: release}

+ defer func() {

+ if len(kept) == 0 {

+ release()

+ return

+ }

+ pins.listedTrees = append(pins.listedTrees, tree)

+ }()

+ keep := func(d *authority.Dir) {

+ for x := d; x != nil && x != top && !kept[x]; x = x.Parent() {

+ kept[x] = true

+ }

+ kept[top] = true

+ }

var walk func(d *authority.Dir, dir string, depth int, hidden bool)

walk = func(d *authority.Dir, dir string, depth int, hidden bool) {

if treeBeforeList != nil {

@@ -1070,14 +1122,19 @@ func strictTreeDBsOn(pins *fleetPins, root string) (dbs, skipped, problems []str

}

walk(child, p, depth+1, hidden || strings.HasPrefix(name, ".") || depth+1 > treeDBWalkDepth)

- child.Close()

+ if kept[child] {

+ tree.dirs = append(tree.dirs, child) // after its own children: deepest last

+ } else {

+ child.Close()

+ }

case authority.Regular:

switch {

case e.Facts.Size == 0:

case strings.HasSuffix(name, ".db") && !hidden && !strings.HasPrefix(name, "."):

dbs = append(dbs, p)

pins.listed[p] = e.Facts

- pins.listedChain[p] = chainFacts(d, top)

+ pins.listedDirs[p] = d

+ keep(d)

case sqliteName(name):

skipped = append(skipped, p)

default:

diff --git a/internal/ops/ops_test.go b/internal/ops/ops_test.go

index 3ffdf8c28..3ceca0d47 100644

--- a/internal/ops/ops_test.go

+++ b/internal/ops/ops_test.go

@@ -809,6 +809,14 @@ dbs:

path: '/data/it''s.db'

- replicas: []

path: /data/last.db

+- path: /data/file-replica.db

+ replicas:

+ - type: file

+ path: /backup/a

+- replicas:

+ - type: file

+ path: /backup/b

+ path: /data/replicas-first.db

logging:

level: info

`

@@ -816,7 +824,7 @@ logging:

if err != nil {

t.Fatal(err)

}

- want := []string{"/data/retired.db", "/data/with space.db", "/data/it's.db", "/data/last.db"}

+ want := []string{"/data/retired.db", "/data/with space.db", "/data/it's.db", "/data/last.db", "/data/file-replica.db", "/data/replicas-first.db"}

if strings.Join(got, ",") != strings.Join(want, ",") {

t.Fatalf("ParseLitestreamDBs = %q, want %q", got, want)

}

@@ -827,6 +835,10 @@ logging:

"bad quote": "dbs:\n - path: \"/a.db\n",

"stray path": "replicas:\n - path: /a.db\n",

"not in a list": "dbs:\n path: /a.db\n",

+ "quoted dbs": "\"dbs\":\n - path: /a.db\n",

+ "quoted path": "dbs:\n - \"path\": /a.db\n",

+ "flow document": "{\"dbs\": [{\"path\": \"/a.db\"}]}\n",

+ "nested dbs": "top:\n dbs:\n - path: /a.db\n",

} {

if got, err := ParseLitestreamDBs(bad); err == nil {

t.Fatalf("%s: read as %q, want an error", name, got)

diff --git a/internal/usage/durabilityadmissions_test.go b/internal/usage/durabilityadmissions_test.go

index cb2c4cbfa..9e8cafc26 100644

--- a/internal/usage/durabilityadmissions_test.go

+++ b/internal/usage/durabilityadmissions_test.go

@@ -441,12 +441,17 @@ func TestAdmissionClosedTokensAndReadBack(t *testing.T) {

if err := l.Admit("", "fleet-1", "alice", "app", at); err == nil {

t.Fatal("admitted with no token")

}

- if err := l.Admit(tok, "", "alice", "app", at); err == nil {

+ fresh1, _ := NewAdmissionToken()

+ if err := l.Admit(fresh1, "", "alice", "app", at); err == nil {

t.Fatal("admitted with no fleet")

}

- if err := l.Admit(tok, "fleet-1", "", "app", at); err == nil {

+ fresh2, _ := NewAdmissionToken()

+ if err := l.Admit(fresh2, "fleet-1", "", "app", at); err == nil {

t.Fatal("admitted with no owner")

}

+ if _, err := l.ReadAdmission(fresh1); !errors.Is(err, ErrAdmissionUnknown) {

+ t.Fatalf("a refused admission wrote: %v", err)

+ }

// a commit that fails after the insert leaves nothing: the read-back

// says unknown and the caller cancels conclusively

tok2, _ := NewAdmissionToken()

diff --git a/internal/usage/durabilityfleets.go b/internal/usage/durabilityfleets.go

index 5aa5853b9..c18786cc9 100644

--- a/internal/usage/durabilityfleets.go

+++ b/internal/usage/durabilityfleets.go

@@ -83,10 +83,10 @@ func (l *DurabilityLedger) OpenFleet(fleetID, box string, witnessGeneration int6

}

defer tx.Rollback() //nolint:errcheck

if f, err := readFleet(tx, fleetID); err == nil {

- if f.Predecessor == predecessor {

- return nil

+ if f.Predecessor == predecessor && f.Box == box && f.WitnessGeneration == witnessGeneration && strings.Join(f.Owners, "\n") == strings.Join(sorted, "\n") && f.OpenedAt.Equal(at) {

+ return nil // the same open again: it landed

}

- return fmt.Errorf("usage ledger: fleet %s is already open against %q", fleetID, f.Predecessor)

+ return fmt.Errorf("usage ledger: fleet %s is already open with other facts", fleetID)

} else if !errors.Is(err, ErrFleetUnknown) {

return err

}

@@ -94,6 +94,16 @@ func (l *DurabilityLedger) OpenFleet(fleetID, box string, witnessGeneration int6

if _, err := readFleet(tx, predecessor); err != nil {

return fmt.Errorf("usage ledger: predecessor %s: %w", predecessor, err)

}

+ // a predecessor has one successor: the chain is a line, as the

+ // model's Build always links to the tip; a second fleet naming it

+ // could never acknowledge it and its ancestry would never end

+ var successors int

+ if err := tx.QueryRow(`SELECT COUNT(*) FROM fleets WHERE predecessor = ?`, predecessor).Scan(&successors); err != nil {

+ return err

+ }

+ if successors > 0 {

+ return fmt.Errorf("usage ledger: predecessor %s already has a successor", predecessor)

+ }

}

if _, err := tx.Exec(`INSERT INTO fleets (fleet_id, box, witness_generation, owners, opened_at, state, predecessor, acknowledged_by) VALUES (?, ?, ?, ?, ?, ?, ?, '')`,

fleetID, box, witnessGeneration, strings.Join(sorted, "\n"), nsOf(at), FleetOpen, predecessor); err != nil {

@@ -200,6 +210,17 @@ func (l *DurabilityLedger) CloseFleet(fleetID string, at time.Time) error {

return err

}

+ // an open uncertainty with nothing left to settle it (a registration

+ // cancelled after its empty settlement was told, the counter dropping

+ // afterwards) closes here, on the settlement's own test: a fleet is

+ // never closed with an uncertainty nothing will close

+ if id, from, isOpen, err := openUncertainty(tx, fleetID); err != nil {

+ return err

+ } else if isOpen {

+ if err := l.maybeClose(tx, fleetID, id, from, at); err != nil {

+ return err

+ }

if _, err := tx.Exec(`UPDATE fleets SET state = ? WHERE fleet_id = ?`, FleetClosed, fleetID); err != nil {

return err

}

diff --git a/internal/usage/durabilityfleets_test.go b/internal/usage/durabilityfleets_test.go

index effd8b078..4e90ee71a 100644

--- a/internal/usage/durabilityfleets_test.go

+++ b/internal/usage/durabilityfleets_test.go

@@ -4,6 +4,7 @@ import (

"errors"

"math/rand"

"reflect"

+ "strings"

"sync/atomic"

"testing"

"time"

@@ -131,12 +132,76 @@ func TestFleetAncestryReachAndRootFirstAcknowledgement(t *testing.T) {

if err := l.Acknowledge("A", time.Time{}); err != nil {

t.Fatalf("a root acknowledging nothing: %v", err)

}

- if err := l.OpenFleet("D", "box", 1, []string{"dan"}, "A", time.Time{}); err != nil {

+ // a predecessor has one successor: the chain is a line

+ if err := l.OpenFleet("D", "box", 1, []string{"dan"}, "A", time.Time{}); err == nil || !strings.Contains(err.Error(), "already has a successor") {

+ t.Fatalf("a second fleet against A: %v", err)

+ }

+ if _, err := l.ReadFleet("D"); !errors.Is(err, ErrFleetUnknown) {

+ t.Fatalf("the refused fleet has a row: %v", err)

+ }

+}

+

+// Review ten (Opus P2-2): a fleet is never closed with an uncertainty

+// nothing will close. A discovery counts a pending registration; the

+// registration cancels and its empty settlement is told while the counter

+// still says one (the gate slice decrements first, but the ledger does not

+// rely on it); the counter then drops; the closure closes the uncertainty

+// on the settlement's own test before it marks the fleet closed.

+func TestFleetClosureClosesAnUncertaintyNothingWillClose(t *testing.T) {

+ l, set := openLedgerAt(t, at(1, 10, 0))

+ if err := l.OpenFleet("A", "box", 1, []string{"alice"}, "", time.Time{}); err != nil {

+ t.Fatal(err)

+ }

+ var pending atomic.Int64

+ pending.Store(1)

+ l.SetPendingCounter("A", &pending)

+ id, err := l.Discover(Discovery{FleetID: "A", At: at(1, 10, 0), Proof: at(1, 9, 0), LastCleanLook: at(1, 9, 30), Pending: []time.Time{at(1, 9, 50)}})

+ if err != nil {

+ t.Fatal(err)

+ }

+ set(at(1, 10, 5))

+ if err := l.EmptySettlement("A", time.Time{}); err != nil {

t.Fatal(err)

}

- if err := l.Acknowledge("D", time.Time{}); err == nil {

- t.Fatal("D acknowledged A, which B had acknowledged")

+ if u, _ := l.ReadUncertainty(id); !u.Open() {

+ t.Fatal("closed with the counter at one")

+ }

+ pending.Store(0)

+ set(at(1, 10, 10))

+ if err := l.FleetClosing("A"); err != nil {

+ t.Fatal(err)

+ }

+ if err := l.CloseFleet("A", time.Time{}); err != nil {

+ t.Fatal(err)

+ }

+ u, _ := l.ReadUncertainty(id)

+ if u.Open() || !u.To.Equal(at(1, 10, 11)) {

+ t.Fatalf("the closure left the uncertainty open: %+v", u)

+ }

+ for _, e := range mustEffective(t, l, "A") {

+ if e.Open() {

+ t.Fatalf("an open uncertainty on a closed fleet: %+v", e)

+ }

+ // the journal's repeat compares every column

+ if err := l.OpenFleet("A", "box", 1, []string{"alice"}, "", at(1, 10, 0)); err != nil {

+ t.Fatalf("the same open again: %v", err)

+ }

+ if err := l.OpenFleet("A", "box", 1, []string{"bob"}, "", at(1, 10, 0)); err == nil {

+ t.Fatal("A opened again with other owners")

+ }

+ if err := l.OpenFleet("A", "other-box", 1, []string{"alice"}, "", at(1, 10, 0)); err == nil {

+ t.Fatal("A opened again on another box")

+ }

+}

+

+func mustEffective(t *testing.T, l *DurabilityLedger, fleet string) []Uncertainty {

+ t.Helper()

+ eff, err := l.Effective(fleet)

+ if err != nil {

+ t.Fatal(err)

}

+ return eff

}

// Mirrors admissionmodel.TestHealthyChainContributesNothing: routine

@@ -322,8 +387,8 @@ func TestOpenFleetRefusals(t *testing.T) {

"close": func() error { return l.CloseFleet("A", time.Time{}) },

"acknowledge": func() error { return l.Acknowledge("A", time.Time{}) },

} {

- if err := f(); err == nil {

- t.Fatalf("%s on a closed ledger", name)

+ if err := f(); !errors.Is(err, errLedgerClosed) {

+ t.Fatalf("%s on a closed ledger: %v, want the ledger's own refusal", name, err)

}

@@ -334,7 +399,13 @@ func TestOpenFleetRefusals(t *testing.T) {

// closed one whole, an open one through its discovery), a successor that

// acknowledged holds its predecessor's effective set in its inherited

// set, an acknowledged step is out of the ancestry, and a closed fleet has

-// no unsettled admission and no latch.

+// no unsettled admission and no latch; and at the end of the schedule,

+// with everything settled, replayed and closed, no uncertainty of any

+// fleet is open (the model's quiescence check). Holds being box-wide on

+// this branch, the construction backfill's and the acknowledgement's

+// hold writes are the rows the discovery already wrote, so their removal

+// is not something this test can see: they are inspection-only until

+// owner scope, as the plan says.

func TestFleetScheduleInheritance(t *testing.T) {

seeds := 40

if testing.Short() {

@@ -348,6 +419,47 @@ func TestFleetScheduleInheritance(t *testing.T) {

open := map[string][]string{} // fleet -> unsettled tokens

acked := map[string]string{} // predecessor -> successor, as acknowledged in this schedule

closed := map[string]bool{}

+ pred := map[string]string{} // the test's own record of the chain

+ // expectedEffective is the test's own computation of a fleet's

+ // effective set from its record of the chain and the rows'

+ // fleet_id, independent of the ledger's Effective: its own

+ // uncertainties, those copied into inherited, and the own and

+ // inherited of every predecessor reached through steps not yet

+ // acknowledged

+ ownIDs := func(g string) map[int64]bool {

+ rows, err := l.db.Query(`SELECT id FROM uncertainties WHERE fleet_id = ?`, g)

+ if err != nil {

+ t.Fatal(err)

+ }

+ defer rows.Close()

+ out := map[int64]bool{}

+ for rows.Next() {

+ var id int64

+ if err := rows.Scan(&id); err != nil {

+ t.Fatal(err)

+ }

+ out[id] = true

+ }

+ return out

+ }

+ expectedEffective := func(g string) map[int64]bool {

+ want := map[int64]bool{}

+ for cur := g; cur != ""; {

+ for id := range ownIDs(cur) {

+ want[id] = true

+ }

+ inh, _ := l.Inherited(cur)

+ for _, id := range inh {

+ want[id] = true

+ }

+ p := pred[cur]

+ if p == "" || acked[p] == cur {

+ break

+ }

+ cur = p

+ }

+ return want

+ }

step := func(i int, what string) {

t.Helper()

for _, g := range fleets {

@@ -355,6 +467,15 @@ func TestFleetScheduleInheritance(t *testing.T) {

if err != nil {

t.Fatalf("seed %d step %d (%s): %v", seed, i, what, err)

}

+ want := expectedEffective(g)

+ if len(want) != len(eff) {

+ t.Fatalf("seed %d step %d (%s): fleet %s effective %v, the chain says %v", seed, i, what, g, ids(eff), want)

+ }

+ for _, u := range eff {

+ if !want[u.ID] {

+ t.Fatalf("seed %d step %d (%s): fleet %s effective has U%d, the chain does not", seed, i, what, g, u.ID)

+ }

for _, u := range eff {

to := u.To

if u.Open() {

@@ -399,6 +520,17 @@ func TestFleetScheduleInheritance(t *testing.T) {

if len(fleets) > 0 {

tip = fleets[len(fleets)-1]

}

+ // a live fleet, not the tip alone: an older fleet still settling

+ // behind a successor is the chain's ordinary state

+ live := ""

+ if len(fleets) > 0 {

+ for _, g := range []string{fleets[rnd.Intn(len(fleets))], tip} {

+ if !closed[g] {

+ live = g

+ break

+ }

what := ""

switch rnd.Intn(10) {

case 0, 1:

@@ -408,59 +540,60 @@ func TestFleetScheduleInheritance(t *testing.T) {

t.Fatalf("seed %d: open %s: %v", seed, id, err)

}

fleets = append(fleets, id)

+ pred[id] = tip

what = "open " + id

}

case 2, 3:

- if tip != "" && !closed[tip] {

- tok := mustAdmit(t, l, tip, "o"+tip, "app", now)

- open[tip] = append(open[tip], tok)

- what = "admit on " + tip

+ if live != "" {

+ tok := mustAdmit(t, l, live, "o"+live, "app", now)

+ open[live] = append(open[live], tok)

+ what = "admit on " + live

}

case 4, 5:

- if tip != "" && len(open[tip]) > 0 {

- k := rnd.Intn(len(open[tip]))

- tok := open[tip][k]

- open[tip] = append(open[tip][:k], open[tip][k+1:]...)

+ if live != "" && len(open[live]) > 0 {

+ k := rnd.Intn(len(open[live]))

+ tok := open[live][k]

+ open[live] = append(open[live][:k], open[live][k+1:]...)

if rnd.Intn(4) == 0 {

if err := l.Unissued(tok, now); err != nil {

t.Fatalf("seed %d: unissued: %v", seed, err)

}

- if err := l.EmptySettlement(tip, now); err != nil {

+ if err := l.EmptySettlement(live, now); err != nil {

t.Fatal(err)

}

} else {

finishAdmission(t, l, tok)

}

- what = "settle on " + tip

+ what = "settle on " + live

}

case 6:

- if tip != "" && !closed[tip] {

+ if live != "" {

if rnd.Intn(5) == 0 {

discoverFail = func() error { return errDiskFull }

}

- if _, err := l.Discover(Discovery{FleetID: tip, At: now, Proof: now.Add(-3 * time.Hour), LastCleanLook: now.Add(-time.Hour), Reason: "test"}); err != nil && !errors.Is(err, errDiskFull) {

- t.Fatalf("seed %d: discover on %s: %v", seed, tip, err)

+ if _, err := l.Discover(Discovery{FleetID: live, At: now, Proof: now.Add(-3 * time.Hour), LastCleanLook: now.Add(-time.Hour), Reason: "test"}); err != nil && !errors.Is(err, errDiskFull) {

+ t.Fatalf("seed %d: discover on %s: %v", seed, live, err)

}

discoverFail = nil

- what = "discover on " + tip

+ what = "discover on " + live

}

case 7:

- if tip != "" && !closed[tip] {

- if err := l.FleetClosing(tip); err != nil {

+ if live != "" {

+ if err := l.FleetClosing(live); err != nil {

t.Fatal(err)

}

- if _, latched := l.Latched(tip); latched && rnd.Intn(2) == 0 {

- if _, err := l.Replay(tip); err != nil {

- t.Fatalf("seed %d: replay %s: %v", seed, tip, err)

+ if _, latched := l.Latched(live); latched && rnd.Intn(2) == 0 {

+ if _, err := l.Replay(live); err != nil {

+ t.Fatalf("seed %d: replay %s: %v", seed, live, err)

}

- err := l.CloseFleet(tip, now)

+ err := l.CloseFleet(live, now)

if err == nil {

- closed[tip] = true

+ closed[live] = true

} else if !errors.Is(err, ErrFleetUnsettled) {

- t.Fatalf("seed %d: close %s: %v", seed, tip, err)

+ t.Fatalf("seed %d: close %s: %v", seed, live, err)

}

- what = "close " + tip

+ what = "close " + live

}

case 8, 9:

if len(fleets) >= 2 {

@@ -479,6 +612,38 @@ func TestFleetScheduleInheritance(t *testing.T) {

step(i, what)

}

+ // quiescence: settle, replay and close everything, then nothing is

+ // open anywhere

+ now = now.Add(time.Hour)

+ set(now)

+ for _, g := range fleets {

+ for _, tok := range open[g] {

+ finishAdmission(t, l, tok)

+ }

+ open[g] = nil

+ if _, latched := l.Latched(g); latched {

+ if _, err := l.Replay(g); err != nil {

+ t.Fatalf("seed %d: final replay of %s: %v", seed, g, err)

+ }

+ if !closed[g] {

+ if err := l.FleetClosing(g); err != nil {

+ t.Fatal(err)

+ }

+ if err := l.CloseFleet(g, now); err != nil {

+ t.Fatalf("seed %d: final close of %s: %v", seed, g, err)

+ }

+ closed[g] = true

+ }

+ step(60, "quiescence")

+ for _, g := range fleets {

+ for _, u := range mustEffective(t, l, g) {

+ if u.Open() {

+ t.Fatalf("seed %d: U%d (fleet %s) open at quiescence, in fleet %s's effective set", seed, u.ID, u.FleetID, g)

+ }

l.Close(t.Context(), nil)

}

diff --git a/internal/usage/durabilityholds_test.go b/internal/usage/durabilityholds_test.go

index ede5ac1fa..d1f5224fa 100644

--- a/internal/usage/durabilityholds_test.go

+++ b/internal/usage/durabilityholds_test.go

@@ -89,6 +89,14 @@ func TestHoldsAndBarriersRepublishTheirDays(t *testing.T) {

// incomplete; its settlement republishes it complete

clock.set(ledgerDay0.Add(60 * time.Hour))

tok := mustAdmit(t, l, "g", "alice", "app", time.Time{})

+ // the admission alone is a publication event: the open day publishes

+ // barred before any seal

+ if n, err := l.Publish(ctx, store); err != nil || n < 1 {

+ t.Fatalf("publish after an admission on an open day: n=%d err=%v", n, err)

+ }

+ if ok, why := readDurabilityDay(t, store, "box1", day2).Completeness(); ok || !strings.Contains(why, "unsettled") {

+ t.Fatalf("the open day after an admission: complete=%v %q", ok, why)

+ }

clock.set(ledgerDay0.Add(72*time.Hour + time.Second))

if err := l.Seal(); err != nil {

t.Fatal(err)

@@ -140,6 +148,125 @@ func TestHoldsAndBarriersRepublishTheirDays(t *testing.T) {

if ok, why := readDurabilityDay(t, store, "box1", day3).Completeness(); ok || !strings.Contains(why, "hold") {

t.Fatalf("day 3 after the racing hold: complete=%v %q", ok, why)

}

+ // an unresolved latch (review ten, Opus P2-3): a day published complete

+ // long before, clean since, is republished incomplete, because the

+ // unresolved bound bars every day the ledger has

+ clock.set(ledgerDay0.Add(120 * time.Hour))

+ recordOne(t, c, clock, "PutObject", 200)

+ clock.set(ledgerDay0.Add(144*time.Hour + time.Second))

+ if err := l.Seal(); err != nil {

+ t.Fatal(err)

+ }

+ if _, err := l.Publish(ctx, store); err != nil {

+ t.Fatal(err)

+ }

+ day5 := DayOf(ledgerDay0.Add(120 * time.Hour))

+ if ok, why := readDurabilityDay(t, store, "box1", day5).Completeness(); !ok {

+ t.Fatalf("precondition: day 5 complete: %q", why)

+ }

+ rowsErr = func(*sql.Rows) error { return errors.New("read failed") }

+ _, err = l.Discover(Discovery{FleetID: "u", At: clock.now(), Proof: ledgerDay0.Add(144*time.Hour + 30*time.Minute), LastCleanLook: ledgerDay0.Add(145 * time.Hour)}) // the true bound is on day 6: day 5 is barred by the overlay alone

+ rowsErr = func(rows *sql.Rows) error { return rows.Err() }

+ if err == nil {

+ t.Fatal("the unreadable discovery did not say so")

+ }

+ if from, ok := l.Latched("u"); !ok || !from.IsZero() {

+ t.Fatalf("not latched unresolved: %v %v", from, ok)

+ }

+ if n, err := l.Publish(ctx, store); err != nil || n < 1 {

+ t.Fatalf("publish with an unresolved latch: n=%d err=%v", n, err)

+ }

+ if ok, why := readDurabilityDay(t, store, "box1", day5).Completeness(); ok || !strings.Contains(why, "unsettled") {

+ t.Fatalf("day 5 under an unresolved latch: complete=%v %q", ok, why)

+ }

+ if ok, _ := readDurabilityDay(t, store, "box1", day0).Completeness(); ok {

+ t.Fatal("day 0 reads complete under an unresolved latch")

+ }

+ // the overlay's inverse (Astra, review ten): the replay finds the true

+ // bound on day 6 and clears the latch; day 5, published barred by the

+ // overlay alone, is published again and reads complete

+ if _, err := l.Replay("u"); err != nil {

+ t.Fatal(err)

+ }

+ if _, ok := l.Latched("u"); ok {

+ t.Fatal("the latch survived the replay")

+ }

+ if n, err := l.Publish(ctx, store); err != nil || n < 1 {

+ t.Fatalf("publish after the replay cleared the overlay: n=%d err=%v", n, err)

+ }

+ if ok, why := readDurabilityDay(t, store, "box1", day5).Completeness(); !ok {

+ t.Fatalf("day 5 after the overlay cleared: %q", why)

+ }

+}

+

+// Review ten (Opus P2-1): one open uncertainty per fleet. A Replay that

+// read the latch and then waited on the locks while a Discover replayed

+// it finds nothing to do; a replay of a latch while an uncertainty of the

+// fleet is already open folds into it, lowering its bound, and opens no

+// second row.

+func TestReplayNeverOpensASecondUncertainty(t *testing.T) {

+ l, set := openLedgerAt(t, at(2, 9, 0))

+ set(at(2, 23, 58))

+ tok := mustAdmit(t, l, "f", "alice", "app", time.Time{})

+ discoverFail = func() error { return errors.New("disk full") }

+ l.Discover(Discovery{FleetID: "f", At: at(2, 23, 58), Proof: at(2, 23, 0), LastCleanLook: at(2, 23, 30)})

+ discoverFail = nil

+ set(at(3, 0, 10))

+ raced := false

+ replayBeforeLocks = func() {

+ if raced {

+ return

+ }

+ raced = true

+ if _, err := l.Discover(Discovery{FleetID: "f", At: at(3, 0, 10), Proof: at(2, 23, 0), LastCleanLook: at(2, 23, 30)}); err != nil {

+ t.Error(err)

+ }

+ defer func() { replayBeforeLocks = nil }()

+ id, err := l.Replay("f")

+ if err != nil {

+ t.Fatal(err)

+ }

+ if !raced || id != 0 {

+ t.Fatalf("the replay of a latch a discovery had replayed meanwhile: id %d raced %v", id, raced)

+ }

+ if n := countUncertainties(t, l, "f"); n != 1 {

+ t.Fatalf("%d uncertainties for the fleet, want one", n)

+ }

+ // a latch replayed while an uncertainty is open folds into it

+ eff, _ := l.Effective("f")

+ _ = eff

+ var open int64

+ if err := l.db.QueryRow(`SELECT id FROM uncertainties WHERE fleet_id = 'f' AND to_ns IS NULL`).Scan(&open); err != nil {

+ t.Fatalf("the replayed uncertainty is not open: %v", err)

+ }

+ pd := &pendingDiscovery{d: Discovery{FleetID: "f", At: at(2, 23, 50), Proof: at(2, 20, 0), LastCleanLook: at(2, 22, 0)}, from: at(2, 19, 59), counted: map[string]bool{}}

+ l.mu.Lock()

+ fid, err := l.replayLocked("f", pd, at(3, 0, 20))

+ l.mu.Unlock()

+ if err != nil {

+ t.Fatal(err)

+ }

+ if fid != open || countUncertainties(t, l, "f") != 1 {

+ t.Fatalf("the fold opened a second row: %d (open %d), %d rows", fid, open, countUncertainties(t, l, "f"))

+ }

+ if u, _ := l.ReadUncertainty(open); !u.From.Equal(at(2, 19, 59)) || !u.Open() {

+ t.Fatalf("the fold did not lower the bound: %+v", u)

+ }

+ held(t, l, true, "2026-10-02")

+ finishAdmission(t, l, tok)

+ if u, _ := l.ReadUncertainty(open); u.Open() {

+ t.Fatalf("still open after the settlement: %+v", u)

+ }

+}

+

+func countUncertainties(t *testing.T, l *DurabilityLedger, fleet string) int {

+ t.Helper()

+ var n int

+ if err := l.db.QueryRow(`SELECT COUNT(*) FROM uncertainties WHERE fleet_id = ?`, fleet).Scan(&n); err != nil {

+ t.Fatal(err)

+ }

+ return n

}

// Review nine (Astra 1): a snapshot whose database read predates a

@@ -376,8 +503,8 @@ func TestDiscoveryEdges(t *testing.T) {

if _, err := l2.Discover(Discovery{FleetID: "g", At: at(3, 13, 0), Proof: at(3, 9, 0)}); !errors.Is(err, errLedgerClosed) {

t.Fatalf("discover on a closed ledger: %v", err)

}

- if _, err := l2.Replay("g"); err != nil { // nothing latched: nil before the closed check is reached

- t.Fatal(err)

+ if _, err := l2.Replay("g"); !errors.Is(err, errLedgerClosed) {

+ t.Fatalf("replay on a closed ledger: %v", err)

}

if err := l2.EmptySettlement("g", time.Time{}); !errors.Is(err, errLedgerClosed) {

t.Fatalf("empty settlement on a closed ledger: %v", err)

diff --git a/internal/usage/durabilitypublish.go b/internal/usage/durabilitypublish.go

index 80af1705d..f0132be5d 100644

--- a/internal/usage/durabilitypublish.go

+++ b/internal/usage/durabilitypublish.go

@@ -2,6 +2,7 @@ package usage

import (

"context"

+ "database/sql"

"encoding/json"

"errors"

"fmt"

@@ -68,6 +69,13 @@ func (l *DurabilityLedger) publish(ctx context.Context, store release.Store) (in

l.mu.Unlock()

return 0, l.noteFork(ctx, store)

}

+ // the ledger's earliest day: the floor an unresolved latched discovery

+ // is enumerated from (it bars every day this ledger has)

+ var earliest sql.NullString

+ if err := l.db.QueryRow(`SELECT MIN(date) FROM days`).Scan(&earliest); err != nil {

+ l.mu.Unlock()

+ return 0, err

+ }

rows, err := l.db.Query(`SELECT date FROM days WHERE dirty = 1 ORDER BY date`)

if err != nil {

l.mu.Unlock()

@@ -107,12 +115,16 @@ func (l *DurabilityLedger) publish(ctx context.Context, store release.Store) (in

// from the earliest day this ledger has) through today, so the bucket

// never reads complete over a discovery the process already knows of.

floor := l.now().UTC()

- if len(dates) > 0 {

- if f, _, err := DayBounds(dates[0]); err == nil {

+ if earliest.Valid {

+ if f, _, err := DayBounds(earliest.String); err == nil {

floor = f

}

- for _, d := range l.latchedDates(floor, l.now()) {

+ latchedDays, err := l.latchedDates(floor, l.now())

+ if err != nil {

+ return 0, err

+ }

+ for _, d := range latchedDays {

if !seen[d] {

seen[d] = true

dates = append(dates, d)

diff --git a/internal/usage/durabilityuncertainty.go b/internal/usage/durabilityuncertainty.go

index 532e7c564..2b999b77b 100644

--- a/internal/usage/durabilityuncertainty.go

+++ b/internal/usage/durabilityuncertainty.go

@@ -233,11 +233,23 @@ func (l *DurabilityLedger) writeHolds(tx *sql.Tx, from, to time.Time, uncertaint

}

// dirtyInterval dirties every day of [from, to] with a fresh sequence:

-// the barrier appearing (an admission) and lifting (its settlement) are

-// publication events. Caller holds l.mu.

+// the barrier appearing (an admission) and lifting (its settlement), and

+// the overlay a replay removes, are publication events. An interval past

+// the bound is an error (the write fails, conservatively), never a

+// truncated set of days. Caller holds l.mu.

func (l *DurabilityLedger) dirtyInterval(tx *sql.Tx, from, to time.Time) error {

- if err := touchDays(tx, from, to, l.allocSeq()); err != nil {

- return err

+ from, to = from.UTC(), to.UTC()

+ if to.Before(from) {

+ from, to = to, from

+ }

+ if to.Sub(from) > holdsMaxDays*24*time.Hour {

+ return fmt.Errorf("usage ledger: an interval over %s to %s is longer than %d days", from.Format(time.RFC3339), to.Format(time.RFC3339), holdsMaxDays)

+ }

+ seq := l.allocSeq()

+ for d := from.Truncate(24 * time.Hour); !d.After(to); d = d.Add(24 * time.Hour) {

+ if err := touchDay(tx, DayOf(d), seq); err != nil {

+ return err

+ }

}

return l.persistSeq(tx)

}

@@ -354,8 +366,10 @@ func openUncertainty(tx *sql.Tx, fleetID string) (id int64, from time.Time, ok b

var discoverFail func() error

// replayFail is a test seam: an error returned here fails the replay's

-// transaction before its commit.

+// transaction before its commit. replayBeforeLocks runs between Replay's

+// first look at the latch and its locks.

var replayFail func() error

+var replayBeforeLocks func()

// latchDiscovery keeps a failed discovery in memory with the bound and

// counted set its transaction computed; a failure before those could be

@@ -403,10 +417,16 @@ func (l *DurabilityLedger) latchedBarring(dayEnd time.Time) int64 {

return n

}

+// errLatchSpan is a latched discovery whose span publication cannot

+// enumerate in one pass; publication fails loudly rather than publishing

+// a truncated set of the days it bars.

+var errLatchSpan = fmt.Errorf("usage ledger: a latched discovery bars more than %d days; publication waits for its replay", holdsMaxDays)

+

// latchedDates is every day a latched discovery bars, from its bound (or

-// the given floor, for an unresolved one) through now, for publication's

-// enumeration; bounded like a gap's touch.

-func (l *DurabilityLedger) latchedDates(floor, now time.Time) []string {

+// the given floor, the ledger's earliest day, for an unresolved one)

+// through now, for publication's enumeration. A span past the bound is an

+// error, never a truncated list.

+func (l *DurabilityLedger) latchedDates(floor, now time.Time) ([]string, error) {

ds := l.disc()

ds.mu.Lock()

defer ds.mu.Unlock()

@@ -419,15 +439,14 @@ func (l *DurabilityLedger) latchedDates(floor, now time.Time) []string {

if from.IsZero() {

from = now

}

- n := 0

+ if now.Sub(from) > holdsMaxDays*24*time.Hour {

+ return nil, errLatchSpan

+ }

for d := from.UTC().Truncate(24 * time.Hour); !d.After(now.UTC()); d = d.Add(24 * time.Hour) {

- if n++; n > ledgerGapTouchMax {

- break

- }

out = append(out, DayOf(d))

}

- return out

+ return out, nil

}

// Replay is the latched discovery's one transaction: the uncertainty at

@@ -439,14 +458,18 @@ func (l *DurabilityLedger) latchedDates(floor, now time.Time) []string {

// nothing is pending (the same test as a settlement's); the latch clears

// only after the commit. Mirrors admissionmodel.Replay.

func (l *DurabilityLedger) Replay(fleetID string) (int64, error) {

- ds := l.disc()

- ds.mu.Lock()

- pd, ok := ds.latched[fleetID]

- ds.mu.Unlock()

- if !ok {

+ l.mu.Lock()

+ closed := l.closed

+ l.mu.Unlock()

+ if closed {

+ return 0, errLedgerClosed

+ }

+ if _, ok := l.Latched(fleetID); !ok {

return 0, nil

}

- now := l.now().UTC()

+ if replayBeforeLocks != nil {

+ replayBeforeLocks()

+ }

l.drainMu.Lock()

defer l.drainMu.Unlock()

l.mu.Lock()

@@ -454,7 +477,17 @@ func (l *DurabilityLedger) Replay(fleetID string) (int64, error) {

if l.closed {

return 0, errLedgerClosed

}

- return l.replayLocked(fleetID, pd, now)

+ // the latch read again under the locks: a discovery or a closure that

+ // held them meanwhile may have replayed it, and a replay of a cleared

+ // latch would open a second uncertainty

+ ds := l.disc()

+ ds.mu.Lock()

+ pd, ok := ds.latched[fleetID]

+ ds.mu.Unlock()

+ if !ok {

+ return 0, nil

+ }

+ return l.replayLocked(fleetID, pd, l.now().UTC())

}

// replayLocked is Replay under drainMu and l.mu: its own transaction, the

@@ -507,13 +540,30 @@ func (l *DurabilityLedger) replayTx(tx *sql.Tx, fleetID string, pd *pendingDisco

for tok := range pd.counted {

counted[tok] = true

}

- res, err := tx.Exec(`INSERT INTO uncertainties (fleet_id, from_ns, to_ns, discovered_at, reason) VALUES (?, ?, NULL, ?, ?)`, fleetID, nsOf(from), nsOf(pd.d.At), pd.d.Reason)

+ // one open uncertainty per fleet: a replay that finds one open (a

+ // discovery committed while this one was latched) folds into it,

+ // lowering its bound to the earlier, rather than opening a second

+ // that nothing would close

+ id, openFrom, isOpen, err := openUncertainty(tx, fleetID)

if err != nil {

return 0, err

}

- id, err := res.LastInsertId()

- if err != nil {

- return 0, err

+ if isOpen {

+ if openFrom.Before(from) {

+ from = openFrom

+ } else if from.Before(openFrom) {

+ if _, err := tx.Exec(`UPDATE uncertainties SET from_ns = ? WHERE id = ?`, nsOf(from), id); err != nil {

+ return 0, err

+ }

+ } else {

+ res, err := tx.Exec(`INSERT INTO uncertainties (fleet_id, from_ns, to_ns, discovered_at, reason) VALUES (?, ?, NULL, ?, ?)`, fleetID, nsOf(from), nsOf(pd.d.At), pd.d.Reason)

+ if err != nil {

+ return 0, err

+ }

+ if id, err = res.LastInsertId(); err != nil {

+ return 0, err

+ }

}

if err := markStale(tx, fleetID); err != nil {

return 0, err

@@ -521,6 +571,26 @@ func (l *DurabilityLedger) replayTx(tx *sql.Tx, fleetID string, pd *pendingDisco

if err := l.writeHolds(tx, from, pd.d.At, id, now); err != nil {

return 0, err

}

+ // the overlay's inverse: the days the latch barred while it stood

+ // (from its bound, or from the ledger's earliest day while it was

+ // unresolved) are dirtied again, so a day published barred and now

+ // outside the uncertainty is published again

+ barredFrom := pd.from

+ if pd.unresolved || barredFrom.IsZero() {

+ var earliest sql.NullString

+ if err := tx.QueryRow(`SELECT MIN(date) FROM days`).Scan(&earliest); err != nil {

+ return 0, err

+ }

+ barredFrom = now

+ if earliest.Valid {

+ if f, _, err := DayBounds(earliest.String); err == nil {

+ barredFrom = f

+ }

+ if err := l.dirtyInterval(tx, barredFrom, now); err != nil {

+ return 0, err

+ }

for token := range counted {

a, err := readAdmission(tx, token)

if err != nil {

carlos / platform Public