amadan
Recover admission journals while retaining the producer fence
9 files changed,
+1127
−3
internal/ops/litestream_admission.go+8 −0internal/ops/litestream_admission_review11_test.go+52 −0internal/usage/durabilityadmissions.go+4 −0internal/usage/durabilityadmissions_test.go+22 −2internal/usage/durabilityfleets.go+37 −1internal/usage/durabilityfleets_test.go+106 −0internal/usage/durabilityledger.go+36 −0internal/usage/durabilityrecovery.go+290 −0internal/usage/durabilityrecovery_test.go+572 −0
diff --git a/internal/ops/litestream_admission.go b/internal/ops/litestream_admission.go| index 3ad6e6189..dc5c4dc99 100644 |
| --- a/internal/ops/litestream_admission.go |
| +++ b/internal/ops/litestream_admission.go |
| @@ -305,6 +305,14 @@ func (f *MeteredFleet) acknowledge() { |
| if row.Predecessor == "" { |
| break |
| } |
| + predecessor, err := f.ledger.ReadFleet(row.Predecessor) |
| + if err != nil { |
| + f.ledgerWrite("acknowledge", err) |
| + return |
| + } |
| + if predecessor.AcknowledgedBy == id { |
| + break |
| + } |
| id = row.Predecessor |
| chain = append(chain, id) |
| } |
diff --git a/internal/ops/litestream_admission_review11_test.go b/internal/ops/litestream_admission_review11_test.go| index 1d66c055e..ddddc5f75 100644 |
| --- a/internal/ops/litestream_admission_review11_test.go |
| +++ b/internal/ops/litestream_admission_review11_test.go |
| @@ -301,3 +301,55 @@ func TestReview11AmbiguousFleetOpenKeepsCommittedFleet(t *testing.T) { |
| t.Fatalf("lost committed fleet: %+v %v", row, err) |
| } |
| } |
| + |
| +func TestRecoveryFleetAcknowledgementStopsAtCopiedEdge(t *testing.T) { |
| + _, _, _, o := review11Fleet(t) |
| + dir := t.TempDir() |
| + l, err := durability.OpenDurabilityLedger(dir, "box1", time.Now, durability.DurabilityLedgerOptions{}) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + o.Ledger = l |
| + o.Replaces = nil |
| + a, err := MeteredFleetStore(o) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + o.Replaces = a |
| + b, err := MeteredFleetStore(o) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if err := b.Close(); err != nil { |
| + t.Fatal(err) |
| + } |
| + if err := a.Close(); err != nil { |
| + t.Fatal(err) |
| + } |
| + if err := l.Close(t.Context(), nil); err != nil { |
| + t.Fatal(err) |
| + } |
| + l, err = durability.OpenDurabilityLedger(dir, "box1", time.Now, durability.DurabilityLedgerOptions{}) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + defer l.Close(t.Context(), nil) |
| + o.Ledger = l |
| + o.Replaces = nil |
| + c, err := MeteredFleetStore(o) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + defer c.Close() |
| + ar, _ := l.ReadFleet(a.id) |
| + br, _ := l.ReadFleet(b.id) |
| + if ar.AcknowledgedBy != c.id || br.AcknowledgedBy != c.id { |
| + t.Fatalf("recovery did not copy ancestry: %+v %+v", ar, br) |
| + } |
| + if err := c.Open(t.Context()); err != nil { |
| + t.Fatal(err) |
| + } |
| + if !c.acked.Load() || c.LedgerErrors() != 0 { |
| + t.Fatalf("driver replayed already-copied ancestry: acked=%v errors=%d", c.acked.Load(), c.LedgerErrors()) |
| + } |
| +} |
diff --git a/internal/usage/durabilityadmissions.go b/internal/usage/durabilityadmissions.go| index de47064db..642cb5ea5 100644 |
| --- a/internal/usage/durabilityadmissions.go |
| +++ b/internal/usage/durabilityadmissions.go |
| @@ -22,6 +22,7 @@ import ( |
| // admissionSchema is created beside the ledger's own tables. |
| var admissionSchema = []string{ |
| + `CREATE TABLE IF NOT EXISTS fleet_open_inputs (fleet_id TEXT PRIMARY KEY, predecessor TEXT NOT NULL)`, |
| `CREATE TABLE IF NOT EXISTS admissions ( |
| token TEXT PRIMARY KEY, |
| fleet_id TEXT NOT NULL, |
| @@ -144,6 +145,9 @@ func (l *DurabilityLedger) Admit(token, fleetID, account, app string, at time.Ti |
| if l.closed { |
| return errLedgerClosed |
| } |
| + if l.fence == nil { |
| + return ErrProducerFence // an admission is a promise the fence backs: without it nothing is admitted |
| + } |
| tx, err := l.db.Begin() |
| if err != nil { |
| return err |
diff --git a/internal/usage/durabilityadmissions_test.go b/internal/usage/durabilityadmissions_test.go| index 9e8cafc26..f0b508ce2 100644 |
| --- a/internal/usage/durabilityadmissions_test.go |
| +++ b/internal/usage/durabilityadmissions_test.go |
| @@ -192,6 +192,17 @@ func TestAdmissionUnrecordedSettlesAndSurvivesReopen(t *testing.T) { |
| if err := l.Close(t.Context(), nil); err != nil { |
| t.Fatal(err) |
| } |
| + // closed with an admission unsettled: the producer fence stays held |
| + // (a transport of this process may still be sent), and a new process |
| + // cannot recover over it until this one exits |
| + if !l.FenceHeld() { |
| + t.Fatal("the fence was released with an admission unsettled") |
| + } |
| + if _, err := OpenDurabilityLedger(dir, "box", func() time.Time { return now }, DurabilityLedgerOptions{}); !errors.Is(err, ErrProducerFence) { |
| + t.Fatalf("a reopen over a held fence: %v", err) |
| + } |
| + l.ReleaseFenceForTest() // the process exits |
| + now = now.Add(time.Hour) |
| l2, err := OpenDurabilityLedger(dir, "box", func() time.Time { return now }, DurabilityLedgerOptions{}) |
| if err != nil { |
| t.Fatal(err) |
| @@ -201,8 +212,17 @@ func TestAdmissionUnrecordedSettlesAndSurvivesReopen(t *testing.T) { |
| if err != nil || !a.Settled() || a.How != AdmissionUnrecorded { |
| t.Fatalf("after reopen: %+v %v", a, err) |
| } |
| - if open, _ := l2.UnsettledAdmissions(); len(open) != 1 || open[0].Token != pending { |
| - t.Fatalf("the pending row after reopen: %+v", open) |
| + // recovery classified the pending row: recovered-unknown, settled now, |
| + // its own closed uncertainty holding its days |
| + p, err := l2.ReadAdmission(pending) |
| + if err != nil || !p.Settled() || p.How != AdmissionRecoveredUnknown || !p.SettledAt.Equal(now) { |
| + t.Fatalf("the pending row after recovery: %+v %v", p, err) |
| + } |
| + if open, _ := l2.UnsettledAdmissions(); len(open) != 0 { |
| + t.Fatalf("unsettled after recovery: %+v", open) |
| + } |
| + if h, _ := l2.Held(DayOf(p.AdmittedAt)); !h { |
| + t.Fatal("the recovered admission's day is not held") |
| } |
| } |
diff --git a/internal/usage/durabilityfleets.go b/internal/usage/durabilityfleets.go| index cc06295f8..0f17ea7cf 100644 |
| --- a/internal/usage/durabilityfleets.go |
| +++ b/internal/usage/durabilityfleets.go |
| @@ -83,13 +83,42 @@ func (l *DurabilityLedger) OpenFleet(fleetID, box string, witnessGeneration int6 |
| } |
| defer tx.Rollback() //nolint:errcheck |
| if f, err := readFleet(tx, fleetID); err == nil { |
| - if f.Predecessor == predecessor && f.Box == box && f.WitnessGeneration == witnessGeneration && strings.Join(f.Owners, "\n") == strings.Join(sorted, "\n") && f.OpenedAt.Equal(at) { |
| + requested := f.Predecessor |
| + if err := tx.QueryRow(`SELECT predecessor FROM fleet_open_inputs WHERE fleet_id = ?`, fleetID).Scan(&requested); err != nil && !errors.Is(err, sql.ErrNoRows) { |
| + return err |
| + } |
| + if requested == predecessor && f.Box == box && f.WitnessGeneration == witnessGeneration && strings.Join(f.Owners, "\n") == strings.Join(sorted, "\n") && f.OpenedAt.Equal(at) { |
| return nil // the same open again: it landed |
| } |
| return fmt.Errorf("usage ledger: fleet %s is already open with other facts", fleetID) |
| } else if !errors.Is(err, ErrFleetUnknown) { |
| return err |
| } |
| + requestedPredecessor := predecessor |
| + if predecessor == "" { |
| + rows, err := tx.Query(`SELECT f.fleet_id FROM fleets f WHERE f.acknowledged_by = '' AND NOT EXISTS (SELECT 1 FROM fleets s WHERE s.predecessor = f.fleet_id) ORDER BY f.fleet_id`) |
| + if err != nil { |
| + return err |
| + } |
| + var tips []string |
| + for rows.Next() { |
| + var id string |
| + if err := rows.Scan(&id); err != nil { |
| + rows.Close() |
| + return err |
| + } |
| + tips = append(tips, id) |
| + } |
| + err = rows.Err() |
| + rows.Close() |
| + if err != nil { |
| + return err |
| + } |
| + if len(tips) == 1 { |
| + predecessor = tips[0] |
| + } |
| + } |
| + |
| if predecessor != "" { |
| if _, err := readFleet(tx, predecessor); err != nil { |
| return fmt.Errorf("usage ledger: predecessor %s: %w", predecessor, err) |
| @@ -109,6 +138,13 @@ func (l *DurabilityLedger) OpenFleet(fleetID, box string, witnessGeneration int6 |
| fleetID, box, witnessGeneration, strings.Join(sorted, "\n"), nsOf(at), FleetOpen, predecessor); err != nil { |
| return err |
| } |
| + if _, err := tx.Exec(`INSERT INTO fleet_open_inputs (fleet_id, predecessor) VALUES (?, ?)`, fleetID, requestedPredecessor); err != nil { |
| + return err |
| + } |
| + if err := l.inheritRecovered(tx, fleetID, at); 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) |
diff --git a/internal/usage/durabilityfleets_test.go b/internal/usage/durabilityfleets_test.go| index 4e90ee71a..80fe0ba26 100644 |
| --- a/internal/usage/durabilityfleets_test.go |
| +++ b/internal/usage/durabilityfleets_test.go |
| @@ -647,3 +647,109 @@ func TestFleetScheduleInheritance(t *testing.T) { |
| l.Close(t.Context(), nil) |
| } |
| } |
| + |
| +func TestRecoveryFleetTopologyAndRepeat(t *testing.T) { |
| + l, _ := openLedgerAt(t, at(3, 12, 0)) |
| + for i, id := range []string{"z", "a", "m"} { |
| + when := at(3-i, 12, 0) |
| + if i == 1 { |
| + when = at(3, 12, 0) |
| + } |
| + if err := l.OpenFleet(id, "box", int64(i), []string{"alice"}, "", when); err != nil { |
| + t.Fatal(err) |
| + } |
| + if err := l.OpenFleet(id, "box", int64(i), []string{"alice"}, "", when); err != nil { |
| + t.Fatalf("identical inferred repeat: %v", err) |
| + } |
| + } |
| + m, _ := l.ReadFleet("m") |
| + if m.Predecessor != "a" { |
| + t.Fatalf("timestamp chosen over graph: %+v", m) |
| + } |
| + if err := l.OpenFleet("m", "box", 2, []string{"alice"}, "z", at(1, 12, 0)); err == nil { |
| + t.Fatal("changed requested predecessor accepted") |
| + } |
| + for _, id := range []string{"z", "a"} { |
| + if err := l.CloseFleet(id, at(3, 12, 0)); err != nil { |
| + t.Fatal(err) |
| + } |
| + } |
| + if err := l.Acknowledge("m", at(3, 12, 0)); !errors.Is(err, ErrAcknowledgeNotReady) { |
| + t.Fatalf("ordinary acknowledgement skipped intermediate: %v", err) |
| + } |
| + if err := l.Acknowledge("a", at(3, 12, 0)); err != nil { |
| + t.Fatal(err) |
| + } |
| + if err := l.Acknowledge("m", at(3, 12, 0)); err != nil { |
| + t.Fatal(err) |
| + } |
| + z, _ := l.ReadFleet("z") |
| + if z.AcknowledgedBy != "a" { |
| + t.Fatalf("ordinary chain fanned in: %+v", z) |
| + } |
| +} |
| + |
| +func TestRecoveryDisconnectedRootsAtomicInheritance(t *testing.T) { |
| + dir := t.TempDir() |
| + now := at(1, 12, 0) |
| + l, err := OpenDurabilityLedger(dir, "box", func() time.Time { return now }, DurabilityLedgerOptions{}) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + for _, id := range []string{"root-a", "root-b"} { |
| + // Historical ledgers allowed disconnected empty-predecessor roots. |
| + if _, err := l.db.Exec(`INSERT INTO fleets (fleet_id,box,witness_generation,owners,opened_at,state) VALUES (?,'box',1,'alice',?,'open')`, id, nsOf(now)); err != nil { |
| + t.Fatal(err) |
| + } |
| + } |
| + l.abandon() |
| + now = at(2, 12, 0) |
| + l, err = OpenDurabilityLedger(dir, "box", func() time.Time { return now }, DurabilityLedgerOptions{}) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + defer l.Close(t.Context(), nil) |
| + fleetFail = func() error { return errDiskFull } |
| + err = l.OpenFleet("new", "box", 2, []string{"bob"}, "", now) |
| + fleetFail = nil |
| + if err == nil { |
| + t.Fatal("injected transaction did not fail") |
| + } |
| + for _, id := range []string{"root-a", "root-b"} { |
| + f, _ := l.ReadFleet(id) |
| + if f.AcknowledgedBy != "" { |
| + t.Fatalf("partial recovery acknowledgement: %+v", f) |
| + } |
| + } |
| + if err := l.OpenFleet("new", "box", 2, []string{"bob"}, "", now); err != nil { |
| + t.Fatal(err) |
| + } |
| + if err := l.OpenFleet("new", "box", 2, []string{"bob"}, "", now); err != nil { |
| + t.Fatalf("ambiguous/repeat result: %v", err) |
| + } |
| + f, _ := l.ReadFleet("new") |
| + if f.Predecessor != "" { |
| + t.Fatalf("invented link among disconnected roots: %+v", f) |
| + } |
| + inherited, err := l.Inherited("new") |
| + if err != nil || len(inherited) != 2 { |
| + t.Fatalf("lost recovered root: %v %v", inherited, err) |
| + } |
| + for _, id := range []string{"root-a", "root-b"} { |
| + f, _ := l.ReadFleet(id) |
| + if f.AcknowledgedBy != "new" { |
| + t.Fatalf("root not acknowledged: %+v", f) |
| + } |
| + } |
| + if err := l.OpenFleet("next", "box", 3, []string{"carol"}, "", now); err != nil { |
| + t.Fatal(err) |
| + } |
| + next, _ := l.ReadFleet("next") |
| + if next.Predecessor != "new" { |
| + t.Fatalf("next fleet missed recovered join: %+v", next) |
| + } |
| + effective, err := l.Effective("next") |
| + if err != nil || len(effective) != 2 { |
| + t.Fatalf("lost transitive recovery: %v %v", effective, err) |
| + } |
| +} |
diff --git a/internal/usage/durabilityledger.go b/internal/usage/durabilityledger.go| index f22ee0d14..48aaaab20 100644 |
| --- a/internal/usage/durabilityledger.go |
| +++ b/internal/usage/durabilityledger.go |
| @@ -94,6 +94,12 @@ type DurabilityLedger struct { |
| box string |
| now func() time.Time |
| lock *os.File |
| + // fence is the producer fence (Part C): admissions.lock in the |
| + // ledger's directory, flocked before recovery and held while any |
| + // admission of this process may still be sent; nil once released. An |
| + // admission requires it held. |
| + fence *os.File |
| + recoveryFleets []string |
| mu sync.Mutex // serialises every database access and the state below |
| // drainMu spans a failure drain's commit and latch subtraction. A |
| @@ -444,10 +450,25 @@ func OpenDurabilityLedger(dir, box string, now func() time.Time, opts Durability |
| } |
| } |
| } |
| + // The producer fence, before recovery: an older producer still alive |
| + // (a process that closed the ledger with admissions in flight and |
| + // kept the fence) refuses this open rather than letting recovery |
| + // classify admissions that may still be sent. |
| + fence, err := takeProducerFence(dir) |
| + if err != nil { |
| + unlockSpoolFile(lf) |
| + lf.Close() |
| + return nil, err |
| + } |
| + l.fence = fence |
| if err := l.recover(opts.HandoffToken); err != nil { |
| + if l.rdb != nil { |
| + l.rdb.Close() |
| + } |
| if l.db != nil { |
| l.db.Close() |
| } |
| + l.releaseFence() |
| unlockSpoolFile(lf) |
| lf.Close() |
| return nil, err |
| @@ -831,6 +852,11 @@ func (l *DurabilityLedger) recover(handoff string) error { |
| if l.started.After(l.state.maxSeenAt) { |
| l.state.maxSeenAt = l.started |
| } |
| + // Part C's recovery, holding the fence: the admissions and fleets an |
| + // earlier process left unsettled and open. |
| + if err := l.recoverAdmissions(tx); err != nil { |
| + return fmt.Errorf("usage ledger %s: recover admissions: %w", l.dir, err) |
| + } |
| if _, err := tx.Exec(`UPDATE ledger SET run = ?, next_seq = ?, max_seen_at = ?, clean_stop = 0, stop_token = '', pending_from_ns = 0 WHERE id = 1`, l.state.run, l.state.nextSeq, nsOf(l.state.maxSeenAt)); err != nil { |
| return err |
| } |
| @@ -2171,6 +2197,15 @@ func (l *DurabilityLedger) Close(ctx context.Context, store release.Store) error |
| l.stopToken = hex.EncodeToString(b[:]) |
| } |
| } |
| + // The producer fence is released only when nothing of this process can |
| + // still be sent: with an admission unsettled it stays held until the |
| + // process exits, and a new process cannot recover over it. |
| + var unsettled int |
| + if err := l.db.QueryRow(`SELECT COUNT(*) FROM admissions WHERE settled_at IS NULL`).Scan(&unsettled); err == nil && unsettled == 0 { |
| + l.releaseFence() |
| + } else { |
| + retainProducerFence(l.fence) |
| + } |
| l.rdb.Close() |
| l.db.Close() |
| unlockSpoolFile(l.lock) |
| @@ -2250,6 +2285,7 @@ func (l *DurabilityLedger) abandon() { |
| l.db.Close() |
| unlockSpoolFile(l.lock) |
| l.lock.Close() |
| + l.releaseFence() // the process is gone: the kernel would release it |
| } |
| // durabilityAttemptsOnDisk counts attempt rows through a second handle, |
diff --git a/internal/usage/durabilityrecovery.go b/internal/usage/durabilityrecovery.go| new file mode 100644 |
| index 000000000..ceadc7b8b |
| --- /dev/null |
| +++ b/internal/usage/durabilityrecovery.go |
| @@ -0,0 +1,290 @@ |
| +package usage |
| + |
| +import ( |
| + "database/sql" |
| + "errors" |
| + "fmt" |
| + "os" |
| + "path/filepath" |
| + "sync" |
| + "time" |
| +) |
| + |
| +// Part C's producer fence and recovery (the replicator's authority design, |
| +// "The producer fence, and recovery at process start"; plan Task 4's sixth |
| +// slice). The fence is an flock on admissions.lock in the ledger's |
| +// directory, taken before recovery and before any admission; the helper |
| +// process (Part B) does not exist yet, so the agent alone holds it, and it |
| +// stays held past Close while an admission of this process is unsettled: |
| +// the kernel releases it when the process exits, and only then can a new |
| +// process recover. Recovery, holding the fence, in the open's transaction |
| +// before any fleet is built: every unsettled admission is recovered-unknown |
| +// with its missing timestamps now and an uncertainty opened and closed at |
| +// once over [admitted_at - margin, now + margin], its days materialised; |
| +// every fleet not closed is closed with an uncertainty over |
| +// [min(opened_at, its earliest admission) - margin, now + margin] if it had |
| +// none open, or its open uncertainty closed at now + margin; every fleet |
| +// left unacknowledged is acknowledged, root-first, by the first fleet the |
| +// new process builds. OpenFleet copies the recovered histories atomically; |
| +// live handovers still use adjacent root-first Acknowledge calls. Mirrors admissionmodel.Recover, except that the model |
| +// acknowledges through the existing chain and the journal acknowledges by |
| +// the new fleet, as the design has it. |
| + |
| +// ErrProducerFence: the producer fence is held by another process, so an |
| +// older producer may still send; recovery waits for it to exit. |
| +var ErrProducerFence = errors.New("usage ledger: the producer fence is held: an older producer may still be alive") |
| + |
| +const producerFenceFile = "admissions.lock" |
| + |
| +// takeProducerFence flocks admissions.lock, without waiting. |
| +func takeProducerFence(dir string) (*os.File, error) { |
| + f, err := os.OpenFile(filepath.Join(dir, producerFenceFile), os.O_CREATE|os.O_RDWR, 0o644) |
| + if err != nil { |
| + return nil, err |
| + } |
| + if err := lockSpoolFile(f); err != nil { |
| + f.Close() |
| + return nil, fmt.Errorf("%w (%v)", ErrProducerFence, err) |
| + } |
| + return f, nil |
| +} |
| + |
| +// releaseFence lets the fence go; nil afterwards. Caller holds l.mu or is |
| +// the open or close path. |
| +func (l *DurabilityLedger) releaseFence() { |
| + if l.fence == nil { |
| + return |
| + } |
| + retainedProducerFences.Lock() |
| + delete(retainedProducerFences.files, l.fence) |
| + retainedProducerFences.Unlock() |
| + l.fence.Close() |
| + l.fence = nil |
| +} |
| + |
| +// FenceHeld reports whether this ledger holds the producer fence. |
| +func (l *DurabilityLedger) FenceHeld() bool { |
| + l.mu.Lock() |
| + defer l.mu.Unlock() |
| + return l.fence != nil |
| +} |
| + |
| +var retainedProducerFences = struct { |
| + sync.Mutex |
| + files map[*os.File]struct{} |
| +}{files: make(map[*os.File]struct{})} |
| + |
| +func retainProducerFence(f *os.File) { |
| + if f == nil { |
| + return |
| + } |
| + retainedProducerFences.Lock() |
| + retainedProducerFences.files[f] = struct{}{} |
| + retainedProducerFences.Unlock() |
| +} |
| + |
| +// recoverAdmissions is recovery's Part C, inside the open's transaction. |
| +func (l *DurabilityLedger) recoverAdmissions(tx *sql.Tx) error { |
| + now := l.started.UTC() |
| + // every unsettled admission: recovered-unknown, its own closed |
| + // uncertainty |
| + rows, err := tx.Query(`SELECT token, fleet_id, admitted_at, COALESCE(transport_done_at,0), COALESCE(recorder_done_at,0) FROM admissions WHERE settled_at IS NULL ORDER BY admitted_at, token`) |
| + if err != nil { |
| + return err |
| + } |
| + type orphan struct { |
| + token, fleet string |
| + admitted time.Time |
| + settled time.Time |
| + } |
| + var orphans []orphan |
| + for rows.Next() { |
| + var o orphan |
| + var admitted, transport, recorder int64 |
| + if err := rows.Scan(&o.token, &o.fleet, &admitted, &transport, &recorder); err != nil { |
| + rows.Close() |
| + return err |
| + } |
| + o.admitted = time.Unix(0, admitted).UTC() |
| + o.settled = maxTime(now, o.admitted) |
| + if transport != 0 { |
| + o.settled = maxTime(o.settled, time.Unix(0, transport).UTC()) |
| + } |
| + if recorder != 0 { |
| + o.settled = maxTime(o.settled, time.Unix(0, recorder).UTC()) |
| + } |
| + orphans = append(orphans, o) |
| + } |
| + if err := rows.Err(); err != nil { |
| + rows.Close() |
| + return err |
| + } |
| + rows.Close() |
| + ns := nsOf(now) |
| + for _, o := range orphans { |
| + if _, err := tx.Exec(`UPDATE admissions SET transport_done_at = COALESCE(transport_done_at, ?), recorder_done_at = COALESCE(recorder_done_at, ?), settled_at = ?, how = ? WHERE token = ?`, |
| + ns, ns, nsOf(o.settled), AdmissionRecoveredUnknown, o.token); err != nil { |
| + return err |
| + } |
| + from, to := minTime(o.admitted, now).Add(-AdmissionMargin), o.settled.Add(AdmissionMargin) |
| + res, err := tx.Exec(`INSERT INTO uncertainties (fleet_id, from_ns, to_ns, discovered_at, reason) VALUES (?, ?, ?, ?, ?)`, o.fleet, nsOf(from), nsOf(to), ns, "recovered-unknown admission "+o.token) |
| + if err != nil { |
| + return err |
| + } |
| + id, err := res.LastInsertId() |
| + if err != nil { |
| + return err |
| + } |
| + if err := l.writeHolds(tx, from, to, id, now); err != nil { |
| + return err |
| + } |
| + if err := l.dirtyInterval(tx, minTime(o.admitted, now), o.settled); err != nil { |
| + return err |
| + } |
| + } |
| + // every fleet not closed: closed, with an uncertainty |
| + frows, err := tx.Query(`SELECT fleet_id, opened_at FROM fleets WHERE state != ? ORDER BY opened_at, fleet_id`, FleetClosed) |
| + if err != nil { |
| + return err |
| + } |
| + type openFleet struct { |
| + id string |
| + opened time.Time |
| + } |
| + var fleets []openFleet |
| + for frows.Next() { |
| + var f openFleet |
| + var opened int64 |
| + if err := frows.Scan(&f.id, &opened); err != nil { |
| + frows.Close() |
| + return err |
| + } |
| + f.opened = time.Unix(0, opened).UTC() |
| + fleets = append(fleets, f) |
| + } |
| + if err := frows.Err(); err != nil { |
| + frows.Close() |
| + return err |
| + } |
| + frows.Close() |
| + for _, f := range fleets { |
| + latest := maxTime(now, f.opened) |
| + var last sql.NullInt64 |
| + var discovered sql.NullInt64 |
| + if err := tx.QueryRow(`SELECT MAX(discovered_at) FROM uncertainties WHERE fleet_id = ?`, f.id).Scan(&discovered); err != nil { |
| + return err |
| + } |
| + if discovered.Valid { |
| + latest = maxTime(latest, time.Unix(0, discovered.Int64).UTC()) |
| + } |
| + if err := tx.QueryRow(`SELECT MAX(settled_at) FROM admissions WHERE fleet_id = ?`, f.id).Scan(&last); err != nil { |
| + return err |
| + } |
| + if last.Valid { |
| + latest = maxTime(latest, time.Unix(0, last.Int64).UTC()) |
| + } |
| + to := latest.Add(AdmissionMargin) |
| + if id, from, isOpen, err := openUncertainty(tx, f.id); err != nil { |
| + return err |
| + } else if isOpen { |
| + to = maxTime(to, from.Add(AdmissionMargin)) |
| + if _, err := tx.Exec(`UPDATE uncertainties SET to_ns = ? WHERE id = ?`, nsOf(to), id); err != nil { |
| + return err |
| + } |
| + if err := l.writeHolds(tx, from, to, id, now); err != nil { |
| + return err |
| + } |
| + } else { |
| + from := minTime(f.opened, now) |
| + var earliest sql.NullInt64 |
| + if err := tx.QueryRow(`SELECT MIN(admitted_at) FROM admissions WHERE fleet_id = ?`, f.id).Scan(&earliest); err != nil { |
| + return err |
| + } |
| + if earliest.Valid && time.Unix(0, earliest.Int64).UTC().Before(from) { |
| + from = time.Unix(0, earliest.Int64).UTC() |
| + } |
| + from = from.Add(-AdmissionMargin) |
| + res, err := tx.Exec(`INSERT INTO uncertainties (fleet_id, from_ns, to_ns, discovered_at, reason) VALUES (?, ?, ?, ?, ?)`, f.id, nsOf(from), nsOf(to), ns, "recovered: the fleet was not closed") |
| + if err != nil { |
| + return err |
| + } |
| + id, err := res.LastInsertId() |
| + if err != nil { |
| + return err |
| + } |
| + if err := l.writeHolds(tx, from, to, id, now); err != nil { |
| + return err |
| + } |
| + } |
| + if _, err := tx.Exec(`UPDATE fleets SET state = ? WHERE fleet_id = ?`, FleetClosed, f.id); err != nil { |
| + return err |
| + } |
| + } |
| + roots, err := tx.Query(`SELECT fleet_id FROM fleets WHERE acknowledged_by = '' ORDER BY fleet_id`) |
| + if err != nil { |
| + return err |
| + } |
| + defer roots.Close() |
| + for roots.Next() { |
| + var id string |
| + if err := roots.Scan(&id); err != nil { |
| + return err |
| + } |
| + l.recoveryFleets = append(l.recoveryFleets, id) |
| + } |
| + return roots.Err() |
| +} |
| + |
| +// Only rows present at recovery participate in this fan-in. The live |
| +// acknowledgement primitive retains its adjacent root-first contract. |
| +func (l *DurabilityLedger) inheritRecovered(tx *sql.Tx, successor string, at time.Time) error { |
| + pending := make(map[string]FleetRow) |
| + for _, id := range l.recoveryFleets { |
| + f, err := readFleet(tx, id) |
| + if err != nil { |
| + return err |
| + } |
| + if f.AcknowledgedBy != "" { |
| + continue |
| + } |
| + if f.State != FleetClosed { |
| + return ErrAcknowledgeNotReady |
| + } |
| + pending[id] = f |
| + } |
| + // Copy before acknowledging: an acknowledgement contracts ancestry. |
| + for id := range pending { |
| + us, err := effectiveOf(tx, []string{id}) |
| + if err != nil { |
| + return err |
| + } |
| + for _, u := range us { |
| + if u.Open() { |
| + return ErrFleetUnsettled |
| + } |
| + if _, err := tx.Exec(`INSERT OR IGNORE INTO inherited (fleet_id, uncertainty_id) VALUES (?, ?)`, successor, u.ID); err != nil { |
| + return err |
| + } |
| + if err := l.writeHolds(tx, u.From, u.To, u.ID, at); err != nil { |
| + return err |
| + } |
| + } |
| + } |
| + for len(pending) > 0 { |
| + progress := false |
| + for id, f := range pending { |
| + if _, waits := pending[f.Predecessor]; waits { |
| + continue |
| + } |
| + if _, err := tx.Exec(`UPDATE fleets SET acknowledged_by = ? WHERE fleet_id = ?`, successor, id); err != nil { |
| + return err |
| + } |
| + delete(pending, id) |
| + progress = true |
| + } |
| + if !progress { |
| + return ErrAcknowledgeNotReady |
| + } |
| + } |
| + return nil |
| +} |
diff --git a/internal/usage/durabilityrecovery_test.go b/internal/usage/durabilityrecovery_test.go| new file mode 100644 |
| index 000000000..1d1748cd8 |
| --- /dev/null |
| +++ b/internal/usage/durabilityrecovery_test.go |
| @@ -0,0 +1,572 @@ |
| +package usage |
| + |
| +import ( |
| + "bufio" |
| + "context" |
| + "database/sql" |
| + "errors" |
| + "fmt" |
| + "os" |
| + "os/exec" |
| + "path/filepath" |
| + "runtime" |
| + "strings" |
| + "testing" |
| + "time" |
| +) |
| + |
| +// The producer fence: a ledger does not open over a held fence (an older |
| +// producer may still send); it takes the fence at open, admits only while |
| +// it holds it, and releases it at Close only when nothing is unsettled. |
| +func TestProducerFence(t *testing.T) { |
| + dir := t.TempDir() |
| + now := at(1, 10, 0) |
| + held, err := os.OpenFile(filepath.Join(dir, producerFenceFile), os.O_CREATE|os.O_RDWR, 0o644) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if err := lockSpoolFile(held); err != nil { |
| + t.Fatal(err) |
| + } |
| + if _, err := OpenDurabilityLedger(dir, "box", func() time.Time { return now }, DurabilityLedgerOptions{}); !errors.Is(err, ErrProducerFence) { |
| + t.Fatalf("opened over a held fence: %v", err) |
| + } |
| + if _, err := os.Stat(filepath.Join(dir, ledgerDBFile)); !os.IsNotExist(err) { |
| + t.Fatalf("a refused open made the database: %v", err) |
| + } |
| + unlockSpoolFile(held) |
| + held.Close() |
| + l, err := OpenDurabilityLedger(dir, "box", func() time.Time { return now }, DurabilityLedgerOptions{}) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if !l.FenceHeld() { |
| + t.Fatal("the fence is not held after the open") |
| + } |
| + tok := mustAdmit(t, l, "f", "alice", "app", time.Time{}) |
| + // without the fence nothing is admitted |
| + l.ReleaseFenceForTest() |
| + tok2, _ := NewAdmissionToken() |
| + if err := l.Admit(tok2, "f", "alice", "app", now); !errors.Is(err, ErrProducerFence) { |
| + t.Fatalf("admitted without the fence: %v", err) |
| + } |
| + if _, err := l.ReadAdmission(tok2); !errors.Is(err, ErrAdmissionUnknown) { |
| + t.Fatalf("the refused admission wrote: %v", err) |
| + } |
| + finishAdmission(t, l, tok) |
| + if err := l.Close(t.Context(), nil); err != nil { |
| + t.Fatal(err) |
| + } |
| + // everything settled: a new open takes the fence at once |
| + l2, err := OpenDurabilityLedger(dir, "box", func() time.Time { return now.Add(time.Hour) }, DurabilityLedgerOptions{}) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + defer l2.Close(t.Context(), nil) |
| + if !l2.FenceHeld() { |
| + t.Fatal("the new ledger does not hold the fence") |
| + } |
| + if eff, _ := l2.Effective("f"); len(eff) != 0 { |
| + t.Fatalf("recovery with nothing unsettled wrote uncertainties: %v", eff) |
| + } |
| +} |
| + |
| +// Mirrors the model's kill at every transition (TestLandedAmbiguousThenCrash, |
| +// TestFenceOutlivesAgent's recovery): a process dies after each step of an |
| +// admission's life; the next process recovers, holding the fence, in one |
| +// transaction: an unsettled row is recovered-unknown with its missing |
| +// timestamps now and a closed uncertainty over [admitted - margin, now + |
| +// margin]; a settled row is left; a fleet not closed is closed with an |
| +// uncertainty (its open one closed at now + margin, or a new one over |
| +// [min(opened, earliest admission) - margin, now + margin]); and the first |
| +// fleet the new process builds continues the chain and acknowledges it |
| +// root-first, inheriting everything. Later days complete. |
| +func TestRecoveryAtEveryTransition(t *testing.T) { |
| + type step struct { |
| + name string |
| + do func(l *DurabilityLedger, tok string) |
| + } |
| + steps := []step{ |
| + {"admitted", func(*DurabilityLedger, string) {}}, |
| + {"recorder started", func(l *DurabilityLedger, tok string) { |
| + if err := l.RecorderStarted(tok, time.Time{}); err != nil { |
| + t.Fatal(err) |
| + } |
| + }}, |
| + {"transport done", func(l *DurabilityLedger, tok string) { |
| + if err := l.TransportDone(tok, time.Time{}, false); err != nil { |
| + t.Fatal(err) |
| + } |
| + }}, |
| + {"recorder done", func(l *DurabilityLedger, tok string) { |
| + if err := l.RecorderStarted(tok, time.Time{}); err != nil { |
| + t.Fatal(err) |
| + } |
| + if err := l.RecorderDone(tok, time.Time{}, AdmissionFinished); err != nil { |
| + t.Fatal(err) |
| + } |
| + }}, |
| + {"settled", func(l *DurabilityLedger, tok string) { |
| + if err := l.RecorderStarted(tok, time.Time{}); err != nil { |
| + t.Fatal(err) |
| + } |
| + if err := l.RecorderDone(tok, time.Time{}, AdmissionFinished); err != nil { |
| + t.Fatal(err) |
| + } |
| + if err := l.TransportDone(tok, time.Time{}, false); err != nil { |
| + t.Fatal(err) |
| + } |
| + }}, |
| + {"discovered", func(l *DurabilityLedger, tok string) { |
| + if _, err := l.Discover(Discovery{FleetID: "A", At: at(2, 23, 58), Proof: at(2, 9, 0), LastCleanLook: at(2, 23, 0)}); err != nil { |
| + t.Fatal(err) |
| + } |
| + }}, |
| + } |
| + for _, st := range steps { |
| + dir := t.TempDir() |
| + now := at(2, 9, 0) |
| + l, err := OpenDurabilityLedger(dir, "box", func() time.Time { return now }, DurabilityLedgerOptions{}) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if err := l.OpenFleet("A", "box", 1, []string{"alice"}, "", time.Time{}); err != nil { |
| + t.Fatal(err) |
| + } |
| + now = at(2, 23, 57) |
| + tok := mustAdmit(t, l, "A", "alice", "app", time.Time{}) |
| + now = at(2, 23, 58) |
| + st.do(l, tok) |
| + before, _ := l.ReadAdmission(tok) |
| + l.abandon() // the process dies |
| + now = at(3, 0, 10) |
| + l2, err := OpenDurabilityLedger(dir, "box", func() time.Time { return now }, DurabilityLedgerOptions{}) |
| + if err != nil { |
| + t.Fatalf("%s: recovery: %v", st.name, err) |
| + } |
| + a, _ := l2.ReadAdmission(tok) |
| + if before.Settled() { |
| + if a.How != before.How || !a.SettledAt.Equal(before.SettledAt) { |
| + t.Fatalf("%s: a settled row was touched by recovery: %+v", st.name, a) |
| + } |
| + } else { |
| + if !a.Settled() || a.How != AdmissionRecoveredUnknown || !a.SettledAt.Equal(now) { |
| + t.Fatalf("%s: the orphan: %+v", st.name, a) |
| + } |
| + if !before.TransportDoneAt.IsZero() && !a.TransportDoneAt.Equal(before.TransportDoneAt) { |
| + t.Fatalf("%s: a completion recorded before the crash lost its instant: %+v", st.name, a) |
| + } |
| + if before.TransportDoneAt.IsZero() && !a.TransportDoneAt.Equal(now) { |
| + t.Fatalf("%s: a missing timestamp is not now: %+v", st.name, a) |
| + } |
| + } |
| + if open, _ := l2.UnsettledAdmissions(); len(open) != 0 { |
| + t.Fatalf("%s: unsettled after recovery: %+v", st.name, open) |
| + } |
| + row, _ := l2.ReadFleet("A") |
| + if row.State != FleetClosed { |
| + t.Fatalf("%s: the fleet after recovery: %s", st.name, row.State) |
| + } |
| + eff, _ := l2.Effective("A") |
| + if len(eff) == 0 { |
| + t.Fatalf("%s: a fleet not closed was closed by recovery with no uncertainty", st.name) |
| + } |
| + for _, u := range eff { |
| + if u.Open() || !u.To.Equal(now.Add(AdmissionMargin)) { |
| + t.Fatalf("%s: an uncertainty not closed at now plus the margin: %+v", st.name, u) |
| + } |
| + if u.From.After(at(2, 23, 56)) { // the admission (23:57) minus the margin, or earlier |
| + t.Fatalf("%s: the uncertainty does not reach the admission: %+v", st.name, u) |
| + } |
| + } |
| + held(t, l2, true, "2026-10-02", "2026-10-03") |
| + if c, _ := l2.Snapshot("2026-10-03"); c.Barred != 0 || c.Held == 0 { |
| + t.Fatalf("%s: the recovery day: %+v", st.name, c) |
| + } |
| + if c, _ := l2.Snapshot("2026-10-05"); c.Barred != 0 || c.Held != 0 { |
| + t.Fatalf("%s: a later day: %+v", st.name, c) |
| + } |
| + // the new process's first fleet continues the chain and |
| + // acknowledges it root-first |
| + if err := l2.OpenFleet("B", "box", 2, []string{"bob"}, "", time.Time{}); err != nil { |
| + t.Fatal(err) |
| + } |
| + if rb, _ := l2.ReadFleet("B"); rb.Predecessor != "A" { |
| + t.Fatalf("%s: the new fleet's predecessor %q, want the journal's tip", st.name, rb.Predecessor) |
| + } |
| + if err := l2.Acknowledge("B", time.Time{}); err != nil { |
| + t.Fatalf("%s: acknowledging the recovered chain: %v", st.name, err) |
| + } |
| + if ra, _ := l2.ReadFleet("A"); ra.AcknowledgedBy != "B" { |
| + t.Fatalf("%s: A acknowledged by %q", st.name, ra.AcknowledgedBy) |
| + } |
| + if inh, _ := l2.Inherited("B"); len(inh) != len(eff) { |
| + t.Fatalf("%s: B inherited %v, A's effective set %v", st.name, inh, ids(eff)) |
| + } |
| + l2.Close(t.Context(), nil) |
| + } |
| +} |
| + |
| +// A chain an earlier process left (A, B, C: closed by recovery, none |
| +// acknowledged) is acknowledged root-first by the first fleet the new |
| +// process builds, in one transaction, each ancestor's effective set copied |
| +// onto it; a fleet in the chain still open refuses the acknowledgement |
| +// until it closes. |
| +func TestRecoveredChainAcknowledgedRootFirst(t *testing.T) { |
| + dir := t.TempDir() |
| + now := at(1, 9, 0) |
| + l, err := OpenDurabilityLedger(dir, "box", func() time.Time { return now }, DurabilityLedgerOptions{}) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + for i, id := range []string{"A", "B", "C"} { |
| + now = at(1, 10+i, 0) |
| + if err := l.OpenFleet(id, "box", int64(i), []string{"o" + id}, "", time.Time{}); err != nil { |
| + t.Fatal(err) |
| + } |
| + if _, err := l.Discover(Discovery{FleetID: id, At: now, Proof: now.Add(-30 * time.Minute), LastCleanLook: now.Add(-20 * time.Minute)}); err != nil { |
| + t.Fatal(err) |
| + } |
| + } |
| + if rc, _ := l.ReadFleet("C"); rc.Predecessor != "B" { |
| + t.Fatalf("C's predecessor %q: a fleet with none continues the journal's chain", rc.Predecessor) |
| + } |
| + l.abandon() |
| + now = at(1, 14, 0) |
| + l2, err := OpenDurabilityLedger(dir, "box", func() time.Time { return now }, DurabilityLedgerOptions{}) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + defer l2.Close(t.Context(), nil) |
| + for _, id := range []string{"A", "B", "C"} { |
| + if r, _ := l2.ReadFleet(id); r.State != FleetClosed || r.AcknowledgedBy != "" { |
| + t.Fatalf("%s after recovery: %+v", id, r) |
| + } |
| + } |
| + if err := l2.OpenFleet("D", "box", 3, []string{"oD"}, "", time.Time{}); err != nil { |
| + t.Fatal(err) |
| + } |
| + if anc, _ := l2.Ancestry("D"); len(anc) != 0 { |
| + t.Fatalf("D's ancestry %v", anc) |
| + } |
| + if err := l2.Acknowledge("D", time.Time{}); err != nil { |
| + t.Fatal(err) |
| + } |
| + for _, id := range []string{"A", "B", "C"} { |
| + if r, _ := l2.ReadFleet(id); r.AcknowledgedBy != "D" { |
| + t.Fatalf("%s acknowledged by %q, want D", id, r.AcknowledgedBy) |
| + } |
| + } |
| + // each fleet has two of its own: its discovery, closed at once, and |
| + // the one recovery wrote when it closed the fleet |
| + const want = 6 |
| + inh, _ := l2.Inherited("D") |
| + if len(inh) != want { |
| + t.Fatalf("D inherited %v, want the three fleets' %d uncertainties", inh, want) |
| + } |
| + if anc, _ := l2.Ancestry("D"); len(anc) != 0 { |
| + t.Fatalf("D's ancestry after acknowledging: %v", anc) |
| + } |
| + // a later successor finds everything on D |
| + if err := l2.OpenFleet("E", "box", 4, []string{"oE"}, "", time.Time{}); err != nil { |
| + t.Fatal(err) |
| + } |
| + if eff, _ := l2.Effective("E"); len(eff) != want { |
| + t.Fatalf("E's effective set %v, want %d", ids(eff), want) |
| + } |
| + // a chain with an open fleet in it: not ready |
| + if err := l2.OpenFleet("F", "box", 5, []string{"oF"}, "", time.Time{}); err != nil { |
| + t.Fatal(err) |
| + } |
| + if err := l2.Acknowledge("F", time.Time{}); !errors.Is(err, ErrAcknowledgeNotReady) { |
| + t.Fatalf("F acknowledging E, which is open: %v", err) |
| + } |
| +} |
| + |
| +func (l *DurabilityLedger) ReleaseFenceForTest() { |
| + l.mu.Lock() |
| + defer l.mu.Unlock() |
| + l.releaseFence() |
| +} |
| + |
| +func TestRecoverySubprocess(t *testing.T) { |
| + mode := os.Getenv("CARLOS_RECOVERY_TEST_MODE") |
| + if mode == "" { |
| + return |
| + } |
| + dir := os.Getenv("CARLOS_RECOVERY_TEST_DIR") |
| + if mode == "inherit" { |
| + f := os.NewFile(3, "inherited fence") |
| + fmt.Println("ready") |
| + var b [1]byte |
| + os.Stdin.Read(b[:]) |
| + f.Close() |
| + os.Exit(0) |
| + } |
| + func() { |
| + now := at(2, 23, 58) |
| + l, err := OpenDurabilityLedger(dir, "box", func() time.Time { return now }, DurabilityLedgerOptions{}) |
| + if err != nil { |
| + panic(err) |
| + } |
| + if err := l.OpenFleet("old", "box", 1, []string{"alice"}, "", now); err != nil { |
| + panic(err) |
| + } |
| + if err := l.Admit("orphan", "old", "alice", "app", now); err != nil { |
| + panic(err) |
| + } |
| + if mode == "gc" { |
| + if err := l.Close(context.Background(), nil); err != nil { |
| + panic(err) |
| + } |
| + return |
| + } |
| + if mode == "recorder" || mode == "transport" || mode == "settled" { |
| + if err := l.RecorderStarted("orphan", now); err != nil { |
| + panic(err) |
| + } |
| + } |
| + if mode == "recorder" || mode == "settled" { |
| + if err := l.RecorderDone("orphan", now, AdmissionFinished); err != nil { |
| + panic(err) |
| + } |
| + } |
| + if mode == "transport" || mode == "settled" { |
| + if err := l.TransportDone("orphan", now, false); err != nil { |
| + panic(err) |
| + } |
| + } |
| + if mode == "discovery" { |
| + if _, err := l.Discover(Discovery{FleetID: "old", At: now, Proof: now.Add(-time.Minute)}); err != nil { |
| + panic(err) |
| + } |
| + } |
| + if mode == "closing" { |
| + if err := l.FleetClosing("old"); err != nil { |
| + panic(err) |
| + } |
| + } |
| + }() |
| + for i := 0; i < 5; i++ { |
| + runtime.GC() |
| + runtime.Gosched() |
| + } |
| + fmt.Println("ready") |
| + var b [1]byte |
| + os.Stdin.Read(b[:]) |
| + os.Exit(0) |
| +} |
| + |
| +func recoveryChild(t *testing.T, dir, mode string, fence *os.File) *exec.Cmd { |
| + t.Helper() |
| + cmd := exec.Command(os.Args[0], "-test.run=^TestRecoverySubprocess$") |
| + cmd.Dir = dir |
| + cmd.Env = append(os.Environ(), "CARLOS_RECOVERY_TEST_MODE="+mode, "CARLOS_RECOVERY_TEST_DIR="+dir) |
| + if fence != nil { |
| + cmd.ExtraFiles = []*os.File{fence} |
| + } |
| + out, err := cmd.StdoutPipe() |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + in, err := cmd.StdinPipe() |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + var stderr strings.Builder |
| + cmd.Stderr = &stderr |
| + if err := cmd.Start(); err != nil { |
| + t.Fatal(err) |
| + } |
| + t.Cleanup(func() { |
| + in.Close() |
| + if cmd.ProcessState == nil { |
| + cmd.Process.Kill() |
| + cmd.Wait() |
| + } |
| + }) |
| + ready := make(chan error, 1) |
| + go func() { |
| + line, err := bufio.NewReader(out).ReadString('\n') |
| + if err == nil && line != "ready\n" { |
| + err = fmt.Errorf("unexpected child output %q", line) |
| + } |
| + ready <- err |
| + }() |
| + select { |
| + case err := <-ready: |
| + if err != nil { |
| + t.Fatalf("child readiness: %v %s", err, stderr.String()) |
| + } |
| + case <-time.After(10 * time.Second): |
| + t.Fatal("child readiness timeout") |
| + } |
| + return cmd |
| +} |
| + |
| +func TestProducerFenceSurvivesGCAndInheritedClose(t *testing.T) { |
| + if runtime.GOOS == "windows" { |
| + t.Skip("Unix producer fence") |
| + } |
| + for _, mode := range []string{"gc", "inherit"} { |
| + t.Run(mode, func(t *testing.T) { |
| + dir := t.TempDir() |
| + var owner *DurabilityLedger |
| + var fence *os.File |
| + if mode == "inherit" { |
| + var err error |
| + owner, err = OpenDurabilityLedger(dir, "box", time.Now, DurabilityLedgerOptions{}) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + fence = owner.fence |
| + } |
| + child := recoveryChild(t, dir, mode, fence) |
| + if owner != nil { |
| + if err := owner.Close(t.Context(), nil); err != nil { |
| + t.Fatal(err) |
| + } |
| + } |
| + if l, err := OpenDurabilityLedger(dir, "box", time.Now, DurabilityLedgerOptions{}); !errors.Is(err, ErrProducerFence) { |
| + if l != nil { |
| + l.Close(t.Context(), nil) |
| + } |
| + t.Fatalf("opened over living producer: %v", err) |
| + } |
| + child.Process.Kill() |
| + child.Wait() |
| + l, err := OpenDurabilityLedger(dir, "box", time.Now, DurabilityLedgerOptions{}) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + defer l.Close(t.Context(), nil) |
| + if mode == "gc" { |
| + a, err := l.ReadAdmission("orphan") |
| + if err != nil || !a.Settled() || a.How != AdmissionRecoveredUnknown { |
| + t.Fatalf("orphan after producer exit: %+v %v", a, err) |
| + } |
| + } |
| + }) |
| + } |
| +} |
| + |
| +func TestRecoveryAfterRealProcessCrash(t *testing.T) { |
| + if runtime.GOOS == "windows" { |
| + t.Skip("Unix producer fence") |
| + } |
| + for _, mode := range []string{"admitted", "recorder", "transport", "settled", "discovery", "closing"} { |
| + t.Run(mode, func(t *testing.T) { |
| + dir := t.TempDir() |
| + child := recoveryChild(t, dir, mode, nil) |
| + child.Process.Kill() |
| + child.Wait() |
| + now := at(3, 0, 10) |
| + l, err := OpenDurabilityLedger(dir, "box", func() time.Time { return now }, DurabilityLedgerOptions{}) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + defer l.Close(t.Context(), nil) |
| + a, err := l.ReadAdmission("orphan") |
| + if err != nil || !a.Settled() { |
| + t.Fatalf("crashed admission: %+v %v", a, err) |
| + } |
| + want := AdmissionRecoveredUnknown |
| + if mode == "settled" { |
| + want = AdmissionFinished |
| + } |
| + if a.How != want { |
| + t.Fatalf("outcome %s want %s", a.How, want) |
| + } |
| + f, err := l.ReadFleet("old") |
| + if err != nil || f.State != FleetClosed { |
| + t.Fatalf("fleet after crash: %+v %v", f, err) |
| + } |
| + held(t, l, true, "2026-10-02", "2026-10-03") |
| + if c, err := l.Snapshot("2026-10-05"); err != nil || c.Barred != 0 || c.Held != 0 { |
| + t.Fatalf("later day %+v %v", c, err) |
| + } |
| + }) |
| + } |
| +} |
| + |
| +func TestRecoveryClockRollbackKeepsParticipantMaximum(t *testing.T) { |
| + dir := t.TempDir() |
| + now := at(3, 12, 0) |
| + l, err := OpenDurabilityLedger(dir, "box", func() time.Time { return now }, DurabilityLedgerOptions{}) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if err := l.OpenFleet("f", "box", 1, []string{"alice"}, "", now); err != nil { |
| + t.Fatal(err) |
| + } |
| + tok := mustAdmit(t, l, "f", "alice", "app", now) |
| + future := at(4, 0, 1) |
| + if err := l.TransportDone(tok, future, false); err != nil { |
| + t.Fatal(err) |
| + } |
| + l.abandon() |
| + now = at(2, 23, 59) |
| + l, err = OpenDurabilityLedger(dir, "box", func() time.Time { return now }, DurabilityLedgerOptions{}) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + defer l.Close(t.Context(), nil) |
| + a, err := l.ReadAdmission(tok) |
| + if err != nil || !a.SettledAt.Equal(future) || !a.TransportDoneAt.Equal(future) || !a.RecorderDoneAt.Equal(now) { |
| + t.Fatalf("rollback destroyed completion maximum: %+v %v", a, err) |
| + } |
| + held(t, l, true, "2026-10-02", "2026-10-03", "2026-10-04") |
| +} |
| + |
| +func TestRecoveryTransactionFailureRollsBackAndReleasesFence(t *testing.T) { |
| + dir := t.TempDir() |
| + now := at(1, 12, 0) |
| + l, err := OpenDurabilityLedger(dir, "box", func() time.Time { return now }, DurabilityLedgerOptions{}) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if err := l.OpenFleet("f", "box", 1, []string{"alice"}, "", now); err != nil { |
| + t.Fatal(err) |
| + } |
| + tok := mustAdmit(t, l, "f", "alice", "app", now) |
| + if _, err := l.db.Exec(`CREATE TRIGGER refuse_recovery BEFORE UPDATE OF state ON fleets BEGIN SELECT RAISE(ABORT,'injected recovery fault'); END`); err != nil { |
| + t.Fatal(err) |
| + } |
| + l.abandon() |
| + now = at(2, 12, 0) |
| + if reopened, err := OpenDurabilityLedger(dir, "box", func() time.Time { return now }, DurabilityLedgerOptions{}); err == nil { |
| + reopened.Close(t.Context(), nil) |
| + t.Fatal("recovery ignored injected fault") |
| + } |
| + db, err := sql.Open("sqlite", filepath.Join(dir, ledgerDBFile)) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + var settled sql.NullInt64 |
| + if err := db.QueryRow(`SELECT settled_at FROM admissions WHERE token = ?`, tok).Scan(&settled); err != nil { |
| + t.Fatal(err) |
| + } |
| + if settled.Valid { |
| + t.Fatal("failed recovery committed orphan terminal") |
| + } |
| + var count int |
| + if err := db.QueryRow(`SELECT COUNT(*) FROM uncertainties`).Scan(&count); err != nil { |
| + t.Fatal(err) |
| + } |
| + if count != 0 { |
| + t.Fatal("failed recovery left partial holds") |
| + } |
| + if _, err := db.Exec(`DROP TRIGGER refuse_recovery`); err != nil { |
| + t.Fatal(err) |
| + } |
| + db.Close() |
| + l, err = OpenDurabilityLedger(dir, "box", func() time.Time { return now }, DurabilityLedgerOptions{}) |
| + if err != nil { |
| + t.Fatalf("failed recovery leaked a lock: %v", err) |
| + } |
| + defer l.Close(t.Context(), nil) |
| + a, err := l.ReadAdmission(tok) |
| + if err != nil || !a.Settled() { |
| + t.Fatalf("retry failed: %+v %v", a, err) |
| + } |
| +} |