amadan
Plan Task 4's fifth slice: the gate and the fleet; the request gate returns an admission whose participants the middleware completes
lease: RequestGate returns an Admission. The gate middleware writes the transport's completion when the round trip returns; the meter writes the recorder's beside its own writes, in the worker: registered when Start lands (before the result is handed over), finished when Finish lands, abandoned with a late Start or a failed Finish, unrecorded when Start failed; the request path says so only when no worker took the call. A gate that admits with no admission is refused; ProofGate and Unadmitted are the gate of a fleet with no ledger. ops: the fleet's gate is the proof, then the admission: 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 a budget, the recheck under the mutex after the commit (unissued and an empty settlement on refusal), a definite failure cancelled, an ambiguous one read back by its token within a bound and otherwise left pending and counted, ErrAdmissionNotDurable past the budget with the writer resolving its own outcome. 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 with the inventory's owners, the replaced fleet as predecessor and the witness generation; closing at Close after the final look; transport-lost for every issued admission whose transport never finalised; closed within a bound and left closing past it; acknowledgement at Open and at every watch tick. The helper process does not exist yet: all of it runs in the agent. Tests: every send admitted and settled with 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 closed ledger; the transport lost at Close with the recorder pending; the middleware's participants in both option orders, a failed Start unrecorded, a nil admission refused.
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
10 files changed,
+988
−28
docs/superpowers/plans/2026-10-03-replicator-authority-plan.md+1 −1internal/lease/requestgate.go+87 −5internal/lease/requestgate_test.go+97 −10internal/lease/requestmeter.go+55 −11internal/ops/litestream_admission.go+371 −0internal/ops/litestream_admission_test.go+300 −0internal/ops/litestream_metered.go+51 −1internal/registry/observe.go+3 −0internal/usage/durabilityadmissions.go+20 −0internal/usage/durabilityledger.go+3 −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 425103823..fabecd7ec 100644 |
| --- a/docs/superpowers/plans/2026-10-03-replicator-authority-plan.md |
| +++ b/docs/superpowers/plans/2026-10-03-replicator-authority-plan.md |
| @@ -36,7 +36,7 @@ The admission model (`internal/ops/admissionmodel`) is the executable contract; |
| 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). |
| 4. **The fleet journal, construction backfill and acknowledgement.** `OpenFleet(predecessor, owners, witness)` writes the `open` row and the backfill of every interval its unacknowledged ancestry holds; `stale`, `closing`, `closed` as the design's transactions (closure carries a pending replay and refuses with an unsettled admission); `Acknowledge(successor)` root-first, copying the effective set into `inherited` and writing the holds for the closed ones. The ledger's reach query replaces the model's `reach`. Mirrors: `TestAncestryReachAndRootFirstAcknowledgement`, `TestHealthyChainContributesNothing`, the per-step inheritance invariant as a property test over a recorded schedule. |
| -5. **The gate and the fleet.** Contract (review nine, Opus P2-2; review ten, Opus P2-2): the ledger is never called with the fleet's discovery mutex held, and the pending counter the ledger reads is the lock-free `*atomic.Int64`, so a settlement under the ledger's mutex never waits on the mutex a gate recheck holds while it writes `unissued`; a registration that resolves decrements the counter before it tells the ledger (`EmptySettlement`, or the row's own transition), so the closure test sees the counter already down, and `CloseFleet` closes an open uncertainty on the same test regardless; the ledger's `Discover` does not check the fleet's journal state (the model returns unless open), which is this slice's to enforce before it calls. `MeteredFleet.admit` becomes the proof (unchanged) plus the admission: registration in the pending set under the discovery mutex, the ledger's `Admit`, the recheck, the token handed to the SDK middleware (`lease.RequestGate` returns the token; the meter's `Start` carries it; the finalize step writes `TransportDone`). `lookNow` is the discovery when it finds a change; `closeNow` is `closing` then `closed`. The helper process does not exist yet (Part B): the transport and the recorder run in the agent, so the lifecycle guard is the fleet's own, `transport-lost` is written at `Close` for a transport that never finalised, and `unrecorded` where the meter's `Start` failed. Mirrors: `TestHelperDeathAndRecorderGuard` (as the agent's close), `TestSendNeedsDurableIssuedAdmission` (the middleware refuses without a token), `TestReadBackOnlyAfterTheHandleCloses`, `TestBetweenCommitAndAnswer`. |
| +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`. |
| 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/lease/requestgate.go b/internal/lease/requestgate.go| index 6b3bfe345..cfea00884 100644 |
| --- a/internal/lease/requestgate.go |
| +++ b/internal/lease/requestgate.go |
| @@ -4,6 +4,7 @@ import ( |
| "context" |
| "errors" |
| "fmt" |
| + "time" |
| "github.com/aws/aws-sdk-go-v2/service/s3" |
| "github.com/aws/smithy-go/middleware" |
| @@ -24,13 +25,86 @@ import ( |
| // |
| // The gate is given the request's context and decides for itself how long |
| // it is willing to take. An attempt waiting on its gate has not been sent. |
| -type RequestGate func(ctx context.Context) error |
| +// |
| +// A gate that admits returns the attempt's Admission (the replicator's |
| +// 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 |
| +// 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 |
| +// answers with no admission and no error is a wiring error, and the |
| +// attempt is refused: a request is never sent without an admission. A |
| +// caller with no ledger to admit into says so with Unadmitted. |
| +type RequestGate func(ctx context.Context) (Admission, error) |
| + |
| +// Admission is one admitted attempt's two participants, written by the |
| +// middleware as each completes. Every method is called at most once per |
| +// 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(at time.Time) |
| + // RecorderStarted is the meter's Start landed: the recorder participant |
| + // is registered and will finish, abandon or be abandoned. |
| + RecorderStarted(at time.Time) |
| + // RecorderDone is the recorder's outcome: AdmissionFinished, |
| + // AdmissionAbandoned or AdmissionUnrecorded. |
| + RecorderDone(at time.Time, outcome string) |
| +} |
| + |
| +// The recorder participant's outcomes, as the ledger names them. |
| +const ( |
| + AdmissionFinished = "finished" |
| + AdmissionAbandoned = "abandoned" |
| + AdmissionUnrecorded = "unrecorded" |
| +) |
| + |
| +// Unadmitted is the admission of a gate with no ledger: the attempt is |
| +// sent with no admission row, and its completions go nowhere. A box whose |
| +// gates answer with it cannot attest that every write was admitted; it is |
| +// for the proof-only gate of a fleet built without a ledger, and for |
| +// tests. |
| +var Unadmitted Admission = unadmitted{} |
| + |
| +type unadmitted struct{} |
| + |
| +func (unadmitted) TransportDone(time.Time) {} |
| +func (unadmitted) RecorderStarted(time.Time) {} |
| +func (unadmitted) RecorderDone(time.Time, string) {} |
| + |
| +// ProofGate is a RequestGate from a proof alone (nil: the attempt may be |
| +// sent): it admits with Unadmitted. For a fleet with no ledger, and tests. |
| +func ProofGate(prove func(ctx context.Context) error) RequestGate { |
| + return func(ctx context.Context) (Admission, error) { |
| + if err := prove(ctx); err != nil { |
| + return nil, err |
| + } |
| + return Unadmitted, nil |
| + } |
| +} |
| // ErrRequestRefused: a request gate stopped an attempt before dispatch. |
| var ErrRequestRefused = errors.New("lease: replica request refused before dispatch") |
| +// ErrNoAdmission: a gate admitted without an admission, which is a wiring |
| +// error; the attempt is refused. |
| +var ErrNoAdmission = errors.New("lease: the request gate admitted without an admission") |
| + |
| var errRequestGateRequired = errors.New("replication_gate_required") |
| +type admissionKey struct{} |
| + |
| +// admissionOf is the attempt's admission, put in the context by the gate |
| +// middleware for the meter inside it; Unadmitted where there is no gate. |
| +func admissionOf(ctx context.Context) Admission { |
| + if a, ok := ctx.Value(admissionKey{}).(Admission); ok && a != nil { |
| + return a |
| + } |
| + return Unadmitted |
| +} |
| + |
| const ( |
| requestGateMiddlewareID = "CarlosReplicationRequestGate" |
| requestMeterMiddlewareID = "CarlosReplicationRequestMeter" |
| @@ -52,10 +126,18 @@ func requestGateOption(gate RequestGate) (func(*s3.Options), error) { |
| return fmt.Errorf("%w: two request gates on one client", ErrRequestRefused) |
| } |
| hook := middleware.DeserializeMiddlewareFunc(requestGateMiddlewareID, func(ctx context.Context, in middleware.DeserializeInput, next middleware.DeserializeHandler) (middleware.DeserializeOutput, middleware.Metadata, error) { |
| - if err := askRequestGate(ctx, gate); err != nil { |
| + adm, err := askRequestGate(ctx, gate) |
| + if err != nil { |
| return middleware.DeserializeOutput{}, middleware.Metadata{}, fmt.Errorf("%w: %w", ErrRequestRefused, err) |
| } |
| - return next.HandleDeserialize(ctx, in) |
| + if adm == nil { |
| + 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 |
| + // or not; the transport participant is complete |
| + adm.TransportDone(time.Now().UTC()) |
| + return out, md, err |
| }) |
| if _, metered := stack.Deserialize.Get(requestMeterMiddlewareID); metered { |
| return stack.Deserialize.Insert(hook, requestMeterMiddlewareID, middleware.Before) |
| @@ -66,10 +148,10 @@ func requestGateOption(gate RequestGate) (func(*s3.Options), error) { |
| } |
| // askRequestGate asks the gate, and reads a gate that panics as a refusal. |
| -func askRequestGate(ctx context.Context, gate RequestGate) (err error) { |
| +func askRequestGate(ctx context.Context, gate RequestGate) (adm Admission, err error) { |
| defer func() { |
| if recover() != nil { |
| - err = errors.New("request gate panicked") |
| + adm, err = nil, errors.New("request gate panicked") |
| } |
| }() |
| return gate(ctx) |
diff --git a/internal/lease/requestgate_test.go b/internal/lease/requestgate_test.go| index b5b758d94..f8f0924cd 100644 |
| --- a/internal/lease/requestgate_test.go |
| +++ b/internal/lease/requestgate_test.go |
| @@ -7,6 +7,7 @@ import ( |
| "net/http" |
| "net/http/httptest" |
| "path/filepath" |
| + "sync" |
| "sync/atomic" |
| "testing" |
| "time" |
| @@ -17,7 +18,7 @@ import ( |
| "github.com/aws/aws-sdk-go-v2/service/s3" |
| ) |
| -func admitEveryRequest(context.Context) error { return nil } |
| +func admitEveryRequest(context.Context) (Admission, error) { return Unadmitted, nil } |
| // gatedTestClient is a real S3 client over a test server with a gate and a |
| // meter on it, the two options applied in the order given. |
| @@ -64,12 +65,12 @@ func TestRequestGateStopsAnAttemptBeforeItIsSentOrCounted(t *testing.T) { |
| client := gatedTestClient(t, func(w http.ResponseWriter, r *http.Request) { |
| received.Add(1) |
| w.WriteHeader(http.StatusOK) |
| - }, func(context.Context) error { |
| + }, func(context.Context) (Admission, error) { |
| asked.Add(1) |
| if refuse.Load() { |
| - return stale |
| + return nil, stale |
| } |
| - return nil |
| + return Unadmitted, nil |
| }, recorder, gateFirst) |
| err := meterPut(context.Background(), client) |
| if !errors.Is(err, ErrRequestRefused) || !errors.Is(err, stale) { |
| @@ -113,12 +114,12 @@ func TestRequestGateIsAskedAgainForARetry(t *testing.T) { |
| refuse.Store(true) |
| w.WriteHeader(http.StatusInternalServerError) |
| _, _ = w.Write([]byte(`<Error><Code>InternalError</Code></Error>`)) |
| - }, func(context.Context) error { |
| + }, func(context.Context) (Admission, error) { |
| asked.Add(1) |
| if refuse.Load() { |
| - return errors.New("closed") |
| + return nil, errors.New("closed") |
| } |
| - return nil |
| + return Unadmitted, nil |
| }, recorder, true) |
| err := meterPut(context.Background(), client) |
| if !errors.Is(err, ErrRequestRefused) { |
| @@ -141,7 +142,7 @@ func TestRequestGateThatPanicsRefuses(t *testing.T) { |
| client := gatedTestClient(t, func(w http.ResponseWriter, r *http.Request) { |
| received.Add(1) |
| w.WriteHeader(http.StatusOK) |
| - }, func(context.Context) error { panic("gate") }, recorder, true) |
| + }, func(context.Context) (Admission, error) { panic("gate") }, recorder, true) |
| if err := meterPut(context.Background(), client); !errors.Is(err, ErrRequestRefused) { |
| t.Fatalf("a panicking gate returned %v", err) |
| } |
| @@ -189,12 +190,12 @@ func TestMeteredStoreClientIsStoppedByItsGate(t *testing.T) { |
| one, two := &memoryRequestRecorder{}, &memoryRequestRecorder{} |
| var closed atomic.Bool |
| store, err := NewMeteredStore("test-bucket", "eu-west-1", []MeteredDB{ |
| - {Path: filepath.Join(dir, "one.db"), Prefix: "replicas/bab/one", Meter: one, MetaDir: filepath.Join(dir, ".replicator", "one"), Gate: func(context.Context) error { |
| + {Path: filepath.Join(dir, "one.db"), Prefix: "replicas/bab/one", Meter: one, MetaDir: filepath.Join(dir, ".replicator", "one"), Gate: ProofGate(func(context.Context) error { |
| if closed.Load() { |
| return errors.New("closed") |
| } |
| return nil |
| - }}, |
| + })}, |
| {Path: filepath.Join(dir, "two.db"), Prefix: "replicas/bac/two", Meter: two, MetaDir: filepath.Join(dir, ".replicator", "two"), Gate: admitEveryRequest}, |
| }) |
| if err != nil { |
| @@ -223,3 +224,89 @@ func TestMeteredStoreClientIsStoppedByItsGate(t *testing.T) { |
| t.Fatalf("%d requests reached the bucket, want 2: %v", got, f.requests()) |
| } |
| } |
| + |
| +// recordingAdmission remembers what the middleware wrote for one attempt. |
| +type recordingAdmission struct { |
| + mu sync.Mutex |
| + transport int |
| + started int |
| + outcomes []string |
| +} |
| + |
| +func (r *recordingAdmission) TransportDone(time.Time) { r.mu.Lock(); r.transport++; r.mu.Unlock() } |
| +func (r *recordingAdmission) RecorderStarted(time.Time) { r.mu.Lock(); r.started++; r.mu.Unlock() } |
| +func (r *recordingAdmission) RecorderDone(_ time.Time, how string) { |
| + r.mu.Lock() |
| + r.outcomes = append(r.outcomes, how) |
| + r.mu.Unlock() |
| +} |
| +func (r *recordingAdmission) snapshot() (int, int, []string) { |
| + r.mu.Lock() |
| + defer r.mu.Unlock() |
| + return r.transport, r.started, append([]string(nil), r.outcomes...) |
| +} |
| + |
| +// failingStartRecorder is a recorder whose Start never lands. |
| +type failingStartRecorder struct{ requestRecorder } |
| + |
| +func (failingStartRecorder) Start(string, time.Time) (uint64, error) { |
| + return 0, errors.New("ledger: no") |
| +} |
| + |
| +// The admission's two participants (Part C): the gate middleware writes |
| +// the transport's completion when the round trip returns; the meter writes |
| +// the recorder's registration when its Start lands and its outcome when |
| +// its Finish lands, in the worker; a Start that fails is unrecorded, with |
| +// no registration; a gate that admits with no admission is a refusal; the |
| +// proof-only gate admits with Unadmitted. |
| +func TestRequestGateCompletesTheAdmission(t *testing.T) { |
| + for _, gateFirst := range []bool{true, false} { |
| + adm := &recordingAdmission{} |
| + recorder := &memoryRequestRecorder{} |
| + var received atomic.Int32 |
| + client := gatedTestClient(t, func(w http.ResponseWriter, r *http.Request) { |
| + received.Add(1) |
| + w.WriteHeader(http.StatusOK) |
| + }, func(context.Context) (Admission, error) { return adm, nil }, recorder, gateFirst) |
| + if err := meterPut(context.Background(), client); err != nil { |
| + t.Fatalf("gateFirst=%v: %v", gateFirst, err) |
| + } |
| + waitFor(t, "the admission's participants", func() bool { tr, st, out := adm.snapshot(); return tr == 1 && st == 1 && len(out) == 1 }) |
| + if tr, st, out := adm.snapshot(); tr != 1 || st != 1 || len(out) != 1 || out[0] != AdmissionFinished { |
| + t.Fatalf("gateFirst=%v: transport %d started %d outcomes %v", gateFirst, tr, st, out) |
| + } |
| + } |
| + // a Start that fails: unrecorded, never registered; the request is sent |
| + adm := &recordingAdmission{} |
| + var received atomic.Int32 |
| + client := gatedTestClient(t, func(w http.ResponseWriter, r *http.Request) { |
| + received.Add(1) |
| + w.WriteHeader(http.StatusOK) |
| + }, func(context.Context) (Admission, error) { return adm, nil }, failingStartRecorder{&memoryRequestRecorder{}}, true) |
| + if err := meterPut(context.Background(), client); err != nil { |
| + t.Fatal(err) |
| + } |
| + waitFor(t, "the unrecorded outcome", func() bool { tr, _, out := adm.snapshot(); return tr == 1 && len(out) == 1 }) |
| + if tr, st, out := adm.snapshot(); tr != 1 || st != 0 || len(out) != 1 || out[0] != AdmissionUnrecorded || received.Load() != 1 { |
| + t.Fatalf("a failed Start: transport %d started %d outcomes %v received %d", tr, st, out, received.Load()) |
| + } |
| + // no admission: refused before the send |
| + received.Store(0) |
| + client = gatedTestClient(t, func(w http.ResponseWriter, r *http.Request) { |
| + received.Add(1) |
| + w.WriteHeader(http.StatusOK) |
| + }, func(context.Context) (Admission, error) { return nil, nil }, &memoryRequestRecorder{}, true) |
| + if err := meterPut(context.Background(), client); !errors.Is(err, ErrNoAdmission) || received.Load() != 0 { |
| + t.Fatalf("a gate with no admission: %v, %d received", err, received.Load()) |
| + } |
| + // the proof-only gate |
| + proved := 0 |
| + gate := ProofGate(func(context.Context) error { proved++; return nil }) |
| + if a, err := gate(context.Background()); err != nil || a != Unadmitted || proved != 1 { |
| + t.Fatalf("ProofGate: %v %v %d", a, err, proved) |
| + } |
| + refused := errors.New("stale") |
| + if a, err := ProofGate(func(context.Context) error { return refused })(context.Background()); !errors.Is(err, refused) || a != nil { |
| + t.Fatalf("ProofGate refusing: %v %v", a, err) |
| + } |
| +} |
diff --git a/internal/lease/requestmeter.go b/internal/lease/requestmeter.go| index ac4baf2a2..319daf928 100644 |
| --- a/internal/lease/requestmeter.go |
| +++ b/internal/lease/requestmeter.go |
| @@ -134,21 +134,46 @@ func meterAttempt(ctx context.Context, in middleware.DeserializeInput, next midd |
| a.release = recorder.Hold(started) |
| a.refs.Store(1) |
| defer a.done() |
| + adm := admissionOf(ctx) |
| operation := middleware.GetOperationName(ctx) |
| if operation == "" { |
| a.fail() |
| } |
| - token, recordErr := a.call(slots, func() (uint64, error) { return recorder.Start(operation, started) }, func(token uint64, err error, late bool) { |
| + // The recorder participant of the attempt's admission (Part C) is |
| + // written in the worker, beside the meter's own write, so the request |
| + // path never waits on it and the registration is on disk before any |
| + // outcome: a Start that lands registers the recorder (before the |
| + // result is handed to the request path, so a Finish cannot overtake |
| + // it); a Start that lands late is abandoned, and so is its |
| + // registration; a Start that fails in the worker is unrecorded there; |
| + // one that never reached a worker (no slot) is unrecorded by the |
| + // request path. The participant is never synthesised while its worker |
| + // can still run. |
| + token, recordErr, ran := a.call(slots, func() (uint64, error) { |
| + token, err := recorder.Start(operation, started) |
| + if err == nil { |
| + adm.RecorderStarted(started) |
| + } |
| + return token, err |
| + }, func(token uint64, err error, late bool) { |
| // A Start that committed after the request path stopped waiting |
| // for it: the row exists, nobody will finish it. Abandon it here, |
| // in the worker, so the request path never waits on a second |
| // database write. |
| - if late && err == nil { |
| - recorder.Abandon(token, time.Now().UTC()) |
| + now := time.Now().UTC() |
| + switch { |
| + case late && err == nil: |
| + recorder.Abandon(token, now) |
| + adm.RecorderDone(now, AdmissionAbandoned) |
| + case err != nil: |
| + adm.RecorderDone(now, AdmissionUnrecorded) |
| } |
| }) |
| if recordErr != nil { |
| a.fail() |
| + if !ran { |
| + adm.RecorderDone(time.Now().UTC(), AdmissionUnrecorded) // no worker took the call: nothing else will say so |
| + } |
| } |
| // Count every attempt, including Err != nil: a partial write may |
| // already have reached S3, so an ambiguous write can't be discounted. |
| @@ -181,15 +206,31 @@ func meterAttempt(ctx context.Context, in middleware.DeserializeInput, next midd |
| return out, metadata, err |
| } |
| result.Incompletes = a.seal() |
| - if _, finishErr := a.call(slots, func() (uint64, error) { return 0, recorder.Finish(token, result) }, func(_ uint64, err error, _ bool) { |
| + if _, finishErr, ran := a.call(slots, func() (uint64, error) { |
| + err := recorder.Finish(token, result) |
| + if err == nil { |
| + adm.RecorderDone(time.Now().UTC(), AdmissionFinished) |
| + } |
| + return 0, err |
| + }, func(_ uint64, err error, _ bool) { |
| // A Finish that failed, late or not, leaves the row unfinished: |
| // mark it unknown now rather than at the next open. A Finish that |
| // merely arrived late but committed needs nothing. |
| if err != nil { |
| - recorder.Abandon(token, time.Now().UTC()) |
| + now := time.Now().UTC() |
| + recorder.Abandon(token, now) |
| + adm.RecorderDone(now, AdmissionAbandoned) |
| } |
| }); finishErr != nil { |
| recorder.Incomplete(started) |
| + if !ran { |
| + // no worker took the Finish: the row stays started, nobody |
| + // will finish it, and nothing else will say so (a worker that |
| + // ran and failed, late or not, says so itself) |
| + now := time.Now().UTC() |
| + recorder.Abandon(token, now) |
| + adm.RecorderDone(now, AdmissionAbandoned) |
| + } |
| } |
| return out, metadata, err |
| } |
| @@ -200,14 +241,17 @@ func meterAttempt(ctx context.Context, in middleware.DeserializeInput, next midd |
| // getting the budget to queue plus the budget to run. The worker always |
| // runs settle with what fn did once the result has been handed over (or |
| // once the request path has stopped waiting: late=true), and holds a |
| -// reference on the attempt until then. |
| -func (a *meteredAttempt) call(slots chan struct{}, fn func() (uint64, error), settle func(token uint64, err error, late bool)) (uint64, error) { |
| +// reference on the attempt until then. ran reports that a worker took |
| +// the call (and has run, or will run, settle); false with an error means |
| +// no worker ran at all (no slot within the budget), and whatever the |
| +// call was for is the request path's to say. |
| +func (a *meteredAttempt) call(slots chan struct{}, fn func() (uint64, error), settle func(token uint64, err error, late bool)) (token uint64, err error, ran bool) { |
| timer := time.NewTimer(requestMeterBudget) |
| defer timer.Stop() |
| select { |
| case slots <- struct{}{}: |
| case <-timer.C: |
| - return 0, errRequestMeterUnavailable |
| + return 0, errRequestMeterUnavailable, false |
| } |
| type result struct { |
| token uint64 |
| @@ -238,13 +282,13 @@ func (a *meteredAttempt) call(slots chan struct{}, fn func() (uint64, error), se |
| }() |
| select { |
| case value := <-done: |
| - return value.token, value.err |
| + return value.token, value.err, true |
| case <-timer.C: |
| if claimed.CompareAndSwap(false, true) { |
| - return 0, errRequestMeterUnavailable |
| + return 0, errRequestMeterUnavailable, true |
| } |
| // The worker claimed first: its result is already in the channel. |
| value := <-done |
| - return value.token, value.err |
| + return value.token, value.err, true |
| } |
| } |
diff --git a/internal/ops/litestream_admission.go b/internal/ops/litestream_admission.go| new file mode 100644 |
| index 000000000..dfaedd256 |
| --- /dev/null |
| +++ b/internal/ops/litestream_admission.go |
| @@ -0,0 +1,371 @@ |
| +package ops |
| + |
| +import ( |
| + "context" |
| + "crypto/rand" |
| + "encoding/hex" |
| + "errors" |
| + "fmt" |
| + "log" |
| + "sort" |
| + "strconv" |
| + "sync/atomic" |
| + "time" |
| + |
| + "github.com/carlosframework/platform/internal/lease" |
| + durability "github.com/carlosframework/platform/internal/usage" |
| +) |
| + |
| +// The fleet's admission (the replicator's authority design, Part C, plan |
| +// Task 4's fifth slice): no request is sent until a row naming it is |
| +// durable. The request gate is the proof (admit: the witness unchanged, |
| +// the fleet not stale) followed by an admission: a registration in the |
| +// fleet's pending set under the discovery mutex, the row committed in the |
| +// durability ledger within a budget, and a recheck under the mutex after |
| +// the commit (stale, or a discovery begun since registration: the row is |
| +// unissued and the request refused). A discovery, under the same mutex, |
| +// counts the pending set. The admission's two participants, the transport |
| +// and the recorder, are written by the SDK middleware (lease.Admission). |
| +// The helper process does not exist yet (Part B): the transport and the |
| +// recorder run in the agent, so transport-lost is written at Close for a |
| +// transport that never finalised and unrecorded where the meter's Start |
| +// failed. Everything here holds with the fleet's discovery mutex released |
| +// before the ledger is called (the plan's contract): the mutex decides, |
| +// the ledger is told afterwards. |
| + |
| +// ErrAdmissionNotDurable: the admission's commit did not return within the |
| +// gate's budget. The request is refused; the commit, if it lands, is |
| +// marked unissued by the goroutine that awaited it. |
| +var ErrAdmissionNotDurable = errors.New("ops: the replica request's admission was not durable within the budget") |
| + |
| +// admissionBudget bounds the wait for an admission's commit on the |
| +// request path. Past it the request is refused (litestream sees a failed |
| +// upload and does not advance), and the writer's outcome is resolved by |
| +// 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 |
| + token string |
| + done atomic.Bool // the transport finalised |
| +} |
| + |
| +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)) |
| +} |
| + |
| +func (a *fleetAdmission) RecorderStarted(at time.Time) { |
| + a.f.ledgerWrite("recorder started", 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)) |
| +} |
| + |
| +// ledgerWrite notes a participant write the ledger refused. The row stays |
| +// as it was, unsettled, barring its days: conservative, and counted so a |
| +// test or a status line can see it. |
| +func (f *MeteredFleet) ledgerWrite(what string, err error) { |
| + if err == nil { |
| + return |
| + } |
| + f.ledgerErrors.Add(1) |
| + log.Printf("replicator: fleet %s: %s: %v", f.id, what, err) |
| +} |
| + |
| +// forget drops an issued admission from the live set (its transport has |
| +// finalised). |
| +func (f *MeteredFleet) forget(token string) { |
| + f.liveMu.Lock() |
| + delete(f.live, token) |
| + f.liveMu.Unlock() |
| +} |
| + |
| +// gateFor is the request gate of one database: the proof, then the |
| +// admission for its owner. With no ledger it is the proof alone, admitting |
| +// with lease.Unadmitted (a fleet built without a ledger: tests, and a box |
| +// that does not attest its admissions). |
| +func (f *MeteredFleet) gateFor(owned OwnedDB) lease.RequestGate { |
| + if f.ledger == nil { |
| + return lease.ProofGate(f.admit) |
| + } |
| + return func(ctx context.Context) (lease.Admission, error) { |
| + return f.gate(ctx, owned) |
| + } |
| +} |
| + |
| +// admissionBeforeCommit is a test seam between the registration and the |
| +// row's commit; admissionDelay delays the writer, for the budget's test. |
| +var admissionBeforeCommit func(f *MeteredFleet) |
| +var admissionDelay func() |
| + |
| +func (f *MeteredFleet) gate(ctx context.Context, owned OwnedDB) (lease.Admission, error) { |
| + if err := f.admit(ctx); err != nil { |
| + return nil, err |
| + } |
| + // registration, under the discovery mutex: a discovery in progress or |
| + // begun before this point counts it or has refused it |
| + token, err := durability.NewAdmissionToken() |
| + if err != nil { |
| + return nil, err |
| + } |
| + at := time.Now().UTC() |
| + f.disc.Lock() |
| + if f.closed.Load() || f.stale.Load() || f.inDiscovery || f.latched.Load() { |
| + f.disc.Unlock() |
| + return nil, ErrOwnershipStale |
| + } |
| + f.pending[token] = at |
| + f.pendingN.Add(1) |
| + epoch := f.epoch |
| + f.disc.Unlock() |
| + if admissionBeforeCommit != nil { |
| + admissionBeforeCommit(f) |
| + } |
| + // the writer: one goroutine owning the commit; the request path waits |
| + // 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) |
| + go func() { |
| + if admissionDelay != nil { |
| + admissionDelay() |
| + } |
| + done <- f.ledger.Admit(token, f.id, owned.Account, owned.App, at) |
| + }() |
| + timer := time.NewTimer(admissionBudget) |
| + defer timer.Stop() |
| + select { |
| + case err := <-done: |
| + return f.resolve(token, epoch, err, false) |
| + case <-timer.C: |
| + go func() { |
| + f.resolve(token, epoch, <-done, true) //nolint:errcheck |
| + }() |
| + return nil, ErrAdmissionNotDurable |
| + } |
| +} |
| + |
| +// resolve is the registration's outcome once the writer has returned |
| +// (the design's table): committed, the recheck under the mutex issues the |
| +// token or marks the row unissued; a definite failure cancels the |
| +// registration; an ambiguous one is read back by its token (the writer |
| +// has returned, so nothing it failed to find can be committed afterwards) |
| +// 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) { |
| + 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 { |
| + f.cancel(token) |
| + f.ledgerWrite("empty settlement", 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 |
| + } |
| + // committed: the row carries the responsibility; the recheck |
| + f.disc.Lock() |
| + 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() |
| + if refused { |
| + f.ledgerWrite("unissued", f.ledger.Unissued(token, now)) |
| + f.ledgerWrite("empty settlement", f.ledger.EmptySettlement(f.id, now)) |
| + if late { |
| + return nil, ErrAdmissionNotDurable |
| + } |
| + return nil, ErrOwnershipStale |
| + } |
| + adm := &fleetAdmission{f: f, token: token} |
| + f.liveMu.Lock() |
| + f.live[token] = adm |
| + f.liveMu.Unlock() |
| + return adm, nil |
| +} |
| + |
| +// cancel is a conclusive cancellation: the registration leaves the |
| +// pending set. |
| +func (f *MeteredFleet) cancel(token string) { |
| + f.disc.Lock() |
| + if _, ok := f.pending[token]; ok { |
| + delete(f.pending, token) |
| + f.pendingN.Add(-1) |
| + } |
| + 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) |
| + } |
| +} |
| + |
| +// discover tells the ledger the fleet's authority was found changed, once |
| +// per fleet, under the discovery mutex's flag: registrations made while it |
| +// runs are refused at their recheck, and the ones pending are counted. A |
| +// discovery whose write fails is latched in the ledger; the fleet refuses |
| +// requests meanwhile (it is stale regardless) and the watch replays it. |
| +func (f *MeteredFleet) discover(at time.Time) { |
| + if f.ledger == nil || !f.discovered.CompareAndSwap(false, true) { |
| + return |
| + } |
| + f.disc.Lock() |
| + f.inDiscovery = true |
| + f.epoch++ |
| + pending := make([]time.Time, 0, len(f.pending)) |
| + for _, t := range f.pending { |
| + pending = append(pending, t) |
| + } |
| + 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() |
| + _, 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) |
| + log.Printf("replicator: fleet %s: discovery latched: %v", f.id, err) |
| + } |
| + f.disc.Lock() |
| + f.inDiscovery = false |
| + f.disc.Unlock() |
| +} |
| + |
| +// replayLatched replays a latched discovery, from the watch and at Close. |
| +func (f *MeteredFleet) replayLatched() { |
| + if f.ledger == nil || !f.latched.Load() { |
| + return |
| + } |
| + if _, err := f.ledger.Replay(f.id); err == nil { |
| + f.latched.Store(false) |
| + } |
| +} |
| + |
| +// acknowledge is the root-first acknowledgement of the fleet this one |
| +// replaces, tried at Open and at every watch tick until it lands. |
| +func (f *MeteredFleet) acknowledge() { |
| + if f.ledger == nil || f.acked.Load() { |
| + return |
| + } |
| + err := f.ledger.Acknowledge(f.id, time.Now().UTC()) |
| + if err == nil { |
| + f.acked.Store(true) |
| + return |
| + } |
| + if !errors.Is(err, durability.ErrAcknowledgeNotReady) { |
| + f.ledgerWrite("acknowledge", err) |
| + } |
| +} |
| + |
| +// journalClose is the journal's side of Close: closing once the store has |
| +// 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. |
| +func (f *MeteredFleet) journalClose() { |
| + if f.ledger == nil || f.id == "" { |
| + return |
| + } |
| + now := time.Now().UTC() |
| + f.ledgerWrite("closing", f.ledger.FleetClosing(f.id)) |
| + f.liveMu.Lock() |
| + lost := make([]*fleetAdmission, 0, len(f.live)) |
| + for _, a := range f.live { |
| + lost = append(lost, a) |
| + } |
| + 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)) |
| + } |
| + } |
| + f.replayLatched() |
| + deadline := time.Now().Add(closeSettleWait) |
| + for { |
| + err := f.ledger.CloseFleet(f.id, time.Now().UTC()) |
| + if err == nil { |
| + f.ledger.SetPendingCounter(f.id, nil) |
| + return |
| + } |
| + if !errors.Is(err, durability.ErrFleetUnsettled) || time.Now().After(deadline) { |
| + log.Printf("replicator: fleet %s stays closing: %v", f.id, err) |
| + return |
| + } |
| + time.Sleep(closeSettlePoll) |
| + } |
| +} |
| + |
| +// closeSettleWait bounds Close's wait for the last admissions to settle; |
| +// past it the journal row stays closing. closeSettlePoll paces the wait. |
| +var ( |
| + closeSettleWait = 10 * time.Second |
| + closeSettlePoll = 50 * time.Millisecond |
| +) |
| + |
| +// newFleetID names a fleet: the box, the instant and a random suffix. |
| +func newFleetID(box string) (string, error) { |
| + var b [4]byte |
| + if _, err := rand.Read(b[:]); err != nil { |
| + return "", err |
| + } |
| + return box + "-" + strconv.FormatInt(time.Now().UnixNano(), 36) + "-" + hex.EncodeToString(b[:]), nil |
| +} |
| + |
| +// fleetOwners is the owners an inventory names, each once, sorted. |
| +func fleetOwners(owned []OwnedDB) []string { |
| + seen := map[string]bool{} |
| + var out []string |
| + for _, o := range owned { |
| + k := o.Account + "/" + o.App |
| + if !seen[k] { |
| + seen[k] = true |
| + out = append(out, k) |
| + } |
| + } |
| + sort.Strings(out) |
| + return out |
| +} |
| + |
| +// FleetID is the fleet's journal id ("" without a ledger). |
| +func (f *MeteredFleet) FleetID() string { return f.id } |
| + |
| +// LedgerErrors is how many participant or journal writes the ledger |
| +// refused for this fleet. |
| +func (f *MeteredFleet) LedgerErrors() int64 { return f.ledgerErrors.Load() } |
diff --git a/internal/ops/litestream_admission_test.go b/internal/ops/litestream_admission_test.go| new file mode 100644 |
| index 000000000..33eb9a269 |
| --- /dev/null |
| +++ b/internal/ops/litestream_admission_test.go |
| @@ -0,0 +1,300 @@ |
| +package ops |
| + |
| +import ( |
| + "context" |
| + "errors" |
| + "os" |
| + "path/filepath" |
| + "testing" |
| + "time" |
| + |
| + "github.com/carlosframework/platform/internal/lease" |
| + "github.com/carlosframework/platform/internal/registry" |
| + durability "github.com/carlosframework/platform/internal/usage" |
| +) |
| + |
| +func openLedger(t *testing.T) *durability.DurabilityLedger { |
| + t.Helper() |
| + l, err := durability.OpenDurabilityLedger(t.TempDir(), "box1", time.Now, durability.DurabilityLedgerOptions{}) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + t.Cleanup(func() { l.Close(context.Background(), nil) }) |
| + return l |
| +} |
| + |
| +func waitUntil(t *testing.T, what string, cond func() bool) { |
| + t.Helper() |
| + deadline := time.Now().Add(5 * time.Second) |
| + for !cond() { |
| + if time.Now().After(deadline) { |
| + t.Fatalf("timed out waiting for %s", what) |
| + } |
| + time.Sleep(5 * time.Millisecond) |
| + } |
| +} |
| + |
| +// Part C's gate on the fleet: every send is admitted first (a durable row |
| +// for the database's owner), the transport and the recorder settle it, no |
| +// day is held or barred afterwards; the journal has the fleet open, then |
| +// closed at Close; a replacement acknowledges it at Open and a healthy |
| +// chain holds nothing. |
| +func TestMeteredFleetAdmitsEverySendAndJournals(t *testing.T) { |
| + 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 |
| + fleet, err := MeteredFleetStore(o) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if fleet.FleetID() == "" { |
| + t.Fatal("no journal id") |
| + } |
| + row, err := ledger.ReadFleet(fleet.FleetID()) |
| + if err != nil || row.State != durability.FleetOpen || row.Predecessor != "" || len(row.Owners) != 2 { |
| + t.Fatalf("the open row: %+v %v", row, err) |
| + } |
| + if err := send(t, fleet, db, 2); err != nil { |
| + t.Fatalf("a send: %v", err) |
| + } |
| + waitUntil(t, "the admission to settle", func() bool { |
| + open, _ := ledger.UnsettledAdmissions(fleet.FleetID()) |
| + return len(open) == 0 |
| + }) |
| + var settled []durability.Admission |
| + for _, tok := range admissionTokens(t, ledger, fleet.FleetID()) { |
| + a, _ := ledger.ReadAdmission(tok) |
| + settled = append(settled, a) |
| + } |
| + if len(settled) != 1 || !settled[0].Settled() || settled[0].How != durability.AdmissionFinished || settled[0].Account != "bac" || settled[0].App != "one" { |
| + t.Fatalf("the send's admission: %+v", settled) |
| + } |
| + today := durability.DayOf(time.Now()) |
| + if c, _ := ledger.Snapshot(today); c.Barred != 0 || c.Held != 0 { |
| + t.Fatalf("after a settled admission on an unchanged fleet: %+v", c) |
| + } |
| + if fleet.LedgerErrors() != 0 { |
| + t.Fatalf("%d ledger writes refused", fleet.LedgerErrors()) |
| + } |
| + if err := fleet.Close(); err != nil { |
| + t.Fatal(err) |
| + } |
| + if row, _ := ledger.ReadFleet(fleet.FleetID()); row.State != durability.FleetClosed { |
| + t.Fatalf("after Close: %s", row.State) |
| + } |
| + // a replacement: linked at construction, acknowledging at Open |
| + o.Replaces = fleet |
| + next, err := MeteredFleetStore(o) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + defer next.Close() |
| + if row, _ := ledger.ReadFleet(next.FleetID()); row.Predecessor != fleet.FleetID() { |
| + t.Fatalf("the replacement's row: %+v", row) |
| + } |
| + if err := next.Open(context.Background()); err != nil { |
| + t.Fatal(err) |
| + } |
| + if row, _ := ledger.ReadFleet(fleet.FleetID()); row.AcknowledgedBy != next.FleetID() { |
| + t.Fatalf("the predecessor after the replacement opened: acknowledged by %q", row.AcknowledgedBy) |
| + } |
| + if eff, _ := ledger.Effective(next.FleetID()); len(eff) != 0 { |
| + t.Fatalf("a healthy chain's effective set: %v", eff) |
| + } |
| + if h, _ := ledger.Held(today); h { |
| + t.Fatal("a healthy chain held the day") |
| + } |
| +} |
| + |
| +func admissionTokens(t *testing.T, l *durability.DurabilityLedger, fleet string) []string { |
| + t.Helper() |
| + // the ledger exposes unsettled rows; settled ones are read by token |
| + // through the fleet's own records in these tests, so the tokens are |
| + // collected from the admissions table directly |
| + return l.AdmissionTokensForTest(fleet) |
| +} |
| + |
| +// Mirrors TestCommitAfterDiscoveryIsUnissued and TestBetweenCommitAndAnswer: |
| +// a change found between a registration and its commit is a discovery |
| +// that counts the registration; the recheck after the commit finds the |
| +// fleet stale and the row is unissued, the request refused; the |
| +// uncertainty reaches the registration's instant and closes with the |
| +// empty settlement; one discovery per fleet. |
| +func TestMeteredFleetDiscoveryRefusesTheCommitAfterIt(t *testing.T) { |
| + 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 |
| + fleet, err := MeteredFleetStore(o) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + defer fleet.Close() |
| + var registeredAt time.Time |
| + admissionBeforeCommit = func(f *MeteredFleet) { |
| + f.disc.Lock() |
| + for _, at := range f.pending { |
| + registeredAt = at |
| + } |
| + f.disc.Unlock() |
| + if err := os.Chmod(filepath.Join(state, "data"), 0o700); err != nil { |
| + t.Error(err) |
| + } |
| + f.lookAgain() // the discovery, counting the pending registration |
| + } |
| + t.Cleanup(func() { admissionBeforeCommit = nil }) |
| + err = send(t, fleet, db, 2) |
| + admissionBeforeCommit = nil |
| + if !errors.Is(err, lease.ErrRequestRefused) || !errors.Is(err, ErrOwnershipStale) { |
| + t.Fatalf("the send after a discovery between registration and commit: %v", err) |
| + } |
| + toks := admissionTokens(t, ledger, fleet.FleetID()) |
| + if len(toks) != 1 { |
| + t.Fatalf("%d admissions, want the one unissued", len(toks)) |
| + } |
| + a, _ := ledger.ReadAdmission(toks[0]) |
| + if !a.Settled() || a.How != durability.AdmissionUnissued { |
| + t.Fatalf("the admission after the recheck: %+v", a) |
| + } |
| + eff, _ := ledger.Effective(fleet.FleetID()) |
| + if len(eff) != 1 || eff[0].Open() || eff[0].From.After(registeredAt.Add(-durability.AdmissionMargin)) { |
| + t.Fatalf("the uncertainty: %+v (registered at %v)", eff, registeredAt) |
| + } |
| + if h, _ := ledger.Held(durability.DayOf(time.Now())); !h { |
| + t.Fatal("the discovery held no day") |
| + } |
| + if fleet.pendingN.Load() != 0 { |
| + t.Fatalf("pending count %d after the registration resolved", fleet.pendingN.Load()) |
| + } |
| + // one discovery per fleet: a later finding on the same fleet (another |
| + // change, the final look at Close) tells the ledger nothing new |
| + fleet.discover(time.Now()) |
| + fleet.lookAgain() |
| + if eff, _ := ledger.Effective(fleet.FleetID()); len(eff) != 1 { |
| + t.Fatalf("a second finding on the fleet opened a second uncertainty: %v", eff) |
| + } |
| + if err := send(t, fleet, db, 3); !errors.Is(err, ErrOwnershipStale) { |
| + t.Fatalf("a send on a stale fleet: %v", err) |
| + } |
| +} |
| + |
| +// 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. |
| +func TestMeteredFleetAdmissionBudgetAndFailure(t *testing.T) { |
| + 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 |
| + fleet, err := MeteredFleetStore(o) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + defer fleet.Close() |
| + oldBudget := admissionBudget |
| + admissionBudget = 50 * time.Millisecond |
| + t.Cleanup(func() { admissionBudget = oldBudget }) |
| + admissionDelay = func() { time.Sleep(200 * time.Millisecond) } |
| + err = send(t, fleet, db, 2) |
| + admissionDelay = nil |
| + if !errors.Is(err, ErrAdmissionNotDurable) { |
| + t.Fatalf("a commit past the budget: %v", err) |
| + } |
| + waitUntil(t, "the late commit to be unissued", func() bool { |
| + toks := admissionTokens(t, ledger, fleet.FleetID()) |
| + if len(toks) != 1 { |
| + return false |
| + } |
| + a, _ := ledger.ReadAdmission(toks[0]) |
| + return a.Settled() && a.How == durability.AdmissionUnissued |
| + }) |
| + if fleet.pendingN.Load() != 0 { |
| + t.Fatalf("pending count %d after the late commit resolved", fleet.pendingN.Load()) |
| + } |
| + 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 }) |
| + 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()) |
| + } |
| +} |
| + |
| +// Close writes transport-lost for an issued admission whose transport |
| +// never finalised; with its recorder still pending the journal row stays |
| +// closing, its admission barring, until the recorder's outcome lands. |
| +func TestMeteredFleetCloseWritesTransportLost(t *testing.T) { |
| + newFakeBucket(t) |
| + dir := platformDir(t) |
| + reg := openRegistry(t, filepath.Join(dir, "registry.db")) |
| + state := filepath.Join(platformDir(t), "unit-a") |
| + 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 |
| + fleet, err := MeteredFleetStore(o) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + tok, _ := durability.NewAdmissionToken() |
| + if err := ledger.Admit(tok, fleet.FleetID(), "bac", "one", time.Now()); err != nil { |
| + t.Fatal(err) |
| + } |
| + fleet.liveMu.Lock() |
| + fleet.live[tok] = &fleetAdmission{f: fleet, token: tok} |
| + fleet.liveMu.Unlock() |
| + oldWait := closeSettleWait |
| + closeSettleWait = 100 * time.Millisecond |
| + t.Cleanup(func() { closeSettleWait = oldWait }) |
| + if err := fleet.Close(); err != nil { |
| + t.Fatal(err) |
| + } |
| + a, _ := ledger.ReadAdmission(tok) |
| + if a.TransportDoneAt.IsZero() || a.Settled() { |
| + t.Fatalf("after Close: %+v (want the transport lost, the recorder pending)", a) |
| + } |
| + if row, _ := ledger.ReadFleet(fleet.FleetID()); row.State != durability.FleetClosing { |
| + t.Fatalf("the journal row: %s, want closing with the recorder pending", row.State) |
| + } |
| + 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) |
| + } |
| +} |
diff --git a/internal/ops/litestream_metered.go b/internal/ops/litestream_metered.go| index 6b9efa5e7..5e8a75d07 100644 |
| --- a/internal/ops/litestream_metered.go |
| +++ b/internal/ops/litestream_metered.go |
| @@ -17,6 +17,7 @@ import ( |
| "github.com/carlosframework/platform/internal/lease" |
| "github.com/carlosframework/platform/internal/registry" |
| + durability "github.com/carlosframework/platform/internal/usage" |
| ) |
| // ErrUnmeterable: the box replicates a database that cannot be metered |
| @@ -107,6 +108,11 @@ type MeteredFleetOptions struct { |
| // hold not written: it is said in the refusal, since the day is then |
| // not held durably. |
| Hold func(at time.Time) error |
| + // Ledger is the box's durability ledger, where the fleet's admissions, |
| + // its journal row and its discoveries are written (Part C). nil builds |
| + // a fleet whose gate is the proof alone (lease.Unadmitted): a box that |
| + // does not attest that every replica write was admitted. |
| + Ledger *durability.DurabilityLedger |
| // Replaces is the fleet this one is built to take over from, if any. |
| // The old fleet must be closed before this one is opened, and closing |
| // it looks at the files and asks the witness one last time each (a |
| @@ -161,6 +167,24 @@ type MeteredFleet struct { |
| // Tests set it to hold a reading open. |
| witness func(context.Context) (bool, error) |
| + // Part C: the ledger, the journal id, the discovery mutex and what it |
| + // 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 |
| + |
| stale atomic.Bool |
| drift atomic.Bool |
| closed atomic.Bool |
| @@ -328,6 +352,7 @@ func (f *MeteredFleet) lastCleanLook() time.Time { |
| // on a fleet opened long after it was built (its first clean look is the |
| // constructor's inventory). |
| func (f *MeteredFleet) goStaleFrom(lookStart, at time.Time) { |
| + f.discover(at) // the ledger's discovery: the uncertainty and its holds (Part C), once |
| sound := time.Unix(0, f.proven.Load()) |
| if !lookStart.IsZero() && lookStart.Before(sound) { |
| sound = lookStart |
| @@ -569,6 +594,7 @@ func (f *MeteredFleet) Open(ctx context.Context) error { |
| // fleet never had. Read at build it would be missed whenever the new |
| // fleet is built before the old one notices. |
| f.inheritHold() |
| + f.acknowledge() |
| if err := f.admit(ctx); err != nil { |
| return err |
| } |
| @@ -600,6 +626,8 @@ func (f *MeteredFleet) watch() { |
| return |
| case <-owner.C: |
| _ = f.admit(context.Background()) |
| + f.acknowledge() |
| + f.replayLatched() |
| case <-files.C: |
| f.lookAgain() |
| } |
| @@ -830,6 +858,7 @@ func (f *MeteredFleet) closeNow() error { |
| // A fleet closed without ever opening still passes on what the fleet |
| // it replaced held. |
| f.inheritHold() |
| + f.journalClose() |
| f.held.Close() |
| f.closed.Store(true) |
| err := f.obs.Close() |
| @@ -898,6 +927,7 @@ func MeteredFleetStore(o MeteredFleetOptions) (fleet *MeteredFleet, err error) { |
| built := &MeteredFleet{ |
| obs: obs, routes: obs.Routes, registryPath: o.Registry.Path(), dataDir: filepath.Clean(o.DataDir), |
| replaces: o.Replaces, stop: make(chan struct{}), |
| + ledger: o.Ledger, pending: map[string]time.Time{}, live: map[string]*fleetAdmission{}, |
| } |
| built.proven.Store(time.Now().UnixNano()) |
| built.lastClean.Store(time.Now().UnixNano()) |
| @@ -990,13 +1020,33 @@ func MeteredFleetStore(o MeteredFleetOptions) (fleet *MeteredFleet, err error) { |
| if prefix != "" { |
| replica = prefix + "/" + key |
| } |
| - dbs = append(dbs, lease.MeteredDB{Path: owned.Path, Prefix: replica, Meter: rec, Gate: built.admit, MetaDir: metas[key], SyncInterval: o.SyncInterval}) |
| + dbs = append(dbs, lease.MeteredDB{Path: owned.Path, Prefix: replica, Meter: rec, Gate: built.gateFor(owned), MetaDir: metas[key], SyncInterval: o.SyncInterval}) |
| } |
| store, err := lease.NewMeteredStore(o.Bucket, o.Region, dbs) |
| if err != nil { |
| inv.Close() |
| return nil, err |
| } |
| + // the journal's open row, before any admission, with the construction |
| + // backfill of what the chain already holds; a fleet that cannot be |
| + // journalled is not built |
| + if o.Ledger != nil { |
| + id, err := newFleetID(o.Ledger.Box()) |
| + if err != nil { |
| + inv.Close() |
| + return nil, err |
| + } |
| + pred := "" |
| + 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) |
| + } |
| + built.id = id |
| + o.Ledger.SetPendingCounter(id, &built.pendingN) |
| + } |
| // The fleet's recovery window, not litestream's one-day default: the |
| // external replicator this replaces keeps a week of snapshots. |
| store.SnapshotInterval, store.SnapshotRetention = SnapshotInterval, SnapshotRetention |
diff --git a/internal/registry/observe.go b/internal/registry/observe.go| index 751bcda79..1386ed11c 100644 |
| --- a/internal/registry/observe.go |
| +++ b/internal/registry/observe.go |
| @@ -132,6 +132,9 @@ func (r *Registry) ObserveRoutes(ctx context.Context) (obs *RouteObservation, er |
| // |
| // As with Unchanged: false with a nil error is a witnessed change, and an |
| // error is a witness that could not be read, which is never "unchanged". |
| +// Generation is the ownership generation the rows were read under. |
| +func (o *RouteObservation) Generation() int64 { return o.generation } |
| + |
| func (o *RouteObservation) OwnershipUnchanged(ctx context.Context) (bool, error) { |
| if o.conn == nil || o.file == nil { |
| return false, errors.New("registry: the observation is closed") |
diff --git a/internal/usage/durabilityadmissions.go b/internal/usage/durabilityadmissions.go| index 1435ba78c..de47064db 100644 |
| --- a/internal/usage/durabilityadmissions.go |
| +++ b/internal/usage/durabilityadmissions.go |
| @@ -443,6 +443,26 @@ func readAdmission(tx *sql.Tx, token string) (Admission, error) { |
| return a, nil |
| } |
| +// AdmissionTokensForTest lists every admission token of a fleet, settled |
| +// or not, in admission order. For tests in other packages. |
| +func (l *DurabilityLedger) AdmissionTokensForTest(fleetID string) []string { |
| + l.mu.Lock() |
| + defer l.mu.Unlock() |
| + rows, err := l.db.Query(`SELECT token FROM admissions WHERE fleet_id = ? ORDER BY admitted_at, token`, fleetID) |
| + if err != nil { |
| + return nil |
| + } |
| + defer rows.Close() |
| + var out []string |
| + for rows.Next() { |
| + var t string |
| + if rows.Scan(&t) == nil { |
| + out = append(out, t) |
| + } |
| + } |
| + return out |
| +} |
| + |
| // UnsettledAdmissions lists the unsettled rows of the given fleets (all |
| // fleets when none is given), oldest first: what the barrier and a |
| // discovery count. |
diff --git a/internal/usage/durabilityledger.go b/internal/usage/durabilityledger.go| index 6eaa44bf0..0d5ff581f 100644 |
| --- a/internal/usage/durabilityledger.go |
| +++ b/internal/usage/durabilityledger.go |
| @@ -2070,6 +2070,9 @@ func sortDurabilityRows(rows []DurabilityRow) { |
| }) |
| } |
| +// Box is the box this ledger is the ledger of. |
| +func (l *DurabilityLedger) Box() string { return l.box } |
| + |
| // Close waits, bounded by ctx, for the attempts this run has in flight; |
| // then seals (forced, which drains latched failures), publishes once if |
| // store is non-nil, and records a clean stop — with a fresh handoff |