amadan

carlos / platform Public

Fix the durability ledger epoch read racing failed-transaction rollbacks

The epoch and created-at were fields of ledgerState, which is restored
whole (l.state = before) under the mutex whenever a Start, Finish, Abandon,
drain, attestation or seal transaction fails. Epoch(), CreatedAt(), every
snapshot and publication read the epoch without the mutex, so each
rollback was a write racing those reads. Both values are now set once by
recover, before the ledger is shared, and never written again.

Verified with a new -race regression that forces the rollback path on
Start and Finish while other goroutines read Epoch and snapshot the day;
go test ./internal/usage -run Durability -race passes.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
claude f4865344b9e5856cd7851f532827e976e996301d parent cf0b77b
3 files changed, +129 −14
  • internal/usage/durabilityfailures_test.go +108 −0
  • internal/usage/durabilityledger.go +18 −11
  • internal/usage/durabilitypublish.go +3 −3

CI passed — run details

diff --git a/internal/usage/durabilityfailures_test.go b/internal/usage/durabilityfailures_test.go
index 98dee44c9..375165994 100644
--- a/internal/usage/durabilityfailures_test.go
+++ b/internal/usage/durabilityfailures_test.go
@@ -515,3 +515,111 @@ func TestDurabilityLatchOverflowHoldsWholeRange(t *testing.T) {
t.Fatalf("day +5 after drain: %+v", content)
}
}
+
+// The epoch is fixed when the ledger is opened, and it is read without the
+// ledger mutex by Epoch, by every snapshot and by publication. A Start or
+// Finish whose transaction fails rolls the in-memory state back under the
+// mutex; if the epoch lived in that state, the rollback would be a write
+// racing every one of those reads. Run with -race: it forces the rollback
+// path (a trigger refuses the ledger-row update that persists the
+// sequence) on both Start and Finish while other goroutines read the
+// epoch and snapshot the day.
+func TestDurabilityEpochReadsDoNotRaceFailedRecordingRollback(t *testing.T) {
+ clock := newLedgerClock(ledgerDay0.Add(time.Hour))
+ l := openTestLedger(t, t.TempDir(), "box1", clock, true)
+ defer func() {
+ ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+ defer cancel()
+ l.Close(ctx, nil)
+ }()
+ epoch := l.Epoch()
+ created := l.CreatedAt()
+ c, err := l.Bind(tenantAttr)
+ if err != nil {
+ t.Fatal(err)
+ }
+ // Tokens started before the trigger exists: their Finish runs into it.
+ var toks []uint64
+ for i := 0; i < 20; i++ {
+ tok, err := c.Start("PutObject", clock.now())
+ if err != nil {
+ t.Fatal(err)
+ }
+ toks = append(toks, tok)
+ }
+ l.mu.Lock()
+ _, err = l.db.Exec(`CREATE TRIGGER force_ledger_fail BEFORE UPDATE ON ledger BEGIN SELECT RAISE(ABORT, 'forced'); END`)
+ l.mu.Unlock()
+ if err != nil {
+ t.Fatal(err)
+ }
+ date := DayOf(clock.now())
+ var wg sync.WaitGroup
+ stop := make(chan struct{})
+ wg.Add(2)
+ go func() {
+ defer wg.Done()
+ for {
+ select {
+ case <-stop:
+ return
+ default:
+ }
+ if got := l.Epoch(); got != epoch {
+ t.Errorf("epoch changed: %s -> %s", epoch, got)
+ return
+ }
+ if got := l.CreatedAt(); !got.Equal(created) {
+ t.Errorf("created changed: %s -> %s", created, got)
+ return
+ }
+ }
+ }()
+ go func() {
+ defer wg.Done()
+ for {
+ select {
+ case <-stop:
+ return
+ default:
+ }
+ content, _, err := l.snapshot(context.Background(), date)
+ if err != nil {
+ t.Errorf("snapshot: %v", err)
+ return
+ }
+ if content.Epoch != epoch {
+ t.Errorf("snapshot epoch %s, want %s", content.Epoch, epoch)
+ return
+ }
+ }
+ }()
+ startFails, finishFails := 0, 0
+ for i := 0; i < 50; i++ {
+ if _, err := c.Start("PutObject", clock.now()); err != nil {
+ startFails++
+ }
+ }
+ for _, tok := range toks {
+ if err := c.Finish(tok, RequestOutcome{At: clock.now(), Status: 200, Answered: true}); err != nil {
+ finishFails++
+ }
+ }
+ close(stop)
+ wg.Wait()
+ if startFails != 50 || finishFails != len(toks) {
+ t.Fatalf("forced failures: start %d/50, finish %d/%d", startFails, finishFails, len(toks))
+ }
+ l.mu.Lock()
+ _, err = l.db.Exec(`DROP TRIGGER force_ledger_fail`)
+ l.mu.Unlock()
+ if err != nil {
+ t.Fatal(err)
+ }
+ for _, tok := range toks {
+ c.Abandon(tok, clock.now())
+ }
+ if got := l.Epoch(); got != epoch {
+ t.Fatalf("epoch after rollbacks: %s, want %s", got, epoch)
+ }
+}
diff --git a/internal/usage/durabilityledger.go b/internal/usage/durabilityledger.go
index 2c27cde45..175f0ffc7 100644
--- a/internal/usage/durabilityledger.go
+++ b/internal/usage/durabilityledger.go
@@ -107,9 +107,16 @@ type DurabilityLedger struct {
// after the rows have been read, so a test can hold a snapshot open
// while it records.
snapshotPause func(date string)
- state ledgerState
- fresh bool
- started time.Time
+ // epoch and createdAt are fixed by recover, before the ledger is shared,
+ // and never written again. They live OUTSIDE state on purpose: state is
+ // rolled back whole (l.state = before) when a transaction fails, and
+ // that write would race every lock-free read of the epoch (Epoch,
+ // snapshotOnce, publishDay, fork).
+ epoch string
+ createdAt time.Time
+ state ledgerState
+ fresh bool
+ started time.Time
// session is the current coverage session's seq, and sessionFrom its
// start (for validation of attestation changes).
session int64
@@ -310,8 +317,6 @@ func (la *failureLatch) subtract(days map[string]int64, overflow ledgerFailureRa
// ledgerState mirrors the single row of the ledger table.
type ledgerState struct {
- epoch string
- createdAt time.Time
run int64
nextSeq int64 // the last seq handed out; the next is nextSeq+1
lastSealAt time.Time
@@ -683,15 +688,17 @@ func (l *DurabilityLedger) recover(handoff string) error {
// box's sequence onto another box's keys.
return fmt.Errorf("usage ledger %s: database belongs to box %q, this process is %q: move the ledger aside", l.dir, box, l.box)
}
- l.state = ledgerState{epoch: epoch, createdAt: timeOf(createdAt), run: run + 1, nextSeq: nextSeq, lastSealAt: timeOf(lastSeal), maxSeenAt: timeOf(maxSeen), pendingFrom: timeOf(pendingFrom), cleanStop: cleanStop != 0, stopToken: stopToken, forkedAt: timeOf(forkedAt), forkReason: forkReason, forkDate: forkDate, forkNoted: forkNoted != 0}
+ l.epoch, l.createdAt = epoch, timeOf(createdAt)
+ l.state = ledgerState{run: run + 1, nextSeq: nextSeq, lastSealAt: timeOf(lastSeal), maxSeenAt: timeOf(maxSeen), pendingFrom: timeOf(pendingFrom), cleanStop: cleanStop != 0, stopToken: stopToken, forkedAt: timeOf(forkedAt), forkReason: forkReason, forkDate: forkDate, forkNoted: forkNoted != 0}
case errors.Is(err, sql.ErrNoRows):
l.fresh = true
var b [8]byte
if _, err := rand.Read(b[:]); err != nil {
return err
}
- l.state = ledgerState{epoch: hex.EncodeToString(b[:]), createdAt: l.started, run: 1}
- if _, err := tx.Exec(`INSERT INTO ledger (id, v, box, epoch, created_at, run, next_seq, last_seal_at, max_seen_at, clean_stop) VALUES (1, 1, ?, ?, ?, 1, 0, 0, 0, 0)`, l.box, l.state.epoch, nsOf(l.started)); err != nil {
+ l.epoch, l.createdAt = hex.EncodeToString(b[:]), l.started
+ l.state = ledgerState{run: 1}
+ if _, err := tx.Exec(`INSERT INTO ledger (id, v, box, epoch, created_at, run, next_seq, last_seal_at, max_seen_at, clean_stop) VALUES (1, 1, ?, ?, ?, 1, 0, 0, 0, 0)`, l.box, l.epoch, nsOf(l.started)); err != nil {
return err
}
default:
@@ -912,14 +919,14 @@ func normalizeUnmetered(set []string) []string {
}
// Epoch is the ledger generation.
-func (l *DurabilityLedger) Epoch() string { return l.state.epoch }
+func (l *DurabilityLedger) Epoch() string { return l.epoch }
// Fresh reports whether this open minted the ledger.
func (l *DurabilityLedger) Fresh() bool { return l.fresh }
// CreatedAt is when the database was minted: the earliest instant this
// directory can vouch for the box's existence (lifecycle evidence).
-func (l *DurabilityLedger) CreatedAt() time.Time { return l.state.createdAt }
+func (l *DurabilityLedger) CreatedAt() time.Time { return l.createdAt }
// Forked reports whether publication found a lineage this ledger does not
// own (ErrDurabilityForked). Persistent: survives reopen.
@@ -1688,7 +1695,7 @@ func (l *DurabilityLedger) snapshotOnce(ctx context.Context, date string, before
return DurabilityContent{}, dayLineage{}, err
}
defer tx.Rollback() //nolint:errcheck
- c := DurabilityContent{V: durabilityContentVersion, Box: l.box, Date: date, Epoch: l.state.epoch}
+ c := DurabilityContent{V: durabilityContentVersion, Box: l.box, Date: date, Epoch: l.epoch}
if err := tx.QueryRow(`SELECT next_seq FROM ledger WHERE id = 1`).Scan(&c.Watermark); err != nil {
return DurabilityContent{}, dayLineage{}, err
}
diff --git a/internal/usage/durabilitypublish.go b/internal/usage/durabilitypublish.go
index 2d0a57c4b..a966b2194 100644
--- a/internal/usage/durabilitypublish.go
+++ b/internal/usage/durabilitypublish.go
@@ -172,7 +172,7 @@ func (l *DurabilityLedger) publishDay(ctx context.Context, store release.Store,
if day.Epochs == nil {
day.Epochs = map[string]*DurabilitySnapshot{}
}
- snap := day.Epochs[l.state.epoch]
+ snap := day.Epochs[l.epoch]
switch {
case snap == nil:
if local.pubRevision > 0 {
@@ -222,7 +222,7 @@ func (l *DurabilityLedger) publishDay(ctx context.Context, store release.Store,
if err != nil {
return false, err
}
- day.Epochs[l.state.epoch] = &DurabilitySnapshot{Revision: revision, PrevHash: prev, Hash: hash, Content: json.RawMessage(raw), PublishedAt: l.now().UTC()}
+ day.Epochs[l.epoch] = &DurabilitySnapshot{Revision: revision, PrevHash: prev, Hash: hash, Content: json.RawMessage(raw), PublishedAt: l.now().UTC()}
out, err := json.Marshal(day)
if err != nil {
return false, err
@@ -245,7 +245,7 @@ func (l *DurabilityLedger) publishDay(ctx context.Context, store release.Store,
// tries to record the conflict on the day. Returns ErrDurabilityForked.
func (l *DurabilityLedger) fork(ctx context.Context, store release.Store, date, reason string, local dayLineage, published *DurabilitySnapshot) error {
now := l.now().UTC()
- cf := DurabilityConflict{Epoch: l.state.epoch, At: now, Reason: reason, LocalRevision: local.pubRevision, LocalHash: local.pubHash}
+ cf := DurabilityConflict{Epoch: l.epoch, At: now, Reason: reason, LocalRevision: local.pubRevision, LocalHash: local.pubHash}
if published != nil {
cf.PublishedRevision, cf.PublishedHash = published.Revision, published.Hash
}