amadan

oficina / docs Public

Plan 5 fix round: give the equivalence test teeth, and pin supersession

The equivalence test was near-vacuous, for exactly the reason the
concurrency test already had a fix for. Every update in it was the same
simple.bin bytes, which Yjs merges idempotently, so the tail after the
snapshot rebuilt the whole document on its own: an empty snapshot, or
one stranding any row, rendered identically and the test passed. What it
pinned was that a snapshot row exists and replay does not crash.

It is now property-shaped, as the spec's section 8 asks — two documents
at two cadences, one with its distinct row deep inside the snapshot's
range and one with it on the boundary itself, the last row the surviving
snapshot must contain. Storing the snapshot of an empty document at the
claimed seq now fails both cases; before, it passed.

Both concurrency tests read their fixtures on the test goroutine now.
validUpdate and ReadFragmentFixture call t.Fatalf, and t.Fatalf from a
racer marks the test failed and Goexits that one goroutine while the
other seven, the WaitGroup and every later assertion carry on. The
distinct racer's error comes back rather than being dropped, too: if
that single POST ever failed, the test would lose its only
discriminating content and pass with nothing left to detect.

Supersession had nothing pinning it. A test now writes a snapshot at
seq 30 over hand-seeded rows and then, as the slower of two boundary
writers would, one at seq 25 — the fresher row must survive, and another
document's rows must not be touched. Both simplifications are caught:
deleting every row that is not mine, and dropping the doc_id scope.

That test also corrected the comment above the predicate, which claimed
"delete everything that is not mine" would leave the document with no
snapshot at all. It would not: the insert and the delete share a
transaction on a single-connection write pool, so two writers serialise
and the second supersedes the first. The real reason for the narrow
predicate is that it never deletes a row this write does not supersede,
including one at a higher seq, and that is what it says now.

ErrUpdateLogFull's comment said a retry gets through in both of the two
cases it names. True for a document that predates snapshots, which gets
one on the POST it fails; false for a document whose snapshot writes
keep failing, whose retry fails the same way. The attempt is made in
both; only one heals.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Paul Campbell pushed by paul@keymail.dev fb483b6ec3826514b7f100cf26dd2f99528f14e0 parent 1cdd18a
3 files changed, +261 −72
  • internal/core/core.go +11 −2
  • internal/docs/snapshots.go +19 −6
  • internal/docs/snapshots_test.go +231 −64
diff --git a/internal/core/core.go b/internal/core/core.go
index 5ec562f..b817f15 100644
--- a/internal/core/core.go
+++ b/internal/core/core.go
@@ -159,9 +159,18 @@ const MaxRetainedUpdates = maxRetainedUpdates
// 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
+// not.
+//
+// The two cases do not recover alike, and only one of them heals on its
+// own. Every attempt the host makes tries to write a snapshot from the
// replayed state before it applies anything (the snapshots design, D3),
-// so a caller that retries gets through.
+// so the document that predates snapshots gets one on the very POST it
+// fails, and its caller's retry gets through — one visible failure, then
+// healed. The document whose snapshot writes keep failing gets no such
+// rescue: the retry's write fails for whatever reason the last fifteen
+// did, and it stays here until that cause is fixed. Those failures are
+// logged at Error on the way past, which is the only warning anyone
+// gets that a document is walking towards this.
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.
diff --git a/internal/docs/snapshots.go b/internal/docs/snapshots.go
index 69dad70..e632f65 100644
--- a/internal/docs/snapshots.go
+++ b/internal/docs/snapshots.go
@@ -115,12 +115,25 @@ func (s *Store) writeSnapshot(ctx context.Context, doc *Document, h *core.Handle
//
// "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.
+ // So what survives is always what newestSnapshot would return,
+ // and a row this write does not supersede is never touched —
+ // including one at a HIGHER seq.
+ //
+ // That last part is the reason not to write the simpler
+ // "delete every other row for this document". Two writers can
+ // cross the boundary together, and the one that commits second
+ // is not necessarily the one that replayed furthest: a writer
+ // working from a prefix that stopped at seq 260 can commit its
+ // snapshot after another has already committed one at seq 300.
+ // Under this predicate the seq 300 row is neither lower-seq nor
+ // an equal-seq lower id, so it stays and newestSnapshot keeps
+ // preferring it. Under "everything that is not mine" it would
+ // be deleted, and the document would quietly go back to
+ // replaying forty updates it had already been spared. Rows at
+ // the same seq are ordinary here rather than a collision to
+ // resolve — migration 0004 makes (doc_id, seq) non-unique for
+ // exactly this reason, since two boundary writers snapshotting
+ // the same prefix produce equivalent rows.
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)
diff --git a/internal/docs/snapshots_test.go b/internal/docs/snapshots_test.go
index 70b6422..151de13 100644
--- a/internal/docs/snapshots_test.go
+++ b/internal/docs/snapshots_test.go
@@ -2,6 +2,7 @@ package docs
import (
"context"
+ "slices"
"sync"
"testing"
"time"
@@ -108,48 +109,91 @@ func TestNewestSnapshotIsNilWhenThereAreNone(t *testing.T) {
// 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.
+//
+// Property-shaped rather than one fixture at one cadence, as §8 asks:
+// two documents, one snapshot apart and two snapshots apart, and each
+// carries one update whose content nothing else in its log holds.
+//
+// That last part is what gives the comparison teeth, and leaving it out
+// makes this test near-vacuous. Every other update here is the same
+// simple.bin bytes, which Yjs merges idempotently, so a log of nothing
+// but those reconstructs the whole document from the tail alone: an
+// empty snapshot, or one stranding any row at all, would render
+// identically and this test would pass. The distinct row sits in the
+// range only the snapshot can carry — everything at or below its seq —
+// so if the snapshot does not hold it, nothing does.
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)
- }
+ // distinctAt is a 0-based index into the run of updates; the row it
+ // lands in gets seq distinctAt+1.
+ cases := []struct {
+ name string
+ updates int
+ distinctAt int
+ }{
+ // One snapshot at seq snapshotCadence and a five-row tail. The
+ // distinct row is deep inside the snapshot's range.
+ {"one cadence, the distinct row deep inside the snapshot", snapshotCadence + 5, 3},
+ // Two cadences: the snapshot at seq snapshotCadence is
+ // superseded by one at 2*snapshotCadence, and the distinct row
+ // is the very last row that surviving snapshot must contain —
+ // the boundary itself.
+ {"two cadences, the distinct row on the boundary", 2*snapshotCadence + 3, 2*snapshotCadence - 1},
+ }
+
+ for _, tc := range cases {
+ t.Run(tc.name, func(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")
- }
+ ordinary := validUpdate(t)
+ distinct := ReadFragmentFixture(t, "concurrent")
+ for i := 0; i < tc.updates; i++ {
+ update := ordinary
+ if i == tc.distinctAt {
+ update = distinct
+ }
+ if _, err := app.Store().ApplyUpdate(ctx, doc, update); err != nil {
+ t.Fatalf("ApplyUpdate %d: %v", i, err)
+ }
+ }
- withSnapshot, err := app.Store().HTML(ctx, doc)
- if err != nil {
- t.Fatalf("HTML: %v", 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")
+ }
+ if want := int64(tc.distinctAt + 1); snap.Seq < want {
+ t.Fatalf("newest snapshot is at seq %d, below the distinct row at seq %d — the tail would carry that content on its own and this test would prove nothing", snap.Seq, want)
+ }
- // 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")
+ 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.Errorf("snapshot-then-tail and full replay disagree about the document:\n through the snapshot: %q\n full replay: %q", withSnapshot, fullReplay)
+ }
+ })
}
}
@@ -193,20 +237,9 @@ func TestConcurrentPostsAtTheCadenceBoundaryDoNotStrandContent(t *testing.T) {
}
}
- 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)
+ if err := raceEightPosts(t, app, doc); err != nil {
+ t.Fatalf("the racer carrying the distinct content failed, so this test has nothing to detect a stranding with: %v", err)
}
- wg.Wait()
// Every racer measured a distance of at least snapshotCadence, so
// a snapshot must exist — otherwise the comparison below is
@@ -259,20 +292,9 @@ func TestASnapshotContainsEveryRowItsSeqClaims(t *testing.T) {
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)
+ if err := raceEightPosts(t, app, doc); err != nil {
+ t.Fatalf("the racer carrying the distinct content failed, so this test has nothing to detect a stranding with: %v", err)
}
- wg.Wait()
var snaps []DocSnapshot
if err := app.Store().db.WithContext(ctx).
@@ -292,6 +314,151 @@ func TestASnapshotContainsEveryRowItsSeqClaims(t *testing.T) {
}
}
+// Supersession, which nothing else here pins. The delete predicate is
+// narrow on purpose — it removes the rows the new snapshot supersedes
+// by newestSnapshot's own (seq, id) ordering, and nothing else — and
+// there are two obvious simplifications of it that both do damage.
+//
+// "Delete every other row for this document" throws away a FRESHER
+// snapshot. Two writers cross the boundary together; the faster one
+// commits at seq 30, and the slower one — working from a replay it
+// began earlier — commits at seq 25 afterwards and takes the seq 30
+// row with it. The document then replays five updates it did not need
+// to, silently, and on a long document that gap is the difference
+// between a snapshot and the cliff.
+//
+// "Delete every row below this seq" without scoping to the document
+// is not a performance question at all: it deletes other documents'
+// snapshots on every write.
+func TestWritingASnapshotDeletesOnlyWhatItSupersedes(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, "Superseded")
+ if err != nil {
+ t.Fatalf("Create: %v", err)
+ }
+ other, err := app.Store().Create(ctx, owner, "A different document")
+ if err != nil {
+ t.Fatalf("Create: %v", err)
+ }
+
+ seed := func(d *Document, seq int64) {
+ t.Helper()
+ row := &DocSnapshot{DocID: d.ID, Seq: seq, State: []byte("an older snapshot"), ABIVersion: coreABIVersion, CreatedAt: time.Now().UTC()}
+ if err := app.Store().db.WithContext(ctx).Create(row).Error; err != nil {
+ t.Fatalf("seed snapshot at seq %d: %v", seq, err)
+ }
+ }
+ seed(doc, 10)
+ seed(doc, 20)
+ seed(other, 10)
+ seed(other, 20)
+
+ if _, err := app.Store().ApplyUpdate(ctx, doc, validUpdate(t)); err != nil {
+ t.Fatalf("ApplyUpdate: %v", err)
+ }
+ h, _, _, err := app.Store().replay(ctx, doc)
+ if err != nil {
+ t.Fatalf("replay: %v", err)
+ }
+ defer h.Close(ctx)
+
+ // A real snapshot at seq 30 supersedes both hand-written rows for
+ // this document, and neither of the other document's.
+ if err := app.Store().writeSnapshot(ctx, doc, h, 30); err != nil {
+ t.Fatalf("writeSnapshot at seq 30: %v", err)
+ }
+ assertSnapshotSeqs(t, app, doc, []int64{30})
+ assertSnapshotSeqs(t, app, other, []int64{10, 20})
+
+ // Now the slower of two boundary writers, committing afterwards
+ // from a prefix that stopped earlier. Its own row goes in; the
+ // fresher seq 30 row stays, and newestSnapshot still prefers it.
+ //
+ // The seq passed here is one no caller would ever pass — ApplyUpdate
+ // only ever hands over the seq its own replay actually reached —
+ // but the delete predicate is what is under test, not the caller.
+ if err := app.Store().writeSnapshot(ctx, doc, h, 25); err != nil {
+ t.Fatalf("writeSnapshot at seq 25: %v", err)
+ }
+ assertSnapshotSeqs(t, app, doc, []int64{25, 30})
+ assertSnapshotSeqs(t, app, other, []int64{10, 20})
+
+ newest, err := app.Store().newestSnapshot(ctx, doc.ID)
+ if err != nil || newest == nil {
+ t.Fatalf("newestSnapshot = (%v, %v), want the seq 30 row", newest, err)
+ }
+ if newest.Seq != 30 {
+ t.Errorf("newestSnapshot returned seq %d, want the fresher seq 30 row the slower writer must not have deleted", newest.Seq)
+ }
+}
+
+// assertSnapshotSeqs pins exactly which snapshot rows a document has
+// left, in seq order.
+func assertSnapshotSeqs(t *testing.T, app *App, doc *Document, want []int64) {
+ t.Helper()
+ var rows []DocSnapshot
+ if err := app.Store().db.WithContext(context.Background()).
+ Where("doc_id = ?", doc.ID).Order("seq ASC, id ASC").Find(&rows).Error; err != nil {
+ t.Fatalf("load snapshots for %q: %v", doc.Title, err)
+ }
+ got := make([]int64, 0, len(rows))
+ for _, r := range rows {
+ got = append(got, r.Seq)
+ }
+ if !slices.Equal(got, want) {
+ t.Errorf("%q has snapshots at seqs %v, want %v", doc.Title, got, want)
+ }
+}
+
+// raceEightPosts fires eight concurrent ApplyUpdates at doc, exactly
+// one of which carries content nothing else in the log holds, and
+// returns that one's error.
+//
+// Both fixtures are read here, on the test goroutine, before any racer
+// starts. validUpdate and ReadFragmentFixture call t.Fatalf if the
+// file cannot be read, and t.Fatalf from a non-test goroutine does not
+// do what the call site looks like it is asking for: it marks the test
+// failed and calls runtime.Goexit on that one goroutine, so the
+// remaining racers, the WaitGroup and every assertion after it carry
+// on running against a document nobody meant to build.
+//
+// The distinct racer's error comes back rather than being dropped on
+// the floor. If that single POST ever failed, its content would never
+// reach the log, every remaining row would say exactly what every
+// other row says, and the caller's comparison could no longer fail no
+// matter what the implementation did — the test would pass by having
+// nothing left to detect.
+func raceEightPosts(t *testing.T, app *App, doc *Document) error {
+ t.Helper()
+ ctx := context.Background()
+ ordinary := validUpdate(t)
+ distinct := ReadFragmentFixture(t, "concurrent")
+
+ const carriesTheDistinctContent = 3
+ var distinctErr error
+ var wg sync.WaitGroup
+ for i := 0; i < 8; i++ {
+ wg.Add(1)
+ go func(i int) {
+ defer wg.Done()
+ if i == carriesTheDistinctContent {
+ // Written by this one goroutine and read only after
+ // wg.Wait below, which is the happens-before edge.
+ _, distinctErr = app.Store().ApplyUpdate(ctx, doc, distinct)
+ return
+ }
+ _, _ = app.Store().ApplyUpdate(ctx, doc, ordinary)
+ }(i)
+ }
+ wg.Wait()
+ return distinctErr
+}
+
// 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