amadan
Plan Task 4's fourth slice: the fleet journal, construction backfill and root-first acknowledgement in the durability ledger
internal/usage/durabilityfleets.go: OpenFleet writes the open row with construction backfill of every interval its unacknowledged ancestry holds; FleetClosing; CloseFleet refuses with an unsettled admission or a pending registration and carries a latched replay in its own transaction, so a fleet is closed with that uncertainty committed or not closed at all; Acknowledge is root-first and terminal, copying the predecessor's effective set into the successor's inherited set and writing holds for the closed ones; Ancestry, Effective and Reach are the design's queries over the chain, computed at write time, never stored; the stale state is written inside the discovery's transaction. Holds are box-wide on this branch, so the backfill and acknowledgement writes are the rows a discovery already wrote; owner scope is these calls with the owners' rows. Tests mirror the model's ancestry-and-root-first sequence and the healthy chain, the closure's refusals and its carried replay, the journal's refusals, and a recorded random schedule asserting the per-step inheritance invariant after every step.
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
3 files changed,
+992
−1
docs/superpowers/plans/2026-10-03-replicator-authority-plan.md+1 −1internal/usage/durabilityfleets.go+507 −0internal/usage/durabilityfleets_test.go+484 −0
CI passed — run details
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 7d54f0ac5..78dffab06 100644 |
| --- a/docs/superpowers/plans/2026-10-03-replicator-authority-plan.md |
| +++ b/docs/superpowers/plans/2026-10-03-replicator-authority-plan.md |
| @@ -35,7 +35,7 @@ The admission model (`internal/ops/admissionmodel`) is the executable contract; |
| 1. **The tables and the admission's row (built: `internal/usage/durabilityadmissions.go`).** `admissions(token TEXT PRIMARY KEY, fleet_id, account, app, admitted_at, transport_done_at, recorder_started_at, recorder_done_at, settled_at, how)`, `fleets(fleet_id PRIMARY KEY, box, witness_generation, owners, opened_at, state, predecessor, acknowledged_by)`, `uncertainties(id, fleet_id, from_ns, to_ns NULL, reason)`, `inherited(fleet_id, uncertainty_id)`, `holds(day, account, app, uncertainty_id, written_at)`, all `IF NOT EXISTS` beside today's tables. The `DurabilityLedger` gains `Admit(token, fleet, owner, app, at) error` (the token allocated by the caller before the write, `NewAdmissionToken`, so an ambiguous commit is read back by it; one transaction, `synchronous=FULL`; an exact repeat is the write that landed, a repeat with other facts is refused; the gate's budget, `ErrAdmissionNotDurable` past it and the goroutine staying to mark a late commit `unissued`, are slice 5's, the gate's, and are not built here), `TransportDone(token, at, lost bool)` (`lost` is not recorded: transport-lost is never a value of `how`), `RecorderStarted(token, at)`, `RecorderDone(token, at, how)` (`finished` and `abandoned` need the recorder registered), settlement by whichever lands last in its own transaction at the latest of the two completions and now, and `Unissued(token, at)` (an edge from `admitted` alone: refused after any participant activity); every transition refuses an instant before the admission's, is idempotent for an exact repeat, and refuses on a closed ledger. Mirrors: `TestUnchangedFleetHoldsNothing` and `TestThreeMidnightsBothOrders` (the row's settlement in both orders; the interval written at settlement is slice 3's), `TestCommitAfterDiscoveryIsUnissued`, `TestHistoricalTermCountsUnrecordedSettlements`; `TestBudgetExpiryWithoutDiscovery` is slice 5's. |
| 2. **The barrier (built: `DurabilityContent.Barred`, counted in the snapshot's read transaction; `CompletenessFor` refuses while it is positive; box-wide, so the ancestry query waits for owner scope).** The publisher's snapshot (`snapshotOnce`) treats a day incomplete for an owner while a durable unsettled admission of that owner, or of any fleet whose unacknowledged ancestry includes the admission's fleet, is admitted on or before that day (box-wide: any owner); it is a query over `admissions` and `fleets`, writes nothing, and lifts at settlement. Mirrors: the model's barrier invariant and `TestSuccessorBarredByPredecessorAdmission`, `TestLandedAmbiguousCommitBarsBeforeReadBack`. |
| 3. **Discovery, the uncertainty and the latch (built in the ledger, with review nine's corrections: every hold write and every admission and settlement dirty their days through `touchDay` with a fresh sequence and `publish` enumerates the latched discoveries' days (`latchedDates`), so publication revisits a day a bound reaches or a barrier leaves; `discoveryState.generation` makes a snapshot read again across a replay; a failed discovery latches the bound and counted set it computed or latches unresolved; `Replay` recomputes the bound in its transaction and closes on `maybeClose`'s test; `uncertainties.discovered_at` floors every closure; `SetPendingCounter` takes an `*atomic.Int64`, lock-free by contract. `Discover` with the fixed bound and the counted set, `uncertainties` and box-wide `holds` rows in one transaction, closed at once with nothing counted or pending; the settlement's hook writing `[admitted_at, settled_at]` and closing when no row and no pending registration remains, the fleet's pending count by `SetPendingCounter`; `EmptySettlement`; a failed transaction latched in memory with its bound and counted tokens, barring publication through the snapshot's `Barred` until `Replay`, one transaction; `DurabilityContent.Held` refusing completeness; the fleet's discovery mutex, registration and the latch's refusal of requests are the gate slice).** `Discover(fleet, at, proof, lastCleanLook)`: under the fleet's discovery mutex (new in `MeteredFleet`: registration, the recheck and the start of a discovery serialise on it, the pending set counted), the fixed bound, the counted set, the `uncertainties` row and the holds of `[from, now]` for the owners reached in one transaction, closed at once with nothing pending; a failed write latches the pending discovery in the existing failure latch (extended with the bound, the owners reached and the counted tokens; the publisher's overlay reaches successors through the ancestry query) and the fleet refuses requests until `Replay`, one transaction as the design has it. Mirrors: `TestZeroPendingClosesAtOnceAcrossMidnight`, `TestFailedDiscoveryLatchBarsUntilReplay`, `TestFailedDiscoveryReplayKeepsSettlements`, `TestReplayHoldsTheLowerBoundWhileOpen`, `TestReplayWritesSettledTokenWhileOpen`, `TestLatchOverlayReachesSuccessor`, the lower-bound tests (the in-flight request across a clean look, the proof term, the earlier-of threshold, the fixed cutoff in both orders, the admission instant, the exact bound, the empty terminals excluded). |
| -4. **The fleet journal, construction backfill and acknowledgement.** `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. |
| +4. **The fleet journal, construction backfill and acknowledgement (built: `internal/usage/durabilityfleets.go`: `OpenFleet(fleetID, box, witnessGeneration, owners, predecessor, at)` with construction backfill of the unacknowledged ancestry's effective set; `FleetClosing`; `CloseFleet`, refusing with an unsettled row or a pending registration and carrying a latched replay in its own transaction; `Acknowledge`, root-first and terminal, copying the predecessor's effective set into `inherited` and writing the closed ones' holds; `ReadFleet`, `Ancestry`, `Effective`, `Reach`, `Inherited`; `stale` written inside the discovery's transaction; holds being box-wide on this branch, the backfill and acknowledgement writes are the rows a discovery already wrote, and the owner-scoped writes are these same calls with the owners' rows. Mirrors: the ancestry-and-root-first sequence, the healthy chain, and a recorded random schedule asserting the per-step inheritance invariant.)** `OpenFleet(predecessor, owners, witness)` writes the `open` row and the backfill of every interval its unacknowledged ancestry holds; `stale`, `closing`, `closed` as the design's transactions (closure carries a pending replay and refuses with an unsettled admission); `Acknowledge(successor)` root-first, copying the effective set into `inherited` and writing the holds for the closed ones. The ledger's reach query replaces the model's `reach`. Mirrors: `TestAncestryReachAndRootFirstAcknowledgement`, `TestHealthyChainContributesNothing`, the per-step inheritance invariant as a property test over a recorded schedule. |
| 5. **The gate and the fleet.** Contract (review nine, Opus P2-2): the ledger is never called with the fleet's discovery mutex held, and the pending counter the ledger reads is the lock-free `*atomic.Int64`, so a settlement under the ledger's mutex never waits on the mutex a gate recheck holds while it writes `unissued`. `MeteredFleet.admit` becomes the proof (unchanged) plus the admission: registration in the pending set under the discovery mutex, the ledger's `Admit`, the recheck, the token handed to the SDK middleware (`lease.RequestGate` returns the token; the meter's `Start` carries it; the finalize step writes `TransportDone`). `lookNow` is the discovery when it finds a change; `closeNow` is `closing` then `closed`. The helper process does not exist yet (Part B): the transport and the recorder run in the agent, so the lifecycle guard is the fleet's own, `transport-lost` is written at `Close` for a transport that never finalised, and `unrecorded` where the meter's `Start` failed. Mirrors: `TestHelperDeathAndRecorderGuard` (as the agent's close), `TestSendNeedsDurableIssuedAdmission` (the middleware refuses without a token), `TestReadBackOnlyAfterTheHandleCloses`, `TestBetweenCommitAndAnswer`. |
| 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/internal/usage/durabilityfleets.go b/internal/usage/durabilityfleets.go| new file mode 100644 |
| index 000000000..5aa5853b9 |
| --- /dev/null |
| +++ b/internal/usage/durabilityfleets.go |
| @@ -0,0 +1,507 @@ |
| +package usage |
| + |
| +import ( |
| + "database/sql" |
| + "errors" |
| + "fmt" |
| + "sort" |
| + "strings" |
| + "time" |
| +) |
| + |
| +// Part C's fleet journal (the design's "The fleet journal and the |
| +// handover" and "Who an uncertainty reaches, and what a fleet inherits"): |
| +// one row per fleet with its predecessor and who acknowledged it; the |
| +// states open, stale, closing and closed, each written in the transaction |
| +// the design names; construction backfill; and acknowledgement, terminal |
| +// and root-first, copying the predecessor's effective set into the |
| +// successor's inherited set. Reach is a query over the chain at write |
| +// time, never stored per uncertainty. Holds are box-wide on this branch |
| +// (owner scope is the specified ledger change), so the backfill and the |
| +// acknowledgement's hold writes are the same rows a box-wide discovery |
| +// already wrote (INSERT OR IGNORE): the journal's links, states and the |
| +// inherited copy are what this slice adds, and the owner-scoped writes |
| +// are these same calls with the owners' rows. Mirrors |
| +// admissionmodel.Build, Close and Acknowledge. |
| + |
| +// Fleet states, as the design's table has them. |
| +const ( |
| + FleetOpen = "open" |
| + FleetStale = "stale" |
| + FleetClosing = "closing" |
| + FleetClosed = "closed" |
| +) |
| + |
| +var ( |
| + // ErrFleetUnknown is returned for a fleet with no row. |
| + ErrFleetUnknown = errors.New("usage ledger: no such fleet") |
| + // ErrFleetUnsettled is returned by a closure while an admission of the |
| + // fleet is unsettled or a registration is pending: the row stays |
| + // closing, its admissions still barring. |
| + ErrFleetUnsettled = errors.New("usage ledger: the fleet has an unsettled admission or a pending registration") |
| + // ErrAcknowledgeNotReady is returned when the predecessor is not closed |
| + // or has not itself acknowledged its own predecessor (root-first). |
| + ErrAcknowledgeNotReady = errors.New("usage ledger: the predecessor is not closed, or has not acknowledged its own predecessor") |
| +) |
| + |
| +// FleetRow is one journal row, read back. |
| +type FleetRow struct { |
| + FleetID string |
| + Box string |
| + WitnessGeneration int64 |
| + Owners []string |
| + OpenedAt time.Time |
| + State string |
| + Predecessor string |
| + AcknowledgedBy string |
| +} |
| + |
| +// OpenFleet writes the fleet's open row, before any admission of it, with |
| +// construction backfill: in the same transaction, the holds of every |
| +// interval already held by any uncertainty of its unacknowledged |
| +// ancestry, open (through now) or closed (whole), for the new fleet's |
| +// owners. A second open of the same id with the same predecessor is the |
| +// write that landed; another predecessor is refused. |
| +func (l *DurabilityLedger) OpenFleet(fleetID, box string, witnessGeneration int64, owners []string, predecessor string, at time.Time) error { |
| + if fleetID == "" { |
| + return errors.New("usage ledger: a fleet needs an id") |
| + } |
| + at = at.UTC() |
| + if at.IsZero() { |
| + at = l.now().UTC() |
| + } |
| + sorted := append([]string(nil), owners...) |
| + sort.Strings(sorted) |
| + l.mu.Lock() |
| + defer l.mu.Unlock() |
| + if l.closed { |
| + return errLedgerClosed |
| + } |
| + tx, err := l.db.Begin() |
| + if err != nil { |
| + return err |
| + } |
| + defer tx.Rollback() //nolint:errcheck |
| + if f, err := readFleet(tx, fleetID); err == nil { |
| + if f.Predecessor == predecessor { |
| + return nil |
| + } |
| + return fmt.Errorf("usage ledger: fleet %s is already open against %q", fleetID, f.Predecessor) |
| + } else if !errors.Is(err, ErrFleetUnknown) { |
| + return err |
| + } |
| + if predecessor != "" { |
| + if _, err := readFleet(tx, predecessor); err != nil { |
| + return fmt.Errorf("usage ledger: predecessor %s: %w", predecessor, err) |
| + } |
| + } |
| + if _, err := tx.Exec(`INSERT INTO fleets (fleet_id, box, witness_generation, owners, opened_at, state, predecessor, acknowledged_by) VALUES (?, ?, ?, ?, ?, ?, ?, '')`, |
| + fleetID, box, witnessGeneration, strings.Join(sorted, "\n"), nsOf(at), FleetOpen, predecessor); err != nil { |
| + return err |
| + } |
| + // construction backfill: the ancestry's effective set, each interval |
| + // held for this fleet's owners from its first instant |
| + anc, err := unackAncestry(tx, fleetID) |
| + if err != nil { |
| + return err |
| + } |
| + us, err := effectiveOf(tx, anc) |
| + if err != nil { |
| + return err |
| + } |
| + for _, u := range us { |
| + to := u.To |
| + if u.Open() { |
| + to = at |
| + } |
| + if err := l.writeHolds(tx, u.From, to, u.ID, at); err != nil { |
| + return err |
| + } |
| + } |
| + if fleetFail != nil { |
| + if err := fleetFail(); err != nil { |
| + return err |
| + } |
| + } |
| + return tx.Commit() |
| +} |
| + |
| +// fleetFail is a test seam: an error here fails a journal transaction |
| +// before its commit. |
| +var fleetFail func() error |
| + |
| +// FleetClosing marks the fleet closing: the helper has exited and the |
| +// fleet has released its witness. From open or stale only; closing and |
| +// closed are left as they are. |
| +func (l *DurabilityLedger) FleetClosing(fleetID string) error { |
| + l.mu.Lock() |
| + defer l.mu.Unlock() |
| + if l.closed { |
| + return errLedgerClosed |
| + } |
| + tx, err := l.db.Begin() |
| + if err != nil { |
| + return err |
| + } |
| + defer tx.Rollback() //nolint:errcheck |
| + if _, err := readFleet(tx, fleetID); err != nil { |
| + return err |
| + } |
| + if _, err := tx.Exec(`UPDATE fleets SET state = ? WHERE fleet_id = ? AND state IN (?, ?)`, FleetClosing, fleetID, FleetOpen, FleetStale); err != nil { |
| + return err |
| + } |
| + return tx.Commit() |
| +} |
| + |
| +// CloseFleet marks the fleet closed: every admission of it settled, no |
| +// registration pending, and no discovery of it uncommitted. A latched |
| +// discovery is replayed in the closure's own transaction, so the fleet is |
| +// closed with that uncertainty committed or not closed at all. With an |
| +// unsettled admission or a pending registration it refuses |
| +// (ErrFleetUnsettled) and the row stays as it was. Idempotent on a closed |
| +// fleet. Mirrors admissionmodel.Close. |
| +func (l *DurabilityLedger) CloseFleet(fleetID string, at time.Time) error { |
| + at = at.UTC() |
| + if at.IsZero() { |
| + at = l.now().UTC() |
| + } |
| + l.drainMu.Lock() |
| + defer l.drainMu.Unlock() |
| + ds := l.disc() |
| + ds.mu.Lock() |
| + pd := ds.latched[fleetID] |
| + ds.mu.Unlock() |
| + l.mu.Lock() |
| + defer l.mu.Unlock() |
| + if l.closed { |
| + return errLedgerClosed |
| + } |
| + tx, err := l.db.Begin() |
| + if err != nil { |
| + return err |
| + } |
| + defer tx.Rollback() //nolint:errcheck |
| + f, err := readFleet(tx, fleetID) |
| + if err != nil { |
| + return err |
| + } |
| + if f.State == FleetClosed { |
| + return nil |
| + } |
| + var open int |
| + if err := tx.QueryRow(`SELECT COUNT(*) FROM admissions WHERE fleet_id = ? AND settled_at IS NULL`, fleetID).Scan(&open); err != nil { |
| + return err |
| + } |
| + if open > 0 || l.pendingCount(fleetID) > 0 { |
| + return ErrFleetUnsettled |
| + } |
| + if pd != nil { |
| + if _, err := l.replayTx(tx, fleetID, pd, at); err != nil { |
| + return err |
| + } |
| + } |
| + if _, err := tx.Exec(`UPDATE fleets SET state = ? WHERE fleet_id = ?`, FleetClosed, fleetID); err != nil { |
| + return err |
| + } |
| + if fleetFail != nil { |
| + if err := fleetFail(); err != nil { |
| + return err |
| + } |
| + } |
| + if err := tx.Commit(); err != nil { |
| + return err |
| + } |
| + if pd != nil { |
| + l.clearLatch(fleetID) |
| + } |
| + return nil |
| +} |
| + |
| +// Acknowledge is the successor's acknowledgement of its predecessor: |
| +// terminal and root-first. The predecessor must be closed and must itself |
| +// have acknowledged its own predecessor (or have none); then, in one |
| +// transaction, the predecessor's effective set (own and inherited, open |
| +// and closed) is copied into the successor's inherited set, holds are |
| +// written for the closed ones for the successor's owners, and the |
| +// predecessor's row names the successor. A successor with no predecessor |
| +// has nothing to acknowledge and returns nil; one that has already |
| +// acknowledged returns nil. Mirrors admissionmodel.Acknowledge. |
| +func (l *DurabilityLedger) Acknowledge(successorID string, at time.Time) error { |
| + at = at.UTC() |
| + if at.IsZero() { |
| + at = l.now().UTC() |
| + } |
| + l.mu.Lock() |
| + defer l.mu.Unlock() |
| + if l.closed { |
| + return errLedgerClosed |
| + } |
| + tx, err := l.db.Begin() |
| + if err != nil { |
| + return err |
| + } |
| + defer tx.Rollback() //nolint:errcheck |
| + g, err := readFleet(tx, successorID) |
| + if err != nil { |
| + return err |
| + } |
| + if g.Predecessor == "" { |
| + return nil |
| + } |
| + p, err := readFleet(tx, g.Predecessor) |
| + if err != nil { |
| + return err |
| + } |
| + if p.AcknowledgedBy == successorID { |
| + return nil |
| + } |
| + if p.AcknowledgedBy != "" { |
| + return fmt.Errorf("usage ledger: fleet %s was acknowledged by %s", p.FleetID, p.AcknowledgedBy) |
| + } |
| + if p.State != FleetClosed { |
| + return ErrAcknowledgeNotReady |
| + } |
| + if p.Predecessor != "" { |
| + pp, err := readFleet(tx, p.Predecessor) |
| + if err != nil { |
| + return err |
| + } |
| + if pp.AcknowledgedBy != p.FleetID { |
| + return ErrAcknowledgeNotReady |
| + } |
| + } |
| + us, err := effectiveOf(tx, []string{p.FleetID}) |
| + if err != nil { |
| + return err |
| + } |
| + for _, u := range us { |
| + if _, err := tx.Exec(`INSERT OR IGNORE INTO inherited (fleet_id, uncertainty_id) VALUES (?, ?)`, successorID, u.ID); err != nil { |
| + return err |
| + } |
| + if !u.Open() { |
| + if err := l.writeHolds(tx, u.From, u.To, u.ID, at); err != nil { |
| + return err |
| + } |
| + } |
| + } |
| + if _, err := tx.Exec(`UPDATE fleets SET acknowledged_by = ? WHERE fleet_id = ?`, successorID, p.FleetID); err != nil { |
| + return err |
| + } |
| + if fleetFail != nil { |
| + if err := fleetFail(); err != nil { |
| + return err |
| + } |
| + } |
| + return tx.Commit() |
| +} |
| + |
| +// ReadFleet reads one journal row. |
| +func (l *DurabilityLedger) ReadFleet(fleetID string) (FleetRow, error) { |
| + l.mu.Lock() |
| + defer l.mu.Unlock() |
| + tx, err := l.db.Begin() |
| + if err != nil { |
| + return FleetRow{}, err |
| + } |
| + defer tx.Rollback() //nolint:errcheck |
| + return readFleet(tx, fleetID) |
| +} |
| + |
| +// Ancestry is the fleet's unacknowledged ancestry, nearest first: the |
| +// predecessors reached by following the chain through steps that have |
| +// not been acknowledged (an acknowledged step ends the walk, because |
| +// acknowledgement copied). |
| +func (l *DurabilityLedger) Ancestry(fleetID string) ([]string, error) { |
| + l.mu.Lock() |
| + defer l.mu.Unlock() |
| + tx, err := l.db.Begin() |
| + if err != nil { |
| + return nil, err |
| + } |
| + defer tx.Rollback() //nolint:errcheck |
| + return unackAncestry(tx, fleetID) |
| +} |
| + |
| +// Effective is the fleet's effective set: its own uncertainties, its |
| +// inherited ones, and the own and inherited sets of its unacknowledged |
| +// ancestry, by id. |
| +func (l *DurabilityLedger) Effective(fleetID string) ([]Uncertainty, error) { |
| + l.mu.Lock() |
| + defer l.mu.Unlock() |
| + tx, err := l.db.Begin() |
| + if err != nil { |
| + return nil, err |
| + } |
| + defer tx.Rollback() //nolint:errcheck |
| + anc, err := unackAncestry(tx, fleetID) |
| + if err != nil { |
| + return nil, err |
| + } |
| + return effectiveOf(tx, append([]string{fleetID}, anc...)) |
| +} |
| + |
| +// Reach is who an uncertainty of the fleet reaches when it is opened, |
| +// extended or closed: the fleet itself and every fleet whose |
| +// unacknowledged ancestry contains it (a query at write time). Under |
| +// box-wide holds the answer changes nothing about the rows written; under |
| +// owner scope it is the set of owners held. |
| +func (l *DurabilityLedger) Reach(fleetID string) ([]string, error) { |
| + l.mu.Lock() |
| + defer l.mu.Unlock() |
| + tx, err := l.db.Begin() |
| + if err != nil { |
| + return nil, err |
| + } |
| + defer tx.Rollback() //nolint:errcheck |
| + return reaching(tx, fleetID) |
| +} |
| + |
| +// Inherited is the ids copied into the fleet's inherited set. |
| +func (l *DurabilityLedger) Inherited(fleetID string) ([]int64, error) { |
| + l.mu.Lock() |
| + defer l.mu.Unlock() |
| + rows, err := l.db.Query(`SELECT uncertainty_id FROM inherited WHERE fleet_id = ? ORDER BY uncertainty_id`, fleetID) |
| + if err != nil { |
| + return nil, err |
| + } |
| + defer rows.Close() |
| + var out []int64 |
| + for rows.Next() { |
| + var id int64 |
| + if err := rows.Scan(&id); err != nil { |
| + return nil, err |
| + } |
| + out = append(out, id) |
| + } |
| + return out, rows.Err() |
| +} |
| + |
| +func readFleet(tx *sql.Tx, fleetID string) (FleetRow, error) { |
| + var f FleetRow |
| + var owners string |
| + var opened int64 |
| + err := tx.QueryRow(`SELECT fleet_id, box, witness_generation, owners, opened_at, state, predecessor, acknowledged_by FROM fleets WHERE fleet_id = ?`, fleetID). |
| + Scan(&f.FleetID, &f.Box, &f.WitnessGeneration, &owners, &opened, &f.State, &f.Predecessor, &f.AcknowledgedBy) |
| + if errors.Is(err, sql.ErrNoRows) { |
| + return FleetRow{}, fmt.Errorf("%w: %s", ErrFleetUnknown, fleetID) |
| + } |
| + if err != nil { |
| + return FleetRow{}, err |
| + } |
| + if owners != "" { |
| + f.Owners = strings.Split(owners, "\n") |
| + } |
| + f.OpenedAt = time.Unix(0, opened).UTC() |
| + return f, nil |
| +} |
| + |
| +// unackAncestry walks predecessors from the fleet while the step has not |
| +// been acknowledged. Mirrors admissionmodel.unackAncestry. |
| +func unackAncestry(tx *sql.Tx, fleetID string) ([]string, error) { |
| + var anc []string |
| + seen := map[string]bool{fleetID: true} |
| + cur, err := readFleet(tx, fleetID) |
| + if err != nil { |
| + return nil, err |
| + } |
| + for cur.Predecessor != "" { |
| + p, err := readFleet(tx, cur.Predecessor) |
| + if err != nil { |
| + return nil, err |
| + } |
| + if p.AcknowledgedBy == cur.FleetID { |
| + break // an acknowledged step ends the walk: the copy carries on |
| + } |
| + if seen[p.FleetID] { |
| + return nil, fmt.Errorf("usage ledger: the fleet chain loops at %s", p.FleetID) |
| + } |
| + seen[p.FleetID] = true |
| + anc = append(anc, p.FleetID) |
| + cur = p |
| + } |
| + return anc, nil |
| +} |
| + |
| +// effectiveOf is the union of the own and inherited uncertainties of the |
| +// given fleets, by id, in id order. |
| +func effectiveOf(tx *sql.Tx, fleets []string) ([]Uncertainty, error) { |
| + seen := map[int64]bool{} |
| + var out []Uncertainty |
| + for _, f := range fleets { |
| + rows, err := tx.Query(`SELECT id, fleet_id, from_ns, to_ns, discovered_at, reason FROM uncertainties WHERE fleet_id = ? |
| + UNION SELECT u.id, u.fleet_id, u.from_ns, u.to_ns, u.discovered_at, u.reason FROM uncertainties u JOIN inherited i ON i.uncertainty_id = u.id WHERE i.fleet_id = ? |
| + ORDER BY 1`, f, f) |
| + if err != nil { |
| + return nil, err |
| + } |
| + for rows.Next() { |
| + var u Uncertainty |
| + var from, discovered int64 |
| + var to sql.NullInt64 |
| + if err := rows.Scan(&u.ID, &u.FleetID, &from, &to, &discovered, &u.Reason); err != nil { |
| + rows.Close() |
| + return nil, err |
| + } |
| + if seen[u.ID] { |
| + continue |
| + } |
| + seen[u.ID] = true |
| + u.From = time.Unix(0, from).UTC() |
| + u.DiscoveredAt = time.Unix(0, discovered).UTC() |
| + if to.Valid { |
| + u.To = time.Unix(0, to.Int64).UTC() |
| + } |
| + out = append(out, u) |
| + } |
| + if err := rows.Err(); err != nil { |
| + rows.Close() |
| + return nil, err |
| + } |
| + rows.Close() |
| + } |
| + sort.Slice(out, func(i, j int) bool { return out[i].ID < out[j].ID }) |
| + return out, nil |
| +} |
| + |
| +// reaching is every fleet whose unacknowledged ancestry contains the |
| +// fleet, and the fleet itself. Mirrors admissionmodel.reach. |
| +func reaching(tx *sql.Tx, fleetID string) ([]string, error) { |
| + if _, err := readFleet(tx, fleetID); err != nil { |
| + return nil, err |
| + } |
| + rows, err := tx.Query(`SELECT fleet_id FROM fleets ORDER BY opened_at, fleet_id`) |
| + if err != nil { |
| + return nil, err |
| + } |
| + var all []string |
| + for rows.Next() { |
| + var id string |
| + if err := rows.Scan(&id); err != nil { |
| + rows.Close() |
| + return nil, err |
| + } |
| + all = append(all, id) |
| + } |
| + if err := rows.Err(); err != nil { |
| + rows.Close() |
| + return nil, err |
| + } |
| + rows.Close() |
| + out := []string{fleetID} |
| + for _, g := range all { |
| + if g == fleetID { |
| + continue |
| + } |
| + anc, err := unackAncestry(tx, g) |
| + if err != nil { |
| + return nil, err |
| + } |
| + for _, a := range anc { |
| + if a == fleetID { |
| + out = append(out, g) |
| + break |
| + } |
| + } |
| + } |
| + return out, nil |
| +} |
diff --git a/internal/usage/durabilityfleets_test.go b/internal/usage/durabilityfleets_test.go| new file mode 100644 |
| index 000000000..effd8b078 |
| --- /dev/null |
| +++ b/internal/usage/durabilityfleets_test.go |
| @@ -0,0 +1,484 @@ |
| +package usage |
| + |
| +import ( |
| + "errors" |
| + "math/rand" |
| + "reflect" |
| + "sync/atomic" |
| + "testing" |
| + "time" |
| +) |
| + |
| +var errDiskFull = errors.New("disk full") |
| + |
| +func ids(us []Uncertainty) []int64 { |
| + var out []int64 |
| + for _, u := range us { |
| + out = append(out, u.ID) |
| + } |
| + return out |
| +} |
| + |
| +// Mirrors admissionmodel.TestAncestryReachAndRootFirstAcknowledgement |
| +// (Astra's sequence): A has U open and an admission unsettled; B is built |
| +// against A and closes unopened without acknowledging; C is built against |
| +// B, so its unacknowledged ancestry is B then A; the settlement on the 4th |
| +// closes U and reaches every fleet in the chain; C cannot acknowledge B |
| +// before B acknowledges A (root-first); after A closes, B acknowledges A |
| +// and C acknowledges B and inherits U. |
| +func TestFleetAncestryReachAndRootFirstAcknowledgement(t *testing.T) { |
| + l, set := openLedgerAt(t, at(2, 9, 0)) |
| + if err := l.OpenFleet("A", "box", 1, []string{"alice"}, "", time.Time{}); err != nil { |
| + t.Fatal(err) |
| + } |
| + if f, _ := l.ReadFleet("A"); f.State != FleetOpen || f.Predecessor != "" || !reflect.DeepEqual(f.Owners, []string{"alice"}) || !f.OpenedAt.Equal(at(2, 9, 0)) { |
| + t.Fatalf("A: %+v", f) |
| + } |
| + set(at(2, 23, 57)) |
| + a := mustAdmit(t, l, "A", "alice", "app", time.Time{}) |
| + set(at(2, 23, 58)) |
| + uid, err := l.Discover(Discovery{FleetID: "A", At: at(2, 23, 58), Proof: at(2, 9, 0), LastCleanLook: at(2, 23, 0)}) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if f, _ := l.ReadFleet("A"); f.State != FleetStale { |
| + t.Fatalf("A after a discovery: %s, want stale", f.State) |
| + } |
| + set(at(2, 23, 59)) |
| + if err := l.OpenFleet("B", "box", 1, []string{"bob"}, "A", time.Time{}); err != nil { |
| + t.Fatal(err) |
| + } |
| + if anc, _ := l.Ancestry("B"); !reflect.DeepEqual(anc, []string{"A"}) { |
| + t.Fatalf("B's ancestry %v", anc) |
| + } |
| + if r, _ := l.Reach("A"); !reflect.DeepEqual(r, []string{"A", "B"}) { |
| + t.Fatalf("U reaches %v, want A and B", r) |
| + } |
| + if eff, _ := l.Effective("B"); !reflect.DeepEqual(ids(eff), []int64{uid}) { |
| + t.Fatalf("B's effective set %v (construction backfill of the open interval)", ids(eff)) |
| + } |
| + held(t, l, true, "2026-10-02") |
| + set(at(3, 0, 1)) |
| + if err := l.FleetClosing("B"); err != nil { |
| + t.Fatal(err) |
| + } |
| + if err := l.CloseFleet("B", time.Time{}); err != nil { |
| + t.Fatalf("B did not close unopened: %v", err) |
| + } |
| + if f, _ := l.ReadFleet("B"); f.State != FleetClosed { |
| + t.Fatalf("B: %s", f.State) |
| + } |
| + if err := l.OpenFleet("C", "box", 1, []string{"carol"}, "B", time.Time{}); err != nil { |
| + t.Fatal(err) |
| + } |
| + if anc, _ := l.Ancestry("C"); !reflect.DeepEqual(anc, []string{"B", "A"}) { |
| + t.Fatalf("C's ancestry %v, want B then A (B has not acknowledged)", anc) |
| + } |
| + if eff, _ := l.Effective("C"); !reflect.DeepEqual(ids(eff), []int64{uid}) { |
| + t.Fatalf("C's effective set %v", ids(eff)) |
| + } |
| + if r, _ := l.Reach("A"); !reflect.DeepEqual(r, []string{"A", "B", "C"}) { |
| + t.Fatalf("U reaches %v, want A, B and C", r) |
| + } |
| + held(t, l, true, "2026-10-02", "2026-10-03") |
| + set(at(4, 0, 2)) |
| + finishAdmission(t, l, a) |
| + if u, _ := l.ReadUncertainty(uid); u.Open() { |
| + t.Fatal("U still open after the settlement") |
| + } |
| + held(t, l, true, "2026-10-02", "2026-10-03", "2026-10-04") |
| + if err := l.Acknowledge("C", time.Time{}); !errors.Is(err, ErrAcknowledgeNotReady) { |
| + t.Fatalf("C acknowledged B before B acknowledged A: %v", err) |
| + } |
| + if err := l.Acknowledge("B", time.Time{}); !errors.Is(err, ErrAcknowledgeNotReady) { |
| + t.Fatalf("B acknowledged A before A closed: %v", err) |
| + } |
| + if err := l.FleetClosing("A"); err != nil { |
| + t.Fatal(err) |
| + } |
| + if err := l.CloseFleet("A", time.Time{}); err != nil { |
| + t.Fatal(err) |
| + } |
| + if err := l.Acknowledge("B", time.Time{}); err != nil { |
| + t.Fatal(err) |
| + } |
| + if f, _ := l.ReadFleet("A"); f.AcknowledgedBy != "B" { |
| + t.Fatalf("A acknowledged by %q", f.AcknowledgedBy) |
| + } |
| + if inh, _ := l.Inherited("B"); !reflect.DeepEqual(inh, []int64{uid}) { |
| + t.Fatalf("B inherited %v", inh) |
| + } |
| + if anc, _ := l.Ancestry("B"); len(anc) != 0 { |
| + t.Fatalf("B's ancestry after acknowledging: %v", anc) |
| + } |
| + if anc, _ := l.Ancestry("C"); !reflect.DeepEqual(anc, []string{"B"}) { |
| + t.Fatalf("C's ancestry after B acknowledged: %v, want B alone (the copy carries on)", anc) |
| + } |
| + if err := l.Acknowledge("C", time.Time{}); err != nil { |
| + t.Fatal(err) |
| + } |
| + if inh, _ := l.Inherited("C"); !reflect.DeepEqual(inh, []int64{uid}) { |
| + t.Fatalf("C did not inherit U: %v", inh) |
| + } |
| + if r, _ := l.Reach("A"); !reflect.DeepEqual(r, []string{"A"}) { |
| + t.Fatalf("after acknowledgement U reaches %v, want A alone", r) |
| + } |
| + // terminal and idempotent; a fleet with no predecessor has nothing to |
| + // acknowledge; a predecessor acknowledged by another is refused |
| + if err := l.Acknowledge("C", time.Time{}); err != nil { |
| + t.Fatalf("acknowledging again: %v", err) |
| + } |
| + if err := l.Acknowledge("A", time.Time{}); err != nil { |
| + t.Fatalf("a root acknowledging nothing: %v", err) |
| + } |
| + if err := l.OpenFleet("D", "box", 1, []string{"dan"}, "A", time.Time{}); err != nil { |
| + t.Fatal(err) |
| + } |
| + if err := l.Acknowledge("D", time.Time{}); err == nil { |
| + t.Fatal("D acknowledged A, which B had acknowledged") |
| + } |
| +} |
| + |
| +// Mirrors admissionmodel.TestHealthyChainContributesNothing: routine |
| +// replacement of a healthy chain holds no day, inherits nothing. |
| +func TestFleetHealthyChainContributesNothing(t *testing.T) { |
| + l, set := openLedgerAt(t, at(1, 10, 0)) |
| + if err := l.OpenFleet("A", "box", 1, []string{"alice"}, "", time.Time{}); err != nil { |
| + t.Fatal(err) |
| + } |
| + a := mustAdmit(t, l, "A", "alice", "app", time.Time{}) |
| + set(at(1, 11, 0)) |
| + finishAdmission(t, l, a) |
| + if err := l.FleetClosing("A"); err != nil { |
| + t.Fatal(err) |
| + } |
| + if err := l.CloseFleet("A", time.Time{}); err != nil { |
| + t.Fatal(err) |
| + } |
| + set(at(1, 12, 0)) |
| + if err := l.OpenFleet("B", "box", 2, []string{"bob"}, "A", time.Time{}); err != nil { |
| + t.Fatal(err) |
| + } |
| + if err := l.Acknowledge("B", time.Time{}); err != nil { |
| + t.Fatal(err) |
| + } |
| + held(t, l, false, "2026-10-01", "2026-10-02") |
| + if eff, _ := l.Effective("B"); len(eff) != 0 { |
| + t.Fatalf("B's effective set %v", ids(eff)) |
| + } |
| + if inh, _ := l.Inherited("B"); len(inh) != 0 { |
| + t.Fatalf("B inherited %v", inh) |
| + } |
| + if c, _ := l.Snapshot("2026-10-01"); c.Barred != 0 || c.Held != 0 { |
| + t.Fatalf("the day after a healthy chain: %+v", c) |
| + } |
| +} |
| + |
| +// Closure refuses with an unsettled admission or a pending registration |
| +// (the row stays, its admissions still barring); a latched discovery is |
| +// carried by the closure's own transaction, so the fleet is closed with |
| +// its uncertainty committed or not closed at all. |
| +func TestFleetClosureRefusesUnsettledAndCarriesTheReplay(t *testing.T) { |
| + l, set := openLedgerAt(t, at(1, 10, 0)) |
| + if err := l.OpenFleet("A", "box", 1, []string{"alice"}, "", time.Time{}); err != nil { |
| + t.Fatal(err) |
| + } |
| + a := mustAdmit(t, l, "A", "alice", "app", time.Time{}) |
| + if err := l.FleetClosing("A"); err != nil { |
| + t.Fatal(err) |
| + } |
| + if err := l.CloseFleet("A", time.Time{}); !errors.Is(err, ErrFleetUnsettled) { |
| + t.Fatalf("closed with an admission unsettled: %v", err) |
| + } |
| + if f, _ := l.ReadFleet("A"); f.State != FleetClosing { |
| + t.Fatalf("A: %s, want closing", f.State) |
| + } |
| + if c, _ := l.Snapshot("2026-10-01"); c.Barred != 1 { |
| + t.Fatalf("the refused closure lifted the barrier: %+v", c) |
| + } |
| + set(at(1, 11, 0)) |
| + finishAdmission(t, l, a) |
| + var pending atomic.Int64 |
| + pending.Store(1) |
| + l.SetPendingCounter("A", &pending) |
| + if err := l.CloseFleet("A", time.Time{}); !errors.Is(err, ErrFleetUnsettled) { |
| + t.Fatalf("closed with a registration pending: %v", err) |
| + } |
| + pending.Store(0) |
| + // a discovery whose write fails: latched; the closure carries it |
| + set(at(1, 12, 0)) |
| + discoverFail = func() error { return errors.New("disk full") } |
| + if _, err := l.Discover(Discovery{FleetID: "A", At: at(1, 12, 0), Proof: at(1, 10, 0), LastCleanLook: at(1, 11, 30)}); err == nil { |
| + t.Fatal("the failed discovery did not say so") |
| + } |
| + discoverFail = nil |
| + if _, ok := l.Latched("A"); !ok { |
| + t.Fatal("the failed discovery is not latched") |
| + } |
| + if c, _ := l.Snapshot("2026-10-01"); c.Barred == 0 { |
| + t.Fatal("the latched discovery does not bar") |
| + } |
| + // the closure's transaction failing leaves the latch and the state |
| + set(at(1, 12, 30)) |
| + fleetFail = func() error { return errors.New("disk full") } |
| + err := l.CloseFleet("A", time.Time{}) |
| + fleetFail = nil |
| + if err == nil { |
| + t.Fatal("a closure whose transaction failed returned nil") |
| + } |
| + if _, ok := l.Latched("A"); !ok { |
| + t.Fatal("the latch was cleared by a closure that did not commit") |
| + } |
| + if f, _ := l.ReadFleet("A"); f.State != FleetClosing { |
| + t.Fatalf("A after a failed closure: %s", f.State) |
| + } |
| + if err := l.CloseFleet("A", time.Time{}); err != nil { |
| + t.Fatal(err) |
| + } |
| + if _, ok := l.Latched("A"); ok { |
| + t.Fatal("the latch survived the closure") |
| + } |
| + f, _ := l.ReadFleet("A") |
| + if f.State != FleetClosed { |
| + t.Fatalf("A: %s", f.State) |
| + } |
| + eff, _ := l.Effective("A") |
| + if len(eff) != 1 || eff[0].Open() || !eff[0].From.Equal(at(1, 10, 0).Add(-AdmissionMargin)) || !eff[0].To.Equal(at(1, 12, 30).Add(AdmissionMargin)) { |
| + t.Fatalf("the replayed uncertainty: %+v", eff) |
| + } |
| + held(t, l, true, "2026-10-01") |
| + if c, _ := l.Snapshot("2026-10-01"); c.Barred != 0 || c.Held == 0 { |
| + t.Fatalf("after the closure: %+v", c) |
| + } |
| + // closed is terminal and idempotent; closing from closed stays closed |
| + if err := l.CloseFleet("A", time.Time{}); err != nil { |
| + t.Fatal(err) |
| + } |
| + if err := l.FleetClosing("A"); err != nil { |
| + t.Fatal(err) |
| + } |
| + if f, _ := l.ReadFleet("A"); f.State != FleetClosed { |
| + t.Fatalf("A after closing on closed: %s", f.State) |
| + } |
| +} |
| + |
| +// The journal's refusals: no id, an unknown predecessor, a second open |
| +// with another predecessor; a repeat with the same is the write that |
| +// landed; an unknown fleet is said; a failed open writes nothing; owners |
| +// are kept sorted; a closed ledger refuses. |
| +func TestOpenFleetRefusals(t *testing.T) { |
| + l, _ := openLedgerAt(t, at(1, 10, 0)) |
| + if err := l.OpenFleet("", "box", 1, nil, "", time.Time{}); err == nil { |
| + t.Fatal("a fleet with no id") |
| + } |
| + if err := l.OpenFleet("B", "box", 1, nil, "nope", time.Time{}); !errors.Is(err, ErrFleetUnknown) { |
| + t.Fatalf("an unknown predecessor: %v", err) |
| + } |
| + if _, err := l.ReadFleet("B"); !errors.Is(err, ErrFleetUnknown) { |
| + t.Fatalf("the refused fleet has a row: %v", err) |
| + } |
| + fleetFail = func() error { return errors.New("disk full") } |
| + err := l.OpenFleet("A", "box", 1, []string{"zed", "alice"}, "", time.Time{}) |
| + fleetFail = nil |
| + if err == nil { |
| + t.Fatal("an open whose transaction failed returned nil") |
| + } |
| + if _, err := l.ReadFleet("A"); !errors.Is(err, ErrFleetUnknown) { |
| + t.Fatalf("the failed open wrote a row: %v", err) |
| + } |
| + if err := l.OpenFleet("A", "box", 1, []string{"zed", "alice"}, "", time.Time{}); err != nil { |
| + t.Fatal(err) |
| + } |
| + if err := l.OpenFleet("A", "box", 1, []string{"zed", "alice"}, "", time.Time{}); err != nil { |
| + t.Fatalf("the same open again: %v", err) |
| + } |
| + if err := l.OpenFleet("A", "box", 1, nil, "B", time.Time{}); err == nil { |
| + t.Fatal("A opened again against another predecessor") |
| + } |
| + if f, _ := l.ReadFleet("A"); !reflect.DeepEqual(f.Owners, []string{"alice", "zed"}) || f.Box != "box" || f.WitnessGeneration != 1 { |
| + t.Fatalf("A: %+v", f) |
| + } |
| + for name, f := range map[string]func() error{ |
| + "closing": func() error { return l.FleetClosing("nope") }, |
| + "close": func() error { return l.CloseFleet("nope", time.Time{}) }, |
| + "acknowledge": func() error { return l.Acknowledge("nope", time.Time{}) }, |
| + } { |
| + if err := f(); !errors.Is(err, ErrFleetUnknown) { |
| + t.Fatalf("%s of an unknown fleet: %v", name, err) |
| + } |
| + } |
| + if _, err := l.Ancestry("nope"); !errors.Is(err, ErrFleetUnknown) { |
| + t.Fatalf("the ancestry of an unknown fleet: %v", err) |
| + } |
| + if _, err := l.Reach("nope"); !errors.Is(err, ErrFleetUnknown) { |
| + t.Fatalf("the reach of an unknown fleet: %v", err) |
| + } |
| + if err := l.Close(t.Context(), nil); err != nil { |
| + t.Fatal(err) |
| + } |
| + for name, f := range map[string]func() error{ |
| + "open": func() error { return l.OpenFleet("C", "box", 1, nil, "", time.Time{}) }, |
| + "closing": func() error { return l.FleetClosing("A") }, |
| + "close": func() error { return l.CloseFleet("A", time.Time{}) }, |
| + "acknowledge": func() error { return l.Acknowledge("A", time.Time{}) }, |
| + } { |
| + if err := f(); err == nil { |
| + t.Fatalf("%s on a closed ledger", name) |
| + } |
| + } |
| +} |
| + |
| +// The per-step inheritance invariant as a property over recorded random |
| +// schedules (the model's check, against the ledger): after every step, |
| +// each fleet's effective set is held for every day of each interval (a |
| +// closed one whole, an open one through its discovery), a successor that |
| +// acknowledged holds its predecessor's effective set in its inherited |
| +// set, an acknowledged step is out of the ancestry, and a closed fleet has |
| +// no unsettled admission and no latch. |
| +func TestFleetScheduleInheritance(t *testing.T) { |
| + seeds := 40 |
| + if testing.Short() { |
| + seeds = 8 |
| + } |
| + for seed := 1; seed <= seeds; seed++ { |
| + rnd := rand.New(rand.NewSource(int64(seed))) |
| + l, set := openLedgerAt(t, at(1, 9, 0)) |
| + now := at(1, 9, 0) |
| + var fleets []string |
| + open := map[string][]string{} // fleet -> unsettled tokens |
| + acked := map[string]string{} // predecessor -> successor, as acknowledged in this schedule |
| + closed := map[string]bool{} |
| + step := func(i int, what string) { |
| + t.Helper() |
| + for _, g := range fleets { |
| + eff, err := l.Effective(g) |
| + if err != nil { |
| + t.Fatalf("seed %d step %d (%s): %v", seed, i, what, err) |
| + } |
| + for _, u := range eff { |
| + to := u.To |
| + if u.Open() { |
| + to = u.DiscoveredAt // the row's own: a replayed discovery keeps the original instant |
| + } |
| + for d := u.From.Truncate(24 * time.Hour); !d.After(to); d = d.Add(24 * time.Hour) { |
| + if h, _ := l.Held(DayOf(d)); !h { |
| + t.Fatalf("seed %d step %d (%s): fleet %s not held for U%d on %s", seed, i, what, g, u.ID, DayOf(d)) |
| + } |
| + } |
| + } |
| + if closed[g] { |
| + if un, _ := l.UnsettledAdmissions(g); len(un) != 0 { |
| + t.Fatalf("seed %d step %d (%s): fleet %s closed with %d unsettled", seed, i, what, g, len(un)) |
| + } |
| + if _, latched := l.Latched(g); latched { |
| + t.Fatalf("seed %d step %d (%s): fleet %s closed with a latch", seed, i, what, g) |
| + } |
| + } |
| + } |
| + for p, s := range acked { |
| + peff, _ := l.Effective(p) |
| + inh, _ := l.Inherited(s) |
| + have := map[int64]bool{} |
| + for _, id := range inh { |
| + have[id] = true |
| + } |
| + for _, u := range peff { |
| + if !have[u.ID] { |
| + t.Fatalf("seed %d step %d (%s): %s acknowledged %s without inheriting U%d", seed, i, what, s, p, u.ID) |
| + } |
| + } |
| + if anc, _ := l.Ancestry(s); len(anc) != 0 && anc[0] == p { |
| + t.Fatalf("seed %d step %d (%s): %s still in %s's ancestry after acknowledgement", seed, i, what, p, s) |
| + } |
| + } |
| + } |
| + for i := 0; i < 60; i++ { |
| + now = now.Add(time.Duration(rnd.Intn(90)) * time.Minute) |
| + set(now) |
| + tip := "" |
| + if len(fleets) > 0 { |
| + tip = fleets[len(fleets)-1] |
| + } |
| + what := "" |
| + switch rnd.Intn(10) { |
| + case 0, 1: |
| + if tip == "" || closed[tip] || rnd.Intn(3) == 0 { |
| + id := string(rune('A' + len(fleets))) |
| + if err := l.OpenFleet(id, "box", int64(len(fleets)), []string{"o" + id}, tip, now); err != nil { |
| + t.Fatalf("seed %d: open %s: %v", seed, id, err) |
| + } |
| + fleets = append(fleets, id) |
| + what = "open " + id |
| + } |
| + case 2, 3: |
| + if tip != "" && !closed[tip] { |
| + tok := mustAdmit(t, l, tip, "o"+tip, "app", now) |
| + open[tip] = append(open[tip], tok) |
| + what = "admit on " + tip |
| + } |
| + case 4, 5: |
| + if tip != "" && len(open[tip]) > 0 { |
| + k := rnd.Intn(len(open[tip])) |
| + tok := open[tip][k] |
| + open[tip] = append(open[tip][:k], open[tip][k+1:]...) |
| + if rnd.Intn(4) == 0 { |
| + if err := l.Unissued(tok, now); err != nil { |
| + t.Fatalf("seed %d: unissued: %v", seed, err) |
| + } |
| + if err := l.EmptySettlement(tip, now); err != nil { |
| + t.Fatal(err) |
| + } |
| + } else { |
| + finishAdmission(t, l, tok) |
| + } |
| + what = "settle on " + tip |
| + } |
| + case 6: |
| + if tip != "" && !closed[tip] { |
| + if rnd.Intn(5) == 0 { |
| + discoverFail = func() error { return errDiskFull } |
| + } |
| + if _, err := l.Discover(Discovery{FleetID: tip, At: now, Proof: now.Add(-3 * time.Hour), LastCleanLook: now.Add(-time.Hour), Reason: "test"}); err != nil && !errors.Is(err, errDiskFull) { |
| + t.Fatalf("seed %d: discover on %s: %v", seed, tip, err) |
| + } |
| + discoverFail = nil |
| + what = "discover on " + tip |
| + } |
| + case 7: |
| + if tip != "" && !closed[tip] { |
| + if err := l.FleetClosing(tip); err != nil { |
| + t.Fatal(err) |
| + } |
| + if _, latched := l.Latched(tip); latched && rnd.Intn(2) == 0 { |
| + if _, err := l.Replay(tip); err != nil { |
| + t.Fatalf("seed %d: replay %s: %v", seed, tip, err) |
| + } |
| + } |
| + err := l.CloseFleet(tip, now) |
| + if err == nil { |
| + closed[tip] = true |
| + } else if !errors.Is(err, ErrFleetUnsettled) { |
| + t.Fatalf("seed %d: close %s: %v", seed, tip, err) |
| + } |
| + what = "close " + tip |
| + } |
| + case 8, 9: |
| + if len(fleets) >= 2 { |
| + s := fleets[rnd.Intn(len(fleets)-1)+1] |
| + f, _ := l.ReadFleet(s) |
| + err := l.Acknowledge(s, now) |
| + if err == nil { |
| + acked[f.Predecessor] = s |
| + } else if !errors.Is(err, ErrAcknowledgeNotReady) { |
| + t.Fatalf("seed %d: acknowledge %s: %v", seed, s, err) |
| + } |
| + what = "acknowledge " + s |
| + } |
| + } |
| + if what != "" { |
| + step(i, what) |
| + } |
| + } |
| + l.Close(t.Context(), nil) |
| + } |
| +} |