amadan

oficina / docs Public

Task 5b: a long log no longer bricks a document without a snapshot

Snapshots removed the ceiling that used to keep doc_updates under
core.MaxRetainedUpdates: ApplyUpdate refused once the handle filled, so
no row past 4096 was ever written. Now a log can be arbitrarily long —
and when its only snapshot becomes unusable (a corrupt row, or the first
ABI bump, which newestSnapshot's exact-match filter invalidates every
snapshot with at once) replay fell back to the whole log, met
ErrUpdateLogFull at row 4097 and returned an error. Text, HTML and JSON
all go through replay, so the document was unreadable rather than slow;
and ApplyUpdate returned before the writeSnapshot that would have
repaired it, so it never healed on its own.

replay's tail loop now re-seeds instead of failing: it snapshots the
handle it has (a valid prefix), opens a fresh one from the document
row's guid and client id, applies that state into it, and carries on —
as often as the log requires. Purely in memory: replay is the read path,
and a read must not write. The persisted snapshot is still refreshed the
ordinary way, by the next write past the cadence, at the current ABI.

If the mid-replay snapshot itself fails, the error is returned rather
than swallowed — carrying on with a full handle would reintroduce the
same failure one row later — and every message names the document and
the seq the replay had reached.

Four tests, each watched failing first: a past-the-cap log with no
snapshot reads; the same document still accepts writes; the ABI-bump
case specifically; and the re-seeded replay renders exactly what the
same content renders in a handle that never crossed the cap.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Paul Campbell pushed by paul@keymail.dev b4d2ed314550a466a562c70cf705e56143a4ccf2 parent 8f4ed42
2 files changed, +351 −2
  • internal/docs/snapshots_test.go +257 −0
  • internal/docs/store.go +94 −2
diff --git a/internal/docs/snapshots_test.go b/internal/docs/snapshots_test.go
index 151de13..8547c58 100644
--- a/internal/docs/snapshots_test.go
+++ b/internal/docs/snapshots_test.go
@@ -3,6 +3,7 @@ package docs
import (
"context"
"slices"
+ "strings"
"sync"
"testing"
"time"
@@ -624,3 +625,259 @@ func TestADocumentPastTheRetainedCapKeepsAcceptingWrites(t *testing.T) {
}
}
}
+
+// seedLongLog writes total rows straight into doc_updates, bypassing
+// ApplyUpdate so no snapshot is ever taken along the way, and returns
+// the distinct update blobs it wrote, in seq order of first appearance.
+//
+// Every row is validUpdate's bytes except the seqs named in distinct,
+// which get the blob given for them. Duplicate Yjs updates merge
+// idempotently, so those distinct rows are the only ones carrying
+// content nothing else in the log holds — which makes the returned
+// slice the same document expressed in a handful of updates, short
+// enough to replay without approaching core.MaxRetainedUpdates at all.
+// That is what the equivalence test below compares against.
+func seedLongLog(t *testing.T, app *App, doc *Document, total int, distinct map[int][]byte) [][]byte {
+ t.Helper()
+ ordinary := validUpdate(t)
+ rows := make([]DocUpdate, 0, total)
+ unique := make([][]byte, 0, len(distinct)+1)
+ seen := make(map[string]bool, len(distinct)+1)
+ for seq := 1; seq <= total; seq++ {
+ update := ordinary
+ if d, ok := distinct[seq]; ok {
+ update = d
+ }
+ rows = append(rows, DocUpdate{DocID: doc.ID, Seq: int64(seq), UpdateBin: update, CreatedAt: time.Now().UTC()})
+ if !seen[string(update)] {
+ seen[string(update)] = true
+ unique = append(unique, update)
+ }
+ }
+ if err := app.Store().db.WithContext(context.Background()).CreateInBatches(rows, 500).Error; err != nil {
+ t.Fatalf("seed a %d-row log: %v", total, err)
+ }
+ return unique
+}
+
+// pastTheCapMarker is text the concurrent fixture contributes and
+// validUpdate's bytes do not, so its presence in a rendered document
+// says the row carrying that fixture was really applied.
+// assertHoldsTheWholeLog checks that claim rather than trusting it.
+const pastTheCapMarker = "BBB"
+
+// longLogDistinctRows is the shape every test below seeds: a log past
+// core.MaxRetainedUpdates whose one distinct update sits PAST the cap,
+// among rows only a replay that carries on beyond core.ErrUpdateLogFull
+// ever applies. That placement is what makes "still reads" mean "reads
+// the whole document" rather than "reads its first 4096 rows" — a
+// replay that stopped at the cap and returned what it had would render
+// without pastTheCapMarker and fail.
+//
+// One distinct fixture, not several. The fragment fixtures are each
+// generated from their own yjs document, and their struct ids overlap,
+// so merging two of them into one document does not concatenate their
+// content — the equivalence test above uses exactly one for the same
+// reason.
+func longLogDistinctRows(t *testing.T) (total int, distinct map[int][]byte) {
+ t.Helper()
+ return core.MaxRetainedUpdates + 50, map[int][]byte{
+ core.MaxRetainedUpdates + 40: ReadFragmentFixture(t, "concurrent"),
+ }
+}
+
+// assertHoldsTheWholeLog fails unless html holds both ends of a log
+// seeded by seedLongLog with longLogDistinctRows' placement: the
+// ordinary rows' own paragraphs, and pastTheCapMarker from the row
+// beyond core.MaxRetainedUpdates.
+//
+// It first renders the ordinary update on its own and requires that
+// NOT to contain the marker. Without that, a fixture whose content the
+// ordinary rows happened to carry too would make the interesting half
+// of this assertion vacuous.
+func assertHoldsTheWholeLog(t *testing.T, app *App, doc *Document, html string) {
+ t.Helper()
+ ordinaryOnly := renderInAFreshHandle(t, app, doc, [][]byte{validUpdate(t)})
+ if strings.Contains(ordinaryOnly, pastTheCapMarker) {
+ t.Fatalf("the ordinary update already renders %q, so finding it below would prove nothing:\n%s", pastTheCapMarker, ordinaryOnly)
+ }
+ for _, want := range []string{"First paragraph.", pastTheCapMarker} {
+ if !strings.Contains(html, want) {
+ t.Errorf("the rendered document is missing %q:\n%s", want, html)
+ }
+ }
+}
+
+// A document whose log is longer than core.MaxRetainedUpdates and has
+// no usable snapshot must still READ. This is the failure snapshots
+// themselves created: before them ApplyUpdate refused once the handle
+// filled, so no log could ever exceed the cap; now one can, and when
+// its only snapshot goes away — a corrupt row, or the first ABI bump,
+// which newestSnapshot's exact-match filter invalidates every snapshot
+// with at once — replay runs the whole log into one handle, meets
+// ErrUpdateLogFull at row 4096 and returns an error. Text, HTML and
+// JSON all go through replay, so such a document is not slow, it is
+// unreadable.
+func TestALongLogWithNoUsableSnapshotStillReads(t *testing.T) {
+ if testing.Short() {
+ t.Skip("replays more than 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, "Long log, no snapshot")
+ if err != nil {
+ t.Fatalf("Create: %v", err)
+ }
+
+ total, distinct := longLogDistinctRows(t)
+ seedLongLog(t, app, doc, total, distinct)
+ if snap, err := app.Store().newestSnapshot(ctx, doc.ID); err != nil || snap != nil {
+ t.Fatalf("newestSnapshot = (%v, %v), want no snapshot at all — that is the whole premise of this test", snap, err)
+ }
+
+ html, err := app.Store().HTML(ctx, doc)
+ if err != nil {
+ t.Fatalf("HTML of a %d-row log with no snapshot: %v", total, err)
+ }
+ assertHoldsTheWholeLog(t, app, doc, string(html))
+}
+
+// …and it must still accept WRITES. The read failure above takes the
+// self-repair down with it: ApplyUpdate returns the moment replay
+// fails, long before the writeSnapshot block that rescues a capped
+// legacy document, so nothing about such a document ever improves on
+// its own.
+func TestALongLogWithNoUsableSnapshotStillAcceptsWrites(t *testing.T) {
+ if testing.Short() {
+ t.Skip("replays more than 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, "Long log, still writable")
+ if err != nil {
+ t.Fatalf("Create: %v", err)
+ }
+
+ total, distinct := longLogDistinctRows(t)
+ seedLongLog(t, app, doc, total, distinct)
+
+ seq, err := app.Store().ApplyUpdate(ctx, doc, validUpdate(t))
+ if err != nil {
+ t.Fatalf("ApplyUpdate on a %d-row log with no snapshot: %v", total, err)
+ }
+ if want := int64(total + 1); seq != want {
+ t.Errorf("ApplyUpdate wrote seq %d, want %d", seq, want)
+ }
+}
+
+// The ABI-bump case, spelled out because it is the one that will
+// actually happen rather than the one that might: newestSnapshot
+// filters on an exact abi_version match, so the first bump after this
+// ships invalidates every snapshot in the database simultaneously and
+// sends every long document to a full-log replay on the same deploy.
+// The snapshot written here is a real one — genuine bytes for this
+// document, at a seq the log really reached — unusable for the single
+// reason that its ABI is not ours.
+func TestALongLogWhoseOnlySnapshotIsFromAnotherABIStillReadsAndWrites(t *testing.T) {
+ if testing.Short() {
+ t.Skip("replays more than 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, "Long log, yesterday's ABI")
+ if err != nil {
+ t.Fatalf("Create: %v", err)
+ }
+
+ total, distinct := longLogDistinctRows(t)
+ unique := seedLongLog(t, app, doc, total, distinct)
+
+ // A real snapshot of this very document, stored under the previous
+ // ABI — what every existing row becomes the moment core's ABI is
+ // bumped.
+ h, err := app.Store().core.NewDoc(ctx, doc.GUID, doc.ClientID)
+ if err != nil {
+ t.Fatalf("NewDoc: %v", err)
+ }
+ for i, u := range unique {
+ if err := h.Apply(ctx, u); err != nil {
+ t.Fatalf("apply update %d of %d: %v", i, len(unique), err)
+ }
+ }
+ state, err := h.Snapshot(ctx)
+ if err != nil {
+ t.Fatalf("Snapshot: %v", err)
+ }
+ h.Close(ctx)
+ row := &DocSnapshot{DocID: doc.ID, Seq: int64(total), State: state, ABIVersion: coreABIVersion - 1, CreatedAt: time.Now().UTC()}
+ if err := app.Store().db.WithContext(ctx).Create(row).Error; err != nil {
+ t.Fatalf("insert an older-ABI snapshot: %v", err)
+ }
+ if snap, err := app.Store().newestSnapshot(ctx, doc.ID); err != nil || snap != nil {
+ t.Fatalf("newestSnapshot = (%v, %v), want nil — an older ABI's row must be unusable, or this test proves nothing", snap, err)
+ }
+
+ html, err := app.Store().HTML(ctx, doc)
+ if err != nil {
+ t.Fatalf("HTML after an ABI bump invalidated the snapshot: %v", err)
+ }
+ assertHoldsTheWholeLog(t, app, doc, string(html))
+
+ if _, err := app.Store().ApplyUpdate(ctx, doc, validUpdate(t)); err != nil {
+ t.Fatalf("ApplyUpdate after an ABI bump invalidated the snapshot: %v", err)
+ }
+}
+
+// Not failing is the smaller half; the property that makes re-seeding
+// mid-replay SAFE is that it changes nothing. A replay that crossed the
+// cap — and so continued from a snapshot of its own partial state,
+// possibly more than once — must render exactly what the same content
+// renders when replayed the ordinary way, in a handle that never came
+// near the cap at all.
+//
+// The comparison has teeth because the log carries a distinct update
+// past seq 4096: a re-seed that dropped the remaining rows, or one that
+// continued from an empty or stale snapshot, differs here rather than
+// merely being slower.
+func TestReSeedingMidReplayRendersTheSameDocument(t *testing.T) {
+ if testing.Short() {
+ t.Skip("replays more than 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, "Re-seeded, unchanged")
+ if err != nil {
+ t.Fatalf("Create: %v", err)
+ }
+
+ total, distinct := longLogDistinctRows(t)
+ unique := seedLongLog(t, app, doc, total, distinct)
+ if len(unique) >= core.MaxRetainedUpdates {
+ t.Fatalf("the comparison replay is %d updates long, which crosses the cap itself", len(unique))
+ }
+
+ reSeeded, err := app.Store().HTML(ctx, doc)
+ if err != nil {
+ t.Fatalf("HTML of a %d-row log: %v", total, err)
+ }
+ want := renderInAFreshHandle(t, app, doc, unique)
+ if string(reSeeded) != want {
+ t.Errorf("re-seeding mid-replay changed the document:\n re-seeded replay: %q\n never past the cap: %q", reSeeded, want)
+ }
+}
diff --git a/internal/docs/store.go b/internal/docs/store.go
index 454ce4b..f980cde 100644
--- a/internal/docs/store.go
+++ b/internal/docs/store.go
@@ -362,6 +362,18 @@ func (s *Store) Get(ctx context.Context, owner int64, sqid string) (*Document, e
// instead. That fallback is slower and always correct, which is the
// bar every failure path in this file is held to.
//
+// Slower, and correct even for a log longer than
+// core.MaxRetainedUpdates — which snapshots themselves made possible,
+// since before them ApplyUpdate refused once a handle filled and no log
+// could grow past the cap. A full-log replay of such a document would
+// meet core.ErrUpdateLogFull partway through and fail, leaving the
+// document unreadable AND unwritable (ApplyUpdate returns here, before
+// the writeSnapshot that would repair it). It would take one ABI bump
+// to do that to every long document at once, newestSnapshot's ABI
+// filter being exact. So the tail loop below does not fail there: it
+// re-seeds a fresh handle from the state it has reached, in memory,
+// and carries on. See reseedForReplay.
+//
// 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
@@ -422,7 +434,26 @@ func (s *Store) replay(ctx context.Context, doc *Document) (h *core.Handle, repl
}
for _, u := range updates {
- if err := h.Apply(ctx, u.UpdateBin); err != nil {
+ applyErr := h.Apply(ctx, u.UpdateBin)
+ if errors.Is(applyErr, core.ErrUpdateLogFull) {
+ // This handle has retained core.MaxRetainedUpdates updates
+ // and will accept no more, but the document is not over —
+ // there are rows left, and reaching them is the difference
+ // between a slow document and an unreadable one. So the
+ // replay continues in a fresh handle re-seeded from this
+ // one's state; see reseedForReplay.
+ //
+ // u itself was NOT applied: Handle.Apply's cap check is its
+ // first statement and returns before it touches the guest,
+ // so this row is still owed and is applied below, into the
+ // fresh handle, exactly once.
+ h, err = s.reseedForReplay(ctx, doc, h, u.Seq)
+ if err != nil {
+ return nil, 0, 0, err
+ }
+ applyErr = h.Apply(ctx, u.UpdateBin)
+ }
+ if applyErr != nil {
h.Close(ctx)
// Surfaced, not swallowed: whether this is a clean
// rejection, core.ErrUpdateLogFull, or a *core.RecoveredError
@@ -431,13 +462,74 @@ func (s *Store) replay(ctx context.Context, doc *Document) (h *core.Handle, repl
// 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, 0, 0, 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, applyErr)
}
replayedSeq = u.Seq
}
return h, replayedSeq, lastSnapshotSeq, nil
}
+// reseedForReplay returns a fresh handle holding everything full holds,
+// so a replay that has exhausted full's retained-update budget
+// (core.MaxRetainedUpdates, core.ErrUpdateLogFull) can carry on instead
+// of failing. full is closed either way; on any error the fresh handle
+// is closed too and nothing is returned for the caller to close.
+//
+// The mechanism is the one this whole feature is built on, used
+// in memory rather than through the database: full.Snapshot encodes its
+// entire state as one Yjs update (Handle.Snapshot, ABI v4's
+// docs_core_snapshot_len/read), and applying that to an empty document
+// reproduces it — the same property doc_snapshots relies on, and the
+// one TestReSeedingMidReplayRendersTheSameDocument pins here. The fresh
+// handle's retained count is 1 afterwards, so the next
+// core.MaxRetainedUpdates rows fit and a log of any length gets through
+// in as many re-seeds as it needs. That also puts a ceiling on what a
+// replay retains regardless of how long the log is, which is what the
+// cap was protecting in the first place.
+//
+// Three things this deliberately does not do:
+//
+// It writes nothing. replay is the read path — Text, HTML and JSON all
+// go through it — and a read that wrote to the database would be a
+// different change with different consequences (D3 keeps snapshot
+// writes on the write path, where a request has already proved the
+// instance is awake). The persisted snapshot is still repaired the
+// ordinary way, and getting here guarantees it: reaching the cap means
+// more than core.MaxRetainedUpdates rows sat past whatever this replay
+// started from, so replayedSeq-lastSnapshotSeq is past snapshotCadence
+// by a factor of sixteen, and this document's next ApplyUpdate runs
+// writeSnapshot — at the current ABI, which is what an ABI bump needs.
+//
+// It invents no identity. The fresh handle is opened exactly as replay
+// opens its first one, from the stored row's GUID and ClientID, so the
+// CRDT library is never a source of entropy here either (see Document).
+//
+// It never continues on a failure. If the snapshot, the fresh handle or
+// its re-seeding fails, this returns the error rather than carrying on
+// with full — carrying on would meet ErrUpdateLogFull again on the very
+// next row and reintroduce the failure one row later, with the cause
+// two frames further away. Every message names the document and the seq
+// the replay had reached, because this is the one path here where an
+// operator has to go and look at a specific document.
+func (s *Store) reseedForReplay(ctx context.Context, doc *Document, full *core.Handle, atSeq int64) (*core.Handle, error) {
+ state, err := full.Snapshot(ctx)
+ if err != nil {
+ full.Close(ctx)
+ return nil, fmt.Errorf("docs: replay of document %d reached the retained-update cap at seq %d and could not snapshot its own state to continue: %w", doc.ID, atSeq, err)
+ }
+
+ fresh, err := s.core.NewDoc(ctx, doc.GUID, doc.ClientID)
+ full.Close(ctx)
+ if err != nil {
+ return nil, fmt.Errorf("docs: replay of document %d reached the retained-update cap at seq %d and could not open a fresh core handle to continue: %w", doc.ID, atSeq, err)
+ }
+ if err := fresh.Apply(ctx, state); err != nil {
+ fresh.Close(ctx)
+ return nil, fmt.Errorf("docs: replay of document %d reached the retained-update cap at seq %d and could not re-seed a fresh handle from its own state: %w", doc.ID, atSeq, err)
+ }
+ return fresh, nil
+}
+
// Text returns doc's current text — read from the real CRDT core by
// replaying its update log, never from a database column. Sealing
// (spec §2 — per-row encryption of update_bin, same as document