amadan

oficina / slides Public

fix: replay long deck logs in bounded batches and transient checkpoints

claude pushed by paul@keymail.dev 2b17d9696344f51fa2d7d978074a874866691b6d parent 0a51b3b
3 files changed, +264 −6
  • internal/slides/decks.go +43 −3
  • internal/slides/replay_long_test.go +212 −0
  • internal/slides/snapshot.go +9 −3

CI passed — run details

Intent

full ledger: fix/bounded-log-replay

diff --git a/internal/slides/decks.go b/internal/slides/decks.go
index 4ce323a..6e88a41 100644
--- a/internal/slides/decks.go
+++ b/internal/slides/decks.go
@@ -200,7 +200,11 @@ func (a *app) openDeck(ctx context.Context, d Deck) (*core.Handle, *deck.Deck, e
h.Close(ctx)
return nil, nil, err
}
+ var since int64
+ retained := 0
if snap != nil {
+ since = snap.Seq
+ retained = 1
if err := h.Apply(ctx, snap.Bytes); err != nil {
h.Close(ctx)
a.logger.Error("slides: read a deck's newest snapshot, replaying the plain log instead",
@@ -209,6 +213,8 @@ func (a *app) openDeck(ctx context.Context, d Deck) (*core.Handle, *deck.Deck, e
if err != nil {
return nil, nil, fmt.Errorf("slides: open deck %s: %w", d.Sqid, err)
}
+ since = 0
+ retained = 0
updates, err = a.replayLog(ctx, d, 0)
if err != nil {
h.Close(ctx)
@@ -216,12 +222,46 @@ func (a *app) openDeck(ctx context.Context, d Deck) (*core.Handle, *deck.Deck, e
}
}
}
- for _, u := range updates {
- if err := h.Apply(ctx, u.Bytes); err != nil {
+ for len(updates) > 0 {
+ for _, u := range updates {
+ if u.Seq != since+1 {
+ h.Close(ctx)
+ return nil, nil, fmt.Errorf("slides: incomplete deck %s log at seq %d", d.Sqid, since+1)
+ }
+ if retained >= replayCheckpointCadence {
+ state, err := h.State(ctx)
+ if err != nil {
+ h.Close(ctx)
+ return nil, nil, fmt.Errorf("slides: checkpoint deck %s replay: %w", d.Sqid, err)
+ }
+ h.Close(ctx)
+ h, err = a.core.NewDoc(ctx, d.Sqid, serverClientID)
+ if err != nil {
+ return nil, nil, err
+ }
+ if err := h.Apply(ctx, state); err != nil {
+ h.Close(ctx)
+ return nil, nil, err
+ }
+ retained = 1
+ }
+ if err := h.Apply(ctx, u.Bytes); err != nil {
+ h.Close(ctx)
+ return nil, nil, fmt.Errorf("slides: replay deck %s update %d: %w", d.Sqid, u.Seq, err)
+ }
+ since = u.Seq
+ retained++
+ }
+ updates, err = a.replayLog(ctx, d, since)
+ if err != nil {
h.Close(ctx)
- return nil, nil, fmt.Errorf("slides: replay deck %s update %d: %w", d.Sqid, u.Seq, err)
+ return nil, nil, err
}
}
+ if since != d.Seq {
+ h.Close(ctx)
+ return nil, nil, fmt.Errorf("slides: incomplete deck %s log at seq %d", d.Sqid, since+1)
+ }
view, err := h.JSON(ctx)
if err != nil {
diff --git a/internal/slides/replay_long_test.go b/internal/slides/replay_long_test.go
new file mode 100644
index 0000000..4229d81
--- /dev/null
+++ b/internal/slides/replay_long_test.go
@@ -0,0 +1,212 @@
+package slides
+
+import (
+ "context"
+ "fmt"
+ "gorm.io/gorm"
+ "strings"
+ "testing"
+ "time"
+
+ "amadan.net/oficina/slides/internal/slides/core"
+)
+
+func TestAuthoritativeLogBeyondCoreRetentionLimit(t *testing.T) {
+ db := testDB(t)
+ a, _, err := newApp(db, Config{Env: EnvDevelopment, Logger: quietLogger()})
+ if err != nil {
+ t.Fatal(err)
+ }
+ t.Cleanup(func() { a.Close() })
+ ctx := context.Background()
+ owner, _, err := ensureUser(ctx, db.G, "long-log@example.test")
+ if err != nil {
+ t.Fatal(err)
+ }
+ row, err := a.createDeckRow(ctx, owner, "Long history", defaultTheme)
+ if err != nil {
+ t.Fatal(err)
+ }
+ h, err := a.core.NewDoc(ctx, row.Sqid, serverClientID)
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer func() { h.Close(ctx) }()
+ updates := make([]DeckUpdate, 0, core.MaxRetainedUpdates+19)
+ for seq := int64(1); seq <= core.MaxRetainedUpdates+19; seq++ {
+ if seq%1000 == 0 {
+ state, err := h.State(ctx)
+ if err != nil {
+ t.Fatal(err)
+ }
+ h.Close(ctx)
+ h, err = a.core.NewDoc(ctx, row.Sqid, serverClientID)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if err := h.Apply(ctx, state); err != nil {
+ t.Fatal(err)
+ }
+ }
+ batch, err := opsJSON([]op{{"op": "set_meta", "title": fmt.Sprintf("Revision %d", seq)}})
+ if err != nil {
+ t.Fatal(err)
+ }
+ if seq == 1 {
+ batch, err = openingOps("Long history", defaultTheme)
+ if err != nil {
+ t.Fatal(err)
+ }
+ }
+ update, err := h.Ops(ctx, batch)
+ if err != nil {
+ t.Fatal(err)
+ }
+ updates = append(updates, DeckUpdate{DeckID: row.ID, Seq: seq, Actor: actorFor(owner), Bytes: update, CreatedAt: time.Now().UTC()})
+ row.Seq = seq
+ }
+ if err := db.G.CreateInBatches(updates, 128).Error; err != nil {
+ t.Fatal(err)
+ }
+ if err := db.G.Model(&Deck{}).Where("id = ?", row.ID).Update("seq", row.Seq).Error; err != nil {
+ t.Fatal(err)
+ }
+ want, err := h.JSON(ctx)
+ if err != nil {
+ t.Fatal(err)
+ }
+ got := viewJSON(t, a, row)
+ if string(got) != string(want) {
+ t.Fatal("long replay changed document content")
+ }
+ t.Run("bounded_queries", func(t *testing.T) {
+ queries := 0
+ name := "test:bounded_replay"
+ if err := db.G.Callback().Query().After("gorm:query").Register(name, func(tx *gorm.DB) {
+ if tx.Statement.Table == "deck_updates" {
+ queries++
+ if tx.RowsAffected > 256 {
+ t.Errorf("unbounded database batch: %d rows", tx.RowsAffected)
+ }
+ }
+ }); err != nil {
+ t.Fatal(err)
+ }
+ defer db.G.Callback().Query().Remove(name)
+ if string(viewJSON(t, a, row)) != string(want) {
+ t.Fatal("bounded query changed content")
+ }
+ if queries < 2 {
+ t.Fatal("long log was not paged")
+ }
+ })
+ t.Run("unreadable_cache", func(t *testing.T) {
+ snap := DeckSnapshot{DeckID: row.ID, Seq: 128, Bytes: []byte{0xff}, CreatedAt: time.Now().UTC()}
+ if err := db.G.Create(&snap).Error; err != nil {
+ t.Fatal(err)
+ }
+ defer db.G.Delete(&snap)
+ if string(viewJSON(t, a, row)) != string(want) {
+ t.Fatal("corrupt cache changed long replay")
+ }
+ })
+ t.Run("future_cache_and_update", func(t *testing.T) {
+ batch, err := opsJSON([]op{{"op": "set_meta", "title": "Future revision"}})
+ if err != nil {
+ t.Fatal(err)
+ }
+ update, err := h.Ops(ctx, batch)
+ if err != nil {
+ t.Fatal(err)
+ }
+ state, err := h.State(ctx)
+ if err != nil {
+ t.Fatal(err)
+ }
+ future := DeckUpdate{DeckID: row.ID, Seq: row.Seq + 1, Actor: actorFor(owner), Bytes: update, CreatedAt: time.Now().UTC()}
+ if err := db.G.Create(&future).Error; err != nil {
+ t.Fatal(err)
+ }
+ defer db.G.Delete(&future)
+ snap := DeckSnapshot{DeckID: row.ID, Seq: row.Seq + 1, Bytes: state, CreatedAt: time.Now().UTC()}
+ if err := db.G.Create(&snap).Error; err != nil {
+ t.Fatal(err)
+ }
+ defer db.G.Delete(&snap)
+ if string(viewJSON(t, a, row)) != string(want) {
+ t.Fatal("replay included a revision after its captured coordinate")
+ }
+ })
+ for _, missing := range []int64{256, row.Seq} {
+ t.Run(fmt.Sprintf("missing_%d", missing), func(t *testing.T) {
+ u := updates[missing-1]
+ if err := db.G.Delete(&u).Error; err != nil {
+ t.Fatal(err)
+ }
+ defer func() {
+ if err := db.G.Create(&u).Error; err != nil {
+ t.Error(err)
+ }
+ }()
+ replayed, view, err := a.openDeck(ctx, row)
+ if replayed != nil {
+ replayed.Close(ctx)
+ }
+ if err == nil || replayed != nil || view != nil {
+ t.Fatal("incomplete log returned a partial deck")
+ }
+ })
+ }
+ t.Run("canceled_checkpoint", func(t *testing.T) {
+ cancelCtx, cancel := context.WithCancel(ctx)
+ defer cancel()
+ queries := 0
+ name := "test:cancel_checkpoint"
+ if err := db.G.Callback().Query().After("gorm:query").Register(name, func(tx *gorm.DB) {
+ if tx.Statement.Table == "deck_updates" {
+ queries++
+ if queries == 9 {
+ cancel()
+ }
+ }
+ }); err != nil {
+ t.Fatal(err)
+ }
+ defer db.G.Callback().Query().Remove(name)
+ replayed, view, err := a.openDeck(cancelCtx, row)
+ if replayed != nil {
+ replayed.Close(ctx)
+ }
+ if err == nil || replayed != nil || view != nil {
+ t.Fatal("canceled replay returned a partial deck")
+ }
+ if !strings.Contains(err.Error(), "checkpoint") {
+ t.Fatalf("did not reach the canceled State call: %v", err)
+ }
+ })
+ opened, _, err := a.openDeck(ctx, row)
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer opened.Close(ctx)
+ batch, err := opsJSON([]op{{"op": "set_meta", "title": "After replay"}})
+ if err != nil {
+ t.Fatal(err)
+ }
+ if _, err := opened.Ops(ctx, batch); err != nil {
+ t.Fatalf("replayed handle cannot accept next edit: %v", err)
+ }
+ var count int64
+ if err := db.G.Model(&DeckUpdate{}).Where("deck_id = ?", row.ID).Count(&count).Error; err != nil {
+ t.Fatal(err)
+ }
+ if count != row.Seq {
+ t.Fatal("read rewrote authoritative history")
+ }
+ if err := db.G.Model(&DeckSnapshot{}).Where("deck_id = ?", row.ID).Count(&count).Error; err != nil {
+ t.Fatal(err)
+ }
+ if count != 0 {
+ t.Fatal("read persisted an accelerator snapshot")
+ }
+}
diff --git a/internal/slides/snapshot.go b/internal/slides/snapshot.go
index c038012..04cd3df 100644
--- a/internal/slides/snapshot.go
+++ b/internal/slides/snapshot.go
@@ -33,6 +33,9 @@ import (
// logged at Error for exactly that reason.
const snapshotCadence = 200
+const replayBatchSize = 256
+const replayCheckpointCadence = core.MaxRetainedUpdates / 2
+
// latestSnapshot returns the newest usable snapshot for deckID, or nil
// when there is none.
//
@@ -56,7 +59,7 @@ func (a *app) latestSnapshot(ctx context.Context, deckID int64) (*DeckSnapshot,
}
// replayPlan is what openDeck has to apply to reconstitute d: the
-// newest snapshot (or nil) and every update after it, in seq order.
+// newest eligible snapshot (or nil) and the first bounded update batch.
//
// It is one function rather than two calls at the call site so that the
// invariant tying them together — the tail starts exactly where the
@@ -75,6 +78,9 @@ func (a *app) replayPlan(ctx context.Context, d Deck) (*DeckSnapshot, []DeckUpda
snap = nil
}
+ if snap != nil && (snap.Seq <= 0 || snap.Seq > d.Seq) {
+ snap = nil
+ }
var since int64
if snap != nil {
since = snap.Seq
@@ -86,8 +92,8 @@ func (a *app) replayPlan(ctx context.Context, d Deck) (*DeckSnapshot, []DeckUpda
func (a *app) replayLog(ctx context.Context, d Deck, since int64) ([]DeckUpdate, error) {
var updates []DeckUpdate
if err := a.db.G.WithContext(ctx).
- Where("deck_id = ? AND seq > ?", d.ID, since).
- Order("seq").Find(&updates).Error; err != nil {
+ Where("deck_id = ? AND seq > ? AND seq <= ?", d.ID, since, d.Seq).
+ Order("seq").Limit(replayBatchSize).Find(&updates).Error; err != nil {
return nil, fmt.Errorf("slides: read deck %s log: %w", d.Sqid, err)
}
return updates, nil