fix(materialize): recovery never writes a blank body over a stored one (BUG-3316) (#1693)

The pre-rollout dry run of op-log recovery (TASK-2198 U4) over the live
DB's 175 candidates found one that would have been wiped. A load-test item's
op-log of 90 tiny rows, with no base document, replays to an EMPTY document,
and recovery would have written "" over its 1,285-byte body. A tab never
does this: its lazy seed (TASK-1261) sees the empty fragment and seeds FROM
items.content.

MaterializeFlush now refuses a blank recovered body over a non-blank
stored one, decided in the precheck against the row locked for the write,
as the new outcome empty_document. Nothing is written: no body, version,
seq or watermark. Recovery treats it as terminal until the op-log changes
(exhaust, which now takes its kind), so the sweep does not replay it every
minute, and it logs the one exhaustion WARN with
last_error_kind=empty_document.

Stated divergence: a tab refuses only when the fragment is EMPTY, so a user
who truly cleared an item's text and then crashed would have flushed "".
Recovery cannot see the fragment, so it keeps the old body there too, which
is the conservative side. No shrink threshold, by ruling: a tab would flush
a shorter body, and the dry run found no such case.

Gate: the dry run re-run on this branch, flushing into a disposable COPY of
the same snapshot. All 175 items produce the same recovered body and class
as the first run. TASK-1270 is refused (empty_document) with its body
intact; 156 are stamped, 17 applied, and 1 (a deadline failure) is never
flushed. Legs: store (blank and whitespace-only, nothing written) and
recovery (refused, terminal, one WARN, reset when the op-log grows).
Mutants: a guard that never fires (2 red), no exhaust (1), wrong kind (1).

Claude-Session: https://claude.ai/code/session_01H63zaWhkb5pAafx2WGjrVC

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,

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)
 	}
 }
 

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)

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

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)
+			}
+		}
+	})
+}