1dc9a6c7b · carlos/platform

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

index 3f65c2a0d..4f594faa7 100644

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

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

@@ -199,3 +199,20 @@ Opus: accept with fixes, no P1, three P2. Astra: reject, one P1, five P2, one P3

| 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. |

+

+## Review eleven: Codex continuation on 2026-10-03

+

+Source reviewed: `358b25ee`. Astra rejected one P1 and five P2s; Opus accepted with fixes (three P2s). The user's transfer authorizes independent Codex review while Claude is unavailable. A new Opus review remains pending; no Opus acceptance is inferred.

+

+- Admission remains refused after discovery commits and before the legacy stale flag is stored. Registration and issuance both check the permanent discovery flag.

+- Issuance checks the monotonic deadline and request cancellation under the discovery mutex. Refused rows and ambiguous commit outcomes have an independent retry owner; request refusal never waits for cleanup I/O.

+- Participant writes remain ordered by admission token. Transport completion retains the live token until durable success. Closure retries survive the fleet's watch/store shutdown, and a successor acknowledges its recorded ancestry root-first, including unopened intermediate fleets.

+- Discovery retains the captured registration instants for its lower bound, but closes against current rows and the live counter. If captured registrations resolved before the transaction, closure includes the current ledger instant plus margin. Tests distinguish cancellation and unissued settlement across three calendar days.

+- Replay commit and latch deletion share the overlay mutex in Replay, Discover and CloseFleet. The deterministic test proves snapshot consistency at that boundary; it does not claim a separately reproduced permanent publication failure.

+- A nonempty unsupported YAML document cannot become an empty database set. Escaped keys are rejected by parser and coverage guard. Supported empty forms remain accepted.

+- A fleet open whose commit answer was lost reads its unique id back before abandoning construction.

+- Lease tests observe admission callbacks for failed Start/Finish, late persistence and saturation; the Close test no longer repairs its journal by hand.

+

+Independent review found two defects in the first local retry draft (synchronous cleanup and a backdated cancellation) and one remaining discovery upper-bound defect. All were corrected before the full gate. Evidence and exact checkpoint state are recorded in the run root's `logs/codex-review11-*` and `logs/codex-replroot-review12.md`.

+

+Still outstanding: persisted admission-token association in meter Start rows; slice 6's producer fence/recovery integration and validation; slice 7's reader marker; Part B's descriptor-based data opens; remaining authority/flag inspection-only claims and previously recorded policy decisions. The saved slice-6 copies were inspected byte-for-byte and left intact. Their reported six killed mutants have no located transcript, so that report is not reused as verified test evidence. This checkpoint does not establish real AWS or Stripe operation, complete metering, or launch readiness. Live billing remains off.

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 fabecd7ec..f7509cd50 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 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).

+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, returning an error if the interval exceeds 4000 days; 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 (built: `internal/ops/litestream_admission.go` and the fleet's wiring in `litestream_metered.go`; `lease.RequestGate` returns a `lease.Admission` whose transport the gate middleware completes at the SDK's finalize step and whose recorder the meter completes beside its own writes, in the worker: registered when `Start` lands, finished when `Finish` lands, abandoned with a late `Start` or a failed `Finish`, unrecorded when `Start` failed; a gate with no ledger is `lease.ProofGate`, admitting with `lease.Unadmitted`, and a gate answering with no admission is refused. The fleet's gate: the proof, a registration in the pending set under the discovery mutex (refused while stale, latched or in discovery), the row committed by one writer goroutine within `admissionBudget`, the recheck under the mutex (stale, latched, in discovery or an epoch passed: `unissued`), a definite failure cancelled with its empty settlement, an ambiguous one read back by its token within `admissionReadBackWait` and otherwise left pending and counted; past the budget `ErrAdmissionNotDurable` and the writer's outcome resolved by the writer. Discovery once per fleet from `goStaleFrom`, with the pending instants, the proof and the last clean look; a latched one replayed from the watch and at Close. The journal: the open row at build (a fleet that cannot be journalled is not built), `closing` at Close after the final look, `transport-lost` for every issued admission whose transport never finalised, `closed` waited for within `closeSettleWait` and left `closing` past it; acknowledgement at Open and at every watch tick until it lands. The helper process does not exist yet, so all of it runs in the agent. Tests: every send admitted and settled, the journal's rows and a replacement's acknowledgement, a discovery between registration and commit refusing the commit at its recheck with the registration counted, the budget and a writer failing against a ledger that cannot answer, transport-lost at Close with the recorder pending.)** 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`.

+5. **The gate and the fleet (built: `internal/ops/litestream_admission.go` and the fleet's wiring in `litestream_metered.go`; `lease.RequestGate` returns a `lease.Admission` whose transport the gate middleware completes at the SDK's finalize step and whose recorder the meter completes beside its own writes, in the worker: registered when `Start` lands, finished when `Finish` lands, abandoned with a late `Start` or a failed `Finish`, unrecorded when `Start` failed; a gate with no ledger is `lease.ProofGate`, admitting with `lease.Unadmitted`, and a gate answering with no admission is refused. The fleet's gate: the proof, a registration in the pending set under the discovery mutex (refused while stale, latched or in discovery), the row committed by one writer goroutine within `admissionBudget`, the recheck under the mutex (stale, latched, in discovery or an epoch passed: `unissued`), a definite failure cancelled with its empty settlement, an ambiguous one read back by its token by the independent retry worker until the ledger answers, left pending and counted meanwhile; past the budget `ErrAdmissionNotDurable` and the writer's outcome resolved by the writer. Discovery once per fleet from `goStaleFrom`, with the pending instants, the proof and the last clean look; a latched one replayed from the watch and at Close. The journal: the open row at build (a fleet that cannot be journalled is not built), `closing` at Close after the final look, `transport-lost` for every issued admission whose transport never finalised, `closed` waited for within `closeSettleWait` with closure retained by the retry worker past it; acknowledgement at Open and at every watch tick until it lands. The helper process does not exist yet, so all of it runs in the agent. Tests: every send admitted and settled, the journal's rows and a replacement's acknowledgement, a discovery between registration and commit refusing the commit at its recheck with the registration counted, the budget and a writer failing against a ledger that cannot answer, transport-lost at Close with the recorder pending.)** 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 callbacks identify it, but the persisted meter attempt still lacks its admission-token association (outstanding); the Deserialize handler return 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 4a759848f..e25e1e694 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 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.

+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. Parts A and D required coverage and Part C slices 1–5 are implemented on this branch. Slice 6 is preserved in a separate source copy; slice 7 and Part B remain outstanding. Nothing is wired or deployed.

## What stands today

@@ -42,7 +42,7 @@ The VFS is the plan's first spike and it must prove, before anything else of Par

## Part C. The durable admission, two kinds of hold, and a handover on disk (Astra 3, 4)

-**Proof and admission are two things.** Fleet construction, `Open` and every watch tick ask the ownership *proof* (today's `admit`: the witness unchanged, the fleet not stale) and create nothing; only the request gate (`lease.RequestGate`, asked by the helper for each physical request) performs an *admission*, which is a proof followed by a row, so an idle healthy fleet has no admission and nothing bars its days. **The admission is a row, written before the send, with one ordering point against discovery.** Under a *discovery mutex* on the fleet (a new requirement: today `admit` and `lookAgain` do not share one, and the admission comments allow a discovery to run unlocked; this design serialises registration, the recheck and the start of a discovery on it) the gate checks the proof and, if the fleet is not stale and no discovery is in progress, registers the admission in the fleet's in-memory *pending set* with its `admitted_at`; it releases the mutex and commits the row in the durability ledger's own database (`admissions`: `token`, `fleet_id`, `owner account and app`, `admitted_at`, `transport_done_at`, `recorder_started_at`, `recorder_done_at`, `settled_at`, `how`), `synchronous=FULL`, within a bounded budget; after the commit it takes the mutex again and, if the fleet went stale or a discovery began since registration, marks the row `unissued` and refuses. Discovery, under the same mutex, sets the in-progress flag and counts the pending set, durable or not yet, as unsettled admissions when it computes its lower bound, and a registered admission that the discovery has counted and that then commits is inside the uncertainty (it was counted); one that commits after the discovery began is refused. So an admission is either counted by every discovery that follows its registration or refused, and Astra's 23:58:10 sequence (proof, discovery, commit, a send on the 4th) ends in a refusal at the recheck, not a send outside the uncertainty. The gate waits for the commit. A commit that does not return within the budget refuses the request (`ErrAdmissionNotDurable`; litestream sees a failed upload and does not advance) and the goroutine that awaited it stays to see the commit land and then marks the row `how = unissued, settled_at = now` in its own transaction, so a late commit never authorises anything and never blocks a day; if the process dies between, recovery classifies the row as below. The meter's `Start` row carries the token (`Start` may still fail or be late, as today; what is now mandatory is the admission, not the `Start`). An admission has two participants, the transport and the recorder, and is settled by whichever finishes last. The transport writes `transport_done_at` from the SDK middleware's finalize step, or `transport-lost` when its helper is proven gone (below). The recorder writes `recorder_done_at` with its outcome: `finished` (`Finish`), `abandoned` (`Abandon`), or `unrecorded` (the request path, when `Start` failed and made no attempt row, which today calls neither; it now writes this terminal); the recorder participant is never synthesised while its worker can still run: today's meter lets recording workers outlive the request path, and the admission stays unsettled (barring) until that worker finishes or abandons. The recorder runs in the agent, fed by the helper's messages over the fleet socket (`Start`, `Finish`, `Abandon` with the token), and the agent *registers* the recorder participant when it receives `Start` (`recorder_started_at`, under an agent-owned lifecycle guard per helper); a `Start` that the agent receives after it has begun the helper's termination handling is refused (the guard is closed first, then the socket drained, so registration and exit handling cannot interleave). The participant that finds the other already done writes `settled_at = max(transport_done_at, recorder_done_at, now)` (the two completions as reported, and no earlier than the instant the ledger learns of the second: a completion reported late never settles a row in the past) and `how`, and performs, in that same transaction, the uncertainty extension or closure below. **The transition table is authoritative and is the plan's model:** states `registered` (in memory only), `admitted` (durable), `transport done`, `recorder done`, `settled`; edges `registered → admitted` (the commit), `admitted → unissued` (the recheck after the commit found the fleet stale or a discovery begun, or the gate's budget elapsed before the answer was returned: the row is rewritten as the terminal and the helper never receives the token), `registered → cancelled` (the admission writer has returned with a definite failure, or the writer's handle is closed and a fresh read then finds no row: the registration leaves the pending set), `admitted → transport done | recorder done`, `transport done → settled` (the recorder's completion), `recorder done → settled` (the transport's completion or `transport-lost`), and `admitted | transport done | recorder done → recovered-unknown` at process recovery only (all missing timestamps `now`); `transport-lost` is a transport completion, not a terminal, so an admission whose recorder is still running after its helper died stays `recorder pending`. **The pending set has one contract.** It holds every registration not yet resolved, and an entry leaves it in exactly two ways, both under the discovery mutex: (a) the admission row is durable, at which point the row carries the responsibility (the removal follows the commit, so a discovery sees the entry or the row and never neither); (b) the cancellation is *conclusive*, at which point the empty-settlement result is recorded for every discovery that counted it (the pending-discovery latch's counted set, or the open uncertainty's closure test, which ranges over rows and pending registrations alike), so an uncertainty that waited only on it closes then. The same empty-settlement rule applies to a counted admission that ends `unissued`. **A cancellation is conclusive only when the admission writer can no longer commit:** the writer is one goroutine owning the transaction, the gate's budget expiring refuses the request but does not resolve the registration, and the registration waits for the writer to return. A writer that returns having committed takes `admitted` (then `unissued`, since the request was refused); one that returns with a definite failure takes `cancelled`; one that cannot tell (the connection lost around the commit) is resolved by a read-back by `token` that begins only after the writer's handle is closed, on a read transaction opened after that close, so nothing it fails to find can be committed afterwards: a row found takes `admitted → unissued`; none found takes `cancelled`; the read is retried until it answers. There is no edge out of `cancelled`. Astra's sequence (the writer still running at the budget's expiry and the discovery's count, a read at 23:58:00.040, the commit at 23:58:00.050) cannot declare a cancellation, because the read waits for the writer; the writer's commit resolves it as `unissued`, and the discovery that counted it is told once. Every terminal write is keyed by `token` and idempotent, retried until it commits. Admissions are the ledger's pending state moved to before the send and given an owner.

+**Proof and admission are two things.** Fleet construction, `Open` and every watch tick ask the ownership *proof* (today's `admit`: the witness unchanged, the fleet not stale) and create nothing; only the request gate (`lease.RequestGate`, asked by the helper for each physical request) performs an *admission*, which is a proof followed by a row, so an idle healthy fleet has no admission and nothing bars its days. **The admission is a row, written before the send, with one ordering point against discovery.** Under a *discovery mutex* on the fleet (a new requirement: today `admit` and `lookAgain` do not share one, and the admission comments allow a discovery to run unlocked; this design serialises registration, the recheck and the start of a discovery on it) the gate checks the proof and, if the fleet is not stale and no discovery is in progress, registers the admission in the fleet's in-memory *pending set* with its `admitted_at`; it releases the mutex and commits the row in the durability ledger's own database (`admissions`: `token`, `fleet_id`, `owner account and app`, `admitted_at`, `transport_done_at`, `recorder_started_at`, `recorder_done_at`, `settled_at`, `how`), `synchronous=FULL`, within a bounded budget; after the commit it takes the mutex again and, if the fleet went stale or a discovery began since registration, marks the row `unissued` and refuses. Discovery, under the same mutex, sets the in-progress flag and counts the pending set, durable or not yet, as unsettled admissions when it computes its lower bound, and a registered admission that the discovery has counted and that then commits is inside the uncertainty (it was counted); one that commits after the discovery began is refused. So an admission is either counted by every discovery that follows its registration or refused, and Astra's 23:58:10 sequence (proof, discovery, commit, a send on the 4th) ends in a refusal at the recheck, not a send outside the uncertainty. The gate waits for the commit. A commit that does not return within the budget refuses the request (`ErrAdmissionNotDurable`; litestream sees a failed upload and does not advance) and the goroutine that awaited it stays to see the commit land and then marks the row `how = unissued, settled_at = now` in its own transaction, so a late commit never authorises anything and never blocks a day; if the process dies between, recovery classifies the row as below. The meter's `Start` row carries the token (`Start` may still fail or be late, as today; what is now mandatory is the admission, not the `Start`). An admission has two participants, the transport and the recorder, and is settled by whichever finishes last. The transport writes `transport_done_at` from the SDK middleware's finalize step, or `transport-lost` when its helper is proven gone (below). The recorder writes `recorder_done_at` with its outcome: `finished` (`Finish`), `abandoned` (`Abandon`), or `unrecorded` (the request path, when `Start` failed and made no attempt row, which today calls neither; it now writes this terminal); the recorder participant is never synthesised while its worker can still run: today's meter lets recording workers outlive the request path, and the admission stays unsettled (barring) until that worker finishes or abandons. The recorder runs in the agent, fed by the helper's messages over the fleet socket (`Start`, `Finish`, `Abandon` with the token), and the agent *registers* the recorder participant when it receives `Start` (`recorder_started_at`, under an agent-owned lifecycle guard per helper); a `Start` that the agent receives after it has begun the helper's termination handling is refused (the guard is closed first, then the socket drained, so registration and exit handling cannot interleave). The participant that finds the other already done writes `settled_at = max(transport_done_at, recorder_done_at, now)` (the two completions as reported, and no earlier than the instant the ledger learns of the second: a completion reported late never settles a row in the past) and `how`, and performs, in that same transaction, the uncertainty extension or closure below. **The transition table is authoritative and is the plan's model:** states `registered` (in memory only), `admitted` (durable), `transport done`, `recorder done`, `settled`; edges `registered → admitted` (the commit), `admitted → unissued` (the recheck after the commit found the fleet stale or a discovery begun, or the gate's budget elapsed before the answer was returned: the row is rewritten as the terminal and the helper never receives the token), `registered → cancelled` (the admission writer has returned with a definite failure, or the writer's handle is closed and a fresh read then finds no row: the registration leaves the pending set), `admitted → transport done | recorder done`, `transport done → settled` (the recorder's completion), `recorder done → settled` (the transport's completion or `transport-lost`), and `admitted | transport done | recorder done → recovered-unknown` at process recovery only (all missing timestamps `now`); `transport-lost` is a transport completion, not a terminal, so an admission whose recorder is still running after its helper died stays `recorder pending`. **The pending set has one contract.** It holds every registration not yet resolved, and an entry leaves it in exactly two ways, both under the discovery mutex: (a) the admission row is durable, at which point the row carries the responsibility (the removal follows the commit, so a discovery sees the entry or the row and never neither); (b) the cancellation is *conclusive*, at which point the empty-settlement result is recorded for every discovery that counted it (the pending-discovery latch's counted set, or the open uncertainty's closure test, which ranges over rows and pending registrations alike), so an uncertainty that waited only on it closes then. The same empty-settlement rule applies to a counted admission that ends `unissued`. If the counted registration resolves before the discovery transaction exists, that transaction closes against current rows and the live counter, retaining the captured instants for its lower bound. When captured pending registrations have already resolved, the upper bound is conservatively no earlier than the ledger clock at discovery commit plus the margin. **A cancellation is conclusive only when the admission writer can no longer commit:** the writer is one goroutine owning the transaction, the gate's budget expiring refuses the request but does not resolve the registration, and the registration waits for the writer to return. A writer that returns having committed takes `admitted` (then `unissued`, since the request was refused); one that returns with a definite failure takes `cancelled`; one that cannot tell (the connection lost around the commit) is resolved by a read-back by `token` that begins only after the writer's handle is closed, on a read transaction opened after that close, so nothing it fails to find can be committed afterwards: a row found takes `admitted → unissued`; none found takes `cancelled`; the read is retried until it answers. There is no edge out of `cancelled`. Astra's sequence (the writer still running at the budget's expiry and the discovery's count, a read at 23:58:00.040, the commit at 23:58:00.050) cannot declare a cancellation, because the read waits for the writer; the writer's commit resolves it as `unissued`, and the discovery that counted it is told once. Every terminal write is keyed by `token` and idempotent, retried until it commits. Admissions are the ledger's pending state moved to before the send and given an owner.

**Two kinds of hold.** A *barrier* is the ledger's completeness rule: a day is not published complete for an admission's owner while an admission of that owner admitted on or before that day is unsettled (box-wide while Part C2's owner scope is not in use); it is a query over `admissions`, writes nothing, and lifts at settlement. An *uncertainty* is a row (`uncertainties`: `id`, `fleet_id`, `from`, `to` null while open, `reason`) written when a fleet's authority is found changed (by the witness, by `Recheck`, by a changed file), in the transaction that also writes the holds for the days of `[from, now]` for every owner the uncertainty reaches: `bound = min(the registry proof, the last clean look's start)`; `from = min(bound, the earliest admitted_at among the fleet's admissions that are unsettled, or settled with settled_at >= bound and how not in {unissued, cancelled}) - margin`, the bound fixed before the rows are read and never the running minimum (revision 9: the first two terms alone missed an admission admitted before the look, in flight across the change, and settled clean before the discovery; such a settlement wrote nothing permanent, so its admission instant must bound the uncertainty; revision 10: the threshold is the bound itself, not the look alone; revision 11: every settlement whose transport ran counts, `unrecorded` included, and only the empty terminals, which never sent, are excluded; `transport-lost` is a transport completion, never a value of `how`; a `recovered-unknown` row has its own interval and need not be counted here). **What a look must do for its start to be the bound:** the look's start instant is stamped before its first observation, and every observation it makes (the witness, `Recheck` of every retained chain, the route's recorded invocation against systemd's, the files) sees any change that persists, so a look that finds nothing is clean against every kind of change from its start; the order of the observations within a look does not matter for that; the earlier-of rule (the proof and the look) is kept as a backstop for a look built without the witness observation only, since a registry proof can stand in for a missed witness reading and for nothing else (the chains, the invocation and the files are mandatory observations). If discovery finds no admission of the fleet unsettled, the uncertainty is opened and closed in that transaction (`to = discovery + margin`). While an uncertainty is open, settlements of that fleet's admissions extend it: each settlement, in its own transaction, writes the holds for every day of `[its admitted_at, its settled_at]` for every owner the uncertainty reaches, and the last settlement closes it with `to = max(discovery, every settlement of the fleet at or after the bound) + margin` (the discovery instant is stored on the row; a settlement reported out of order never closes it before an interval already written) and materialises every day of the closed interval, the margin's extra day across midnight included; recovery uses the same helper and advances the publication watermark past the days it materialised. An admission that settles while no uncertainty of its fleet is open writes nothing permanent: the ordinary day of an unchanged fleet becomes complete when its barrier lifts. The multiple-midnights sequence (admitted on the 2nd, the change at 23:58:10, sent on the 3rd, finished on the 4th) holds the 2nd, 3rd and 4th through the settlement's interval, whichever participant finished last; the request completed before the change was noticed is inside the lower bound as today. **If the discovery transaction fails** the barrier is not enough (every admission may already be settled): discovery first latches a *pending discovery* in the ledger's in-memory failure overlay (`latchFailure`, extended to carry the lower bound and the owners reached), which the publisher's snapshot already overlays and detects across a drain (today's `snapshotOnce` and `drainFailures` behaviour, extended rather than replaced: a failed drain must not stop the overlay from making the snapshot incomplete), so every day from the bound reads incomplete for those owners until the uncertainty commits; the fleet refuses every request meanwhile and retries the write. The latch carries the lower bound, the owners reached and the tokens of the admissions the discovery counted; settlements of those admissions meanwhile find no uncertainty open and write no hold, but their rows record `settled_at` as always. **The replay is one transaction:** it opens the uncertainty with the original `from`; for each counted token whose row is settled it writes the holds for `[its admitted_at, its settled_at]` for every owner reached (and for a token that was cancelled or unissued, nothing); if any counted token is still unsettled the uncertainty stays open and the usual settlement rule closes it later; otherwise it closes with `to = max(discovery, replay instant) + margin` and materialises every day of the closed interval; the replay's closure test is the settlement's (no unsettled row of the fleet, nothing pending), not the set counted at the discovery, since a registration pending then may be a row by now; a latch keeps the bound and counted set its failed transaction computed, and one whose rows could not be read is unresolved and bars every day until the replay computes them; the overlay clears only after that commit, and a snapshot whose read the commit crossed reads again, so no snapshot sees neither the latch nor the row. Every hold written, and every admission and settlement, dirties the days it touches with a sequence above any snapshot's watermark, and publication enumerates a latched discovery's days from its bound, so a day already published complete is published again when a bound reaches it or a barrier lifts. Astra's sequence (T counted at 23:58:10 on the 2nd, the write failing, T settling at 00:02 on the 4th, the replay at 00:03) writes the 2nd, 3rd and 4th through T's interval and closes at 00:04 on the 4th, and a successor constructed after the replay backfills that closed interval for its owners; the fleet cannot be marked `closed` while the discovery is uncommitted (above); a process that dies with the latch held leaves a fleet not `closed`, which recovery covers from the fleet's `opened_at`.

diff --git a/internal/lease/requestgate.go b/internal/lease/requestgate.go

index cfea00884..4e12c83d9 100644

--- a/internal/lease/requestgate.go

+++ b/internal/lease/requestgate.go

@@ -30,7 +30,7 @@ import (

// authority design, Part C): the durable row written before the send,

// whose two participants, the transport and the recorder, the middleware

// completes. The gate middleware writes the transport's completion when

-// the attempt's round trip has returned (the SDK's finalize step), and the

+// the attempt's round trip has returned (the Deserialize handler return), and the

// meter writes the recorder's: registered when its Start lands, finished

// when its Finish lands, abandoned when a late Start or a failed Finish is

// abandoned, unrecorded when Start failed and made no row. A gate that

@@ -44,7 +44,7 @@ type RequestGate func(ctx context.Context) (Admission, error)

// participant and must not block the request path on more than the

// ledger's own write.

type Admission interface {

- // TransportDone is the SDK's finalize step: the round trip returned.

+ // TransportDone is the Deserialize handler return: the round trip returned.

TransportDone(at time.Time)

// RecorderStarted is the meter's Start landed: the recorder participant

// is registered and will finish, abandon or be abandoned.

@@ -134,7 +134,7 @@ func requestGateOption(gate RequestGate) (func(*s3.Options), error) {

return middleware.DeserializeOutput{}, middleware.Metadata{}, fmt.Errorf("%w: %w", ErrRequestRefused, ErrNoAdmission)

}

out, md, err := next.HandleDeserialize(context.WithValue(ctx, admissionKey{}, adm), in)

- // the finalize step: the round trip has returned, answered

+ // The round trip has returned from Deserialize, answered

// or not; the transport participant is complete

adm.TransportDone(time.Now().UTC())

return out, md, err

diff --git a/internal/lease/requestmeter_test.go b/internal/lease/requestmeter_test.go

index 1ac7ebcd7..ba50b1473 100644

--- a/internal/lease/requestmeter_test.go

+++ b/internal/lease/requestmeter_test.go

@@ -148,15 +148,30 @@ func TestRequestMeterCountsSDKAttempts(t *testing.T) {

func TestRequestMeterRecordingFailureDoesNotBlockReplication(t *testing.T) {

for _, stage := range []string{"start", "finish"} {

t.Run(stage, func(t *testing.T) {

+ adm := &recordingAdmission{}

recorder := &memoryRequestRecorder{startErr: stage == "start", finishErr: stage == "finish"}

var received atomic.Int32

- client := requestMeterTestClient(t, func(w http.ResponseWriter, r *http.Request) { received.Add(1); w.Header().Set("ETag", `"fixture"`) }, recorder)

+ client := gatedTestClient(t, func(w http.ResponseWriter, r *http.Request) { received.Add(1); w.Header().Set("ETag", `"fixture"`) }, func(context.Context) (Admission, error) {

+ return adm, nil

+ }, recorder, true)

if err := meterPut(context.Background(), client); err != nil {

t.Fatal(err)

}

if received.Load() != 1 || recorder.incomplete != 1 {

t.Fatalf("received=%d incomplete=%d", received.Load(), recorder.incomplete)

}

+ waitFor(t, "the failed recorder operation's admission outcome", func() bool {

+ tr, _, out := adm.snapshot()

+ return tr == 1 && len(out) == 1

+ })

+ tr, started, outcomes := adm.snapshot()

+ wantStarted, wantOutcome := 1, AdmissionAbandoned

+ if stage == "start" {

+ wantStarted, wantOutcome = 0, AdmissionUnrecorded

+ }

+ if tr != 1 || started != wantStarted || len(outcomes) != 1 || outcomes[0] != wantOutcome {

+ t.Fatalf("admission: transport=%d started=%d outcomes=%v, want 1, %d, [%s]", tr, started, outcomes, wantStarted, wantOutcome)

+ }

})

}

@@ -247,15 +262,16 @@ func (r *blockingRequestRecorder) Finish(token uint64, out requestOutcome) error

func TestRequestMeterStalledPersistenceDoesNotDelayReplication(t *testing.T) {

for _, stage := range []string{"start", "finish"} {

t.Run(stage, func(t *testing.T) {

+ adm := &recordingAdmission{}

recorder := &blockingRequestRecorder{stage: stage, release: make(chan struct{}), exited: make(chan struct{})}

var once sync.Once

release := func() { once.Do(func() { close(recorder.release) }) }

defer release()

received := make(chan struct{}, 1)

- client := requestMeterTestClient(t, func(w http.ResponseWriter, r *http.Request) {

+ client := gatedTestClient(t, func(w http.ResponseWriter, r *http.Request) {

received <- struct{}{}

w.Header().Set("ETag", `"fixture"`)

- }, recorder)

+ }, func(context.Context) (Admission, error) { return adm, nil }, recorder, true)

done := make(chan error, 1)

go func() { done <- meterPut(context.Background(), client) }()

select {

@@ -278,10 +294,23 @@ func TestRequestMeterStalledPersistenceDoesNotDelayReplication(t *testing.T) {

t.Fatal("recorder did not exit after release")

}

recorder.mu.Lock()

- defer recorder.mu.Unlock()

if recorder.incomplete == 0 {

+ recorder.mu.Unlock()

t.Fatal("stalled persistence was treated as complete")

}

+ recorder.mu.Unlock()

+ waitFor(t, "the late recorder operation's admission outcome", func() bool {

+ tr, _, out := adm.snapshot()

+ return tr == 1 && len(out) == 1

+ })

+ tr, started, outcomes := adm.snapshot()

+ wantOutcome := AdmissionFinished

+ if stage == "start" {

+ wantOutcome = AdmissionAbandoned

+ }

+ if tr != 1 || started != 1 || len(outcomes) != 1 || outcomes[0] != wantOutcome {

+ t.Fatalf("admission: transport=%d started=%d outcomes=%v, want 1, 1, [%s]", tr, started, outcomes, wantOutcome)

+ }

})

}

@@ -303,14 +332,22 @@ func (r *saturatedRequestRecorder) Start(op string, at time.Time) (uint64, error

func TestRequestMeterSaturationBoundsWorkersAndKeepsReplicationRunning(t *testing.T) {

recorder := &saturatedRequestRecorder{release: make(chan struct{})}

+ var admissionsMu sync.Mutex

+ var admissions []*recordingAdmission

var once sync.Once

release := func() { once.Do(func() { close(recorder.release) }); recorder.workers.Wait() }

defer release()

var received atomic.Int32

- client := requestMeterTestClient(t, func(w http.ResponseWriter, r *http.Request) {

+ client := gatedTestClient(t, func(w http.ResponseWriter, r *http.Request) {

received.Add(1)

w.Header().Set("ETag", `"fixture"`)

- }, recorder)

+ }, func(context.Context) (Admission, error) {

+ adm := &recordingAdmission{}

+ admissionsMu.Lock()

+ admissions = append(admissions, adm)

+ admissionsMu.Unlock()

+ return adm, nil

+ }, recorder, true)

for i := 0; i < 12; i++ {

ctx, cancel := context.WithTimeout(context.Background(), time.Second)

err := meterPut(ctx, client)

@@ -327,15 +364,52 @@ func TestRequestMeterSaturationBoundsWorkersAndKeepsReplicationRunning(t *testin

}

release()

recorder.mu.Lock()

- defer recorder.mu.Unlock()

if recorder.incomplete != 12 {

- t.Fatalf("incomplete=%d, want 12", recorder.incomplete)

+ got := recorder.incomplete

+ recorder.mu.Unlock()

+ t.Fatalf("incomplete=%d, want 12", got)

}

for _, request := range recorder.requests {

if request.finished {

+ recorder.mu.Unlock()

t.Fatal("late start falsely gained a known outcome")

}

+ recorder.mu.Unlock()

+ waitFor(t, "every saturated admission outcome", func() bool {

+ admissionsMu.Lock()

+ defer admissionsMu.Unlock()

+ if len(admissions) != 12 {

+ return false

+ }

+ for _, adm := range admissions {

+ tr, _, out := adm.snapshot()

+ if tr != 1 || len(out) != 1 {

+ return false

+ }

+ return true

+ })

+ admissionsMu.Lock()

+ defer admissionsMu.Unlock()

+ var abandoned, unrecorded int

+ for i, adm := range admissions {

+ tr, started, outcomes := adm.snapshot()

+ if tr != 1 || len(outcomes) != 1 {

+ t.Fatalf("admission %d: transport=%d started=%d outcomes=%v", i, tr, started, outcomes)

+ }

+ switch {

+ case started == 1 && outcomes[0] == AdmissionAbandoned:

+ abandoned++

+ case started == 0 && outcomes[0] == AdmissionUnrecorded:

+ unrecorded++

+ default:

+ t.Fatalf("admission %d: started=%d outcomes=%v, want a late abandoned Start or an unrecorded no-slot call", i, started, outcomes)

+ }

+ if abandoned != 4 || unrecorded != 8 {

+ t.Fatalf("admission outcomes: abandoned=%d unrecorded=%d, want 4 and 8", abandoned, unrecorded)

+ }

}

func TestRequestMeterRedirectsHoldEvidenceWithoutBlockingReplication(t *testing.T) {

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

index 3e5b8dc01..d668a8011 100644

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

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

@@ -77,7 +77,7 @@ func (p *fleetPins) Close() {

// directories beneath it that listed a database.

type listedTree struct {

top *authority.Dir

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

+ dirs []*authority.Dir // children before parents

release func()

}

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

index de60822c1..6311e9847 100644

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

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

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

// 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).

+ sawContent := false

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

trim := strings.TrimSpace(raw)

switch {

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

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)

}

+ sawContent = true

}

var dbs []string

inDBs := false

@@ -284,7 +286,7 @@ 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") {

+ if !foundDBs && sawContent {

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

}

return dbs, nil

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

index dfaedd256..32f6963f6 100644

--- a/internal/ops/litestream_admission.go

+++ b/internal/ops/litestream_admission.go

@@ -44,15 +44,6 @@ var ErrAdmissionNotDurable = errors.New("ops: the replica request's admission wa

// the goroutine that stayed with it.

var admissionBudget = 5 * time.Second

-// admissionReadBackDelay paces the read-back of an ambiguous commit, and

-// admissionReadBackWait bounds it: past it the registration stays pending

-// (the counter keeps any uncertainty open) and recovery classifies the

-// row, if one landed.

-var (

- admissionReadBackDelay = 50 * time.Millisecond

- admissionReadBackWait = 2 * time.Second

-)

-

// fleetAdmission is one issued admission, as the middleware completes it.

type fleetAdmission struct {

f *MeteredFleet

@@ -64,16 +55,25 @@ func (a *fleetAdmission) TransportDone(at time.Time) {

if !a.done.CompareAndSwap(false, true) {

return

}

- a.f.forget(a.token)

- a.f.ledgerWrite("transport done", a.f.ledger.TransportDone(a.token, at, false))

+ a.transportDone(at, false)

}

func (a *fleetAdmission) RecorderStarted(at time.Time) {

- a.f.ledgerWrite("recorder started", a.f.ledger.RecorderStarted(a.token, at))

+ a.f.retryWrite(a.token, "recorder started", func() error { return a.f.ledger.RecorderStarted(a.token, at) })

}

func (a *fleetAdmission) RecorderDone(at time.Time, how string) {

- a.f.ledgerWrite("recorder done", a.f.ledger.RecorderDone(a.token, at, how))

+ a.f.retryWrite(a.token, "recorder done", func() error { return a.f.ledger.RecorderDone(a.token, at, how) })

+}

+

+func (a *fleetAdmission) transportDone(at time.Time, lost bool) {

+ a.f.retryWrite(a.token, "transport done", func() error {

+ if err := a.f.ledger.TransportDone(a.token, at, lost); err != nil {

+ return err

+ }

+ a.f.forget(a.token)

+ return nil

+ })

}

// ledgerWrite notes a participant write the ledger refused. The row stays

@@ -87,8 +87,7 @@ func (f *MeteredFleet) ledgerWrite(what string, err error) {

log.Printf("replicator: fleet %s: %s: %v", f.id, what, err)

}

-// forget drops an issued admission from the live set (its transport has

-// finalised).

+// forget releases the transport retry owner after durable completion.

func (f *MeteredFleet) forget(token string) {

f.liveMu.Lock()

delete(f.live, token)

@@ -112,6 +111,7 @@ func (f *MeteredFleet) gateFor(owned OwnedDB) lease.RequestGate {

// row's commit; admissionDelay delays the writer, for the budget's test.

var admissionBeforeCommit func(f *MeteredFleet)

var admissionDelay func()

+var fleetOpenResult func(error) error

func (f *MeteredFleet) gate(ctx context.Context, owned OwnedDB) (lease.Admission, error) {

if err := f.admit(ctx); err != nil {

@@ -123,9 +123,10 @@ func (f *MeteredFleet) gate(ctx context.Context, owned OwnedDB) (lease.Admission

if err != nil {

return nil, err

}

+ deadline := time.Now().Add(admissionBudget)

at := time.Now().UTC()

f.disc.Lock()

- if f.closed.Load() || f.stale.Load() || f.inDiscovery || f.latched.Load() {

+ if f.closed.Load() || f.stale.Load() || f.discovered.Load() || f.inDiscovery || f.latched.Load() {

f.disc.Unlock()

return nil, ErrOwnershipStale

}

@@ -140,20 +141,24 @@ func (f *MeteredFleet) gate(ctx context.Context, owned OwnedDB) (lease.Admission

// for it within the budget, and past the budget the writer's outcome

// is still resolved, by the writer itself

done := make(chan error, 1)

+ delay := admissionDelay

go func() {

- if admissionDelay != nil {

- admissionDelay()

+ if delay != nil {

+ delay()

}

done <- f.ledger.Admit(token, f.id, owned.Account, owned.App, at)

}()

- timer := time.NewTimer(admissionBudget)

+ timer := time.NewTimer(time.Until(deadline))

defer timer.Stop()

select {

case err := <-done:

- return f.resolve(token, epoch, err, false)

+ return f.resolve(ctx, token, epoch, err, false, deadline)

+ case <-ctx.Done():

+ go func() { f.resolve(ctx, token, epoch, <-done, true, deadline) }()

+ return nil, ctx.Err()

case <-timer.C:

go func() {

- f.resolve(token, epoch, <-done, true) //nolint:errcheck

+ f.resolve(ctx, token, epoch, <-done, true, deadline) //nolint:errcheck

}()

return nil, ErrAdmissionNotDurable

}

@@ -167,35 +172,48 @@ func (f *MeteredFleet) gate(ctx context.Context, owned OwnedDB) (lease.Admission

// and taken as committed or cancelled. late is a writer the request path

// stopped waiting for: whatever it did, the request was refused, so a

// committed row is unissued.

-func (f *MeteredFleet) resolve(token string, epoch uint64, err error, late bool) (lease.Admission, error) {

+func (f *MeteredFleet) resolve(ctx context.Context, token string, epoch uint64, err error, late bool, deadline time.Time) (lease.Admission, error) {

now := time.Now().UTC()

if err != nil {

- landed, rerr := f.readBack(token)

- if rerr != nil {

- // the read-back could not answer and the fleet is closed:

- // the registration stays pending (the counter keeps any

- // uncertainty open) and recovery classifies the row, if any

- f.ledgerWrite("read-back", rerr)

- return nil, err

- }

- if !landed {

+ if errors.Is(err, durability.ErrLedgerClosed) {

f.cancel(token)

- f.ledgerWrite("empty settlement", f.ledger.EmptySettlement(f.id, now))

+ f.retryWrite(token, "empty settlement", func() error { return f.ledger.EmptySettlement(f.id, now) })

return nil, fmt.Errorf("%w: %v", ErrOwnershipStale, err)

}

- // landed: the row carries the responsibility now, and the

- // request has been refused (the caller heard an error)

- late = true

+ var resolvedAt time.Time

+ var landed bool

+ f.retryWrite(token, "read-back", func() error {

+ if resolvedAt.IsZero() {

+ _, readErr := f.readAdmission(token)

+ if readErr != nil && !errors.Is(readErr, durability.ErrAdmissionUnknown) {

+ return readErr

+ }

+ landed = readErr == nil

+ resolvedAt = time.Now().UTC()

+ f.cancel(token)

+ }

+ if landed {

+ if e := f.ledger.Unissued(token, resolvedAt); e != nil {

+ return e

+ }

+ return f.ledger.EmptySettlement(f.id, resolvedAt)

+ })

+ return nil, fmt.Errorf("%w: %v", ErrOwnershipStale, err)

}

// committed: the row carries the responsibility; the recheck

+ if f.admissionRecheck != nil {

+ f.admissionRecheck()

+ }

f.disc.Lock()

+ late = late || !time.Now().Before(deadline) || ctx.Err() != nil

delete(f.pending, token)

f.pendingN.Add(-1)

- refused := late || f.closed.Load() || f.stale.Load() || f.inDiscovery || f.latched.Load() || f.epoch != epoch

- f.disc.Unlock()

+ refused := late || f.closed.Load() || f.stale.Load() || f.discovered.Load() || f.inDiscovery || f.latched.Load() || f.epoch != epoch

if refused {

- f.ledgerWrite("unissued", f.ledger.Unissued(token, now))

- f.ledgerWrite("empty settlement", f.ledger.EmptySettlement(f.id, now))

+ f.disc.Unlock()

+ f.retryWrite(token, "unissued", func() error { return f.ledger.Unissued(token, now) })

+ f.retryWrite(token, "empty settlement", func() error { return f.ledger.EmptySettlement(f.id, now) })

if late {

return nil, ErrAdmissionNotDurable

}

@@ -205,6 +223,7 @@ func (f *MeteredFleet) resolve(token string, epoch uint64, err error, late bool)

f.liveMu.Lock()

f.live[token] = adm

f.liveMu.Unlock()

+ f.disc.Unlock()

return adm, nil

}

@@ -219,23 +238,11 @@ func (f *MeteredFleet) cancel(token string) {

f.disc.Unlock()

}

-// readBack asks the ledger whether a token's row landed, trying again

-// while the ledger cannot answer, within a bound.

-func (f *MeteredFleet) readBack(token string) (bool, error) {

- deadline := time.Now().Add(admissionReadBackWait)

- for {

- _, err := f.ledger.ReadAdmission(token)

- switch {

- case err == nil:

- return true, nil

- case errors.Is(err, durability.ErrAdmissionUnknown):

- return false, nil

- }

- if time.Now().After(deadline) {

- return false, err

- }

- time.Sleep(admissionReadBackDelay)

+func (f *MeteredFleet) readAdmission(token string) (durability.Admission, error) {

+ if f.admissionRead != nil {

+ return f.admissionRead(token)

}

+ return f.ledger.ReadAdmission(token)

}

// discover tells the ledger the fleet's authority was found changed, once

@@ -257,6 +264,9 @@ func (f *MeteredFleet) discover(at time.Time) {

f.disc.Unlock()

sort.Slice(pending, func(i, j int) bool { return pending[i].Before(pending[j]) })

proof := time.Unix(0, f.proven.Load()).UTC()

+ if f.discoveryBeforeWrite != nil {

+ f.discoveryBeforeWrite()

+ }

_, err := f.ledger.Discover(durability.Discovery{FleetID: f.id, At: at.UTC(), Proof: proof, LastCleanLook: f.lastCleanLook(), Pending: pending, Reason: "the fleet's authority was found changed"})

if err != nil {

f.latched.Store(true)

@@ -283,7 +293,25 @@ func (f *MeteredFleet) acknowledge() {

if f.ledger == nil || f.acked.Load() {

return

}

- err := f.ledger.Acknowledge(f.id, time.Now().UTC())

+ chain := []string{f.id}

+ for id := f.id; ; {

+ row, err := f.ledger.ReadFleet(id)

+ if err != nil {

+ f.ledgerWrite("acknowledge", err)

+ return

+ }

+ if row.Predecessor == "" {

+ break

+ }

+ id = row.Predecessor

+ chain = append(chain, id)

+ }

+ var err error

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

+ if err = f.ledger.Acknowledge(chain[i], time.Now().UTC()); err != nil {

+ break

+ }

if err == nil {

f.acked.Store(true)

return

@@ -297,13 +325,13 @@ func (f *MeteredFleet) acknowledge() {

// stopped and the witness is about to be released; transport-lost for

// every issued admission whose transport never finalised; then closed,

// waited for within a bound (the recorder's workers may still be

-// finishing), and left closing past it, its admissions still barring.

+// finishing). The retry worker retains closure past that bound.

func (f *MeteredFleet) journalClose() {

if f.ledger == nil || f.id == "" {

return

}

now := time.Now().UTC()

- f.ledgerWrite("closing", f.ledger.FleetClosing(f.id))

+ f.retryWrite("journal", "closing", func() error { return f.ledger.FleetClosing(f.id) })

f.liveMu.Lock()

lost := make([]*fleetAdmission, 0, len(f.live))

for _, a := range f.live {

@@ -312,8 +340,7 @@ func (f *MeteredFleet) journalClose() {

f.liveMu.Unlock()

for _, a := range lost {

if a.done.CompareAndSwap(false, true) {

- f.forget(a.token)

- f.ledgerWrite("transport lost", f.ledger.TransportDone(a.token, now, true))

+ a.transportDone(now, true)

}

f.replayLatched()

@@ -326,6 +353,14 @@ func (f *MeteredFleet) journalClose() {

}

if !errors.Is(err, durability.ErrFleetUnsettled) || time.Now().After(deadline) {

log.Printf("replicator: fleet %s stays closing: %v", f.id, err)

+ f.retryWrite("journal", "closed", func() error {

+ f.replayLatched()

+ if err := f.ledger.CloseFleet(f.id, time.Now().UTC()); err != nil {

+ return err

+ }

+ f.ledger.SetPendingCounter(f.id, nil)

+ return nil

+ })

return

}

time.Sleep(closeSettlePoll)

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

new file mode 100644

index 000000000..94a1ac6f5

--- /dev/null

+++ b/internal/ops/litestream_admission_retry.go

@@ -0,0 +1,69 @@

+package ops

+

+import "time"

+

+type admissionWrite struct {

+ what string

+ write func() error

+ reported bool

+}

+

+// Writes stay ordered within a token. Enqueuing never waits for the ledger;

+// its worker survives the request and the fleet's store/watch shutdown.

+func (f *MeteredFleet) retryWrite(key, what string, write func() error) {

+ f.retryMu.Lock()

+ defer f.retryMu.Unlock()

+ if f.retries == nil {

+ f.retries = make(map[string][]*admissionWrite)

+ }

+ f.retries[key] = append(f.retries[key], &admissionWrite{what: what, write: write})

+ if !f.retrying {

+ f.retrying = true

+ go f.retryAdmissions()

+ }

+}

+

+func (f *MeteredFleet) retryAdmissions() {

+ for {

+ closed := f.ledger.Closed()

+ f.retryMu.Lock()

+ if closed || len(f.retries) == 0 {

+ f.retries = nil

+ f.retrying = false

+ f.retryMu.Unlock()

+ return

+ }

+ batch := make(map[string]*admissionWrite, len(f.retries))

+ for key, queue := range f.retries {

+ batch[key] = queue[0]

+ }

+ f.retryMu.Unlock()

+ progress := false

+ for key, w := range batch {

+ var err error

+ if f.admissionWriteFail != nil {

+ err = f.admissionWriteFail(w.what)

+ }

+ if err == nil {

+ err = w.write()

+ }

+ if err != nil {

+ if !w.reported {

+ f.ledgerWrite(w.what, err)

+ w.reported = true

+ }

+ continue

+ }

+ f.retryMu.Lock()

+ f.retries[key] = f.retries[key][1:]

+ if len(f.retries[key]) == 0 {

+ delete(f.retries, key)

+ }

+ f.retryMu.Unlock()

+ progress = true

+ }

+ if !progress {

+ time.Sleep(50 * time.Millisecond)

+ }

+}

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

new file mode 100644

index 000000000..1d66c055e

--- /dev/null

+++ b/internal/ops/litestream_admission_review11_test.go

@@ -0,0 +1,303 @@

+package ops

+

+import (

+ "context"

+ "errors"

+ "path/filepath"

+ "sync/atomic"

+ "testing"

+ "time"

+

+ "github.com/carlosframework/platform/internal/registry"

+ durability "github.com/carlosframework/platform/internal/usage"

+)

+

+func review11Fleet(t *testing.T) (*MeteredFleet, *durability.DurabilityLedger, OwnedDB, MeteredFleetOptions) {

+ t.Helper()

+ newFakeBucket(t)

+ dir := platformDir(t)

+ reg := openRegistry(t, filepath.Join(dir, "registry.db"))

+ state := filepath.Join(platformDir(t), "unit-a")

+ db := writeFile(t, filepath.Join(state, "data", "main.db"))

+ upsert(t, reg, registry.Route{Host: "a.example.com", Account: "bac", App: "one", DBPath: state, Backing: registry.BackingUnit, Version: "v1"})

+ ledger := openLedger(t)

+ o := fleetOptions(reg, dir)

+ o.Ledger = ledger

+ f, err := MeteredFleetStore(o)

+ if err != nil {

+ t.Fatal(err)

+ }

+ t.Cleanup(func() { f.Close() })

+ for _, owned := range f.Owned {

+ if owned.Path == db {

+ return f, ledger, owned, o

+ }

+ t.Fatal("missing owned database")

+ return nil, nil, OwnedDB{}, o

+}

+

+func retryQueueEmpty(f *MeteredFleet) bool {

+ f.retryMu.Lock()

+ defer f.retryMu.Unlock()

+ return len(f.retries) == 0

+}

+

+func TestReview11DiscoveryCannotReopenAdmission(t *testing.T) {

+ f, ledger, owned, _ := review11Fleet(t)

+ f.discover(time.Now())

+ if f.stale.Load() {

+ t.Fatal("fixture must pause before stale is stored")

+ }

+ adm, err := f.gate(context.Background(), owned)

+ if adm != nil || !errors.Is(err, ErrOwnershipStale) {

+ t.Fatalf("admitted after discovery: %v %v", adm, err)

+ }

+ if tokens := admissionTokens(t, ledger, f.id); len(tokens) != 0 {

+ t.Fatalf("admissions after discovery: %v", tokens)

+ }

+}

+

+func TestReview11DeadlineAtIssuance(t *testing.T) {

+ for _, blocked := range []bool{false, true} {

+ name := "already-expired"

+ if blocked {

+ name = "expires-waiting-for-mutex"

+ }

+ t.Run(name, func(t *testing.T) {

+ f, ledger, _, _ := review11Fleet(t)

+ token, _ := durability.NewAdmissionToken()

+ at := time.Now().UTC()

+ if err := ledger.Admit(token, f.id, "bac", "one", at); err != nil {

+ t.Fatal(err)

+ }

+ f.pending[token] = at

+ f.pendingN.Add(1)

+ deadline := time.Now().Add(-time.Second)

+ if blocked {

+ deadline = time.Now().Add(30 * time.Millisecond)

+ f.disc.Lock()

+ }

+ arrived := make(chan struct{})

+ f.admissionRecheck = func() { close(arrived) }

+ result := make(chan error, 1)

+ go func() {

+ adm, err := f.resolve(context.Background(), token, f.epoch, nil, false, deadline)

+ if adm != nil {

+ err = errors.New("issued an expired admission")

+ }

+ result <- err

+ }()

+ <-arrived

+ if blocked {

+ <-time.After(time.Until(deadline))

+ f.disc.Unlock()

+ }

+ if err := <-result; !errors.Is(err, ErrAdmissionNotDurable) {

+ t.Fatalf("expired answer: %v", err)

+ }

+ waitUntil(t, "expired admission settlement", func() bool {

+ a, _ := ledger.ReadAdmission(token)

+ return a.Settled() && a.How == durability.AdmissionUnissued

+ })

+ }

+}

+

+func TestReview11ParticipantWritesRetainRetryOwner(t *testing.T) {

+ for _, what := range []string{"recorder started", "recorder done", "transport done"} {

+ t.Run(what, func(t *testing.T) {

+ f, ledger, owned, _ := review11Fleet(t)

+ var failed atomic.Bool

+ f.admissionWriteFail = func(w string) error {

+ if w == what && failed.CompareAndSwap(false, true) {

+ return errors.New("injected write failure")

+ }

+ return nil

+ }

+ adm, err := f.gate(context.Background(), owned)

+ if err != nil {

+ t.Fatal(err)

+ }

+ token := adm.(*fleetAdmission).token

+ adm.RecorderStarted(time.Now())

+ adm.RecorderDone(time.Now(), durability.AdmissionFinished)

+ adm.TransportDone(time.Now())

+ waitUntil(t, "retry settlement", func() bool { a, _ := ledger.ReadAdmission(token); return a.Settled() })

+ waitUntil(t, "retry queue empty", func() bool { return retryQueueEmpty(f) })

+ a, _ := ledger.ReadAdmission(token)

+ if !failed.Load() || a.How != durability.AdmissionFinished || a.RecorderStartedAt.IsZero() {

+ t.Fatalf("retry did not preserve registration: %+v failed=%v", a, failed.Load())

+ }

+ f.liveMu.Lock()

+ live := len(f.live)

+ f.liveMu.Unlock()

+ if live != 0 {

+ t.Fatalf("live transports after durable completion: %d", live)

+ }

+ })

+ }

+}

+

+func TestReview11RefusedWritesRetainRetryOwner(t *testing.T) {

+ for _, what := range []string{"unissued", "empty settlement"} {

+ t.Run(what, func(t *testing.T) {

+ f, ledger, owned, _ := review11Fleet(t)

+ var failed atomic.Bool

+ f.admissionWriteFail = func(w string) error {

+ if w == what && failed.CompareAndSwap(false, true) {

+ return errors.New("injected write failure")

+ }

+ return nil

+ }

+ f.admissionRecheck = func() { f.discover(time.Now()) }

+ adm, err := f.gate(context.Background(), owned)

+ if adm != nil || !errors.Is(err, ErrOwnershipStale) {

+ t.Fatalf("refusal: %v %v", adm, err)

+ }

+ waitUntil(t, "refusal cleanup", func() bool { return retryQueueEmpty(f) })

+ tokens := admissionTokens(t, ledger, f.id)

+ if len(tokens) != 1 || !failed.Load() {

+ t.Fatalf("tokens=%v failed=%v", tokens, failed.Load())

+ }

+ a, _ := ledger.ReadAdmission(tokens[0])

+ if a.How != durability.AdmissionUnissued || !a.Settled() {

+ t.Fatalf("not unissued: %+v", a)

+ }

+ eff, _ := ledger.Effective(f.id)

+ if len(eff) != 1 || eff[0].Open() {

+ t.Fatalf("uncertainty after cleanup: %+v", eff)

+ }

+ })

+ }

+}

+

+func TestReview11ReadBackLivesBeyondRequest(t *testing.T) {

+ f, ledger, _, _ := review11Fleet(t)

+ token, _ := durability.NewAdmissionToken()

+ f.pending[token] = time.Now().UTC()

+ f.pendingN.Add(1)

+ f.discover(time.Now())

+ var available atomic.Bool

+ var tried atomic.Bool

+ f.admissionRead = func(tok string) (durability.Admission, error) {

+ tried.Store(true)

+ if !available.Load() {

+ return durability.Admission{}, errors.New("injected read failure")

+ }

+ return ledger.ReadAdmission(tok)

+ }

+ if adm, err := f.resolve(context.Background(), token, f.epoch, errors.New("ambiguous commit"), false, time.Now().Add(time.Second)); adm != nil || err == nil {

+ t.Fatalf("ambiguous answer: %v %v", adm, err)

+ }

+ waitUntil(t, "first read-back", tried.Load)

+ if f.pendingN.Load() != 1 {

+ t.Fatal("lost unresolved registration")

+ }

+ resolvedAfter := time.Now().UTC()

+ available.Store(true)

+ waitUntil(t, "conclusive cancellation", func() bool { return f.pendingN.Load() == 0 && retryQueueEmpty(f) })

+ eff, _ := ledger.Effective(f.id)

+ if len(eff) != 1 || eff[0].Open() || eff[0].To.Before(resolvedAfter.Add(durability.AdmissionMargin)) {

+ t.Fatalf("backdated cancellation: %+v, resolved after %v", eff, resolvedAfter)

+ }

+}

+

+func TestReview11CleanupDoesNotBlockRefusal(t *testing.T) {

+ f, ledger, owned, _ := review11Fleet(t)

+ entered, release := make(chan struct{}), make(chan struct{})

+ defer close(release)

+ f.retryWrite("blocked", "blocked", func() error { close(entered); <-release; return nil })

+ <-entered

+ f.admissionRecheck = func() { f.discover(time.Now()) }

+ done := make(chan error, 1)

+ go func() { _, err := f.gate(context.Background(), owned); done <- err }()

+ select {

+ case err := <-done:

+ if !errors.Is(err, ErrOwnershipStale) {

+ t.Fatal(err)

+ }

+ case <-time.After(time.Second):

+ t.Fatal("refusal waited for cleanup I/O")

+ }

+ if len(admissionTokens(t, ledger, f.id)) != 1 {

+ t.Fatal("fixture did not commit an admission")

+ }

+}

+

+func TestReview11UnopenedIntermediateAcknowledges(t *testing.T) {

+ a, ledger, _, o := review11Fleet(t)

+ o.Replaces = a

+ b, err := MeteredFleetStore(o)

+ if err != nil {

+ t.Fatal(err)

+ }

+ t.Cleanup(func() { b.Close() })

+ o.Replaces = b

+ c, err := MeteredFleetStore(o)

+ if err != nil {

+ t.Fatal(err)

+ }

+ t.Cleanup(func() { c.Close() })

+ if err := b.Close(); err != nil {

+ t.Fatal(err)

+ }

+ if err := a.Close(); err != nil {

+ t.Fatal(err)

+ }

+ if err := c.Open(context.Background()); err != nil {

+ t.Fatal(err)

+ }

+ ar, _ := ledger.ReadFleet(a.id)

+ br, _ := ledger.ReadFleet(b.id)

+ if ar.AcknowledgedBy != b.id || br.AcknowledgedBy != c.id {

+ t.Fatalf("unfinished chain: A=%+v B=%+v", ar, br)

+ }

+}

+

+func TestReview11RegistrationResolvesBeforeDiscoveryWrite(t *testing.T) {

+ f, ledger, owned, _ := review11Fleet(t)

+ committed, resolveNow := make(chan struct{}), make(chan struct{})

+ f.admissionRecheck = func() { close(committed); <-resolveNow }

+ answer := make(chan error, 1)

+ go func() { _, err := f.gate(context.Background(), owned); answer <- err }()

+ <-committed

+ f.discoveryBeforeWrite = func() {

+ close(resolveNow)

+ if err := <-answer; !errors.Is(err, ErrOwnershipStale) {

+ t.Errorf("discovery did not refuse committed row: %v", err)

+ }

+ waitUntil(t, "pre-discovery empty settlement", func() bool { return retryQueueEmpty(f) })

+ eff, _ := ledger.Effective(f.id)

+ if len(eff) != 0 {

+ t.Errorf("discovery committed too soon: %+v", eff)

+ }

+ f.discover(time.Now())

+ eff, _ := ledger.Effective(f.id)

+ if f.pendingN.Load() != 0 || len(eff) != 1 || eff[0].Open() {

+ t.Fatalf("lost early settlement: pending=%d effective=%+v", f.pendingN.Load(), eff)

+ }

+}

+

+func TestReview11AmbiguousFleetOpenKeepsCommittedFleet(t *testing.T) {

+ a, ledger, _, o := review11Fleet(t)

+ o.Replaces = a

+ fleetOpenResult = func(err error) error {

+ if err != nil {

+ return err

+ }

+ return errors.New("lost commit answer")

+ }

+ t.Cleanup(func() { fleetOpenResult = nil })

+ b, err := MeteredFleetStore(o)

+ fleetOpenResult = nil

+ if err != nil {

+ t.Fatal(err)

+ }

+ defer b.Close()

+ row, err := ledger.ReadFleet(b.id)

+ if err != nil || row.Predecessor != a.id {

+ t.Fatalf("lost committed fleet: %+v %v", row, err)

+ }

+}

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

index 33eb9a269..fbbc6db65 100644

--- a/internal/ops/litestream_admission_test.go

+++ b/internal/ops/litestream_admission_test.go

@@ -163,6 +163,7 @@ func TestMeteredFleetDiscoveryRefusesTheCommitAfterIt(t *testing.T) {

if len(toks) != 1 {

t.Fatalf("%d admissions, want the one unissued", len(toks))

}

+ waitUntil(t, "the refused admission to settle", func() bool { a, _ := ledger.ReadAdmission(toks[0]); return a.Settled() })

a, _ := ledger.ReadAdmission(toks[0])

if !a.Settled() || a.How != durability.AdmissionUnissued {

t.Fatalf("the admission after the recheck: %+v", a)

@@ -191,9 +192,7 @@ func TestMeteredFleetDiscoveryRefusesTheCommitAfterIt(t *testing.T) {

// The gate's budget and the writer's failures: a commit that does not

// return within the budget refuses the request and is marked unissued when

-// it lands; a writer that fails against a ledger that cannot answer the

-// read-back leaves the registration pending, counted, and the request

-// refused.

+// it lands. A closed-ledger refusal is definite and cancels registration.

func TestMeteredFleetAdmissionBudgetAndFailure(t *testing.T) {

newFakeBucket(t)

dir := platformDir(t)

@@ -232,19 +231,15 @@ func TestMeteredFleetAdmissionBudgetAndFailure(t *testing.T) {

if c, _ := ledger.Snapshot(durability.DayOf(time.Now())); c.Barred != 0 {

t.Fatalf("an unissued row bars: %+v", c)

}

- // the writer fails and the read-back cannot answer: the registration

- // stays pending

- oldWait := admissionReadBackWait

- admissionReadBackWait = 100 * time.Millisecond

- t.Cleanup(func() { admissionReadBackWait = oldWait })

+ // A closed ledger refuses before writing, so no read-back is needed.

if err := ledger.Close(context.Background(), nil); err != nil {

t.Fatal(err)

}

if err := send(t, fleet, db, 3); !errors.Is(err, lease.ErrRequestRefused) || errors.Is(err, ErrAdmissionNotDurable) {

t.Fatalf("a send with the ledger closed: %v", err)

}

- if fleet.pendingN.Load() != 1 {

- t.Fatalf("pending count %d, want the unresolved registration counted", fleet.pendingN.Load())

+ if fleet.pendingN.Load() != 0 {

+ t.Fatalf("pending count %d after a definite closed-ledger refusal", fleet.pendingN.Load())

}

@@ -265,13 +260,11 @@ func TestMeteredFleetCloseWritesTransportLost(t *testing.T) {

if err != nil {

t.Fatal(err)

}

- tok, _ := durability.NewAdmissionToken()

- if err := ledger.Admit(tok, fleet.FleetID(), "bac", "one", time.Now()); err != nil {

+ adm, err := fleet.gate(context.Background(), fleet.Owned[0])

+ if err != nil {

t.Fatal(err)

}

- fleet.liveMu.Lock()

- fleet.live[tok] = &fleetAdmission{f: fleet, token: tok}

- fleet.liveMu.Unlock()

+ tok := adm.(*fleetAdmission).token

oldWait := closeSettleWait

closeSettleWait = 100 * time.Millisecond

t.Cleanup(func() { closeSettleWait = oldWait })

@@ -288,13 +281,10 @@ func TestMeteredFleetCloseWritesTransportLost(t *testing.T) {

if c, _ := ledger.Snapshot(durability.DayOf(time.Now())); c.Barred == 0 {

t.Fatal("the pending admission does not bar")

}

- if err := ledger.RecorderDone(tok, time.Now(), durability.AdmissionUnrecorded); err != nil {

- t.Fatal(err)

- }

- if a, _ := ledger.ReadAdmission(tok); !a.Settled() {

- t.Fatalf("not settled after the recorder's outcome: %+v", a)

- }

- if err := ledger.CloseFleet(fleet.FleetID(), time.Now()); err != nil {

- t.Fatal(err)

- }

+ adm.RecorderDone(time.Now(), durability.AdmissionUnrecorded)

+ waitUntil(t, "late recorder and autonomous journal closure", func() bool {

+ a, _ := ledger.ReadAdmission(tok)

+ row, _ := ledger.ReadFleet(fleet.FleetID())

+ return a.Settled() && row.State == durability.FleetClosed

+ })

}

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

index 5e8a75d07..fc5d284bf 100644

--- a/internal/ops/litestream_metered.go

+++ b/internal/ops/litestream_metered.go

@@ -171,19 +171,26 @@ type MeteredFleet struct {

// guards (the pending set, the in-progress flag, the epoch), the

// lock-free pending count the ledger reads, the issued admissions

// whose transport has not finalised, and the flags

- ledger *durability.DurabilityLedger

- id string

- disc sync.Mutex

- pending map[string]time.Time

- inDiscovery bool

- epoch uint64

- pendingN atomic.Int64

- liveMu sync.Mutex

- live map[string]*fleetAdmission

- discovered atomic.Bool

- latched atomic.Bool

- acked atomic.Bool

- ledgerErrors atomic.Int64

+ ledger *durability.DurabilityLedger

+ id string

+ disc sync.Mutex

+ pending map[string]time.Time

+ inDiscovery bool

+ epoch uint64

+ pendingN atomic.Int64

+ liveMu sync.Mutex

+ live map[string]*fleetAdmission

+ discovered atomic.Bool

+ latched atomic.Bool

+ acked atomic.Bool

+ ledgerErrors atomic.Int64

+ retryMu sync.Mutex

+ retries map[string][]*admissionWrite

+ retrying bool

+ admissionRecheck func()

+ discoveryBeforeWrite func()

+ admissionWriteFail func(string) error

+ admissionRead func(string) (durability.Admission, error)

stale atomic.Bool

drift atomic.Bool

@@ -1040,9 +1047,18 @@ func MeteredFleetStore(o MeteredFleetOptions) (fleet *MeteredFleet, err error) {

if o.Replaces != nil {

pred = o.Replaces.id

}

- if err := o.Ledger.OpenFleet(id, o.Ledger.Box(), obs.Generation(), fleetOwners(inv.Owned), pred, time.Now().UTC()); err != nil {

- inv.Close()

- return nil, fmt.Errorf("%w: the fleet's journal row: %v", ErrUnmeterable, err)

+ openedAt := time.Now().UTC()

+ openErr := o.Ledger.OpenFleet(id, o.Ledger.Box(), obs.Generation(), fleetOwners(inv.Owned), pred, openedAt)

+ if fleetOpenResult != nil {

+ openErr = fleetOpenResult(openErr)

+ }

+ if openErr != nil {

+ // The random id belongs to this construction alone. A committed row

+ // must retain its fleet even when the commit's answer was lost.

+ if _, readErr := o.Ledger.ReadFleet(id); readErr != nil {

+ inv.Close()

+ return nil, fmt.Errorf("%w: the fleet's journal row: %v", ErrUnmeterable, openErr)

+ }

}

built.id = id

o.Ledger.SetPendingCounter(id, &built.pendingN)

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

index 3ceca0d47..ce80eda71 100644

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

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

@@ -829,16 +829,18 @@ logging:

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

}

for name, bad := range map[string]string{

- "flow style": "dbs: [{path: /data/a.db}]\n",

- "no path": "dbs:\n - replicas: []\n",

- "two paths": "dbs:\n - path: /a.db\n path: /b.db\n",

- "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",

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

+ "no path": "dbs:\n - replicas: []\n",

+ "two paths": "dbs:\n - path: /a.db\n path: /b.db\n",

+ "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",

+ "escaped keys": "\"d\\u0062s\":\n - \"p\\u0061th\": /data/retired.db\n",

+ "unrecognised document": "logging:\n level: info\n",

} {

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

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

@@ -847,7 +849,22 @@ logging:

if _, err := CoverageShortfall("dbs: [{path: /data/a.db}]\n", []string{"/data/a.db"}); err == nil {

t.Fatal("the shrink guard compared against a configuration it could not read")

}

- if got, err := ParseLitestreamDBs("dbs: []\n"); err != nil || len(got) != 0 {

- t.Fatalf("an empty list: %q %v", got, err)

+ escaped := "\"d\\u0062s\":\n - \"p\\u0061th\": /data/retired.db\n"

+ if _, err := CoverageShortfall(escaped, nil); err == nil {

+ t.Fatal("the shrink guard treated escaped database keys as an empty configuration")

+ }

+ for name, empty := range map[string]string{

+ "absent config": "",

+ "comments only": "# no databases configured yet\n",

+ "explicit empty list": "dbs: []\n",

+ "generated empty block": `snapshot:

+ interval: 24h0m0s

+ retention: 168h0m0s

+dbs:

+`,

+ } {

+ if got, err := ParseLitestreamDBs(empty); err != nil || len(got) != 0 {

+ t.Fatalf("%s: %q %v, want an empty configuration", name, got, err)

+ }

}

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

new file mode 100644

index 000000000..51e25cdad

--- /dev/null

+++ b/internal/usage/durability_review11_test.go

@@ -0,0 +1,178 @@

+package usage

+

+import (

+ "context"

+ "sync/atomic"

+ "testing"

+ "time"

+)

+

+func TestReview11DiscoveryAfterEmptySettlement(t *testing.T) {

+ for _, terminal := range []string{"cancelled", "unissued"} {

+ t.Run(terminal, func(t *testing.T) {

+ l, set := openLedgerAt(t, at(1, 23, 50))

+ var pending atomic.Int64

+ pending.Store(1)

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

+ d := Discovery{FleetID: "f", At: at(1, 23, 59), Proof: at(1, 23, 58), Pending: []time.Time{at(1, 23, 50)}}

+ set(d.At)

+ if terminal == "unissued" {

+ token := mustAdmit(t, l, "f", "alice", "app", d.Pending[0])

+ if err := l.Unissued(token, d.At); err != nil {

+ t.Fatal(err)

+ }

+ pending.Store(0)

+ if err := l.EmptySettlement("f", d.At); err != nil {

+ t.Fatal(err)

+ }

+ id, err := l.Discover(d)

+ if err != nil {

+ t.Fatal(err)

+ }

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

+ if err != nil {

+ t.Fatal(err)

+ }

+ if u.Open() || !u.To.Equal(at(2, 0, 0)) || !u.From.Equal(at(1, 23, 49)) {

+ t.Fatalf("resolved snapshot registration stranded uncertainty or lost bound: %+v", u)

+ }

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

+ })

+ }

+}

+

+func TestReview11DiscoveryAfterDelayedEmptySettlement(t *testing.T) {

+ for _, terminal := range []string{"cancelled", "unissued"} {

+ t.Run(terminal, func(t *testing.T) {

+ l, set := openLedgerAt(t, at(1, 23, 50))

+ var pending atomic.Int64

+ pending.Store(1)

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

+ d := Discovery{FleetID: "f", At: at(1, 23, 58), Proof: at(1, 23, 57), Pending: []time.Time{at(1, 23, 50)}}

+ resolved := at(2, 23, 59)

+ set(resolved)

+ if terminal == "unissued" {

+ token := mustAdmit(t, l, "f", "alice", "app", d.Pending[0])

+ if err := l.Unissued(token, resolved); err != nil {

+ t.Fatal(err)

+ }

+ pending.Store(0)

+ if err := l.EmptySettlement("f", resolved); err != nil {

+ t.Fatal(err)

+ }

+ id, err := l.Discover(d)

+ if err != nil {

+ t.Fatal(err)

+ }

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

+ if err != nil {

+ t.Fatal(err)

+ }

+ if u.Open() || !u.To.Equal(at(3, 0, 0)) || !u.From.Equal(at(1, 23, 49)) {

+ t.Fatalf("closure lost delayed empty settlement or its margin: %+v", u)

+ }

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

+ })

+ }

+}

+

+func TestReview11DiscoveryKeepsCurrentPendingAndRows(t *testing.T) {

+ for _, remaining := range []string{"counter", "row", "snapshot_without_counter"} {

+ t.Run(remaining, func(t *testing.T) {

+ l, _ := openLedgerAt(t, at(1, 23, 59))

+ var pending atomic.Int64

+ d := Discovery{FleetID: "f", At: at(1, 23, 59), Proof: at(1, 23, 58)}

+ switch remaining {

+ case "counter":

+ pending.Store(1)

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

+ case "row":

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

+ mustAdmit(t, l, "f", "alice", "app", at(1, 23, 50))

+ case "snapshot_without_counter":

+ d.Pending = []time.Time{at(1, 23, 50)}

+ }

+ id, err := l.Discover(d)

+ if err != nil {

+ t.Fatal(err)

+ }

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

+ if err != nil || !u.Open() {

+ t.Fatalf("closed with unresolved %s: %+v, %v", remaining, u, err)

+ }

+ })

+ }

+}

+

+func TestReview11ReplayCommitHidesOldOverlay(t *testing.T) {

+ for _, caller := range []string{"Replay", "Discover", "CloseFleet"} {

+ t.Run(caller, func(t *testing.T) {

+ l, _ := openLedgerAt(t, at(3, 12, 0))

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

+ t.Fatal(err)

+ }

+ d := Discovery{FleetID: "f", At: at(3, 12, 0), Proof: at(3, 9, 0)}

+ l.latchDiscovery(d, time.Time{}, nil)

+ before, err := l.Snapshot("2026-10-01")

+ if err != nil || before.Barred != 1 {

+ t.Fatalf("unresolved overlay: %+v, %v", before, err)

+ }

+ type result struct {

+ content DurabilityContent

+ err error

+ }

+ done := make(chan result, 1)

+ reachedOverlay := make(chan struct{})

+ releaseOverlay := make(chan struct{})

+ var captured *result

+ snapshotBeforeOverlay = func() { close(reachedOverlay); <-releaseOverlay }

+ replayAfterCommit = func() {

+ go func() {

+ c, _, err := l.snapshotOnce(context.Background(), "2026-10-01", 0, true)

+ done <- result{c, err}

+ }()

+ <-reachedOverlay

+ ds := l.disc()

+ unlocked := ds.mu.TryLock()

+ if unlocked {

+ ds.mu.Unlock()

+ }

+ close(releaseOverlay)

+ if unlocked {

+ r := <-done

+ captured = &r

+ }

+ t.Cleanup(func() { replayAfterCommit = nil; snapshotBeforeOverlay = nil })

+ switch caller {

+ case "Replay":

+ _, err = l.Replay("f")

+ case "Discover":

+ _, err = l.Discover(d)

+ case "CloseFleet":

+ err = l.CloseFleet("f", d.At)

+ }

+ if err != nil {

+ t.Fatal(err)

+ }

+ if captured == nil {

+ select {

+ case r := <-done:

+ captured = &r

+ case <-time.After(5 * time.Second):

+ t.Fatal("snapshot did not resume after replay")

+ }

+ replayAfterCommit = nil

+ snapshotBeforeOverlay = nil

+ if captured.err != nil || captured.content.Barred != 0 || captured.content.Held != 0 {

+ t.Fatalf("snapshot read committed replay with old unresolved overlay: %+v, %v", captured.content, captured.err)

+ }

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

+ t.Fatal("replay left latch installed")

+ }

+ })

+ }

+}

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

index c18786cc9..cc06295f8 100644

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

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

@@ -229,13 +229,10 @@ func (l *DurabilityLedger) CloseFleet(fleetID string, at time.Time) error {

return err

}

- if err := tx.Commit(); err != nil {

- return err

- }

if pd != nil {

- l.clearLatch(fleetID)

+ return l.commitReplay(tx, fleetID)

}

- return nil

+ return tx.Commit()

}

// Acknowledge is the successor's acknowledgement of its predecessor:

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

new file mode 100644

index 000000000..0b2a8a38e

--- /dev/null

+++ b/internal/usage/durabilitylifecycle.go

@@ -0,0 +1,8 @@

+package usage

+

+// Closed reports when recovery must take over unfinished writes.

+func (l *DurabilityLedger) Closed() bool {

+ l.mu.Lock()

+ defer l.mu.Unlock()

+ return l.closed

+}

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

index 2b999b77b..1d878de06 100644

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

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

@@ -29,6 +29,8 @@ const AdmissionMargin = time.Minute

// distinct from the database's own refusal.

var errLedgerClosed = errors.New("usage ledger: closed")

+var ErrLedgerClosed = errLedgerClosed

+

// ErrDiscoveryBound is a discovery with neither lower-bound term.

var ErrDiscoveryBound = errors.New("usage ledger: a discovery needs a proof instant or a last clean look")

@@ -327,9 +329,20 @@ func (l *DurabilityLedger) discover(d Discovery) (int64, time.Time, map[string]b

return 0, from, counted, err

}

to := d.At

- closeNow := len(counted) == 0 && len(d.Pending) == 0

+ ds := l.disc()

+ ds.mu.Lock()

+ n := ds.pending[d.FleetID]

+ ds.mu.Unlock()

+ pending := int64(len(d.Pending))

+ if n != nil {

+ pending = n.Load()

+ }

+ closeNow := len(counted) == 0 && pending == 0

if closeNow {

- to = d.At.Add(AdmissionMargin)

+ if len(d.Pending) > 0 {

+ to = maxTime(to, l.now().UTC())

+ }

+ to = to.Add(AdmissionMargin)

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

return 0, from, counted, err

}

@@ -507,20 +520,27 @@ func (l *DurabilityLedger) replayLocked(fleetID string, pd *pendingDiscovery, no

return 0, err

}

- if err := tx.Commit(); err != nil {

+ if err := l.commitReplay(tx, fleetID); err != nil {

return 0, err

}

- l.clearLatch(fleetID)

return id, nil

}

-// clearLatch forgets a replayed discovery; after its commit only.

-func (l *DurabilityLedger) clearLatch(fleetID string) {

+var replayAfterCommit func()

+

+func (l *DurabilityLedger) commitReplay(tx *sql.Tx, fleetID string) error {

ds := l.disc()

ds.mu.Lock()

+ defer ds.mu.Unlock()

+ if err := tx.Commit(); err != nil {

+ return err

+ }

+ if replayAfterCommit != nil {

+ replayAfterCommit()

+ }

delete(ds.latched, fleetID)

ds.generation++

- ds.mu.Unlock()

+ return nil

}

// replayTx is the replay's writes inside a transaction the caller

carlos / platform Public

Retain admission cleanup ownership and close discovery races