amadan

oficina / docs Public

Plan 5: replay from a snapshot, and write one every 256 updates

Store.replay now loads the newest usable snapshot, applies it, and reads
only the doc_updates rows after it. With no usable snapshot — none
written, an unreadable row, a row from another ABI, bytes the core
refuses — it replays the whole log exactly as before, logs at Error, and
carries on. The log stays complete and authoritative; a snapshot is only
ever an accelerator, so the worst failure here is a slower replay and
never wrong content.

The snapshot is taken in ApplyUpdate from the REPLAYED state, before the
incoming update is applied, and that placement is the whole of its
correctness. Taking it after a successful Apply is wrong twice. The
replay runs outside ApplyUpdate's transaction, so a concurrent POST can
commit a row this handle never saw; a snapshot written there would claim
a seq whose content it does not hold, and the in-core verification
cannot notice, because it compares the candidate against the same stale
handle that produced it. And a document already at MaxRetainedUpdates
replays to a full handle, so Apply refuses with ErrUpdateLogFull before
anything succeeds — an after-success hook is never reached, and the one
case this feature exists for would stay broken forever. From the
replayed state both close by construction: the prefix is gapless and
complete, and the read happens whether or not the update is accepted.

writeSnapshot verifies the candidate in the core before persisting it,
refuses to write on a false verdict, and deletes the snapshots the new
row supersedes — (seq, id)-ordered exactly as newestSnapshot reads, not
"everything that is not mine", which two concurrent boundary writers
would use to delete each other's row and leave the document with none.
No snapshot failure can fail an update: each is logged once at Error and
the update proceeds.

The cadence distance comes from the snapshot row replay already loaded.
No second query.

Tests: equivalence between snapshot-then-tail and a full replay; both of
the concurrency defect's faces, including a direct check that every
snapshot contains every row its seq claims; a document seeded at the cap,
which must leave a snapshot behind on the POST it fails and accept the
retry; a corrupt snapshot falling back to the log; and the headline
claim driven the long way, past MaxRetainedUpdates through the real
path. Verified by mutation — moving the snapshot to after a successful
Apply fails the concurrency and cap tests reliably.

internal/core comments only: ErrUpdateLogFull and the package doc said no
host caller persisted snapshots yet. One does now.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Paul Campbell pushed by paul@keymail.dev 1cdd18ae651c53d43c30af28181ff352cbe6d409 parent 718bc2c
4 files changed, +594 −57
  • internal/core/core.go +42 −30
  • internal/docs/snapshots.go +78 −0
  • internal/docs/snapshots_test.go +358 −0
  • internal/docs/store.go +116 −27
diff --git a/internal/core/core.go b/internal/core/core.go
index bdebd02..5ec562f 100644
--- a/internal/core/core.go
+++ b/internal/core/core.go
@@ -33,18 +33,20 @@
// Spec §5.2 describes recovery as rebuilding "from its latest snapshot" —
// a persisted, compacted state plus a bounded tail of subsequent updates.
// As of ABI v4 the guest module can encode its whole state as one update
-// (docs_core_snapshot_len / docs_core_snapshot_read), but this package
-// does not yet call them or persist anything — that wiring belongs to
-// the snapshot-storage task this one sets up for. Until then, every
-// Handle keeps the updates it has itself applied successfully, in
-// memory, for the life of the Handle, and a discarded instance is
-// rebuilt by replaying that entire log from an empty document. This is
-// a deliberate substitution, not an oversight — it is what the absence
-// of snapshot *storage* (not, any longer, of the export itself) makes
-// necessary — and it is why the log still needs its own bound
-// (maxRetainedUpdates below): without storage wired up, "replay
-// everything" remains the only recovery this package can do,
-// so the log cannot be allowed to grow without limit.
+// (docs_core_snapshot_len / docs_core_snapshot_read) and Handle.Snapshot
+// below returns those bytes, but a Handle's own recovery still does not
+// use them: every Handle keeps the updates it has itself applied
+// successfully, in memory, for the life of the Handle, and a discarded
+// instance is rebuilt by replaying that entire log from an empty
+// document. Nothing in here is persisted, so there is nothing in here
+// to restore from — §5.2's snapshot restore is the host's to perform
+// with what Handle.Snapshot hands it, and internal/docs does now
+// perform it, storing snapshots in doc_snapshots and starting each
+// document's replay from the newest one. This is a deliberate division,
+// not an oversight, and it is why the in-memory log still needs its own
+// bound (maxRetainedUpdates below): within this package "replay
+// everything" remains the only recovery available, so the log cannot be
+// allowed to grow without limit.
package core
import (
@@ -106,20 +108,21 @@ const defaultGuestCallTimeout = 1 * time.Second
//
// The real fix is snapshot support: ABI v4 now exports a state-encode
// pair (docs_core_snapshot_len / docs_core_snapshot_read) that encodes a
-// document's current state as one update, so rebuild could restore a
+// document's current state as one update, so a rebuild restores a
// compacted snapshot plus only the updates since it, matching spec §5.2 and
// the doc_snapshots table spec §4.1 describes — bounded, cheap recovery
-// regardless of a document's total history. Calling that export and
-// persisting the result belongs with the snapshot/compaction work, not
-// here — this package does not do either yet.
+// regardless of a document's total history. This package exposes that
+// export (Handle.Snapshot, below); persisting the result is the host's
+// job, and internal/docs/store.go's Store.replay and Store.ApplyUpdate
+// now do it, writing a snapshot every 256 updates and replaying from
+// the newest one instead of from the whole log.
//
-// Until then, this package does the safe thing available without that
-// storage wired up: once a document's log is full, Apply refuses to
-// accept further updates at all (ErrUpdateLogFull), rather than either
-// growing the log without bound or silently dropping old entries —
-// dropping old entries would make a later rebuild replay an incomplete
-// history and produce silently wrong content, which is worse than
-// refusing.
+// This cap stays regardless, as the floor under that arrangement: once
+// a document's log is full, Apply refuses to accept further updates at
+// all (ErrUpdateLogFull), rather than either growing the log without
+// bound or silently dropping old entries — dropping old entries would
+// make a later rebuild replay an incomplete history and produce
+// silently wrong content, which is worse than refusing.
//
// How reachable this cap actually is, for whoever writes the
// snapshot/compaction plan: the docs app persists one doc_updates row
@@ -144,13 +147,22 @@ const MaxRetainedUpdates = maxRetainedUpdates
// ErrUpdateLogFull is returned by Handle.Apply once a document's retained
// update log has reached maxRetainedUpdates. See that constant's doc
-// comment: ABI v4 exports the snapshot/compaction primitive, and this
-// package now calls it (Handle.Snapshot, Handle.VerifySnapshot, below)
-// — what's still missing is a host caller in internal/docs persisting a
-// snapshot row and using it to compact a document's log before this
-// limit is reached. This is a deliberate, visible failure standing in
-// for that missing internal/docs wiring, not a bug.
-var ErrUpdateLogFull = errors.New("update log exceeds the retained limit; document needs compaction (not yet wired up on the host side) before it can accept further updates")
+// comment: ABI v4 exports the snapshot/compaction primitive, this package
+// calls it (Handle.Snapshot, Handle.VerifySnapshot, below), and
+// internal/docs now persists the result, so an ordinary document replays
+// from a snapshot plus a short tail and never comes near this limit.
+//
+// It is still reachable, and deliberately so rather than raised or
+// removed. A document whose snapshot writes keep failing walks its tail
+// back to the cap; a document that was already past it before snapshots
+// shipped starts there. Both end here, refusing the write, which is the
+// one honest answer available — the alternative, dropping old entries to
+// make room, would make a later rebuild replay an incomplete history and
+// return silently wrong content. Refusing is visible; wrong content is
+// not. In both cases the host's next attempt writes a snapshot from the
+// replayed state before it applies anything (the snapshots design, D3),
+// so a caller that retries gets through.
+var ErrUpdateLogFull = errors.New("update log exceeds the retained limit; the document needs a snapshot before it can accept further updates")
// Option configures a Runtime. See WithGuestCallTimeout.
type Option func(*runtimeOptions)
diff --git a/internal/docs/snapshots.go b/internal/docs/snapshots.go
index 1544651..69dad70 100644
--- a/internal/docs/snapshots.go
+++ b/internal/docs/snapshots.go
@@ -4,6 +4,7 @@ import (
"context"
"errors"
"fmt"
+ "time"
"gorm.io/gorm"
@@ -50,3 +51,80 @@ func (s *Store) newestSnapshot(ctx context.Context, docID int64) (*DocSnapshot,
}
return &row, nil
}
+
+// writeSnapshot encodes h's current state as one Yjs update, asks the
+// core to verify it is this document, and persists it as doc_snapshots
+// row (doc.ID, seq) — then deletes every snapshot for the document
+// that this one supersedes.
+//
+// seq must be the highest seq h actually holds, and the caller is
+// responsible for that being true: see ApplyUpdate, where this is
+// called from the replayed state before the incoming update is
+// applied, which is what makes the number truthful (the snapshots
+// design, D3). Nothing here can check it. Verification compares the
+// candidate against h itself, so a snapshot faithfully copied from a
+// state that is missing a committed row passes perfectly — D5 says so
+// in as many words. Placement closes that hole, not this function.
+//
+// Every error is returned rather than logged here, so the one call
+// site logs each of them once, at Error, with the same doc_id and seq
+// keys: a failed encode, a failed verification and a failed insert all
+// mean the same thing to an operator, which is that this document did
+// not get the snapshot it was due. None of them may fail the update —
+// by the time this matters the log is what the document is, and it is
+// untouched.
+func (s *Store) writeSnapshot(ctx context.Context, doc *Document, h *core.Handle, seq int64) error {
+ candidate, err := h.Snapshot(ctx)
+ if err != nil {
+ return fmt.Errorf("docs: encode snapshot for document %d at seq %d: %w", doc.ID, seq, err)
+ }
+
+ // Verified before it is trusted, inside the core (D5): the
+ // candidate is replayed into a throwaway document there and
+ // compared by state vector, delete set and structural render. A
+ // false verdict is a refusal to write, never a warning attached to
+ // a row that gets written anyway — Store.replay reads whatever is
+ // on disk back and has no way to second-guess it beyond noticing
+ // that the core refused the bytes outright.
+ verified, err := h.VerifySnapshot(ctx, candidate)
+ if err != nil {
+ return fmt.Errorf("docs: verify snapshot for document %d at seq %d: %w", doc.ID, seq, err)
+ }
+ if !verified {
+ return fmt.Errorf("docs: snapshot for document %d at seq %d did not verify as this document; not written", doc.ID, seq)
+ }
+
+ row := &DocSnapshot{
+ DocID: doc.ID,
+ Seq: seq,
+ State: candidate,
+ ABIVersion: coreABIVersion,
+ CreatedAt: time.Now().UTC(),
+ }
+ return s.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
+ if err := tx.Create(row).Error; err != nil {
+ return fmt.Errorf("docs: persist snapshot for document %d at seq %d: %w", doc.ID, seq, err)
+ }
+
+ // Superseded snapshots are derived data, reproducible from a
+ // log that keeps every row, so deleting them deletes no
+ // history — and retaining them would cost one whole document
+ // copy per snapshotCadence updates, which is the difference
+ // between an accelerator and a second, fatter copy of the
+ // archive.
+ //
+ // "Superseded" is (seq, id) ordered exactly as newestSnapshot
+ // reads: strictly lower seq, or the same seq and a lower id.
+ // Not simply "every other row". Two concurrent boundary POSTs
+ // can each insert an equivalent snapshot at the same seq
+ // (migration 0004: (doc_id, seq) is deliberately not unique),
+ // and a rule of "delete everything that is not mine" would
+ // have each of them delete the other's row and leave the
+ // document with none.
+ if err := tx.Where("doc_id = ? AND (seq < ? OR (seq = ? AND id < ?))", doc.ID, row.Seq, row.Seq, row.ID).
+ Delete(&DocSnapshot{}).Error; err != nil {
+ return fmt.Errorf("docs: delete superseded snapshots for document %d at seq %d: %w", doc.ID, seq, err)
+ }
+ return nil
+ })
+}
diff --git a/internal/docs/snapshots_test.go b/internal/docs/snapshots_test.go
index 187f750..70b6422 100644
--- a/internal/docs/snapshots_test.go
+++ b/internal/docs/snapshots_test.go
@@ -2,8 +2,11 @@ package docs
import (
"context"
+ "sync"
"testing"
"time"
+
+ "github.com/carlosframework/docs/internal/core"
)
func TestNewestSnapshotPrefersHighestSeqThenHighestID(t *testing.T) {
@@ -99,3 +102,358 @@ func TestNewestSnapshotIsNilWhenThereAreNone(t *testing.T) {
t.Errorf("newestSnapshot = (%v, %v), want (nil, nil)", got, err)
}
}
+
+// The central property (the snapshots design, D6 and its §8): a
+// document replayed through snapshot-then-tail must render identically
+// to one replayed from the whole log. Asserted against the rendered
+// document rather than against the snapshot row, because the row is
+// exactly what a stale snapshot makes look right.
+func TestSnapshotThenTailRendersIdenticallyToAFullReplay(t *testing.T) {
+ app, err := NewForTest(t)
+ if err != nil {
+ t.Fatalf("NewForTest: %v", err)
+ }
+ ctx := context.Background()
+ owner := app.TestPerson(t, "owner@example.com")
+ doc, err := app.Store().Create(ctx, owner, "Equivalence")
+ if err != nil {
+ t.Fatalf("Create: %v", err)
+ }
+
+ update := validUpdate(t)
+ for i := 0; i < snapshotCadence+5; i++ {
+ if _, err := app.Store().ApplyUpdate(ctx, doc, update); err != nil {
+ t.Fatalf("ApplyUpdate %d: %v", i, err)
+ }
+ }
+ snap, err := app.Store().newestSnapshot(ctx, doc.ID)
+ if err != nil {
+ t.Fatalf("newestSnapshot: %v", err)
+ }
+ if snap == nil {
+ t.Fatal("no snapshot exists past the cadence, so nothing below exercises snapshot-then-tail")
+ }
+
+ withSnapshot, err := app.Store().HTML(ctx, doc)
+ if err != nil {
+ t.Fatalf("HTML: %v", err)
+ }
+
+ // Delete every snapshot and read again: same document, full replay.
+ if err := app.Store().db.WithContext(ctx).
+ Where("doc_id = ?", doc.ID).Delete(&DocSnapshot{}).Error; err != nil {
+ t.Fatalf("delete snapshots: %v", err)
+ }
+ fullReplay, err := app.Store().HTML(ctx, doc)
+ if err != nil {
+ t.Fatalf("HTML after deleting snapshots: %v", err)
+ }
+ if withSnapshot != fullReplay {
+ t.Error("snapshot-then-tail and full replay disagree about the document")
+ }
+}
+
+// The snapshots design's D3, first defect: a snapshot must never claim
+// a seq whose content it does not contain. Asserted against the
+// rendered document, never against the snapshot row — the row is
+// exactly what a stale snapshot makes look right.
+//
+// Two things this test needs that are easy to leave out, and without
+// either of which it passes no matter what the implementation does.
+//
+// It warms up to exactly snapshotCadence rows, not one short of it.
+// Replay runs outside ApplyUpdate's transaction, so with 255 rows in
+// hand all eight racers can (and in practice do) replay before any of
+// them commits, every one of them measures a distance of 255, and not
+// one snapshot is written — the boundary the test is named for is
+// never crossed.
+//
+// And exactly one racer posts different content. The rest post the
+// same bytes as the warmup, which Yjs merges idempotently: if every
+// row said the same thing, a row a stale snapshot dropped would be
+// invisible in the render, and this comparison could not fail. The one
+// odd update is the content a snapshot taken after a successful Apply
+// would strand — its writer's handle never saw that row, but its seq
+// column would cover it, so the tail after the snapshot skips it too.
+func TestConcurrentPostsAtTheCadenceBoundaryDoNotStrandContent(t *testing.T) {
+ app, err := NewForTest(t)
+ if err != nil {
+ t.Fatalf("NewForTest: %v", err)
+ }
+ ctx := context.Background()
+ owner := app.TestPerson(t, "owner@example.com")
+ doc, err := app.Store().Create(ctx, owner, "Boundary")
+ if err != nil {
+ t.Fatalf("Create: %v", err)
+ }
+
+ for i := 0; i < snapshotCadence; i++ {
+ if _, err := app.Store().ApplyUpdate(ctx, doc, validUpdate(t)); err != nil {
+ t.Fatalf("warmup %d: %v", i, err)
+ }
+ }
+
+ distinct := ReadFragmentFixture(t, "concurrent")
+ var wg sync.WaitGroup
+ for i := 0; i < 8; i++ {
+ wg.Add(1)
+ go func(i int) {
+ defer wg.Done()
+ update := validUpdate(t)
+ if i == 3 {
+ update = distinct
+ }
+ _, _ = app.Store().ApplyUpdate(ctx, doc, update)
+ }(i)
+ }
+ wg.Wait()
+
+ // Every racer measured a distance of at least snapshotCadence, so
+ // a snapshot must exist — otherwise the comparison below is
+ // between two identical full replays and proves nothing.
+ snap, err := app.Store().newestSnapshot(ctx, doc.ID)
+ if err != nil || snap == nil {
+ t.Fatalf("newestSnapshot = (%v, %v), want a snapshot written at the boundary", snap, err)
+ }
+
+ withSnapshot, err := app.Store().HTML(ctx, doc)
+ if err != nil {
+ t.Fatalf("HTML: %v", err)
+ }
+ if err := app.Store().db.WithContext(ctx).
+ Where("doc_id = ?", doc.ID).Delete(&DocSnapshot{}).Error; err != nil {
+ t.Fatalf("delete snapshots: %v", err)
+ }
+ fullReplay, err := app.Store().HTML(ctx, doc)
+ if err != nil {
+ t.Fatalf("HTML: %v", err)
+ }
+ if withSnapshot != fullReplay {
+ t.Errorf("a snapshot written during concurrent writes stranded content the log still holds:\n through the snapshot: %q\n full replay: %q", withSnapshot, fullReplay)
+ }
+}
+
+// The same property as above, stated directly rather than through a
+// render that happens to differ: the seq a snapshot claims must be
+// truthful about the rows it contains. Replaying doc_updates through
+// snap.Seq and applying snap alone must produce the same document.
+//
+// This is the assertion D5's in-core verification structurally cannot
+// make — it compares a candidate against the very handle that produced
+// it, so a faithful copy of a stale state passes it perfectly. Only a
+// comparison against the log, which is the authority, can catch that.
+func TestASnapshotContainsEveryRowItsSeqClaims(t *testing.T) {
+ app, err := NewForTest(t)
+ if err != nil {
+ t.Fatalf("NewForTest: %v", err)
+ }
+ ctx := context.Background()
+ owner := app.TestPerson(t, "owner@example.com")
+ doc, err := app.Store().Create(ctx, owner, "Truthful seq")
+ if err != nil {
+ t.Fatalf("Create: %v", err)
+ }
+
+ for i := 0; i < snapshotCadence; i++ {
+ if _, err := app.Store().ApplyUpdate(ctx, doc, validUpdate(t)); err != nil {
+ t.Fatalf("warmup %d: %v", i, err)
+ }
+ }
+ distinct := ReadFragmentFixture(t, "concurrent")
+ var wg sync.WaitGroup
+ for i := 0; i < 8; i++ {
+ wg.Add(1)
+ go func(i int) {
+ defer wg.Done()
+ update := validUpdate(t)
+ if i == 3 {
+ update = distinct
+ }
+ _, _ = app.Store().ApplyUpdate(ctx, doc, update)
+ }(i)
+ }
+ wg.Wait()
+
+ var snaps []DocSnapshot
+ if err := app.Store().db.WithContext(ctx).
+ Where("doc_id = ?", doc.ID).Order("seq ASC, id ASC").Find(&snaps).Error; err != nil {
+ t.Fatalf("load snapshots: %v", err)
+ }
+ if len(snaps) == 0 {
+ t.Fatal("no snapshot was written at the boundary")
+ }
+
+ for _, snap := range snaps {
+ fromSnapshot := renderInAFreshHandle(t, app, doc, [][]byte{snap.State})
+ fromLog := renderInAFreshHandle(t, app, doc, updateBinsThrough(t, app, doc, snap.Seq))
+ if fromSnapshot != fromLog {
+ t.Errorf("snapshot at seq %d does not contain everything the log holds through that seq:\n snapshot: %q\n log: %q", snap.Seq, fromSnapshot, fromLog)
+ }
+ }
+}
+
+// updateBinsThrough returns doc's update_bin blobs with seq <= through,
+// in seq order — the log's own answer to "what is this document at
+// that seq", which is the only authority a snapshot can be checked
+// against.
+func updateBinsThrough(t *testing.T, app *App, doc *Document, through int64) [][]byte {
+ t.Helper()
+ var rows []DocUpdate
+ if err := app.Store().db.WithContext(context.Background()).
+ Where("doc_id = ? AND seq <= ?", doc.ID, through).
+ Order("seq ASC").Find(&rows).Error; err != nil {
+ t.Fatalf("load updates through seq %d: %v", through, err)
+ }
+ bins := make([][]byte, 0, len(rows))
+ for _, r := range rows {
+ bins = append(bins, r.UpdateBin)
+ }
+ return bins
+}
+
+// renderInAFreshHandle applies updates, in order, to a brand new core
+// handle for doc and returns its HTML. Deliberately not Store.HTML:
+// that reads whatever is in the database, and these two callers need
+// to render an explicit list of bytes instead.
+func renderInAFreshHandle(t *testing.T, app *App, doc *Document, updates [][]byte) string {
+ t.Helper()
+ ctx := context.Background()
+ h, err := app.Store().core.NewDoc(ctx, doc.GUID, doc.ClientID)
+ if err != nil {
+ t.Fatalf("NewDoc: %v", err)
+ }
+ defer h.Close(ctx)
+ for i, u := range updates {
+ if err := h.Apply(ctx, u); err != nil {
+ t.Fatalf("apply update %d of %d: %v", i, len(updates), err)
+ }
+ }
+ html, err := h.HTML(ctx)
+ if err != nil {
+ t.Fatalf("HTML: %v", err)
+ }
+ return html
+}
+
+// The snapshots design's D3, second defect, and the case its §1 opens
+// with: a document that is ALREADY at the cap when this ships. Driven
+// through the new path a document never reaches the cap, so this
+// builds the production shape directly — a full log and no snapshot —
+// and asserts it recovers.
+func TestADocumentAlreadyAtTheCapRecovers(t *testing.T) {
+ if testing.Short() {
+ t.Skip("replays a full retained log; run without -short")
+ }
+ app, err := NewForTest(t)
+ if err != nil {
+ t.Fatalf("NewForTest: %v", err)
+ }
+ ctx := context.Background()
+ owner := app.TestPerson(t, "owner@example.com")
+ doc, err := app.Store().Create(ctx, owner, "At the cap")
+ if err != nil {
+ t.Fatalf("Create: %v", err)
+ }
+
+ // Write rows straight into the log, bypassing ApplyUpdate, so no
+ // snapshot is ever taken — exactly the state a document edited
+ // before this feature shipped is in.
+ update := validUpdate(t)
+ rows := make([]DocUpdate, 0, core.MaxRetainedUpdates)
+ for i := 1; i <= core.MaxRetainedUpdates; i++ {
+ rows = append(rows, DocUpdate{DocID: doc.ID, Seq: int64(i), UpdateBin: update, CreatedAt: time.Now().UTC()})
+ }
+ if err := app.Store().db.WithContext(ctx).CreateInBatches(rows, 500).Error; err != nil {
+ t.Fatalf("seed a full log: %v", err)
+ }
+
+ // The first write still fails — the handle is full before the
+ // snapshot can help — but it must leave a snapshot behind.
+ if _, err := app.Store().ApplyUpdate(ctx, doc, update); err == nil {
+ t.Log("first write succeeded; acceptable if the cap was not reached")
+ }
+ snap, err := app.Store().newestSnapshot(ctx, doc.ID)
+ if err != nil {
+ t.Fatalf("newestSnapshot: %v", err)
+ }
+ if snap == nil {
+ t.Fatal("a capped document got no snapshot, so it can never recover")
+ }
+
+ // The retry replays snapshot + empty tail and succeeds.
+ if _, err := app.Store().ApplyUpdate(ctx, doc, update); err != nil {
+ t.Errorf("retry after a snapshot existed still failed: %v", err)
+ }
+}
+
+// A corrupted snapshot row must not corrupt the document: the log is
+// authoritative, so an unreadable snapshot is discarded and the whole
+// log replayed instead. Correct, and merely slower.
+func TestACorruptSnapshotFallsBackToTheLog(t *testing.T) {
+ app, err := NewForTest(t)
+ if err != nil {
+ t.Fatalf("NewForTest: %v", err)
+ }
+ ctx := context.Background()
+ owner := app.TestPerson(t, "owner@example.com")
+ doc, err := app.Store().Create(ctx, owner, "Corrupt")
+ if err != nil {
+ t.Fatalf("Create: %v", err)
+ }
+ for i := 0; i < snapshotCadence+2; i++ {
+ if _, err := app.Store().ApplyUpdate(ctx, doc, validUpdate(t)); err != nil {
+ t.Fatalf("ApplyUpdate %d: %v", i, err)
+ }
+ }
+ want, err := app.Store().HTML(ctx, doc)
+ if err != nil {
+ t.Fatalf("HTML: %v", err)
+ }
+ // Without this, the corruption below would update zero rows and
+ // the whole test would pass vacuously.
+ if snap, err := app.Store().newestSnapshot(ctx, doc.ID); err != nil || snap == nil {
+ t.Fatalf("newestSnapshot = (%v, %v), want a row to corrupt", snap, err)
+ }
+
+ if err := app.Store().db.WithContext(ctx).Model(&DocSnapshot{}).
+ Where("doc_id = ?", doc.ID).
+ Update("state", []byte{0xff, 0x00, 0x13, 0x37}).Error; err != nil {
+ t.Fatalf("corrupt the snapshot: %v", err)
+ }
+
+ got, err := app.Store().HTML(ctx, doc)
+ if err != nil {
+ t.Fatalf("HTML with a corrupt snapshot: %v", err)
+ }
+ if got != want {
+ t.Error("a corrupt snapshot changed the document instead of being discarded")
+ }
+}
+
+// The cliff itself, driven the long way round: more updates than
+// core.MaxRetainedUpdates through the real path, which must keep
+// accepting them. Slow — thousands of wasm applies — so it is skipped
+// under -short, but it is the one test that proves the headline claim
+// rather than a proxy for it.
+func TestADocumentPastTheRetainedCapKeepsAcceptingWrites(t *testing.T) {
+ if testing.Short() {
+ t.Skip("thousands of updates; run without -short")
+ }
+ app, err := NewForTest(t)
+ if err != nil {
+ t.Fatalf("NewForTest: %v", err)
+ }
+ ctx := context.Background()
+ owner := app.TestPerson(t, "owner@example.com")
+ doc, err := app.Store().Create(ctx, owner, "Past the cap")
+ if err != nil {
+ t.Fatalf("Create: %v", err)
+ }
+
+ update := validUpdate(t)
+ for i := 0; i < core.MaxRetainedUpdates+50; i++ {
+ if _, err := app.Store().ApplyUpdate(ctx, doc, update); err != nil {
+ t.Fatalf("ApplyUpdate %d of %d: %v", i, core.MaxRetainedUpdates+50, err)
+ }
+ }
+}
diff --git a/internal/docs/store.go b/internal/docs/store.go
index d2dfbd0..454ce4b 100644
--- a/internal/docs/store.go
+++ b/internal/docs/store.go
@@ -337,33 +337,88 @@ func (s *Store) Get(ctx context.Context, owner int64, sqid string) (*Document, e
}
// replay opens a fresh core.Handle for doc (guid and client id come
-// from the stored row, never from yrs — see Document) and applies
-// every update in doc_updates, in seq order, rebuilding the
-// document's CRDT state from scratch. The caller owns the returned
-// Handle and must Close it.
+// from the stored row, never from yrs — see Document) and rebuilds the
+// document's CRDT state into it: the newest usable snapshot, then only
+// the doc_updates rows after it, in seq order. With no usable snapshot
+// it applies every row instead. The caller owns the returned Handle
+// and must Close it.
//
-// This is a deliberate O(n) full replay on every call, matching
-// internal/core's own recovery strategy (its package doc comment:
-// "Rebuilding is log replay, not snapshot restore" — ABI v1 exports
-// no state-encode function, so there is no snapshot to load instead).
-// That is fine at this plan's scale (a handful of updates per
-// document); a document with a long history needs compaction
-// (ErrUpdateLogFull's doc comment) or ABI v2 snapshot support,
-// neither of which exists yet — replaying long histories here would
-// be papering over that gap rather than living within it.
-func (s *Store) replay(ctx context.Context, doc *Document) (*core.Handle, error) {
- h, err := s.core.NewDoc(ctx, doc.GUID, doc.ClientID)
+// Snapshot-then-tail is the snapshots design's D6. It is what keeps
+// the retained-update count in a core.Handle clear of
+// core.MaxRetainedUpdates: while snapshot writes keep succeeding,
+// replay is one snapshot plus roughly snapshotCadence updates, so the
+// cliff ErrUpdateLogFull describes stops being reachable in the
+// ordinary case. That bound is conditional on those writes, and
+// snapshotCadence's own comment says what it costs when they stop
+// succeeding. The export it restores
+// from (ABI v4's docs_core_snapshot_len/read, via core.Handle.Snapshot)
+// is exactly the one core.go's maxRetainedUpdates comment asks for;
+// ApplyUpdate below is the host caller that persists it.
+//
+// A snapshot is an accelerator, never the source of truth (D2).
+// doc_updates keeps every row, and any snapshot that cannot be read —
+// unloadable row, bytes the core refuses, an ABI newestSnapshot
+// already filters out — is discarded and the whole log replayed
+// instead. That fallback is slower and always correct, which is the
+// bar every failure path in this file is held to.
+//
+// The two seqs come back with the handle rather than being asked for
+// again: replayedSeq is the highest seq this replay actually
+// incorporated (the last row applied, or the snapshot's own seq when
+// the tail was empty), and lastSnapshotSeq is the seq of the snapshot
+// it started from, or 0 when it started from nothing — including when
+// a snapshot existed but was discarded, so the next write repairs it.
+// ApplyUpdate's cadence check is the difference between them, computed
+// from rows already read and never from a second query — the same
+// discipline the live-sync design's D2 sets for Docs-Seq, for the same
+// reason. Text, HTML and JSON below discard both: reading a document
+// never writes a snapshot, because a snapshot is only ever taken on
+// the write path (D3), where a request has already proved the instance
+// is awake.
+func (s *Store) replay(ctx context.Context, doc *Document) (h *core.Handle, replayedSeq int64, lastSnapshotSeq int64, err error) {
+ h, err = s.core.NewDoc(ctx, doc.GUID, doc.ClientID)
if err != nil {
- return nil, fmt.Errorf("docs: open core handle for document %d: %w", doc.ID, err)
+ return nil, 0, 0, fmt.Errorf("docs: open core handle for document %d: %w", doc.ID, err)
+ }
+
+ // A snapshot that cannot even be read out of the database is not
+ // an error to return: the log alone reproduces the document, so
+ // this is logged and abandoned rather than propagated.
+ snap, snapErr := s.newestSnapshot(ctx, doc.ID)
+ if snapErr != nil {
+ s.logger.Error("docs: load newest snapshot, replaying the whole log instead", "doc_id", doc.ID, "err", snapErr)
+ snap = nil
+ }
+ if snap != nil {
+ if applyErr := h.Apply(ctx, snap.State); applyErr != nil {
+ s.logger.Error("docs: apply snapshot, replaying the whole log instead", "doc_id", doc.ID, "snapshot_seq", snap.Seq, "err", applyErr)
+ // The handle is discarded along with the snapshot rather
+ // than reused. Handle.Apply leaves the instance usable
+ // after a clean rejection and rebuilds it after a trap,
+ // but its third shape — a rebuild that itself failed —
+ // makes no such promise, and a fresh NewDoc on this
+ // already-exceptional path costs less than reasoning
+ // about which shape this was.
+ h.Close(ctx)
+ snap = nil
+ h, err = s.core.NewDoc(ctx, doc.GUID, doc.ClientID)
+ if err != nil {
+ return nil, 0, 0, fmt.Errorf("docs: reopen core handle for document %d after a bad snapshot: %w", doc.ID, err)
+ }
+ }
+ }
+
+ tail := s.db.WithContext(ctx).Where("doc_id = ?", doc.ID)
+ if snap != nil {
+ lastSnapshotSeq = snap.Seq
+ replayedSeq = snap.Seq
+ tail = tail.Where("seq > ?", snap.Seq)
}
var updates []DocUpdate
- if err := s.db.WithContext(ctx).
- Where("doc_id = ?", doc.ID).
- Order("seq ASC").
- Find(&updates).Error; err != nil {
+ if err := tail.Order("seq ASC").Find(&updates).Error; err != nil {
h.Close(ctx)
- return nil, fmt.Errorf("docs: load updates for document %d: %w", doc.ID, err)
+ return nil, 0, 0, fmt.Errorf("docs: load updates for document %d: %w", doc.ID, err)
}
for _, u := range updates {
@@ -376,10 +431,11 @@ func (s *Store) replay(ctx context.Context, doc *Document) (*core.Handle, error)
// itself wrote can no longer be trusted to reproduce the
// document, and callers must see that rather than silently
// getting a wrong or truncated text back.
- return nil, fmt.Errorf("docs: replay update seq %d for document %d: %w", u.Seq, doc.ID, err)
+ return nil, 0, 0, fmt.Errorf("docs: replay update seq %d for document %d: %w", u.Seq, doc.ID, err)
}
+ replayedSeq = u.Seq
}
- return h, nil
+ return h, replayedSeq, lastSnapshotSeq, nil
}
// Text returns doc's current text — read from the real CRDT core by
@@ -388,7 +444,7 @@ func (s *Store) replay(ctx context.Context, doc *Document) (*core.Handle, error)
// titles) is out of scope for this plan; this plaintext replay is not
// a decision that content stays unsealed.
func (s *Store) Text(ctx context.Context, doc *Document) (string, error) {
- h, err := s.replay(ctx, doc)
+ h, _, _, err := s.replay(ctx, doc)
if err != nil {
return "", err
}
@@ -416,7 +472,7 @@ func (s *Store) Text(ctx context.Context, doc *Document) (string, error) {
// escaping that earns it, rather than left for views/show.html to
// assert with a {{ ... | html }} pipeline of its own.
func (s *Store) HTML(ctx context.Context, doc *Document) (template.HTML, error) {
- h, err := s.replay(ctx, doc)
+ h, _, _, err := s.replay(ctx, doc)
if err != nil {
return "", err
}
@@ -439,7 +495,7 @@ func (s *Store) HTML(ctx context.Context, doc *Document) (template.HTML, error)
// typed return records has no counterpart here — a JSON string that
// reached a template would need escaping like any other data.
func (s *Store) JSON(ctx context.Context, doc *Document) (string, error) {
- h, err := s.replay(ctx, doc)
+ h, _, _, err := s.replay(ctx, doc)
if err != nil {
return "", err
}
@@ -543,12 +599,45 @@ func (s *Store) ApplyUpdate(ctx context.Context, doc *Document, update []byte) (
return 0, fmt.Errorf("docs: apply update to document %d: verify owner %d: %w", doc.ID, doc.OwnerID, err)
}
- h, err := s.replay(ctx, doc)
+ h, replayedSeq, lastSnapshotSeq, err := s.replay(ctx, doc)
if err != nil {
return 0, err
}
defer h.Close(ctx)
+ // The snapshot is taken HERE — from the replayed state, before the
+ // incoming update is applied — and the placement is the whole of
+ // its correctness (the snapshots design, D3).
+ //
+ // Taking it after a successful Apply is wrong twice. The replay
+ // above runs outside the transaction below, so a concurrent POST
+ // can commit a row this handle never saw; a snapshot taken from
+ // that handle would claim a seq whose content it does not hold,
+ // and D5's verification cannot notice, because it compares the
+ // candidate against this same stale handle. And a document already
+ // at core.MaxRetainedUpdates never reaches an after-success hook
+ // at all — Apply refuses with ErrUpdateLogFull first — so the one
+ // case this feature exists for would stay broken forever. Written
+ // here, that document gets its snapshot, fails this one POST, and
+ // the client's own retry replays snapshot plus an empty tail and
+ // succeeds.
+ //
+ // The replayed prefix is complete by construction: seq is gapless
+ // (MAX+1 inside the single-writer transaction below) and replay
+ // read every row that existed when it ran. So replayedSeq is
+ // unconditionally truthful about what the handle holds.
+ if replayedSeq-lastSnapshotSeq >= snapshotCadence {
+ if err := s.writeSnapshot(ctx, doc, h, replayedSeq); err != nil {
+ // Never fatal: the update below still stands, the log is
+ // untouched, and the next POST tries again — and so does
+ // every one after it, since the distance only grows until
+ // a write succeeds. Error rather than Warn because the
+ // headroom that keeps ErrUpdateLogFull unreachable is
+ // spent by roughly fifteen consecutive failures here (D4).
+ s.logger.Error("docs: write snapshot", "doc_id", doc.ID, "seq", replayedSeq, "err", err)
+ }
+ }
+
if err := h.Apply(ctx, update); err != nil {
return 0, fmt.Errorf("docs: apply update to document %d: %w", doc.ID, &updateRejection{err})
}