diff --git a/docs/deployment.md b/docs/deployment.md index ac1f0403d..7ad2aaa19 100644 --- a/docs/deployment.md +++ b/docs/deployment.md @@ -1256,7 +1256,11 @@ records one version, attributed to `system` with source `recovery` and the summary "recovered from an unsaved editor session" (the op-log records no author, so none is guessed). It does not delete the op-log; the next tab to open the item replays it as before. Items holding edits set aside by an editor -upgrade (`content_state: superseded_set_aside`) are not touched. +upgrade (`content_state: superseded_set_aside`) are not touched. A recovery +never writes a blank body over a stored one: when an item's edits replay to an +empty document (which a tab would answer by showing the stored body), the +stored body is kept, nothing is written, and the item logs the one warning +below with `last_error_kind=empty_document`. A job that fails (timeout, memory limit, a worker death, an editor error) is retried after 1 minute, then 2; after 3 consecutive failures the item is not @@ -1268,7 +1272,8 @@ logged one by one; the sweep's summary line (`op-log recovery sweep`, written only when something changed) carries their running count as `set_aside_skipped_total`. This budget is kept in memory, so a restart grants each item its 3 attempts again. An item whose stored edits were written under -another editor schema version is skipped, and logs the same one warning. +another editor schema version is skipped, and logs the same one warning +(`last_error_kind=schema_version`). With `PAD_MATERIALIZE=off` none of this runs: no worker is ever started, and a tab that closed without saving catches up only when a tab next opens the item, diff --git a/internal/server/materialize_recovery.go b/internal/server/materialize_recovery.go index 63adbe2c2..6bad82d4c 100644 --- a/internal/server/materialize_recovery.go +++ b/internal/server/materialize_recovery.go @@ -127,6 +127,7 @@ const ( mrNothing materializeResult = "nothing_pending" mrRoomOpen materializeResult = "room_open" mrSetAside materializeResult = "set_aside" + mrEmptyRefused materializeResult = "empty_document" // BUG-3316: never a blank body over a stored one mrCursorMoved materializeResult = "cursor_moved" mrGone materializeResult = "gone" mrBudgetExceeded materializeResult = "budget_exhausted" @@ -403,7 +404,7 @@ func (r *materializeRecovery) process(itemID string) materializeResult { } for _, v := range in.SchemaVersions { if v != collab.DefaultSchemaVersion { - r.exhaust(itemID, in.Cursor, fmt.Sprintf("op-log schema version %q, server %q", v, collab.DefaultSchemaVersion)) + r.exhaust(itemID, in.Cursor, "schema_version", fmt.Sprintf("op-log schema version %q, server %q", v, collab.DefaultSchemaVersion)) return mrSchemaVersion } } @@ -474,11 +475,27 @@ func (r *materializeRecovery) process(itemID string) materializeResult { case store.MaterializeSetAside: r.setAsideSkipped.Add(1) return mrSetAside + case store.MaterializeEmptyRefused: + // Terminal until the op-log changes: retrying would replay the same + // empty document every sweep (BUG-3316). + r.exhaust(itemID, in.Cursor, "empty_document", + fmt.Sprintf("the op-log replays to an empty document and the stored body is %d bytes; kept the stored body", storedLen(s, itemID))) + return mrEmptyRefused default: return mrGone } } +// storedLen is the item's stored body length, for the empty_document log +// line only (a rare path, so one extra read). +func storedLen(s *Server, itemID string) int { + it, err := s.store.GetItem(itemID) + if err != nil || it == nil { + return -1 + } + return len(it.Content) +} + // admit applies the failure budget. The budget resets when the op-log has // moved past the MAX(id) recorded at the last failure. func (r *materializeRecovery) admit(itemID string, cursor int64) (materializeResult, bool) { @@ -581,18 +598,19 @@ func (r *materializeRecovery) logExhaustion(level slog.Level, itemID string, fai "item_id", itemID, "failures", failures, "last_error_kind", kind, "last_error", lastErr) } -// exhaust marks the item as not to be tried until its op-log changes. -func (r *materializeRecovery) exhaust(itemID string, cursor int64, why string) { +// exhaust marks the item as not to be tried until its op-log changes, and +// writes its one exhaustion line with kind as last_error_kind. +func (r *materializeRecovery) exhaust(itemID string, cursor int64, kind, why string) { r.mu.Lock() b := r.budgetSlotLocked(itemID) b.failures = r.cfg.maxFailures b.opLogMax = cursor b.lastErr = why - b.lastKind = "schema_version" + b.lastKind = kind logExhaustion := r.claimExhaustionLogLocked(b) r.mu.Unlock() if logExhaustion { - r.logExhaustion(slog.LevelWarn, itemID, r.cfg.maxFailures, "schema_version", why) + r.logExhaustion(slog.LevelWarn, itemID, r.cfg.maxFailures, kind, why) } } diff --git a/internal/server/materialize_recovery_test.go b/internal/server/materialize_recovery_test.go index 4fc202919..f9ddc41d8 100644 --- a/internal/server/materialize_recovery_test.go +++ b/internal/server/materialize_recovery_test.go @@ -836,6 +836,54 @@ func TestMaterializeRecoverySecondPassIsHarmless(t *testing.T) { } } +// BUG-3316, through the recovery worker: an op-log that replays to an empty +// document never blanks the stored body. The item is refused as terminal, is +// not replayed again until its op-log changes, and leaves ONE WARN line naming +// empty_document. +func TestMaterializeRecoveryRefusesAnEmptyDocument(t *testing.T) { + var buf bytes.Buffer + fake := &fakeMaterializer{fn: func(materialize.Job) (string, error) { return "", nil }} + f := newRecoveryFixture(t, fake, time.Minute, materializeRecoveryConfig{ + logger: slog.New(slog.NewTextHandler(&buf, &slog.HandlerOptions{Level: slog.LevelInfo})), + }) + it := f.item(t, "Load test target", "a stored body a tab would seed from") + f.appendRows(t, it.ID, contentFrame(1)) + nv := len(mustVersions(t, f, it.ID)) + + if res := f.r.process(it.ID); res != mrEmptyRefused { + t.Fatalf("process = %q, want %q", res, mrEmptyRefused) + } + got, _ := f.srv.store.GetItem(it.ID) + if got.Content != "a stored body a tab would seed from" || len(mustVersions(t, f, it.ID)) != nv { + t.Fatalf("the stored body was touched: %q, versions %d -> %d", got.Content, nv, len(mustVersions(t, f, it.ID))) + } + for i := 0; i < 3; i++ { + if res := f.r.process(it.ID); res != mrBudgetExceeded { + t.Fatalf("pass %d after the refusal = %q, want %q (terminal until the op-log changes)", i, res, mrBudgetExceeded) + } + } + if n := fake.calls.Load(); n != 1 { + t.Fatalf("the empty document was replayed %d times, want 1", n) + } + var lines []string + for _, l := range strings.Split(buf.String(), "\n") { + if strings.Contains(l, "exhausted its failure budget") { + lines = append(lines, l) + } + } + if len(lines) != 1 || !strings.Contains(lines[0], "level=WARN") || !strings.Contains(lines[0], "last_error_kind=empty_document") || + !strings.Contains(lines[0], "item_id="+it.ID) { + t.Fatalf("exhaustion lines = %q, want one WARN naming empty_document", lines) + } + + // New edits reset it: the op-log grew, so the item is tried again. + f.appendRows(t, it.ID, contentFrame(2)) + fake.fn = func(materialize.Job) (string, error) { return "typed after all", nil } + if res := f.r.process(it.ID); res != mrApplied { + t.Fatalf("after the op-log grew: %q, want %q", res, mrApplied) + } +} + func mustVersions(t *testing.T, f *recoveryFixture, itemID string) []models.Version { t.Helper() vs, err := f.srv.store.ListItemVersions(itemID) diff --git a/internal/store/materialize_flush.go b/internal/store/materialize_flush.go index c0fad8372..494e7db2b 100644 --- a/internal/store/materialize_flush.go +++ b/internal/store/materialize_flush.go @@ -6,6 +6,7 @@ import ( "encoding/hex" "errors" "fmt" + "strings" "time" "github.com/PerpetualSoftware/pad/internal/models" @@ -214,6 +215,14 @@ const ( MaterializeSetAside MaterializeOutcome = "set_aside" // MaterializeGone: the item is missing or soft-deleted. MaterializeGone MaterializeOutcome = "gone" + // MaterializeEmptyRefused: the recovered body is blank and the stored one + // is not, so nothing was written (BUG-3316). A tab never does this: its + // lazy seed (TASK-1261) sees an empty document and seeds FROM + // items.content. Recovery cannot see whether the document is empty or + // merely renders blank, so it keeps the stored body in both cases, which + // is the conservative side of the tab's rule. Decided under the write + // lock, against the row being replaced. + MaterializeEmptyRefused MaterializeOutcome = "empty_document" ) type materializeAbort struct{ outcome MaterializeOutcome } @@ -321,7 +330,10 @@ func (s *Store) MaterializeFlush(itemID string, cursor int64, markdown string) ( Recovered: true, OpLogCursor: &c, } - updated, err := s.UpdateItemWithPreCheck(itemID, input, func(tx *sql.Tx, _ *models.Item) error { + updated, err := s.UpdateItemWithPreCheck(itemID, input, func(tx *sql.Tx, existing *models.Item) error { + if strings.TrimSpace(markdown) == "" && existing != nil && strings.TrimSpace(existing.Content) != "" { + return &materializeAbort{outcome: MaterializeEmptyRefused} + } outcome, err := check(tx, tx) if err != nil { return err diff --git a/internal/store/materialize_flush_test.go b/internal/store/materialize_flush_test.go index 157b9466e..be7b389c5 100644 --- a/internal/store/materialize_flush_test.go +++ b/internal/store/materialize_flush_test.go @@ -3,6 +3,7 @@ package store_test import ( "errors" "fmt" + "strings" "testing" "time" @@ -344,3 +345,35 @@ func TestListMaterializeCandidatesPagesByKeyset(t *testing.T) { } }) } + +// BUG-3316: a recovery that renders BLANK never replaces a stored body. A +// tab's lazy seed (TASK-1261) seeds from items.content when its replayed +// document is empty; recovery keeps the stored body instead, and writes +// nothing at all: not the body, not a version row, not seq, not the watermark. +// (Found by the pre-rollout dry run: a load-test item's op-log of 90 tiny rows +// replays to an empty document over a 1,285-byte body.) +func TestMaterializeFlushRefusesBlankOverAStoredBody(t *testing.T) { + eachBackend(t, func(t *testing.T, s *store.Store) { + for _, blank := range []string{"", " \n\n\t"} { + _, _, item := seedStaleItem(t, s) + cursor := appendFrames(t, s, item.ID, recoveryFrame(1)) + before, _ := s.GetItem(item.ID) + if strings.TrimSpace(before.Content) == "" { + t.Fatalf("precondition: the stored body %q must be non-blank", before.Content) + } + nv := versionCount(t, s, item.ID) + wm, _, _ := s.GetItemContentFlushedOpLogID(item.ID) + + outcome, updated, err := s.MaterializeFlush(item.ID, cursor, blank) + if err != nil || outcome != store.MaterializeEmptyRefused || updated != nil { + t.Fatalf("%q: MaterializeFlush = %q, %v, %v; want %q", blank, outcome, updated, err, store.MaterializeEmptyRefused) + } + got, _ := s.GetItem(item.ID) + wm2, _, _ := s.GetItemContentFlushedOpLogID(item.ID) + if got.Content != before.Content || got.Seq != before.Seq || versionCount(t, s, item.ID) != nv || wm2 != wm { + t.Fatalf("%q: something was written: content %q->%q seq %d->%d versions %d->%d watermark %d->%d", + blank, before.Content, got.Content, before.Seq, got.Seq, nv, versionCount(t, s, item.ID), wm, wm2) + } + } + }) +}