feat(materialize): recover unsaved editor sessions into items.content (TASK-2198 U4) (#1683)

* feat(materialize): recover unsaved editor sessions into items.content in the background (TASK-2198 U4)

Wires the U3 supervisor in. A collab room reclaimed at the end of its
grace TTL with content-bearing rows above the flush watermark (T3), or a
dormant pending item found by a one-minute sweep (T1), is queued; one
worker goroutine materializes the item's whole content-bearing op-log
with the whole-workspace link index and lands it through
store.MaterializeFlush, under the collab item lock with no room open.

The write is a flush: cursor == MAX(op-log id), rows pending, no
set-aside rows, checked in the write's own transaction; watermark to the
cursor only; seq bumped; one version row attributed to system/recovery;
the op-log is never pruned. PAD_MATERIALIZE=off disables it entirely.
A per-item in-memory failure budget (3 failures, doubling backoff, reset
by a new op-log row) keeps untrusted input from being retried in a loop.

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

* test(store): recovery watermark is the cursor, not MAX at UPDATE time (TASK-2198 U4)

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

* test(store): a second recovery still writes its own version row (TASK-2198 U4)

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

* docs(mcp): op-log recovery bounds the stale-body window; tool surface v0.60 (TASK-2198 U4)

content_state applied_pending_flush now clears on its own after the
item's room closes, unless recovery fails or PAD_MATERIALIZE=off. Every
sentence that called the window unbounded, or said nothing server-side
moves op-log content into items.content, is corrected (CLAUDE.md,
instructions.md, README, docs/deployment.md, docs/backup.md, code
comments). BEHAVIOR bump to 0.60; goldens regenerated.

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

* fix(materialize): count ErrNoMemoryCap silently against the per-item budget (TASK-2198 U4)

U3 now refuses every job with ErrNoMemoryCap when no memory cap can be
established, and warns once itself. The recovery worker counts it like
any failure but logs it at debug, not per item. Its log lines go through
a logger seam so the test can tell.

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

* style: gofmt (TASK-2198 U4)

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

* feat(materialize): one log line per budget exhaustion; count set-aside skips on the sweep line (TASK-2198 U4)

An item that uses up its failure budget logs exactly one WARN naming the
item, the failure count and the last error kind (DEBUG when the kind is
no_memory_cap, which the supervisor already warned about once), and logs
again only after op-log growth resets the budget and it is exhausted
anew. The schema-version skip goes through the same line.

Items skipped for set-aside rows are counted and reported on a new
single sweep summary line, written only when it has news; never per item.

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

* fix(materialize): keyset-page the recovery sweep so no candidate starves (TASK-2198 U4, codex r1)

The sweep read the oldest 200 candidates from the start every tick, and
the room check and the failure budget filtered them afterwards. Once 200
sat in backoff or exhausted, every sweep returned the same 200 and the
201st was never queued: filtering above a bound is starvation.

Each tick now reads one page strictly after an in-memory keyset cursor
(newest op-log created_at, item id) and wraps to the start after a short
page, so every candidate is offered within ceil(N/limit)+1 ticks,
whatever sorts ahead of it. Keyset over NOT IN: no parameter-count limit,
and it also covers the room check, which an exclusion list would not.

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

* test(materialize): the sweep cursor wraps; docs for the exhaustion line, the set-aside count and paging (TASK-2198 U4)

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

* docs(deployment): reflow the recovery budget paragraph (TASK-2198 U4)

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

* test(materialize): recovery strips the follows-title marker as a tab's save does (TASK-2198 U4, BUG-3315)

Since #1688 a tab loads a link that follows its target's title with the
follows-title marker in the link's title attribute, so the marker lives in
Y.Docs that recovery replays. Three fixture cases (a follows-title link;
the same link after a rename; a URL edited to an external address) are
generated from documents that really hold the marker, and the existing 8
are unchanged. A new leg recovers each through the production path
(whole-workspace index) and through the runner with a restricted and an
empty index, and asserts no 'pad-follows-title' reaches the body. It goes
red in all 7 ways when the bundle's marker strips are disabled.

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

* test(store): a recovery version row carries #1691's history data (TASK-2198 U4)

Rebased onto PLAN-2348 U2 (#1691), which adds user_id, line counts and a
create marker to version rows. Recovery writes its row through the same
UpdateItem insert, so it gets them automatically: no user_id (a system
row), not a create, and line counts equal to diff.LineCounts of the stale
and recovered bodies. The leg pins that.

Claude-Session: https://claude.ai/code/session_01H63zaWhkb5pAafx2WGjrVC
This commit is contained in:
xarmian
2026-09-30 11:03:07 -04:00
committed by GitHub
parent 7ba823004c
commit cb99879d53
32 changed files with 3036 additions and 56 deletions
+5 -5
View File
File diff suppressed because one or more lines are too long
+2 -2
View File
File diff suppressed because one or more lines are too long
+12 -12
View File
@@ -606,13 +606,13 @@ Steps:
// per item rather than a scan of pre-built bundles, which would mean
// holding every workspace in memory to read one field.
//
// The remedy is the only one there is: a browser tab. No Go code in
// this repo can decode a Yjs update payload (the dumb-relay design),
// so nothing server-side can move op-log content into items.content
// — and the two server-side paths that touch the op-log,
// PruneAndApply and ForceRefreshRoom, write CALLER-supplied content
// and prune, which would DISCARD exactly these edits. That is why
// the escape hatch is named for what it does to the data.
// The remedy is a browser tab, or a RUNNING server's op-log recovery
// (TASK-2198 U4, internal/materialize), which this command cannot
// use: it runs with the server stopped. The two other server-side
// paths that touch the op-log, PruneAndApply and ForceRefreshRoom,
// write CALLER-supplied content and prune, which would DISCARD
// exactly these edits. That is why the escape hatch is named for
// what it does to the data.
pendingByWorkspace := map[string][]store.PendingFlushItem{}
pendingTotal := 0
for _, ws := range workspaces {
@@ -933,11 +933,11 @@ func gateUnflushedEdits(workspaces []models.Workspace, pending map[string][]stor
fmt.Fprintf(&b, " %-12s %s\n", it.Ref, it.Title)
}
}
// The only remedy there is. No Go code in this repo can decode a Yjs update
// payload (the dumb-relay design), so nothing server-side can move op-log
// content into items.content — and the two server-side paths that touch the
// op-log, PruneAndApply and ForceRefreshRoom, write CALLER-supplied content
// and prune, which would DISCARD exactly these edits.
// The remedy a stopped server leaves. A running server's op-log recovery
// (TASK-2198 U4) also moves this content into items.content, but this
// command runs with the server stopped; the other two server-side paths
// that touch the op-log, PruneAndApply and ForceRefreshRoom, write
// CALLER-supplied content and prune, which would DISCARD exactly these edits.
fmt.Fprint(&b, "\nOpen each of those items in the web UI so the tab flushes its pending edits\n")
fmt.Fprint(&b, "into the database, then re-run this command. To migrate anyway and lose those\n")
fmt.Fprint(&b, "edits, re-run with --discard-unflushed-edits.\n")
+2 -2
View File
@@ -73,8 +73,8 @@ func TestGateUnflushedEditsRefusesAndNamesEveryAffectedItem(t *testing.T) {
t.Errorf("refusal does not end by promising nothing was migrated:\n%s", msg)
}
// The remedy and the escape hatch both have to be IN the refusal: a refusal
// with no way forward is a trap, and per BUG-3000 the flush may never happen
// on its own.
// with no way forward is a trap, and the command runs with the server stopped,
// so the server's op-log recovery (TASK-2198 U4) cannot help it.
if !strings.Contains(msg, "web UI") {
t.Errorf("refusal does not name the only remedy (open the item so a tab flushes):\n%s", msg)
}
+7 -6
View File
@@ -4427,12 +4427,13 @@ func warnStaleEditSeed(item *models.Item) {
// scripts and a warning on stdout would corrupt the JSON they parse.
//
// It deliberately states no duration, and BUG-3000's day-64 measurement is why
// that must stay: the flush is NOT guaranteed to land at all. Nothing server-side
// moves applier content into items.content — the op-log GC's flush-coverage guard
// deliberately retains those rows rather than flushing them — so the row catches up
// only when a tab next opens the item. Any number here would be a claim this
// command cannot support. (This comment previously said the question was "not
// established"; it now is.)
// that must stay: the flush is NOT guaranteed to land at all. The tab flushes it,
// or, once the item's room has closed, the server's op-log recovery does
// (TASK-2198 U4); that recovery can fail, and an operator can turn it off
// (PAD_MATERIALIZE=off), in which case the row catches up only when a tab next
// opens the item. Any number here would be a claim this command cannot
// support. (This comment previously said the question was "not established";
// it now is.)
func warnContentPendingFlush(item *models.Item) {
if item == nil || item.Warnings == nil {
return
+19
View File
@@ -1,10 +1,12 @@
package main
import (
"log/slog"
"os"
pad "github.com/PerpetualSoftware/pad"
"github.com/PerpetualSoftware/pad/internal/materialize"
"github.com/PerpetualSoftware/pad/internal/server"
"github.com/spf13/cobra"
)
@@ -33,3 +35,20 @@ func materializeWorkerCmd() *cobra.Command {
},
}
}
// newMaterializerFromEnv builds the op-log recovery worker's Materializer
// (TASK-2198 U4): a materialize.Supervisor configured from
// PAD_MATERIALIZE_TIMEOUT / PAD_MATERIALIZE_MEM_LIMIT. It returns nil, and
// never calls build, when PAD_MATERIALIZE=off: then no Supervisor exists and
// no worker process is ever spawned. build is NewSupervisor in production.
func newMaterializerFromEnv(getenv func(string) string, logger *slog.Logger, build func(materialize.SupervisorConfig) server.Materializer) server.Materializer {
if !materialize.Enabled(getenv, logger) {
logger.Info("op-log recovery disabled (" + materialize.EnvSwitch + "=off)")
return nil
}
return build(materialize.ConfigFromEnv(getenv, logger))
}
func newSupervisorMaterializer(cfg materialize.SupervisorConfig) server.Materializer {
return materialize.NewSupervisor(cfg)
}
+51
View File
@@ -2,10 +2,15 @@ package main
import (
"encoding/json"
"errors"
"log/slog"
"os/exec"
"strings"
"testing"
"github.com/PerpetualSoftware/pad/internal/cmdhelp"
"github.com/PerpetualSoftware/pad/internal/materialize"
"github.com/PerpetualSoftware/pad/internal/server"
)
// The materializer worker (TASK-2198) is an internal process entry point: it
@@ -29,3 +34,49 @@ func TestMaterializeWorkerCmdHiddenAndOffCmdhelp(t *testing.T) {
t.Fatal("the cmdhelp document mentions the materializer worker")
}
}
// PAD_MATERIALIZE=off (TASK-2198 U4) must leave the server with no
// Materializer at all: the Supervisor is never built, so no worker process can
// be spawned and StartMaterializeRecovery installs no trigger.
func TestMaterializerDisabledBuildsNothing(t *testing.T) {
for _, v := range []string{"off", "OFF", " off ", "0", "false", "no", "disabled"} {
built := 0
m := newMaterializerFromEnv(func(k string) string {
if k == materialize.EnvSwitch {
return v
}
return ""
}, slog.Default(), func(materialize.SupervisorConfig) server.Materializer {
built++
return materialize.NewSupervisor(materialize.SupervisorConfig{})
})
if m != nil || built != 0 {
t.Errorf("%s=%q: materializer %v, builds %d; want nil, 0", materialize.EnvSwitch, v, m, built)
}
}
for _, v := range []string{"", "on", "1", "true", "yes", "something-else"} {
built, spawned := 0, 0
m := newMaterializerFromEnv(func(k string) string {
if k == materialize.EnvSwitch {
return v
}
return ""
}, slog.Default(), func(cfg materialize.SupervisorConfig) server.Materializer {
built++
cfg.Command = func() (*exec.Cmd, error) {
spawned++
return nil, errors.New("test: no spawn")
}
return materialize.NewSupervisor(cfg)
})
if m == nil || built != 1 {
t.Errorf("%s=%q: materializer %v, builds %d; want one", materialize.EnvSwitch, v, m, built)
}
if spawned != 0 {
t.Errorf("%s=%q: building the supervisor spawned %d workers; it must start none until a job", materialize.EnvSwitch, v, spawned)
}
if c, ok := m.(interface{ Close() error }); ok {
_ = c.Close()
}
}
}
+9
View File
@@ -705,6 +705,15 @@ func serveCmd() *cobra.Command {
}
srv.StartOpLogGC()
// Op-log recovery (TASK-2198 U4). Rebuilds items.content from the
// collaborative op-log when a tab went away without flushing,
// in a capped child process (`pad __materialize-worker`) started
// on the first job, not here. PAD_MATERIALIZE=off disables it
// entirely; PAD_MATERIALIZE_TIMEOUT / PAD_MATERIALIZE_MEM_LIMIT
// tune the worker. docs/deployment.md, "Op-log materializer worker".
srv.SetMaterializer(newMaterializerFromEnv(os.Getenv, slog.Default(), newSupervisorMaterializer))
srv.StartMaterializeRecovery()
// Token reaper (PLAN-1933 DR-5 / TASK-1936). Periodic sweep
// that deletes expired/used email-verification tokens,
// password-reset tokens, sessions, CLI-auth sessions and expired
File diff suppressed because one or more lines are too long
File diff suppressed because one or more lines are too long
+9 -5
View File
@@ -251,11 +251,15 @@ since otherwise no server is running to make one — that workspace is refused t
That refusal does **not** claim nothing has been migrated: earlier workspaces in
the run are already in PostgreSQL, and it says so with the count.
Opening the item in the web UI is the only remedy, and that is structural rather
than an omission: Pad's collaboration server is a dumb relay that stores opaque
Yjs updates without parsing them, so nothing server-side can turn those updates
back into markdown — only a client can. This is also why the escape hatch is
named for what it does to the data:
There are two remedies. Opening each item in the web UI lets the tab flush it.
Or start the Pad server with op-log recovery on (the default; see
[Op-log materializer worker](deployment.md#op-log-materializer-worker)) and
leave it running for a few minutes: it rebuilds each such body from the op-log
in the background, then stop it and re-run the migration. The migration itself
never recovers anything, because it runs with the server stopped, and a
recovery can fail or be turned off (`PAD_MATERIALIZE=off`), so the refusal
stays. This is also why the escape hatch is named for what it does to the
data:
```bash
pad db migrate-to-pg --discard-unflushed-edits
+47 -6
View File
@@ -1224,18 +1224,58 @@ address inside `PAD_TRUSTED_PROXIES` (the proxy sends no `X-Forwarded-For` or
### Op-log materializer worker
**Not active yet.** The worker and the two variables below take effect once
the materializer is wired into the server (TASK-2198 U4); until then they are
inert, and setting them changes nothing.
When a browser tab closes without saving, its last edits are only in the
item's collaborative op-log. Pad can turn that op-log back into the item's
markdown on the server by running the editor's own JavaScript. It does this in
item's collaborative op-log, and the item's stored body (what the API, the
CLI, MCP and search read, marked `content_state: applied_pending_flush`) is
behind them. Pad turns that op-log back into the item's markdown on the server
by running the editor's own JavaScript, and stores the result. It does this in
a **separate worker process** (the same `pad` binary, started as
`pad __materialize-worker`), never inside the server: a crafted update can
make that JavaScript use gigabytes of memory, and in a separate process that
kills one worker instead of the server.
**When recovery happens.** Only in the background, never while a request
waits:
- **When the item's editing room closes.** A room stays open while any tab has
the item open, and for 60 seconds after the last one leaves. When it closes
with unsaved edits in the op-log, the item is queued. A tab that closed
without saving is therefore normally recovered about a minute after it went
away, plus the time the jobs queued ahead of it take.
- **A sweep every minute** picks up items the first trigger could not: unsaved
edits whose newest op-log row is more than 2 minutes old and whose item has
no open room (for example after a server restart). It takes 200 at a time
and continues where the last sweep stopped, so a large backlog is worked
through in turn rather than the same 200 being retried.
A recovery writes the body only if the op-log has not changed since the job
was built and no tab has the item open; otherwise it does nothing and the next
trigger tries again. It bumps the item's `seq`, so a writer holding an older
`expected_seq` is refused rather than overwriting the recovered text, and it
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.
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
tried again until new edits arrive for it. Each failure is logged with its
kind, and an item that gives up logs one warning, `item exhausted its failure
budget`, with its id, failure count and `last_error_kind`, so a stuck item can
be found in the log. Items skipped because they hold set-aside edits are not
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.
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,
which may never happen. The same is true while no memory cap can be
established (see the end of this section): each pending item's job is then
refused, counted against its 3 attempts, and not logged per item.
- The worker starts on the first job, not at server start, and then stays up
while jobs keep coming (loading it takes a few seconds). Jobs run one at a
time.
@@ -1260,6 +1300,7 @@ kills one worker instead of the server.
| Variable | Default | Description |
|----------|---------|-------------|
| `PAD_MATERIALIZE` | `on` | `off` (or `0`, `false`, `no`) disables op-log recovery entirely: no worker process is started and no trigger runs. Any other value leaves it on; an unrecognised one is logged. |
| `PAD_MATERIALIZE_TIMEOUT` | `2s` | Per-job time limit, in Go duration syntax (`2s`, `1500ms`). Clamped to 250ms–60s. |
| `PAD_MATERIALIZE_IDLE_TIMEOUT` | `5m` | How long the worker may sit with no job before it is stopped (it holds about 200 MB resident while it runs), in Go duration syntax. `0` means never stop it. Clamped to 30s–24h. |
| `PAD_MATERIALIZE_MEM_LIMIT` | `2GiB` | How much memory one job may add to the loaded worker (see below): a whole number of bytes, or a whole number followed by `KiB`, `MiB` or `GiB` with no space (`2048MiB`, `2GiB`, `2147483648`). Decimal units such as `GB` are refused. Clamped to 256MiB–16GiB. |
+67 -1
View File
@@ -146,6 +146,65 @@ type RoomManager struct {
// column read in the common (same-process) case.
lastRestoreSeqMu sync.Mutex
lastRestoreSeqs map[string]int64
// idleHook, when set, is called with the item's id after a room has been
// reclaimed at the end of its grace TTL (TASK-2198 U4: the op-log
// recovery trigger). Called outside m.mu, from the grace timer's
// goroutine; it must not block.
idleHookMu sync.Mutex
idleHook func(itemID string)
}
// SetIdleHook installs fn as the room-reclaimed callback (see idleHook). Nil
// removes it.
func (m *RoomManager) SetIdleHook(fn func(itemID string)) {
m.idleHookMu.Lock()
m.idleHook = fn
m.idleHookMu.Unlock()
}
// ErrRoomOpen is returned by UnderItemLockIfNoRoom when the item has a room.
var ErrRoomOpen = errors.New("collab: a room is open for the item")
// UnderItemLockIfNoRoom runs fn under the per-item setup lock, and only when
// no room exists for the item; otherwise it returns ErrRoomOpen without
// calling fn. A room counts from its creation until its grace TTL reclaims
// it, whether or not a connection is attached.
//
// While fn runs nothing can append to the item's op-log: every append
// happens through a room (a connection's readLoop, or the designated
// applier), and Join creates a room only while holding this same lock. The op-log
// recovery write (TASK-2198 U4) relies on that; it also re-checks the op-log
// inside its own transaction.
func (m *RoomManager) UnderItemLockIfNoRoom(itemID string, fn func() error) error {
m.mu.Lock()
if m.closed {
m.mu.Unlock()
return errManagerClosed
}
m.mu.Unlock()
lock := m.itemLock(itemID)
lock.Lock()
defer lock.Unlock()
m.mu.Lock()
closed, open := m.closed, m.rooms[itemID] != nil
m.mu.Unlock()
if closed {
return errManagerClosed
}
if open {
return ErrRoomOpen
}
return fn()
}
// HasRoom reports whether a room exists for the item right now. Racy by
// nature; a caller that acts on the answer re-checks under the lock
// (UnderItemLockIfNoRoom).
func (m *RoomManager) HasRoom(itemID string) bool {
m.mu.Lock()
defer m.mu.Unlock()
return m.rooms[itemID] != nil
}
// SetRestoreBoundary records a restore boundary for itemID, keeping the
@@ -841,9 +900,16 @@ func (m *RoomManager) getOrCreate(itemID string) *Room {
// Join mints a fresh Room.
func (m *RoomManager) markRoomGone(itemID string) {
m.mu.Lock()
defer m.mu.Unlock()
delete(m.rooms, itemID)
m.mu.Unlock()
slog.Debug("collab: room reclaimed after grace TTL", "item_id", itemID)
m.idleHookMu.Lock()
hook := m.idleHook
m.idleHookMu.Unlock()
if hook != nil {
hook(itemID)
}
}
// RoomCount is a test/debug accessor. Production code shouldn't make
+24
View File
@@ -29,6 +29,30 @@ const (
EnvIdleTimeout = "PAD_MATERIALIZE_IDLE_TIMEOUT"
)
// EnvSwitch turns op-log recovery off entirely: PAD_MATERIALIZE=off (or 0,
// false, no; case-insensitive) and the server constructs no Supervisor,
// installs no trigger and never spawns a worker. Unset, or any other value,
// leaves it on; a value that is neither a recognised on nor off word is
// logged and read as on.
const EnvSwitch = "PAD_MATERIALIZE"
// Enabled reads EnvSwitch through getenv.
func Enabled(getenv func(string) string, logger *slog.Logger) bool {
if logger == nil {
logger = slog.Default()
}
v := strings.ToLower(strings.TrimSpace(getenv(EnvSwitch)))
switch v {
case "off", "0", "false", "no", "disabled":
return false
case "", "on", "1", "true", "yes", "enabled":
return true
}
logger.Warn("materialize: unrecognised "+EnvSwitch+" value; op-log recovery stays ON (set it to off to disable)",
"value", getenv(EnvSwitch))
return true
}
const (
DefaultTimeout = 2 * time.Second
MinTimeout = 250 * time.Millisecond
File diff suppressed because one or more lines are too long
+15 -1
View File
@@ -1651,6 +1651,20 @@ const CmdhelpVersion = "0.1"
// this one either, and it is not a promise that the row will catch
// up at all.
//
// 0.60 — TASK-2198 U4, op-log recovery. BEHAVIOR bump on the v0.59 /
// v0.39 grounds: no tool name, action enum or param shape changed, but
// `content_state: "applied_pending_flush"` — on `pad_item` get / full
// list, `pad_playbook` run and the item resource's line above the body —
// now CLEARS ON ITS OWN. When the item's collab room closes (60s after
// its last tab leaves) or a one-minute sweep finds it dormant, the server
// rebuilds the body from the op-log in a worker process and stores it:
// the body changes, `seq` moves (a held `expected_seq` then answers
// `update_conflict`, where it used to answer `content_pending_flush`),
// and the history gains a version by `system` / `recovery`. Before it the
// row caught up only when a tab next opened the item. Not a guarantee: a
// failed job, an item with set-aside rows and PAD_MATERIALIZE=off all
// leave the old behaviour, so "re-read later, never re-send" stands.
//
// 0.59 — BUG-3252, the tombstone (ruled over cascade). BEHAVIOR bump
// on the v0.58 / v0.45 grounds: no name, enum or param shape changed,
// but `pad_item.action=delete-comment` on a comment that still has
@@ -1705,7 +1719,7 @@ const CmdhelpVersion = "0.1"
// this surface can receive it; the entry exists so a future action does
// not collapse it to permission_denied. When an action that can reach
// it is added, that addition is the contract change and owns the bump.
const ToolSurfaceVersion = "0.59"
const ToolSurfaceVersion = "0.60"
// MetaVersionURI is the canonical URI of the queryable version document.
// Lives outside the pad://workspace/{ws}/... namespace because it's a
+23 -2
View File
@@ -272,6 +272,17 @@ const ContentStatePendingFlush = ContentOutcomeAppliedPendingFlush
// stale should ask IsContentStateStale instead.
const ContentStateSetAside = "superseded_set_aside"
// The version-row attribution of an op-log recovery (ItemUpdate.Recovered,
// TASK-2198 U4). item_versions.created_by and .source carry no CHECK
// constraint on either dialect; "system" is already the created_by of
// template-seeded items. The web timeline labels them "System" / "Recovered";
// `pad item history` prints them as stored.
const (
VersionCreatedBySystem = "system"
VersionSourceRecovery = "recovery"
RecoveryChangeSummary = "recovered from an unsaved editor session"
)
// IsContentStateStale reports whether an Item.ContentState value says the
// stored body is behind edits that exist elsewhere, for any reason. It is the
// one helper for doors that mark a body stale without saying what to do.
@@ -1332,8 +1343,18 @@ type ItemUpdate struct {
// boundary is what Join reads to force_refresh a client whose ?content_seq
// seed predates the restore — surviving a server restart, unlike the
// in-memory fast-path. Internal-only (`json:"-"`); no HTTP client can set it.
MarkRestoreBoundary bool `json:"-"`
Comment *string `json:"comment,omitempty"`
MarkRestoreBoundary bool `json:"-"`
// Recovered marks the op-log materializer's write (TASK-2198 U4): the
// server rebuilt items.content from the item's collaborative op-log after
// the tab holding those edits went away without flushing them. The store
// attributes the version row to the system (VersionCreatedBySystem /
// VersionSourceRecovery) rather than to a guessed author, because op-log
// rows carry no user id, and advances the flush watermark only to
// OpLogCursor and only when that cursor is still MAX(op-log id), exactly
// as a collab-snapshot flush does. Internal-only (`json:"-"`); no HTTP
// client can set it.
Recovered bool `json:"-"`
Comment *string `json:"comment,omitempty"`
// OpLogCursor is the highest item_yjs_updates.id the calling client
// has applied into its local Y.Doc (TASK-1319). Used by the
// collab-snapshot flush PATCH to advance the op-log GC watermark
+5
View File
@@ -1990,6 +1990,11 @@ func (s *Server) handleUpdateItem(w http.ResponseWriter, r *http.Request) {
// items.source must not move on an update.
_, input.VersionSource = actorFromRequest(r)
}
// TASK-2198 U4: "recovery" labels the server's own op-log recovery write
// in the version history. A client may not claim it.
if input.VersionSource == models.VersionSourceRecovery {
input.VersionSource = ""
}
// BUG-2542: stamp the writer on single-item updates. Bulk ops already do
// this (handlers_items_bulk.go), but this path did not, so an agent's
@@ -468,8 +468,10 @@ func writeContentNotAppliedError(w http.ResponseWriter, ref string, landedFields
// carrying a version token while the item's op-log holds unflushed
// collaborative edits. 409, but NOT update_conflict — re-reading returns the
// same row and the same seq, so a read-and-retry loop would spin. The message
// says what clears it, because nothing server-side does: an open tab flushes
// within seconds; with no tab, only opening the item or the override.
// says what clears it: an open tab flushes within seconds; with no tab, opening
// the item, the override, or the server's own op-log recovery once the item's
// room has closed (TASK-2198 U4), which can fail or be turned off, so the
// message does not promise it.
//
// BUG-3244: when the item holds rows a schema rebuild set aside, the refusal
// covers every content write and the message must NOT offer opening the item,
+659
View File
@@ -0,0 +1,659 @@
package server
import (
"context"
"errors"
"fmt"
"log/slog"
"runtime/debug"
"sort"
"sync"
"sync/atomic"
"time"
"github.com/PerpetualSoftware/pad/internal/collab"
"github.com/PerpetualSoftware/pad/internal/materialize"
"github.com/PerpetualSoftware/pad/internal/models"
"github.com/PerpetualSoftware/pad/internal/store"
)
// Op-log recovery (TASK-2198 U4): the triggers and the worker.
//
// A tab that dies without flushing leaves its last typing only in the
// op-log, and items.content (what REST, the CLI, MCP and search read) stays
// behind it until a tab next opens the item — before this, possibly never.
// This file closes that window in the background:
//
// - T3, room idle: when a collab room is reclaimed at the end of its grace
// TTL (collab.DefaultGraceTTL, 60s after the last connection left) and the
// item still has content-bearing rows above its flush watermark, the item
// is queued.
// - T1, sweep: every materializeSweepInterval the store is asked for live
// items whose op-log holds such rows, whose newest row is older than
// materializeDormancy, and which have no set-aside rows
// (store.ListMaterializeCandidates); those without an open room are
// queued. It catches what T3 cannot see: a server restart, a room that
// closed while the worker was failing, rows left by an older server.
//
// One goroutine drains a de-duplicated queue into the Materializer (in
// production the single materialize.Supervisor, which runs the job in a
// capped child process). Nothing here runs on a request path.
//
// FAILURE BUDGET. The input is untrusted (any editor can write op-log rows),
// so a job is never retried in a loop: a failed item waits
// materializeBackoffBase, doubling per consecutive failure, and after
// materializeMaxFailures consecutive failures it is not tried again until its
// op-log changes (MAX(id) moves past the value recorded at the last failure).
// The budget is kept in memory, bounded at materializeBudgetCap entries: a
// restart re-grants each item its K attempts, which bounds a poison item at K
// worker deaths per process lifetime. Each failure is logged with its typed
// error (materialize.Err*).
const (
materializeSweepInterval = time.Minute
materializeDormancy = 2 * time.Minute
materializeSweepLimit = 200
materializeMaxFailures = 3
materializeBackoffBase = time.Minute
materializeBudgetCap = 10000
materializeQueueCap = 10000
)
// Materializer runs one materialization. *materialize.Supervisor implements it
// (and *materialize.Runner, which tests use in-process).
type Materializer interface {
Materialize(ctx context.Context, job materialize.Job) (string, error)
}
type materializeBudget struct {
failures int
opLogMax int64
nextAttempt time.Time
lastErr string
lastKind string
// exhaustionLogged: the one exhaustion line for this budget has been
// written. A fresh budget (after the op-log grows) starts false again.
exhaustionLogged bool
}
type materializeRecovery struct {
s *Server
m Materializer
cfg materializeRecoveryConfig
ctx context.Context
cancel context.CancelFunc
mu sync.Mutex
started bool
stopped bool
queued map[string]bool
queue []string
wake chan struct{}
budget map[string]*materializeBudget
// setAsideSkipped counts items the worker skipped because they hold rows a
// schema rebuild set aside (BUG-3244), over the process lifetime; reported
// by the sweep's summary line, never per item. setAsideLogged is the value
// the last summary line carried.
setAsideSkipped atomic.Int64
setAsideLogged int64
// sweepCursor is where the next sweep page starts (sweep goroutine only).
sweepCursor store.MaterializeCursor
// processed, when set by a test, receives each processed item and its
// result after the item is done.
processed func(itemID string, r materializeResult)
}
type materializeRecoveryConfig struct {
sweepInterval time.Duration
dormancy time.Duration
maxFailures int
backoffBase time.Duration
sweepLimit int
now func() time.Time
// logger receives the worker's log lines. Nil: slog.Default().
logger *slog.Logger
}
// materializeResult is what one processing pass did with an item.
type materializeResult string
const (
mrApplied materializeResult = "applied"
mrStamped materializeResult = "stamped"
mrNothing materializeResult = "nothing_pending"
mrRoomOpen materializeResult = "room_open"
mrSetAside materializeResult = "set_aside"
mrCursorMoved materializeResult = "cursor_moved"
mrGone materializeResult = "gone"
mrBudgetExceeded materializeResult = "budget_exhausted"
mrBackoff materializeResult = "backoff"
mrSchemaVersion materializeResult = "schema_version"
mrFailed materializeResult = "failed"
mrStopped materializeResult = "stopped"
)
// SetMaterializer installs the op-log recovery worker's Materializer. Nil (the
// default, and what PAD_MATERIALIZE=off leaves) disables recovery entirely:
// StartMaterializeRecovery then installs no hook and starts no goroutine. Must
// be called before StartMaterializeRecovery. If m also has a Close() error
// method, Stop() calls it after the worker has drained.
func (s *Server) SetMaterializer(m Materializer) {
if m == nil {
s.materializeRecovery = nil
return
}
s.materializeRecovery = newMaterializeRecovery(s, m, materializeRecoveryConfig{})
}
func newMaterializeRecovery(s *Server, m Materializer, cfg materializeRecoveryConfig) *materializeRecovery {
if cfg.sweepInterval <= 0 {
cfg.sweepInterval = materializeSweepInterval
}
if cfg.dormancy <= 0 {
cfg.dormancy = materializeDormancy
}
if cfg.maxFailures <= 0 {
cfg.maxFailures = materializeMaxFailures
}
if cfg.backoffBase <= 0 {
cfg.backoffBase = materializeBackoffBase
}
if cfg.sweepLimit <= 0 {
cfg.sweepLimit = materializeSweepLimit
}
if cfg.now == nil {
cfg.now = time.Now
}
if cfg.logger == nil {
cfg.logger = slog.Default()
}
ctx, cancel := context.WithCancel(context.Background())
return &materializeRecovery{
s: s, m: m, cfg: cfg, ctx: ctx, cancel: cancel,
queued: map[string]bool{},
wake: make(chan struct{}, 1),
budget: map[string]*materializeBudget{},
}
}
// StartMaterializeRecovery starts the op-log recovery worker and its two
// triggers. A no-op without a Materializer (SetMaterializer) or without a
// collab room manager (SetCollabRoomManager), and idempotent.
func (s *Server) StartMaterializeRecovery() {
r := s.materializeRecovery
if r == nil || s.collab == nil {
return
}
r.mu.Lock()
if r.started || r.stopped {
r.mu.Unlock()
return
}
r.started = true
r.mu.Unlock()
s.collab.SetIdleHook(r.onRoomIdle)
r.cfg.logger.Info("op-log recovery started",
"sweep_interval", r.cfg.sweepInterval.String(),
"dormancy", r.cfg.dormancy.String(),
"max_failures", r.cfg.maxFailures)
s.bg.Add(2)
go func() {
defer s.bg.Done()
r.workLoop()
}()
go func() {
defer s.bg.Done()
defer s.recoverSweeper("materialize-sweep")
t := time.NewTicker(r.cfg.sweepInterval)
defer t.Stop()
for {
select {
case <-r.ctx.Done():
return
case <-t.C:
r.sweep()
}
}
}()
}
func (s *Server) stopMaterializeRecovery() {
r := s.materializeRecovery
if r == nil {
return
}
r.mu.Lock()
r.stopped = true
r.mu.Unlock()
if s.collab != nil {
s.collab.SetIdleHook(nil)
}
r.cancel()
}
// closeMaterializer closes the Materializer when it has a Close method (the
// Supervisor: kills its worker process). Called by Stop after bg.Wait.
func (s *Server) closeMaterializer() {
r := s.materializeRecovery
if r == nil {
return
}
if c, ok := r.m.(interface{ Close() error }); ok {
if err := c.Close(); err != nil {
r.cfg.logger.Warn("op-log recovery: closing the materializer failed", "error", err)
}
}
}
// onRoomIdle is T3: the collab room for itemID was reclaimed. Runs on the
// grace timer's goroutine; one indexed read, then a non-blocking enqueue.
func (r *materializeRecovery) onRoomIdle(itemID string) {
pending, err := r.s.store.ItemHasPendingContent(itemID)
if err != nil {
r.cfg.logger.Warn("op-log recovery: pending check on room idle failed", "item_id", itemID, "error", err)
return
}
if pending {
r.enqueue(itemID)
}
}
// sweep is T1.
func (r *materializeRecovery) sweep() {
// One page per tick, continuing from where the last tick stopped and
// wrapping to the start after a short page: every candidate is offered
// within ceil(candidates / sweepLimit) + 1 ticks, however many of those
// ahead of it the room check or the failure budget turn away (see
// ListMaterializeCandidates).
page, err := r.s.store.ListMaterializeCandidates(r.cfg.now().Add(-r.cfg.dormancy), r.sweepCursor, r.cfg.sweepLimit)
if err != nil {
r.cfg.logger.Warn("op-log recovery: sweep query failed", "error", err)
return
}
if len(page) < r.cfg.sweepLimit {
r.sweepCursor = store.MaterializeCursor{}
} else {
last := page[len(page)-1]
r.sweepCursor = store.MaterializeCursor{LastAt: last.LastAt, ItemID: last.ItemID}
}
ids := make([]string, 0, len(page))
queued := 0
for _, c := range page {
ids = append(ids, c.ItemID)
if r.s.collab != nil && r.s.collab.HasRoom(c.ItemID) {
continue
}
if r.enqueue(c.ItemID) {
queued++
}
}
// ONE summary line per sweep, and only when it has news: items queued, or
// a set-aside skip since the last line. Set-aside items never come from
// this query (it excludes them); they reach the worker through the room
// trigger or a race, and are counted there, not logged per item.
skipped := r.setAsideSkipped.Load()
if queued > 0 || skipped != r.setAsideLogged {
r.setAsideLogged = skipped
r.cfg.logger.Info("op-log recovery sweep",
"candidates", len(ids), "queued", queued, "set_aside_skipped_total", skipped)
}
}
// enqueue adds itemID unless it is already queued. Never blocks; a full queue
// drops the item, which the next sweep will offer again.
func (r *materializeRecovery) enqueue(itemID string) bool {
r.mu.Lock()
defer r.mu.Unlock()
if r.stopped || r.queued[itemID] {
return false
}
if len(r.queue) >= materializeQueueCap {
return false
}
r.queued[itemID] = true
r.queue = append(r.queue, itemID)
select {
case r.wake <- struct{}{}:
default:
}
return true
}
func (r *materializeRecovery) dequeue() (string, bool) {
r.mu.Lock()
defer r.mu.Unlock()
if len(r.queue) == 0 {
return "", false
}
id := r.queue[0]
r.queue = r.queue[1:]
delete(r.queued, id)
return id, true
}
func (r *materializeRecovery) workLoop() {
for {
for {
if r.ctx.Err() != nil {
return
}
id, ok := r.dequeue()
if !ok {
break
}
res := r.processSafely(id)
if r.processed != nil {
r.processed(id, res)
}
}
select {
case <-r.ctx.Done():
return
case <-r.wake:
}
}
}
// processSafely is process with a panic fence, so one bad item cannot end the
// worker goroutine.
func (r *materializeRecovery) processSafely(itemID string) (res materializeResult) {
defer func() {
if p := recover(); p != nil {
r.cfg.logger.Error("op-log recovery: panic processing item",
"item_id", itemID, "panic", p, "stack", string(debug.Stack()))
res = mrFailed
}
}()
return r.process(itemID)
}
func (r *materializeRecovery) process(itemID string) materializeResult {
s := r.s
if s.collab == nil {
return mrStopped
}
if s.collab.HasRoom(itemID) {
return mrRoomOpen
}
in, err := s.store.LoadMaterializeInput(itemID)
if err != nil {
r.cfg.logger.Warn("op-log recovery: reading the op-log failed", "item_id", itemID, "error", err)
return mrFailed
}
if in == nil {
r.forget(itemID)
return mrGone
}
if in.SetAside > 0 {
r.setAsideSkipped.Add(1)
return mrSetAside
}
if in.Pending == 0 {
r.forget(itemID)
return mrNothing
}
if res, ok := r.admit(itemID, in.Cursor); !ok {
return res
}
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))
return mrSchemaVersion
}
}
ws, err := s.store.GetWorkspaceByID(in.WorkspaceID)
if err != nil || ws == nil {
r.cfg.logger.Warn("op-log recovery: reading the workspace failed", "item_id", itemID, "error", err)
return mrFailed
}
index, err := s.materializeLinkIndex(in.WorkspaceID)
if err != nil {
r.cfg.logger.Warn("op-log recovery: building the link index failed", "item_id", itemID, "error", err)
return mrFailed
}
job := materialize.Job{
Rows: in.Rows,
SchemaVersion: collab.DefaultSchemaVersion,
LinkIndex: index,
WorkspaceSlug: ws.Slug,
}
started := time.Now()
md, err := r.m.Materialize(r.ctx, job)
if err != nil {
if r.ctx.Err() != nil {
return mrStopped
}
r.recordFailure(itemID, in.Cursor, err)
return mrFailed
}
var (
outcome store.MaterializeOutcome
updated *models.Item
)
err = s.collab.UnderItemLockIfNoRoom(itemID, func() error {
var ferr error
outcome, updated, ferr = s.store.MaterializeFlush(itemID, in.Cursor, md)
return ferr
})
if errors.Is(err, collab.ErrRoomOpen) {
return mrRoomOpen
}
if err != nil {
if r.ctx.Err() != nil {
return mrStopped
}
r.cfg.logger.Warn("op-log recovery: writing the recovered body failed", "item_id", itemID, "error", err)
return mrFailed
}
r.forget(itemID)
switch outcome {
case store.MaterializeApplied:
r.cfg.logger.Info("op-log recovery: recovered an item's unsaved editor session",
"item_id", itemID, "op_log_cursor", in.Cursor, "rows", len(in.Rows),
"pending_rows", in.Pending, "bytes", len(md), "elapsed", time.Since(started).String())
if updated != nil {
s.publishItemEventWithName(sseItemUpdated, updated.WorkspaceID, updated.ID, updated.Title,
updated.CollectionSlug, models.VersionCreatedBySystem, "", models.VersionSourceRecovery, updated.Seq)
}
return mrApplied
case store.MaterializeStamped:
return mrStamped
case store.MaterializeCursorMoved:
return mrCursorMoved
case store.MaterializeNothingPending:
return mrNothing
case store.MaterializeSetAside:
r.setAsideSkipped.Add(1)
return mrSetAside
default:
return mrGone
}
}
// 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) {
r.mu.Lock()
defer r.mu.Unlock()
b := r.budget[itemID]
if b == nil {
return "", true
}
if cursor > b.opLogMax {
delete(r.budget, itemID)
return "", true
}
if b.failures >= r.cfg.maxFailures {
return mrBudgetExceeded, false
}
if r.cfg.now().Before(b.nextAttempt) {
return mrBackoff, false
}
return "", true
}
func (r *materializeRecovery) forget(itemID string) {
r.mu.Lock()
delete(r.budget, itemID)
r.mu.Unlock()
}
func (r *materializeRecovery) budgetSlotLocked(itemID string) *materializeBudget {
b := r.budget[itemID]
if b == nil {
if len(r.budget) >= materializeBudgetCap {
for k := range r.budget { // evict one; it re-earns its K attempts
delete(r.budget, k)
break
}
}
b = &materializeBudget{}
r.budget[itemID] = b
}
return b
}
func (r *materializeRecovery) recordFailure(itemID string, cursor int64, err error) {
r.mu.Lock()
b := r.budgetSlotLocked(itemID)
kind := materializeErrorKind(err)
b.failures++
b.opLogMax = cursor
b.lastErr = err.Error()
b.lastKind = kind
delay := r.cfg.backoffBase << (b.failures - 1)
b.nextAttempt = r.cfg.now().Add(delay)
failures := b.failures
logExhaustion := r.claimExhaustionLogLocked(b)
r.mu.Unlock()
attrs := []any{"item_id", itemID, "op_log_cursor", cursor, "failures", failures,
"max_failures", r.cfg.maxFailures, "kind", kind, "error", err}
// No memory cap could be established, so the supervisor refused the job
// without sending it anywhere, and has already warned once for the whole
// worker. It counts against the budget like any failure, but a line per
// item would repeat that warning for every pending item on every pass —
// so even its exhaustion line is DEBUG.
if errors.Is(err, materialize.ErrNoMemoryCap) {
if logExhaustion {
r.logExhaustion(slog.LevelDebug, itemID, failures, kind, err.Error())
} else {
r.cfg.logger.Debug("op-log recovery: job refused, no memory cap", attrs...)
}
return
}
if logExhaustion {
r.logExhaustion(slog.LevelWarn, itemID, failures, kind, err.Error())
return
}
r.cfg.logger.Warn("op-log recovery: materialization failed; will retry after a backoff",
append(attrs, "retry_after", delay.String())...)
}
// claimExhaustionLogLocked reports whether b has just used up the budget and
// its exhaustion line has not been written yet, and marks it written. Once per
// budget: admit refuses an exhausted item without reaching recordFailure, and
// only op-log growth (which deletes the budget) can start a new one.
func (r *materializeRecovery) claimExhaustionLogLocked(b *materializeBudget) bool {
if b.failures < r.cfg.maxFailures || b.exhaustionLogged {
return false
}
b.exhaustionLogged = true
return true
}
// logExhaustion is the ONE line a stuck item leaves: which item, how many
// failures, and the kind of the last one. WARN, because from here on the item's
// newest edits stay out of its stored body until someone edits it again, which
// an operator should be able to find with a grep and no database column; it
// cannot become noise, since it is written once per exhaustion.
func (r *materializeRecovery) logExhaustion(level slog.Level, itemID string, failures int, kind, lastErr string) {
r.cfg.logger.Log(r.ctx, level, "op-log recovery: item exhausted its failure budget; not retried until its op-log changes",
"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) {
r.mu.Lock()
b := r.budgetSlotLocked(itemID)
b.failures = r.cfg.maxFailures
b.opLogMax = cursor
b.lastErr = why
b.lastKind = "schema_version"
logExhaustion := r.claimExhaustionLogLocked(b)
r.mu.Unlock()
if logExhaustion {
r.logExhaustion(slog.LevelWarn, itemID, r.cfg.maxFailures, "schema_version", why)
}
}
// materializeErrorKind names the typed materialize error, for the log line.
func materializeErrorKind(err error) string {
for _, k := range []struct {
err error
name string
}{
{materialize.ErrNoMemoryCap, "no_memory_cap"},
{materialize.ErrMemoryLimit, "memory_limit"},
{materialize.ErrDeadline, "deadline"},
{materialize.ErrChildDied, "worker_died"},
{materialize.ErrProtocol, "protocol"},
{materialize.ErrStart, "start"},
{materialize.ErrJobFailed, "job_failed"},
{materialize.ErrJobTooLarge, "too_large"},
{materialize.ErrSchemaVersion, "schema_version"},
{materialize.ErrClosed, "closed"},
{materialize.ErrInterrupted, "interrupted"},
} {
if errors.Is(err, k.err) {
return k.name
}
}
return "other"
}
// materializeLinkIndex is the link index a recovery job converts internal
// links against: EVERY live item of the workspace (Dave's ruling), not the
// set any one viewer can see. A tab's index is its viewer's localIndex, but a
// recovery has no viewer, and an index missing an item a link points at would
// leave that link as a raw URL instead of the [[REF]] a full tab stores.
// Titles in the output come only from the document itself: markdownToWikiLinks
// compares an item's title with the link text and never emits it (proven by
// TestMaterializeLinkIndexContributesNoTitle).
//
// ORDER mirrors localIndex.getAll (web/src/lib/stores/localIndex.svelte.ts):
// updated_at DESC, then id ASC, because markdownToWikiLinks takes the FIRST
// match. Sorted here rather than trusting ORDER BY, because Postgres compares
// TEXT under the database collation and the tab compares UTF-16 code units;
// Go's byte order is the tab's order for the ASCII ids the store mints.
func (s *Server) materializeLinkIndex(workspaceID string) ([]materialize.LinkEntry, error) {
items, err := s.store.ListItemsIndex(workspaceID, store.ItemIndexParams{})
if err != nil {
return nil, err
}
sort.SliceStable(items, func(i, j int) bool {
a, b := items[i], items[j]
if !a.UpdatedAt.Equal(b.UpdatedAt) {
return a.UpdatedAt.After(b.UpdatedAt)
}
return a.ID < b.ID
})
out := make([]materialize.LinkEntry, 0, len(items))
for _, it := range items {
e := materialize.LinkEntry{Slug: it.Slug, Title: it.Title, CollectionPrefix: it.CollectionPrefix}
if it.ItemNumber != nil {
e.ItemNumber = *it.ItemNumber
}
out = append(out, e)
}
return out, nil
}
@@ -0,0 +1,283 @@
package server
import (
"bytes"
"context"
"encoding/json"
"os"
"strings"
"testing"
"time"
"github.com/PerpetualSoftware/pad/internal/collab"
"github.com/PerpetualSoftware/pad/internal/materialize"
"github.com/PerpetualSoftware/pad/internal/models"
)
// TASK-2198 U4, legs (i) and (ii) of the link-index ruling. The recovery
// converts internal links against the WHOLE workspace's index (every live
// item, no viewer), so the question is whether that index can put text into a
// recovered body — above all an item's TITLE, which a restricted viewer of the
// body may not be entitled to see.
//
// The documents are in testdata/materialize_links.json, written by
// web/src/lib/collab/materializer/materializerLinkFixture.test.ts: each is a
// real Y.Doc holding a link as a tab stores it after loading. `raw` is the
// same document through the flush pipeline with NO index, i.e. the text the
// document itself contains.
type linkFixtureCase struct {
Name string `json:"name"`
Markdown string `json:"markdown"`
Rows []string `json:"rows"`
Raw string `json:"raw"`
}
func loadLinkFixture(t *testing.T) []linkFixtureCase {
t.Helper()
b, err := os.ReadFile("testdata/materialize_links.json")
if err != nil {
t.Fatal(err)
}
var cases []linkFixtureCase
if err := json.Unmarshal(b, &cases); err != nil {
t.Fatal(err)
}
if len(cases) < 8 {
t.Fatalf("fixture has %d cases", len(cases))
}
return cases
}
const secretTitle = "Secret Launch Codename"
// linkWorld: workspace "ws" holding X = SECR-1 (secretTitle) in "secrets", a
// collection a restricted member or guest would not see, and PUB-2 in
// "public". Item numbers are workspace-sequential, so the order of creation
// fixes both refs; the test checks them.
func newLinkWorld(t *testing.T) (*recoveryFixture, *models.Item, *models.Item) {
t.Helper()
f := newRecoveryFixture(t, recoveryRunner(t), time.Minute, materializeRecoveryConfig{})
secrets, err := f.srv.store.CreateCollection(f.ws.ID, models.CollectionCreate{Name: "Secrets", Prefix: "SECR", Schema: `{"fields":[]}`})
if err != nil {
t.Fatal(err)
}
public, err := f.srv.store.CreateCollection(f.ws.ID, models.CollectionCreate{Name: "Public", Prefix: "PUB", Schema: `{"fields":[]}`})
if err != nil {
t.Fatal(err)
}
x, err := f.srv.store.CreateItem(f.ws.ID, secrets.ID, models.ItemCreate{Title: secretTitle, Fields: `{}`})
if err != nil {
t.Fatal(err)
}
p, err := f.srv.store.CreateItem(f.ws.ID, public.ID, models.ItemCreate{Title: "Public Plan", Fields: `{}`})
if err != nil {
t.Fatal(err)
}
x, _ = f.srv.store.GetItem(x.ID)
p, _ = f.srv.store.GetItem(p.ID)
if x.Ref != "SECR-1" || x.Slug != "secret-launch-codename" || x.CollectionSlug != "secrets" || p.Ref != "PUB-2" {
t.Fatalf("fixture refs: X %s/%s/%s, P %s", x.Ref, x.Slug, x.CollectionSlug, p.Ref)
}
return f, x, p
}
// recoverCase inserts the case's op-log over a stale body and runs one
// recovery pass through the production path (index builder included).
func recoverCase(t *testing.T, f *recoveryFixture, c linkFixtureCase) string {
t.Helper()
it := f.item(t, "Recovering "+c.Name, "stale")
f.appendRows(t, it.ID, decodeRows(t, c.Rows)...)
if res := f.r.process(it.ID); res != mrApplied {
t.Fatalf("%s: process = %q", c.Name, res)
}
got, _ := f.srv.store.GetItem(it.ID)
return got.Content
}
// LEG (i). The STOP condition: no index-sourced TITLE may appear in a
// recovered body. Each form markdownToWikiLinks can emit is covered:
//
// [[REF]] link text == the item's CURRENT title: the title was in
// the document and is REMOVED, not added.
// [[REF|text]] any other text: `text` comes from the document.
// [[Title]] the legacy form for an item with no ref: `text` again.
// (Every item the store mints has a ref, so the legacy
// branch is reached only for rows with a NULL item_number;
// it emits the document's link text either way.)
// [[ws::REF|text]] cross-workspace: the index is not consulted.
//
// What the index CAN contribute is the REF, and only for a link addressed by
// SLUG (`/owner/ws/coll/<slug>` becomes `[[SECR-1|text]]`). The slug itself is
// in the document; the ref is a number the item already exposes in its URL
// wherever it is linked by ref. That is the whole of the index's contribution.
func TestMaterializeLinkIndexContributesNoTitle(t *testing.T) {
f, _, _ := newLinkWorld(t)
index, err := f.srv.materializeLinkIndex(f.ws.ID)
if err != nil {
t.Fatal(err)
}
want := map[string]string{
"ref-link-other-text": "See [[SECR-1|shown text]] here.",
"ref-link-old-title": "See [[SECR-1|Old Codename]] here.",
"slug-link-slug-text": "See [[SECR-1|the plan]] here.",
"text-equals-current-title": "See [[SECR-1]] here.",
"broken-link": "See [[gone text]] here.",
"cross-workspace": "See [[other::SECR-1|xw text]] here.",
"typed-wiki-literal": "Typed [[SECR-1]] literal.",
"public-ref-link": "See [[PUB-2|the public plan]] here.",
// BUG-3315 U2: a follows-title link follows the title, renamed or not.
"marker-follows-title": "See [[SECR-1]] here.",
"marker-renamed": "See [[SECR-1]] here.",
"marker-external-href": "See [Old Codename](https://example.com/x) here.",
}
for _, c := range loadLinkFixture(t) {
got := recoverCase(t, f, c)
if w, ok := want[c.Name]; !ok || got != w {
t.Errorf("%s: recovered %q, want %q", c.Name, got, w)
}
// The general claim, over EVERY title in the index the job was given:
// a title in the output must already be in the document.
for _, e := range index {
if e.Title != "" && strings.Contains(got, e.Title) && !strings.Contains(c.Raw, e.Title) {
t.Errorf("STOP: %s: title %q of %s-%d is in the recovered body but not in the document\nbody: %q\ndoc: %q",
c.Name, e.Title, e.CollectionPrefix, e.ItemNumber, got, c.Raw)
}
}
if strings.Contains(got, secretTitle) {
t.Errorf("STOP: %s: X's title reached the recovered body: %q", c.Name, got)
}
}
}
// LEG (ii). A link to an item a restricted viewer cannot see survives the
// recovery as the same [[REF|text]] a full member's tab would store, because
// the index is the whole workspace. With a VIEWER-RESTRICTED index (what a
// restricted tab's localIndex holds) the same document stores a raw URL
// instead, bound to the owner and workspace slugs, which is the degradation
// the whole-workspace ruling exists to prevent.
func TestMaterializeLinkToRestrictedItemSurvives(t *testing.T) {
f, x, _ := newLinkWorld(t)
var c linkFixtureCase
for _, k := range loadLinkFixture(t) {
if k.Name == "ref-link-other-text" {
c = k
}
}
got := recoverCase(t, f, c)
if got != "See [[SECR-1|shown text]] here." {
t.Fatalf("recovered %q: the link to the restricted item did not survive as [[SECR-1|shown text]]", got)
}
// The counterfactual, through the same bundle: drop X's collection.
full, err := f.srv.materializeLinkIndex(f.ws.ID)
if err != nil {
t.Fatal(err)
}
var restricted []materialize.LinkEntry
for _, e := range full {
if e.CollectionPrefix != "SECR" {
restricted = append(restricted, e)
}
}
if len(restricted) != len(full)-1 {
t.Fatalf("restricted index %d of %d; X %s not singled out", len(restricted), len(full), x.Ref)
}
md, err := recoveryRunner(t).Materialize(context.Background(), materialize.Job{
Rows: decodeRows(t, c.Rows), SchemaVersion: collab.DefaultSchemaVersion, LinkIndex: restricted, WorkspaceSlug: "ws",
})
if err != nil {
t.Fatal(err)
}
if md != "See [shown text](/alice/ws/secrets/SECR-1) here." {
t.Fatalf("restricted-index counterfactual = %q", md)
}
}
// BUG-3315 U2: a tab since #1688 loads a link that follows its target's title
// with the follows-title marker, and the marker lives in the Y.Doc. Recovery
// replays that document, so it must strip the marker exactly as a tab's save
// does: on the production path (whole-workspace index), with a restricted
// index (the target unseen), and with none. The marker must never reach
// items.content.
func TestMaterializeRecoveryStripsFollowsTitleMarker(t *testing.T) {
f, _, _ := newLinkWorld(t)
full, err := f.srv.materializeLinkIndex(f.ws.ID)
if err != nil {
t.Fatal(err)
}
var restricted []materialize.LinkEntry
for _, e := range full {
if e.CollectionPrefix != "SECR" {
restricted = append(restricted, e)
}
}
n := 0
for _, c := range loadLinkFixture(t) {
if !strings.HasPrefix(c.Name, "marker-") {
continue
}
n++
// Guard: the op-log really carries the marker.
carries := false
for _, r := range decodeRows(t, c.Rows) {
carries = carries || bytes.Contains(r, []byte("pad-follows-title"))
}
if !carries {
t.Fatalf("%s: the fixture's op-log does not carry the marker", c.Name)
}
if got := recoverCase(t, f, c); strings.Contains(got, "pad-follows-title") {
t.Errorf("%s: production recovery stored the marker: %q", c.Name, got)
}
for name, idx := range map[string][]materialize.LinkEntry{"restricted": restricted, "empty": {}} {
md, err := recoveryRunner(t).Materialize(context.Background(), materialize.Job{
Rows: decodeRows(t, c.Rows), SchemaVersion: collab.DefaultSchemaVersion, LinkIndex: idx, WorkspaceSlug: "ws",
})
if err != nil {
t.Fatal(err)
}
if strings.Contains(md, "pad-follows-title") {
t.Errorf("%s, %s index: recovery kept the marker: %q", c.Name, name, md)
}
}
}
if n != 3 {
t.Fatalf("%d marker cases in the fixture, want 3", n)
}
}
// The index is every live item, ordered as localIndex.getAll orders a tab's:
// updated_at DESC, then id ASC.
func TestMaterializeLinkIndexOrderAndScope(t *testing.T) {
f, x, p := newLinkWorld(t)
gone := f.item(t, "Archived one", "")
if err := f.srv.store.DeleteItem(gone.ID); err != nil {
t.Fatal(err)
}
// Make X the most recently updated.
title := secretTitle
if _, err := f.srv.store.UpdateItem(x.ID, models.ItemUpdate{Title: &title, Content: strPtr("touched")}); err != nil {
t.Fatal(err)
}
time.Sleep(1100 * time.Millisecond) // updated_at has one-second resolution
body := "later"
if _, err := f.srv.store.UpdateItem(p.ID, models.ItemUpdate{Content: &body}); err != nil {
t.Fatal(err)
}
index, err := f.srv.materializeLinkIndex(f.ws.ID)
if err != nil {
t.Fatal(err)
}
if len(index) != 2 {
t.Fatalf("index = %+v, want the two live items (archived excluded)", index)
}
if index[0].Title != "Public Plan" || index[1].Title != secretTitle {
t.Fatalf("order = [%q, %q], want most recently updated first", index[0].Title, index[1].Title)
}
if index[1].ItemNumber != 1 || index[1].CollectionPrefix != "SECR" || index[1].Slug != "secret-launch-codename" {
t.Fatalf("entry = %+v", index[1])
}
}
func strPtr(s string) *string { return &s }
@@ -0,0 +1,778 @@
package server
import (
"bytes"
"context"
"encoding/base64"
"encoding/json"
"fmt"
"io"
"log/slog"
"net/http"
"net/http/httptest"
"os"
"strings"
"sync"
"sync/atomic"
"testing"
"time"
pad "github.com/PerpetualSoftware/pad"
"github.com/PerpetualSoftware/pad/internal/collab"
"github.com/PerpetualSoftware/pad/internal/materialize"
"github.com/PerpetualSoftware/pad/internal/models"
"github.com/PerpetualSoftware/pad/internal/store"
)
// TASK-2198 U4: the op-log recovery triggers, worker and failure budget, and
// the end-to-end recovery through the real materializer bundle.
// ---------------------------------------------------------------- fixtures
// fakeMaterializer answers every job with fn and counts calls.
type fakeMaterializer struct {
calls atomic.Int64
mu sync.Mutex
jobs []materialize.Job
fn func(job materialize.Job) (string, error)
}
func (f *fakeMaterializer) Materialize(_ context.Context, job materialize.Job) (string, error) {
f.calls.Add(1)
f.mu.Lock()
f.jobs = append(f.jobs, job)
f.mu.Unlock()
if f.fn == nil {
return "recovered body", nil
}
return f.fn(job)
}
// recoveryRunner is ONE in-process Runner over the embedded bundle; loading
// costs about a second. It satisfies Materializer, so the server drives the
// real JavaScript without a child process.
var (
recoveryRunnerOnce sync.Once
recoveryRunnerVal *materialize.Runner
recoveryRunnerErr error
)
func recoveryRunner(t *testing.T) *materialize.Runner {
t.Helper()
recoveryRunnerOnce.Do(func() { recoveryRunnerVal, recoveryRunnerErr = materialize.New(pad.MaterializerJS) })
if recoveryRunnerErr != nil {
t.Fatalf("load the materializer bundle: %v", recoveryRunnerErr)
}
return recoveryRunnerVal
}
type recoveryFixture struct {
srv *Server
rm *collab.RoomManager
r *materializeRecovery
ws *models.Workspace
coll *models.Collection
mu sync.Mutex
processed []string
results map[string]materializeResult
done chan string
}
// newRecoveryFixture: a test server with a room manager (grace TTL grace) and
// an op-log recovery worker over m, NOT started. cfg zero fields default.
func newRecoveryFixture(t *testing.T, m Materializer, grace time.Duration, cfg materializeRecoveryConfig) *recoveryFixture {
t.Helper()
srv := testServer(t)
bus := collab.NewMemoryOpBus()
t.Cleanup(bus.Close)
rm := collab.NewRoomManagerWithConfig(srv.store, bus, collab.RoomManagerConfig{GraceTTL: grace})
t.Cleanup(rm.Close)
srv.SetCollabRoomManager(rm)
f := &recoveryFixture{srv: srv, rm: rm, results: map[string]materializeResult{}, done: make(chan string, 256)}
if m != nil {
f.r = newMaterializeRecovery(srv, m, cfg)
f.r.processed = func(id string, res materializeResult) {
f.mu.Lock()
f.processed = append(f.processed, id)
f.results[id] = res
f.mu.Unlock()
f.done <- id
}
srv.materializeRecovery = f.r
}
ws, err := srv.store.CreateWorkspace(models.WorkspaceCreate{Name: "Recovery WS", Slug: "ws"})
if err != nil {
t.Fatalf("CreateWorkspace: %v", err)
}
coll, err := srv.store.CreateCollection(ws.ID, models.CollectionCreate{Name: "Notes", Prefix: "NOTE", Schema: `{"fields":[]}`})
if err != nil {
t.Fatalf("CreateCollection: %v", err)
}
f.ws, f.coll = ws, coll
return f
}
func (f *recoveryFixture) item(t *testing.T, title, content string) *models.Item {
t.Helper()
it, err := f.srv.store.CreateItem(f.ws.ID, f.coll.ID, models.ItemCreate{Title: title, Content: content, Fields: `{}`})
if err != nil {
t.Fatalf("CreateItem: %v", err)
}
return it
}
func (f *recoveryFixture) appendRows(t *testing.T, itemID string, rows ...[]byte) int64 {
t.Helper()
var last int64
for _, r := range rows {
id, err := f.srv.store.AppendYjsUpdate(itemID, r, collab.DefaultSchemaVersion)
if err != nil {
t.Fatalf("AppendYjsUpdate: %v", err)
}
last = id
}
return last
}
func (f *recoveryFixture) waitProcessed(t *testing.T, itemID string, within time.Duration) materializeResult {
t.Helper()
deadline := time.After(within)
for {
f.mu.Lock()
res, ok := f.results[itemID]
f.mu.Unlock()
if ok {
return res
}
select {
case <-f.done:
case <-deadline:
t.Fatalf("item %s was not processed within %s", itemID, within)
}
}
}
func (f *recoveryFixture) queueSnapshot() []string {
f.r.mu.Lock()
defer f.r.mu.Unlock()
return append([]string(nil), f.r.queue...)
}
// contentFrame is a content-bearing Update frame, distinct per n.
func contentFrame(n byte) []byte { return []byte{0x00, 0x02, 0x05, 0x01, n, 0x00, 0x7F, 0x00} }
func dialAndClose(t *testing.T, baseURL, itemID string) {
t.Helper()
c, resp, err := dialCollab(t, baseURL, itemID, nil, "")
if err != nil {
st := ""
if resp != nil {
st = resp.Status
}
t.Fatalf("dial %s: %v %s", itemID, err, st)
}
// Read until the op_log_cursor text frame: the join has registered.
_ = c.SetReadDeadline(time.Now().Add(3 * time.Second))
for {
mt, _, rerr := c.ReadMessage()
if rerr != nil {
t.Fatalf("read: %v", rerr)
}
if mt == 1 { // websocket.TextMessage
break
}
}
c.Close()
}
// ---------------------------------------------------------------- T3
// T3's decision, unit level: a reclaimed room queues its item only when the
// item still has content-bearing rows above its watermark.
func TestMaterializeRoomIdleEnqueuesOnlyWithPendingRows(t *testing.T) {
f := newRecoveryFixture(t, &fakeMaterializer{}, time.Minute, materializeRecoveryConfig{})
pending := f.item(t, "Pending", "stale")
f.appendRows(t, pending.ID, contentFrame(1))
flushed := f.item(t, "Flushed", "fine")
c := f.appendRows(t, flushed.ID, contentFrame(2))
if err := f.srv.store.SetItemContentFlushedOpLogIDForTesting(flushed.ID, c); err != nil {
t.Fatal(err)
}
empty := f.item(t, "No op-log", "fine")
f.r.onRoomIdle(pending.ID)
f.r.onRoomIdle(flushed.ID)
f.r.onRoomIdle(empty.ID)
q := f.queueSnapshot()
if len(q) != 1 || q[0] != pending.ID {
t.Fatalf("queue after room idle = %v, want only %s", q, pending.ID)
}
// De-duplicated.
f.r.onRoomIdle(pending.ID)
if q := f.queueSnapshot(); len(q) != 1 {
t.Fatalf("queue after a second idle = %v, want one entry", q)
}
}
// T3 end to end: a real collab room is reclaimed at the end of its grace TTL,
// the hook installed by StartMaterializeRecovery fires, and the pending item
// is recovered while the clean one is never handed to the materializer.
func TestMaterializeRoomIdleRecoversThroughTheHook(t *testing.T) {
fake := &fakeMaterializer{}
f := newRecoveryFixture(t, fake, 50*time.Millisecond, materializeRecoveryConfig{sweepInterval: time.Hour})
ts := httptest.NewServer(f.srv)
t.Cleanup(ts.Close)
pending := f.item(t, "Pending", "stale")
f.appendRows(t, pending.ID, contentFrame(1))
clean := f.item(t, "Clean", "fine")
f.srv.StartMaterializeRecovery()
dialAndClose(t, ts.URL, clean.ID)
dialAndClose(t, ts.URL, pending.ID)
if res := f.waitProcessed(t, pending.ID, 5*time.Second); res != mrApplied {
t.Fatalf("pending item result %q, want applied", res)
}
got, _ := f.srv.store.GetItem(pending.ID)
if got.Content != "recovered body" || got.ContentState != "" {
t.Fatalf("after room idle: content %q state %q", got.Content, got.ContentState)
}
// The clean item's room is reclaimed too (both dials closed before the
// pending one was processed); it must not have been queued.
deadline := time.Now().Add(2 * time.Second)
for f.rm.RoomCount() > 0 && time.Now().Before(deadline) {
time.Sleep(10 * time.Millisecond)
}
time.Sleep(100 * time.Millisecond)
f.mu.Lock()
defer f.mu.Unlock()
if _, ok := f.results[clean.ID]; ok {
t.Fatalf("the clean item was queued on room idle")
}
if n := fake.calls.Load(); n != 1 {
t.Fatalf("materializer calls = %d, want 1", n)
}
}
// ---------------------------------------------------------------- T1
func TestMaterializeSweepQueuesDormantPendingItemsWithoutARoom(t *testing.T) {
f := newRecoveryFixture(t, &fakeMaterializer{}, time.Minute, materializeRecoveryConfig{
// The sweep's clock is an hour ahead, so rows written now are dormant.
now: func() time.Time { return time.Now().Add(time.Hour) },
})
ts := httptest.NewServer(f.srv)
t.Cleanup(ts.Close)
dormant := f.item(t, "Dormant", "stale")
f.appendRows(t, dormant.ID, contentFrame(1))
withRoom := f.item(t, "Open room", "stale")
f.appendRows(t, withRoom.ID, contentFrame(2))
flushed := f.item(t, "Flushed", "fine")
c := f.appendRows(t, flushed.ID, contentFrame(3))
if err := f.srv.store.SetItemContentFlushedOpLogIDForTesting(flushed.ID, c); err != nil {
t.Fatal(err)
}
// Hold a room open on withRoom (the connection stays up).
conn, _, err := dialCollab(t, ts.URL, withRoom.ID, nil, "")
if err != nil {
t.Fatalf("dial: %v", err)
}
t.Cleanup(func() { conn.Close() })
deadline := time.Now().Add(3 * time.Second)
for !f.rm.HasRoom(withRoom.ID) && time.Now().Before(deadline) {
time.Sleep(5 * time.Millisecond)
}
if !f.rm.HasRoom(withRoom.ID) {
t.Fatal("no room for the open connection")
}
f.r.sweep()
q := f.queueSnapshot()
if len(q) != 1 || q[0] != dormant.ID {
t.Fatalf("sweep queued %v, want only the dormant item %s", q, dormant.ID)
}
// The worker re-checks: even if an item with a room is queued, it is not
// materialized.
if res := f.r.process(withRoom.ID); res != mrRoomOpen {
t.Fatalf("process(item with a room) = %q, want room_open", res)
}
}
// ---------------------------------------------------------------- budget
func TestMaterializeFailureBudgetStopsAfterKAndResumesOnNewRows(t *testing.T) {
clock := time.Now()
var clockMu sync.Mutex
now := func() time.Time { clockMu.Lock(); defer clockMu.Unlock(); return clock }
advance := func(d time.Duration) { clockMu.Lock(); clock = clock.Add(d); clockMu.Unlock() }
fake := &fakeMaterializer{fn: func(materialize.Job) (string, error) {
return "", fmt.Errorf("%w: exit status 2", materialize.ErrChildDied)
}}
f := newRecoveryFixture(t, fake, time.Minute, materializeRecoveryConfig{now: now, maxFailures: 3, backoffBase: time.Minute})
it := f.item(t, "Poison", "stale")
f.appendRows(t, it.ID, contentFrame(1))
want := []struct {
res materializeResult
calls int64
after time.Duration
}{
{mrFailed, 1, 0},
{mrBackoff, 1, 0}, // inside the 1m backoff
{mrFailed, 2, time.Minute + 1}, // backoff over
{mrBackoff, 2, time.Minute}, // 2m backoff now
{mrFailed, 3, time.Minute + 1}, // K reached
{mrBudgetExceeded, 3, 24 * time.Hour}, // no time heals it
}
for i, w := range want {
advance(w.after)
if res := f.r.process(it.ID); res != w.res || fake.calls.Load() != w.calls {
t.Fatalf("step %d: result %q calls %d, want %q calls %d", i, res, fake.calls.Load(), w.res, w.calls)
}
}
// A new op-log row resets the budget.
f.appendRows(t, it.ID, contentFrame(2))
if res := f.r.process(it.ID); res != mrFailed || fake.calls.Load() != 4 {
t.Fatalf("after a new row: result %q calls %d, want failed, 4", res, fake.calls.Load())
}
// And a success clears it.
fake.fn = nil
advance(time.Hour)
if res := f.r.process(it.ID); res != mrApplied {
t.Fatalf("after recovery of the input: %q, want applied", res)
}
f.r.mu.Lock()
_, left := f.r.budget[it.ID]
f.r.mu.Unlock()
if left {
t.Fatal("budget entry survived a success")
}
}
// Rows written under another editor schema are skipped, not materialized.
func TestMaterializeSkipsForeignSchemaVersion(t *testing.T) {
fake := &fakeMaterializer{}
f := newRecoveryFixture(t, fake, time.Minute, materializeRecoveryConfig{})
it := f.item(t, "Old schema", "stale")
if _, err := f.srv.store.AppendYjsUpdate(it.ID, contentFrame(1), "0-old"); err != nil {
t.Fatal(err)
}
if res := f.r.process(it.ID); res != mrSchemaVersion || fake.calls.Load() != 0 {
t.Fatalf("process = %q calls %d, want schema_version, 0", res, fake.calls.Load())
}
if res := f.r.process(it.ID); res != mrBudgetExceeded {
t.Fatalf("second pass = %q, want budget_exhausted until the op-log changes", res)
}
}
// ---------------------------------------------------------------- disabled
// With no Materializer (PAD_MATERIALIZE=off leaves it nil) nothing is
// installed: a room going idle over a pending item leaves it pending.
func TestMaterializeRecoveryDisabledInstallsNothing(t *testing.T) {
f := newRecoveryFixture(t, nil, 30*time.Millisecond, materializeRecoveryConfig{})
ts := httptest.NewServer(f.srv)
t.Cleanup(ts.Close)
f.srv.SetMaterializer(nil)
f.srv.StartMaterializeRecovery()
if f.srv.materializeRecovery != nil {
t.Fatal("a recovery worker exists with no materializer")
}
it := f.item(t, "Pending", "stale")
f.appendRows(t, it.ID, contentFrame(1))
dialAndClose(t, ts.URL, it.ID)
deadline := time.Now().Add(2 * time.Second)
for f.rm.RoomCount() > 0 && time.Now().Before(deadline) {
time.Sleep(10 * time.Millisecond)
}
time.Sleep(100 * time.Millisecond)
got, _ := f.srv.store.GetItem(it.ID)
if got.Content != "stale" || got.ContentState != models.ContentStatePendingFlush {
t.Fatalf("disabled: content %q state %q, want untouched and pending", got.Content, got.ContentState)
}
}
// ---------------------------------------------------------------- end to end
type corpusCaseE2E struct {
Name string `json:"name"`
Rows []string `json:"rows"`
LinkIndex []materialize.LinkEntry `json:"link_index"`
WorkspaceSlug string `json:"workspace_slug"`
Expected string `json:"expected"`
}
func decodeRows(t *testing.T, rows []string) [][]byte {
t.Helper()
out := make([][]byte, len(rows))
for i, s := range rows {
b, err := base64.StdEncoding.DecodeString(s)
if err != nil {
t.Fatal(err)
}
out[i] = b
}
return out
}
// TestMaterializeRecoveryEndToEnd: real y-protocols op-logs from the U2
// corpus (recorded from two live Editor.svelte mounts) are inserted as items'
// op-logs over a stale items.content; the REAL sweep loop, worker and bundle
// recover them. The stored body must be the live editor's flush output, the
// item must read clean on GET, and a writer still holding the pre-recovery
// seq must be refused rather than overwrite the recovered text.
//
// Only corpus cases recorded with an EMPTY link index are used: the server
// builds its own index from the workspace, and none of these documents link
// to an item that exists here, so the flush pipeline's link conversion
// matches nothing, exactly as with the empty index the expected output was
// recorded with.
func TestMaterializeRecoveryEndToEnd(t *testing.T) {
raw, err := os.ReadFile("../materialize/testdata/corpus.json")
if err != nil {
t.Fatal(err)
}
var corpus []corpusCaseE2E
if err := json.Unmarshal(raw, &corpus); err != nil {
t.Fatal(err)
}
f := newRecoveryFixture(t, recoveryRunner(t), time.Minute, materializeRecoveryConfig{
sweepInterval: 20 * time.Millisecond,
now: func() time.Time { return time.Now().Add(time.Hour) },
})
type seeded struct {
c corpusCaseE2E
item *models.Item
}
var cases []seeded
for _, c := range corpus {
// A cross-workspace link (/-/r/...) is rewritten to [[ws::REF]]
// whenever the index is non-empty, whatever it holds, and a real
// workspace's index always holds at least the item itself; the
// corpus's empty-index cases were recorded with the conversion
// skipped. Cases containing one are left out rather than papered over.
if len(c.LinkIndex) != 0 || strings.Contains(c.Expected, "](/-/r/") {
continue
}
if c.WorkspaceSlug != f.ws.Slug {
t.Fatalf("%s: corpus workspace %q, fixture %q", c.Name, c.WorkspaceSlug, f.ws.Slug)
}
it := f.item(t, "E2E "+c.Name, "stale body before recovery")
f.appendRows(t, it.ID, decodeRows(t, c.Rows)...)
cases = append(cases, seeded{c, it})
}
if len(cases) < 4 {
t.Fatalf("only %d empty-index corpus cases; the instrument needs more", len(cases))
}
// Pre-recovery state, and the refusal a token-guarded write gets while
// the rows are pending (BUG-3133), for contrast with the one after.
first := cases[0].item
pre, _ := f.srv.store.GetItem(first.ID)
if pre.ContentState != models.ContentStatePendingFlush {
t.Fatalf("precondition: content_state %q", pre.ContentState)
}
preSeq := pre.Seq
f.srv.StartMaterializeRecovery()
for _, c := range cases {
if res := f.waitProcessed(t, c.item.ID, 60*time.Second); res != mrApplied {
t.Fatalf("%s: result %q, want applied", c.c.Name, res)
}
}
for _, c := range cases {
rr := doRequest(f.srv, http.MethodGet, "/api/v1/workspaces/ws/items/"+c.item.Slug, nil)
if rr.Code != http.StatusOK {
t.Fatalf("%s: GET %d %s", c.c.Name, rr.Code, rr.Body.String())
}
var got models.Item
parseJSON(t, rr, &got)
if got.Content != c.c.Expected {
t.Errorf("%s: recovered body differs from the live editor's flush\n--- got ---\n%s\n--- want ---\n%s", c.c.Name, got.Content, c.c.Expected)
}
if got.ContentState == models.ContentStatePendingFlush {
t.Errorf("%s: content_state still %q after recovery", c.c.Name, got.ContentState)
}
}
body := map[string]any{"content": "a caller's body from before the recovery", "expected_seq": preSeq}
rr := doRequest(f.srv, http.MethodPatch, "/api/v1/workspaces/ws/items/"+first.Slug, body)
if rr.Code != http.StatusConflict || !strings.Contains(rr.Body.String(), "update_conflict") {
t.Fatalf("stale expected_seq PATCH = %d %s, want 409 update_conflict", rr.Code, rr.Body.String())
}
after, _ := f.srv.store.GetItem(first.ID)
if after.Content != cases[0].c.Expected {
t.Fatalf("the refused PATCH changed the body")
}
t.Logf("%d corpus op-logs recovered end to end", len(cases))
}
// The version history shows the recovery as the system's, on the wire.
func TestMaterializeRecoveryVersionOnTheWire(t *testing.T) {
f := newRecoveryFixture(t, &fakeMaterializer{}, time.Minute, materializeRecoveryConfig{})
it := f.item(t, "Versioned", "stale")
f.appendRows(t, it.ID, contentFrame(1))
if res := f.r.process(it.ID); res != mrApplied {
t.Fatalf("process = %q", res)
}
rr := doRequest(f.srv, http.MethodGet, "/api/v1/workspaces/ws/items/"+it.Slug+"/versions", nil)
if rr.Code != http.StatusOK {
t.Fatalf("versions: %d %s", rr.Code, rr.Body.String())
}
var vs []models.Version
parseJSON(t, rr, &vs)
if len(vs) == 0 || vs[0].CreatedBy != models.VersionCreatedBySystem || vs[0].Source != models.VersionSourceRecovery ||
vs[0].ChangeSummary != models.RecoveryChangeSummary {
t.Fatalf("newest version = %+v", vs)
}
}
// A client cannot claim the recovery label for its own write.
func TestClientCannotClaimRecoveryVersionSource(t *testing.T) {
f := newRecoveryFixture(t, nil, time.Minute, materializeRecoveryConfig{})
it := f.item(t, "Spoof", "before")
rr := doRequest(f.srv, http.MethodPatch, "/api/v1/workspaces/ws/items/"+it.Slug,
map[string]any{"content": "after", "version_source": models.VersionSourceRecovery})
if rr.Code != http.StatusOK {
t.Fatalf("PATCH %d %s", rr.Code, rr.Body.String())
}
vs, err := f.srv.store.ListItemVersions(it.ID)
if err != nil || len(vs) == 0 {
t.Fatalf("versions: %v %v", vs, err)
}
if vs[0].Source == models.VersionSourceRecovery {
t.Fatal("a client write was labelled as a recovery")
}
}
// ErrNoMemoryCap (U3: no cap could be established, so the supervisor refused
// the job and warned once itself) counts against the budget like any failure,
// but is not logged per item. The control is ErrChildDied, which is.
func TestMaterializeNoMemoryCapCountsSilently(t *testing.T) {
for _, tc := range []struct {
err error
wantWarn bool
}{
{fmt.Errorf("%w: no cap mechanism on plan9", materialize.ErrNoMemoryCap), false},
{fmt.Errorf("%w: exit status 2", materialize.ErrChildDied), true},
} {
var buf bytes.Buffer
clock := time.Now()
fake := &fakeMaterializer{fn: func(materialize.Job) (string, error) { return "", tc.err }}
f := newRecoveryFixture(t, fake, time.Minute, materializeRecoveryConfig{
maxFailures: 3, backoffBase: time.Minute,
now: func() time.Time { return clock },
logger: slog.New(slog.NewTextHandler(&buf, &slog.HandlerOptions{Level: slog.LevelInfo})),
})
it := f.item(t, "Uncapped", "stale")
f.appendRows(t, it.ID, contentFrame(1))
for i := 0; i < 3; i++ {
if res := f.r.process(it.ID); res != mrFailed {
t.Fatalf("%v: pass %d = %q, want failed", tc.err, i, res)
}
clock = clock.Add(time.Hour)
}
if res := f.r.process(it.ID); res != mrBudgetExceeded || fake.calls.Load() != 3 {
t.Fatalf("%v: after K failures %q with %d calls, want budget_exhausted, 3", tc.err, res, fake.calls.Load())
}
if got := strings.Contains(buf.String(), "level=WARN"); got != tc.wantWarn {
t.Fatalf("%v: per-item WARN logged = %v, want %v\n%s", tc.err, got, tc.wantWarn, buf.String())
}
}
}
// An item that uses up its budget leaves exactly ONE line naming it, its
// failure count and the last error kind; a budget reset (op-log growth) and a
// second exhaustion leave exactly one more.
func TestMaterializeExhaustionLogsOncePerExhaustion(t *testing.T) {
var buf bytes.Buffer
clock := time.Now()
fake := &fakeMaterializer{fn: func(materialize.Job) (string, error) {
return "", fmt.Errorf("%w: exit status 2", materialize.ErrChildDied)
}}
f := newRecoveryFixture(t, fake, time.Minute, materializeRecoveryConfig{
maxFailures: 3, backoffBase: time.Minute,
now: func() time.Time { return clock },
logger: slog.New(slog.NewTextHandler(&buf, &slog.HandlerOptions{Level: slog.LevelInfo})),
})
it := f.item(t, "Stuck", "stale")
f.appendRows(t, it.ID, contentFrame(1))
const msg = "exhausted its failure budget"
exhaustLines := func() []string {
var out []string
for _, l := range strings.Split(buf.String(), "\n") {
if strings.Contains(l, msg) {
out = append(out, l)
}
}
return out
}
drive := func(passes int) {
for i := 0; i < passes; i++ {
f.r.process(it.ID)
clock = clock.Add(time.Hour)
}
}
drive(3 + 5) // K failures, then five refused passes
lines := exhaustLines()
if len(lines) != 1 {
t.Fatalf("exhaustion lines after one exhaustion = %d, want 1\n%s", len(lines), buf.String())
}
for _, want := range []string{"level=WARN", "item_id=" + it.ID, "failures=3", "last_error_kind=worker_died"} {
if !strings.Contains(lines[0], want) {
t.Fatalf("exhaustion line lacks %q: %s", want, lines[0])
}
}
f.appendRows(t, it.ID, contentFrame(2)) // op-log grows: the budget resets
drive(3 + 5)
if n := len(exhaustLines()); n != 2 {
t.Fatalf("exhaustion lines after a reset and a second exhaustion = %d, want 2\n%s", n, buf.String())
}
if fake.calls.Load() != 6 {
t.Fatalf("materializer calls %d, want 6", fake.calls.Load())
}
// The schema-version skip is an exhaustion too, logged once.
buf.Reset()
old := f.item(t, "Old schema", "stale")
if _, err := f.srv.store.AppendYjsUpdate(old.ID, contentFrame(3), "0-old"); err != nil {
t.Fatal(err)
}
f.r.process(old.ID)
f.r.process(old.ID)
if l := exhaustLines(); len(l) != 1 || !strings.Contains(l[0], "last_error_kind=schema_version") {
t.Fatalf("schema-version exhaustion lines = %v", l)
}
}
// Items skipped for set-aside rows are counted, reported on the sweep's one
// summary line, and never logged per item.
func TestMaterializeSetAsideSkipsAreCountedNotLogged(t *testing.T) {
var buf bytes.Buffer
fake := &fakeMaterializer{}
f := newRecoveryFixture(t, fake, time.Minute, materializeRecoveryConfig{
logger: slog.New(slog.NewTextHandler(&buf, &slog.HandlerOptions{Level: slog.LevelDebug})),
})
it := f.item(t, "Aside", "stale")
f.appendRows(t, it.ID, contentFrame(1))
if _, _, err := f.srv.store.SetAsideAndClearOpLog(it.ID); err != nil {
t.Fatal(err)
}
f.appendRows(t, it.ID, contentFrame(2)) // pending again, over set-aside rows
f.r.onRoomIdle(it.ID) // the room trigger does queue it: its op-log is pending
if q := f.queueSnapshot(); len(q) != 1 {
t.Fatalf("queue = %v", q)
}
for i := 0; i < 3; i++ {
if res := f.r.process(it.ID); res != mrSetAside {
t.Fatalf("process = %q, want set_aside", res)
}
}
if n := f.r.setAsideSkipped.Load(); n != 3 {
t.Fatalf("set-aside skip count = %d, want 3", n)
}
if fake.calls.Load() != 0 {
t.Fatal("a set-aside item reached the materializer")
}
if strings.Contains(buf.String(), it.ID) {
t.Fatalf("set-aside skips logged per item:\n%s", buf.String())
}
f.r.sweep()
f.r.sweep() // no news: no second line
var sweepLines []string
for _, l := range strings.Split(buf.String(), "\n") {
if strings.Contains(l, "op-log recovery sweep") {
sweepLines = append(sweepLines, l)
}
}
if len(sweepLines) != 1 || !strings.Contains(sweepLines[0], "set_aside_skipped_total=3") {
t.Fatalf("sweep summary lines = %v, want one carrying set_aside_skipped_total=3", sweepLines)
}
}
// STARVATION (codex r1): the sweep reads a bounded page, and the room check
// and the failure budget filter it afterwards. With materializeSweepLimit+1
// dormant pending items of which the first materializeSweepLimit fail every
// time, the last one must still be recovered, within
// ceil(N / limit) + 1 = 3 sweeps, at the REAL page size.
func TestMaterializeSweepReachesEveryCandidate(t *testing.T) {
clock := time.Now().Add(time.Hour) // rows written now are dormant
var targetFrame, healedFrame []byte
fake := &fakeMaterializer{fn: func(job materialize.Job) (string, error) {
if len(job.Rows) == 1 && (bytes.Equal(job.Rows[0], targetFrame) || bytes.Equal(job.Rows[0], healedFrame)) {
return "recovered body", nil
}
return "", fmt.Errorf("%w: exit status 2", materialize.ErrChildDied)
}}
f := newRecoveryFixture(t, fake, time.Minute, materializeRecoveryConfig{
now: func() time.Time { return clock },
logger: slog.New(slog.NewTextHandler(io.Discard, nil)),
})
n := materializeSweepLimit + 1
frameOf := map[string][]byte{}
for i := 0; i < n; i++ {
it := f.item(t, fmt.Sprintf("Backlog %03d", i), "stale")
fr := []byte{0x00, 0x02, 0x05, 0x01, byte(i >> 8), byte(i), 0x7F, 0x00}
f.appendRows(t, it.ID, fr)
frameOf[it.ID] = fr
}
all, err := f.srv.store.ListMaterializeCandidates(clock.Add(-materializeDormancy), store.MaterializeCursor{}, 10*n)
if err != nil || len(all) != n {
t.Fatalf("candidates %d (%v), want %d", len(all), err, n)
}
target := all[n-1].ItemID // the one every page read from the start leaves out
targetFrame = frameOf[target]
// runUntil sweeps up to 3 times, draining the queue after each, and
// reports the sweep on which want was recovered (0: never).
runUntil := func(want string) int {
for tick := 1; tick <= 3; tick++ {
f.r.sweep()
hit := false
for {
id, ok := f.r.dequeue()
if !ok {
break
}
if res := f.r.process(id); id == want && res == mrApplied {
hit = true
}
}
clock = clock.Add(time.Hour) // every backoff expires between ticks
if hit {
return tick
}
}
return 0
}
tick := runUntil(target)
if tick == 0 {
t.Fatalf("the %dth candidate was never recovered in 3 sweeps", n)
}
t.Logf("the %dth candidate was recovered on sweep %d", n, tick)
// And the walk WRAPS: the first candidate starts succeeding (its op-log,
// and so its place in the order, unchanged; two failures, so still inside
// its budget) and is reached again after the cursor has passed it.
first := all[0].ItemID
healedFrame = frameOf[first]
if tick := runUntil(first); tick == 0 {
t.Fatal("after the cursor passed it, the first candidate was never offered again")
} else {
t.Logf("the first candidate was recovered on sweep %d after the wrap", tick)
}
}
+12
View File
@@ -264,6 +264,12 @@ type Server struct {
// StartOpLogGC; Stop() signals the loop via stopOpLogGC.
opLogGC opLogGCConfig
// materializeRecovery is the op-log recovery worker (TASK-2198 U4): it
// rebuilds items.content from an item's collaborative op-log after the tab
// holding those edits went away without flushing them. Nil until
// SetMaterializer; started by StartMaterializeRecovery; Stop() stops it.
materializeRecovery *materializeRecovery
// tokenReaper holds the periodic-sweep config + lifecycle for the
// short-lived-credential reaper (PLAN-1933 DR-5 / TASK-1936).
// Mirrors orphanGC/opLogGC. Configured via SetTokenReaperConfig +
@@ -553,6 +559,10 @@ func (s *Server) Stop() {
// Yjs op-log prune sweeper (TASK-1309). Same lifecycle pattern;
// signals BEFORE Wait() so the goroutine sees the close and exits.
s.stopOpLogGC()
// Op-log recovery worker (TASK-2198 U4). Signalled BEFORE collab.Close and
// bg.Wait: cancelling its context kills a job in flight (the supervisor
// kills the worker process), so the goroutine drains promptly.
s.stopMaterializeRecovery()
// Short-lived-credential reaper (PLAN-1933 DR-5 / TASK-1936). Same
// lifecycle pattern; signal BEFORE Wait() so the goroutine exits.
s.stopTokenReaper()
@@ -589,6 +599,8 @@ func (s *Server) Stop() {
s.collab.Close()
}
s.bg.Wait()
// The worker PROCESS goes after bg.Wait, when nothing can hand it a job.
s.closeMaterializer()
// Watch/nudge bus (BUG-2651). Closed AFTER bg.Wait() so a background
// producer cannot publish into a bus that is already tearing down; both
// implementations are safe if one does anyway (MemoryBus finds no
+90
View File
@@ -0,0 +1,90 @@
[
{
"name": "ref-link-other-text",
"markdown": "See [shown text](/alice/ws/secrets/SECR-1) here.",
"rows": [
"AAKxAQEHlhEABwEHZGVmYXVsdAMJcGFyYWdyYXBoBwCWEQAGBACWEQEEU2VlIIaWEQUEbGlua1R7ImhyZWYiOiIvYWxpY2Uvd3Mvc2VjcmV0cy9TRUNSLTEiLCJ0YXJnZXQiOm51bGwsInJlbCI6bnVsbCwiY2xhc3MiOiIiLCJ0aXRsZSI6bnVsbH2ElhEGCnNob3duIHRleHSGlhEQBGxpbmsEbnVsbISWEREGIGhlcmUuAA=="
],
"raw": "See [shown text](/alice/ws/secrets/SECR-1) here."
},
{
"name": "ref-link-old-title",
"markdown": "See [Old Codename](/alice/ws/secrets/SECR-1) here.",
"rows": [
"AAKzAQEHlhEABwEHZGVmYXVsdAMJcGFyYWdyYXBoBwCWEQAGBACWEQEEU2VlIIaWEQUEbGlua1R7ImhyZWYiOiIvYWxpY2Uvd3Mvc2VjcmV0cy9TRUNSLTEiLCJ0YXJnZXQiOm51bGwsInJlbCI6bnVsbCwiY2xhc3MiOiIiLCJ0aXRsZSI6bnVsbH2ElhEGDE9sZCBDb2RlbmFtZYaWERIEbGluawRudWxshJYREwYgaGVyZS4A"
],
"raw": "See [Old Codename](/alice/ws/secrets/SECR-1) here."
},
{
"name": "slug-link-slug-text",
"markdown": "See [the plan](/alice/ws/secrets/secret-launch-codename) here.",
"rows": [
"AAK/AQEHlhEABwEHZGVmYXVsdAMJcGFyYWdyYXBoBwCWEQAGBACWEQEEU2VlIIaWEQUEbGlua2R7ImhyZWYiOiIvYWxpY2Uvd3Mvc2VjcmV0cy9zZWNyZXQtbGF1bmNoLWNvZGVuYW1lIiwidGFyZ2V0IjpudWxsLCJyZWwiOm51bGwsImNsYXNzIjoiIiwidGl0bGUiOm51bGx9hJYRBgh0aGUgcGxhboaWEQ4EbGluawRudWxshJYRDwYgaGVyZS4A"
],
"raw": "See [the plan](/alice/ws/secrets/secret-launch-codename) here."
},
{
"name": "text-equals-current-title",
"markdown": "See [Secret Launch Codename](/alice/ws/secrets/SECR-1) here.",
"rows": [
"AAK9AQEHlhEABwEHZGVmYXVsdAMJcGFyYWdyYXBoBwCWEQAGBACWEQEEU2VlIIaWEQUEbGlua1R7ImhyZWYiOiIvYWxpY2Uvd3Mvc2VjcmV0cy9TRUNSLTEiLCJ0YXJnZXQiOm51bGwsInJlbCI6bnVsbCwiY2xhc3MiOiIiLCJ0aXRsZSI6bnVsbH2ElhEGFlNlY3JldCBMYXVuY2ggQ29kZW5hbWWGlhEcBGxpbmsEbnVsbISWER0GIGhlcmUuAA=="
],
"raw": "See [Secret Launch Codename](/alice/ws/secrets/SECR-1) here."
},
{
"name": "broken-link",
"markdown": "See [gone text](broken) here.",
"rows": [
"AAKeAQEHlhEABwEHZGVmYXVsdAMJcGFyYWdyYXBoBwCWEQAGBACWEQEEU2VlIIaWEQUEbGlua0J7ImhyZWYiOiJicm9rZW4iLCJ0YXJnZXQiOm51bGwsInJlbCI6bnVsbCwiY2xhc3MiOiIiLCJ0aXRsZSI6bnVsbH2ElhEGCWdvbmUgdGV4dIaWEQ8EbGluawRudWxshJYREAYgaGVyZS4A"
],
"raw": "See [[gone text]] here."
},
{
"name": "cross-workspace",
"markdown": "See [xw text](/-/r/other/SECR-1) here.",
"rows": [
"AAKnAQEHlhEABwEHZGVmYXVsdAMJcGFyYWdyYXBoBwCWEQAGBACWEQEEU2VlIIaWEQUEbGlua017ImhyZWYiOiIvLS9yL290aGVyL1NFQ1ItMSIsInRhcmdldCI6bnVsbCwicmVsIjpudWxsLCJjbGFzcyI6IiIsInRpdGxlIjpudWxsfYSWEQYHeHcgdGV4dIaWEQ0EbGluawRudWxshJYRDgYgaGVyZS4A"
],
"raw": "See [xw text](/-/r/other/SECR-1) here."
},
{
"name": "typed-wiki-literal",
"markdown": "Typed \\[\\[SECR-1\\]\\] literal.",
"rows": [
"AAJAAQOWEQAHAQdkZWZhdWx0AwlwYXJhZ3JhcGgHAJYRAAYEAJYRARlUeXBlZCBbW1NFQ1ItMV1dIGxpdGVyYWwuAA=="
],
"raw": "Typed [[SECR-1]] literal."
},
{
"name": "public-ref-link",
"markdown": "See [the public plan](/alice/ws/public/PUB-2) here.",
"rows": [
"AAK0AQEHlhEABwEHZGVmYXVsdAMJcGFyYWdyYXBoBwCWEQAGBACWEQEEU2VlIIaWEQUEbGlua1J7ImhyZWYiOiIvYWxpY2Uvd3MvcHVibGljL1BVQi0yIiwidGFyZ2V0IjpudWxsLCJyZWwiOm51bGwsImNsYXNzIjoiIiwidGl0bGUiOm51bGx9hJYRBg90aGUgcHVibGljIHBsYW6GlhEVBGxpbmsEbnVsbISWERYGIGhlcmUuAA=="
],
"raw": "See [the public plan](/alice/ws/public/PUB-2) here."
},
{
"name": "marker-follows-title",
"markdown": "See [Secret Launch Codename](/alice/ws/secrets/SECR-1 \"pad-follows-title:Secret Launch Codename\") here.",
"rows": [
"AALjAQEHlhEABwEHZGVmYXVsdAMJcGFyYWdyYXBoBwCWEQAGBACWEQEEU2VlIIaWEQUEbGlua3p7ImhyZWYiOiIvYWxpY2Uvd3Mvc2VjcmV0cy9TRUNSLTEiLCJ0YXJnZXQiOm51bGwsInJlbCI6bnVsbCwiY2xhc3MiOiIiLCJ0aXRsZSI6InBhZC1mb2xsb3dzLXRpdGxlOlNlY3JldCBMYXVuY2ggQ29kZW5hbWUifYSWEQYWU2VjcmV0IExhdW5jaCBDb2RlbmFtZYaWERwEbGluawRudWxshJYRHQYgaGVyZS4A"
],
"raw": "See [Secret Launch Codename](/alice/ws/secrets/SECR-1) here."
},
{
"name": "marker-renamed",
"markdown": "See [Old Codename](/alice/ws/secrets/SECR-1 \"pad-follows-title:Old Codename\") here.",
"rows": [
"AALPAQEHlhEABwEHZGVmYXVsdAMJcGFyYWdyYXBoBwCWEQAGBACWEQEEU2VlIIaWEQUEbGlua3B7ImhyZWYiOiIvYWxpY2Uvd3Mvc2VjcmV0cy9TRUNSLTEiLCJ0YXJnZXQiOm51bGwsInJlbCI6bnVsbCwiY2xhc3MiOiIiLCJ0aXRsZSI6InBhZC1mb2xsb3dzLXRpdGxlOk9sZCBDb2RlbmFtZSJ9hJYRBgxPbGQgQ29kZW5hbWWGlhESBGxpbmsEbnVsbISWERMGIGhlcmUuAA=="
],
"raw": "See [Old Codename](/alice/ws/secrets/SECR-1) here."
},
{
"name": "marker-external-href",
"markdown": "See [Old Codename](https://example.com/x \"pad-follows-title:Old Codename\") here.",
"rows": [
"AALMAQEHlhEABwEHZGVmYXVsdAMJcGFyYWdyYXBoBwCWEQAGBACWEQEEU2VlIIaWEQUEbGlua217ImhyZWYiOiJodHRwczovL2V4YW1wbGUuY29tL3giLCJ0YXJnZXQiOm51bGwsInJlbCI6bnVsbCwiY2xhc3MiOiIiLCJ0aXRsZSI6InBhZC1mb2xsb3dzLXRpdGxlOk9sZCBDb2RlbmFtZSJ9hJYRBgxPbGQgQ29kZW5hbWWGlhESBGxpbmsEbnVsbISWERMGIGhlcmUuAA=="
],
"raw": "See [Old Codename](https://example.com/x) here."
}
]
+4 -2
View File
@@ -12,8 +12,10 @@ import (
// BUG-3000: a read that serves items.content must say so when that content is
// behind the item's live collaborative document.
//
// The window is UNBOUNDED, which is what makes the marker worth having rather than
// a documentation note: nothing server-side moves applier content into the row.
// The window was UNBOUNDED when this was written, which is what made the marker
// worth having rather than a documentation note. Since TASK-2198 U4 the server's
// op-log recovery closes it after the item's room closes, when it is on and the
// job succeeds; the op-log GC below still never does.
// PruneSweep's candidate query requires MAX(op-log.id) <= content_flushed_op_log_id,
// so an item in exactly this state is EXCLUDED from the sweep — its op-log rows are
// retained (the content is durable) and the row stays behind until a tab next opens
+12 -1
View File
@@ -2806,6 +2806,12 @@ func (s *Store) updateItemWithParentLinkOnce(
if source == "" {
source = "web"
}
// TASK-2198 U4: an op-log recovery is the system's write, not a
// guessed author's (op-log rows carry no user id).
if input.Recovered {
createdBy = models.VersionCreatedBySystem
source = models.VersionSourceRecovery
}
// ForceVersion (e.g. a version restore) and a title change both bypass
// the per-(actor, source) throttle so a bracketing snapshot is always
@@ -2916,7 +2922,12 @@ func (s *Store) updateItemWithParentLinkOnce(
// cannot delete them. When the cursor is missing (older
// clients, malformed bodies) we behave as before — no
// advancement.
if input.VersionSource != "collab-snapshot" {
//
// An op-log RECOVERY (input.Recovered, TASK-2198 U4) takes the
// cursor-gated arm too: its markdown was rebuilt from the rows up to
// its cursor and no further, so stamping MAX would claim a row that
// committed after the job was built.
if input.VersionSource != "collab-snapshot" && !input.Recovered {
sets = append(sets, "content_flushed_op_log_id = (SELECT COALESCE(MAX(id), 0) FROM item_yjs_updates WHERE item_id = ?)")
args = append(args, id)
} else if input.OpLogCursor != nil {
+348
View File
@@ -0,0 +1,348 @@
package store
import (
"crypto/sha256"
"database/sql"
"encoding/hex"
"errors"
"fmt"
"time"
"github.com/PerpetualSoftware/pad/internal/models"
)
// Op-log recovery (TASK-2198 U4): the store half.
//
// When a tab dies uncleanly its last typing exists only in item_yjs_updates:
// content-bearing rows above items.content_flushed_op_log_id, which
// content_state reports as applied_pending_flush. The server materializes
// those rows into markdown in a worker process (internal/materialize) and
// writes the result here. This file holds the reads that build the job and the
// write that lands it; internal/server/materialize_recovery.go holds the
// triggers and the worker, and is the only caller.
//
// THE WRITE IS A FLUSH, NOT AN EDIT. It carries the same proof a tab's
// collab-snapshot flush carries — the markdown was rebuilt from every op-log
// row up to a cursor, and the cursor is still MAX(op-log id) — so it advances
// the watermark to that cursor and to nothing beyond it. It never prunes the
// op-log: the next tab to open the item replays the op-log, which already holds
// everything this write stored.
// MaterializeInput is what a recovery job is built from, read in ONE statement
// so the rows and the cursor describe the same op-log.
type MaterializeInput struct {
ItemID string
WorkspaceID string
// Rows are the item's CONTENT-BEARING op-log rows, oldest first, over the
// WHOLE op-log, not only those above the watermark: replay rebuilds the
// document from nothing, as a tab's replay (since=0) does. Rows marked
// non-content-bearing provably cannot change the document (BUG-3124), so
// leaving them out gives the same document (U2's parity set).
Rows [][]byte
// Cursor is MAX(id) over ALL of the item's op-log rows, content-bearing or
// not, from the same statement as Rows.
Cursor int64
// SchemaVersions are the distinct schema_version stamps over all rows.
SchemaVersions []string
// Pending is how many content-bearing rows sit above the watermark.
Pending int
// SetAside is how many rows a schema rebuild set aside (BUG-3244).
SetAside int
}
// LoadMaterializeInput reads the item's op-log for a recovery job. It returns
// nil (no error) when the item is missing, soft-deleted, or has no op-log.
func (s *Store) LoadMaterializeInput(itemID string) (*MaterializeInput, error) {
if itemID == "" {
return nil, errors.New("LoadMaterializeInput: itemID is required")
}
var workspaceID string
if err := s.db.QueryRow(s.q(`SELECT workspace_id FROM items WHERE id = ? AND deleted_at IS NULL`), itemID).Scan(&workspaceID); err != nil {
if errors.Is(err, sql.ErrNoRows) {
return nil, nil
}
return nil, fmt.Errorf("materialize input (item): %w", err)
}
// ONE statement: the rows, their stamps and the watermark come from the
// same snapshot on both dialects, so Cursor can never name a row whose
// bytes are not in Rows. Non-content-bearing rows contribute their id and
// stamp but not their bytes.
rows, err := s.db.Query(s.q(`
SELECT u.id, u.content_bearing, u.schema_version,
CASE WHEN u.content_bearing = TRUE THEN u.update_data END,
COALESCE(i.content_flushed_op_log_id, 0)
FROM item_yjs_updates u
JOIN items i ON i.id = u.item_id
WHERE u.item_id = ?
ORDER BY u.id ASC`), itemID)
if err != nil {
return nil, fmt.Errorf("materialize input (op-log): %w", err)
}
defer rows.Close()
in := &MaterializeInput{ItemID: itemID, WorkspaceID: workspaceID}
seen := map[string]bool{}
for rows.Next() {
var (
id int64
bearing bool
version string
data []byte
watermark int64
)
if err := rows.Scan(&id, &bearing, &version, &data, &watermark); err != nil {
return nil, fmt.Errorf("materialize input (scan): %w", err)
}
in.Cursor = id
if !seen[version] {
seen[version] = true
in.SchemaVersions = append(in.SchemaVersions, version)
}
if bearing {
in.Rows = append(in.Rows, data)
if id > watermark {
in.Pending++
}
}
}
if err := rows.Err(); err != nil {
return nil, fmt.Errorf("materialize input (rows): %w", err)
}
if in.Cursor == 0 {
return nil, nil
}
if err := s.db.QueryRow(s.q(`SELECT COUNT(*) FROM item_yjs_updates_set_aside WHERE item_id = ?`), itemID).Scan(&in.SetAside); err != nil {
return nil, fmt.Errorf("materialize input (set-aside): %w", err)
}
return in, nil
}
// ItemHasPendingContent reports whether the item's op-log holds content-bearing
// rows above its flush watermark: content_state's applied_pending_flush
// predicate, for a live item.
func (s *Store) ItemHasPendingContent(itemID string) (bool, error) {
var n int
err := s.db.QueryRow(s.q(`
SELECT COUNT(*) FROM items i
WHERE i.id = ? AND i.deleted_at IS NULL AND `+pendingFlushExistsSQLFor("i")), itemID).Scan(&n)
if err != nil {
return false, fmt.Errorf("item pending content: %w", err)
}
return n > 0, nil
}
// MaterializeCandidate is one row of the recovery sweep's backstop set, with
// the keyset position (LastAt, ItemID) it sorts at.
type MaterializeCandidate struct {
ItemID string
// LastAt is the item's newest op-log created_at, exactly as stored.
LastAt string
}
// MaterializeCursor is a keyset position in the sweep's order. The zero value
// is the start.
type MaterializeCursor struct {
LastAt string
ItemID string
}
// ListMaterializeCandidates returns up to limit live items whose op-log holds
// content-bearing rows above the flush watermark, whose NEWEST op-log row is
// older than before, and which hold no set-aside rows: the recovery sweep's
// backstop set (the shape of ListDormantOpLogItemsBefore, with the flush
// predicate inverted). The caller still checks for an open room.
//
// KEYSET-PAGED, strictly after `after`, in (newest op-log created_at, item id)
// order. The caller filters what comes back (open rooms, and the worker's
// per-item failure budget), so a page read from the start every time would
// hand back the same items forever once `limit` of them sit filtered, and
// every candidate behind them would starve. Walking the set page by page and
// wrapping at the end reaches every candidate however many ineligible ones
// sort ahead of it.
func (s *Store) ListMaterializeCandidates(before time.Time, after MaterializeCursor, limit int) ([]MaterializeCandidate, error) {
if limit <= 0 {
limit = 100
}
cutoff := before.UTC().Format(time.RFC3339)
rows, err := s.db.Query(s.q(`
SELECT u.item_id, MAX(u.created_at)
FROM item_yjs_updates u
JOIN items i ON i.id = u.item_id
WHERE i.deleted_at IS NULL
AND NOT EXISTS (SELECT 1 FROM item_yjs_updates_set_aside sa WHERE sa.item_id = u.item_id)
GROUP BY u.item_id, i.content_flushed_op_log_id
HAVING MAX(u.created_at) < ?
AND SUM(CASE WHEN u.content_bearing = TRUE AND u.id > COALESCE(i.content_flushed_op_log_id, 0) THEN 1 ELSE 0 END) > 0
AND (MAX(u.created_at) > ? OR (MAX(u.created_at) = ? AND u.item_id > ?))
ORDER BY MAX(u.created_at) ASC, u.item_id ASC
LIMIT ?`), cutoff, after.LastAt, after.LastAt, after.ItemID, limit)
if err != nil {
return nil, fmt.Errorf("list materialize candidates: %w", err)
}
defer rows.Close()
var out []MaterializeCandidate
for rows.Next() {
var c MaterializeCandidate
if err := rows.Scan(&c.ItemID, &c.LastAt); err != nil {
return nil, fmt.Errorf("scan materialize candidate: %w", err)
}
out = append(out, c)
}
return out, rows.Err()
}
// MaterializeOutcome is what MaterializeFlush did.
type MaterializeOutcome string
const (
// MaterializeApplied: items.content now holds the markdown; seq and
// updated_at were bumped, one version row was written (when the body
// changed) and the watermark is the cursor.
MaterializeApplied MaterializeOutcome = "applied"
// MaterializeStamped: the body already was the markdown, so only the
// watermark moved (StampContentWatermarkIfCaughtUp): no seq, no version.
MaterializeStamped MaterializeOutcome = "stamped"
// MaterializeCursorMoved: the op-log changed after the job was built
// (MAX(id) is not the cursor). Nothing was written.
MaterializeCursorMoved MaterializeOutcome = "cursor_moved"
// MaterializeNothingPending: no content-bearing row is above the watermark
// any more (a tab flushed, or a direct write replaced the op-log).
MaterializeNothingPending MaterializeOutcome = "nothing_pending"
// MaterializeSetAside: the item holds set-aside rows (BUG-3244). They are
// left untouched and the item is not written.
MaterializeSetAside MaterializeOutcome = "set_aside"
// MaterializeGone: the item is missing or soft-deleted.
MaterializeGone MaterializeOutcome = "gone"
)
type materializeAbort struct{ outcome MaterializeOutcome }
func (e *materializeAbort) Error() string { return "materialize flush: " + string(e.outcome) }
// materializeFlushPrecheckHook, when set by a test, runs inside the write's
// transaction immediately before the checks; materializeFlushAfterCheckHook
// runs after they have passed, before the UPDATE: the window a row committed
// by another connection lands in on Postgres (READ COMMITTED).
var (
materializeFlushPrecheckHook func(itemID string)
materializeFlushAfterCheckHook func(itemID string)
)
// MaterializeFlush writes a recovered body to items.content (TASK-2198 U4).
//
// CALLER CONTRACT: hold the collab per-item setup lock with no room open for
// the item (collab.RoomManager.UnderItemLockIfNoRoom). Every op-log append
// goes through a room, and a room can only be created under that lock, so
// nothing can append while this runs. The checks below do not depend on that:
// they run inside the write's own transaction, under UpdateItem's locks, and
// the watermark moves through the cursor-gated SQL a collab-snapshot flush
// uses, so a row committing after the check is left above the watermark.
//
// It proceeds only when, in the write's transaction: MAX(op-log id) equals
// cursor, at least one content-bearing row is above the watermark, and the
// item holds no set-aside rows. Otherwise nothing is written and the outcome
// says which check failed; that is not an error. The version row is
// attributed to the system (models.ItemUpdate.Recovered). The op-log is never
// pruned.
func (s *Store) MaterializeFlush(itemID string, cursor int64, markdown string) (MaterializeOutcome, *models.Item, error) {
if itemID == "" || cursor < 1 {
return "", nil, errors.New("MaterializeFlush: itemID and a positive cursor are required")
}
var current string
if err := s.db.QueryRow(s.q(`SELECT content FROM items WHERE id = ? AND deleted_at IS NULL`), itemID).Scan(&current); err != nil {
if errors.Is(err, sql.ErrNoRows) {
return MaterializeGone, nil, nil
}
return "", nil, fmt.Errorf("materialize flush (read): %w", err)
}
check := func(q Queryer, tx *sql.Tx) (MaterializeOutcome, error) {
if materializeFlushPrecheckHook != nil {
materializeFlushPrecheckHook(itemID)
}
var maxID int64
if err := q.QueryRow(s.q(`SELECT COALESCE(MAX(id), 0) FROM item_yjs_updates WHERE item_id = ?`), itemID).Scan(&maxID); err != nil {
return "", fmt.Errorf("materialize flush (max): %w", err)
}
if maxID != cursor {
return MaterializeCursorMoved, nil
}
var setAside int
if err := q.QueryRow(s.q(`SELECT COUNT(*) FROM item_yjs_updates_set_aside WHERE item_id = ?`), itemID).Scan(&setAside); err != nil {
return "", fmt.Errorf("materialize flush (set-aside): %w", err)
}
if setAside > 0 {
return MaterializeSetAside, nil
}
var pending int
if tx != nil {
n, err := s.CountPendingContentRowsTx(tx, itemID)
if err != nil {
return "", err
}
pending = n
} else if err := q.QueryRow(s.q(`
SELECT COUNT(*) FROM item_yjs_updates u JOIN items i ON i.id = u.item_id
WHERE u.item_id = ? AND u.id > COALESCE(i.content_flushed_op_log_id, 0) AND u.content_bearing = TRUE`), itemID).Scan(&pending); err != nil {
return "", fmt.Errorf("materialize flush (pending): %w", err)
}
if pending == 0 {
return MaterializeNothingPending, nil
}
return "", nil
}
// The body already is the document: move the watermark only, as a tab's
// watermark stamp would (BUG-3124 unit B). A version row and a seq bump
// for a write that changes nothing would conflict every token-holder for
// no reason.
if current == markdown {
if outcome, err := check(s.db, nil); err != nil || outcome != "" {
return outcome, nil, err
}
sum := sha256.Sum256([]byte(markdown))
moved, err := s.StampContentWatermarkIfCaughtUp(itemID, cursor, hex.EncodeToString(sum[:]))
if err != nil {
return "", nil, err
}
if !moved {
return MaterializeCursorMoved, nil, nil
}
return MaterializeStamped, nil, nil
}
c := cursor
input := models.ItemUpdate{
Content: &markdown,
ChangeSummary: models.RecoveryChangeSummary,
ForceVersion: true,
Recovered: true,
OpLogCursor: &c,
}
updated, err := s.UpdateItemWithPreCheck(itemID, input, func(tx *sql.Tx, _ *models.Item) error {
outcome, err := check(tx, tx)
if err != nil {
return err
}
if outcome != "" {
return &materializeAbort{outcome: outcome}
}
if materializeFlushAfterCheckHook != nil {
materializeFlushAfterCheckHook(itemID)
}
return nil
})
if err != nil {
var abort *materializeAbort
if errors.As(err, &abort) {
return abort.outcome, nil, nil
}
return "", nil, err
}
if updated == nil {
return MaterializeGone, nil, nil
}
return MaterializeApplied, updated, nil
}
@@ -0,0 +1,64 @@
package store
import (
"testing"
"github.com/PerpetualSoftware/pad/internal/models"
)
// TASK-2198 U4: the recovery write's watermark is its CURSOR, never MAX at
// UPDATE time. On Postgres (READ COMMITTED) a row another connection commits
// after the write's checks is visible to the UPDATE's own subqueries, so a
// write that stamped MAX would claim a row its markdown never saw, and
// content_state would read clean over an unrecovered edit.
//
// Postgres only: on SQLite the hook's append would wait on the write
// transaction's lock, so the window does not exist there.
func TestMaterializeFlushWatermarkIsTheCursorNotMax(t *testing.T) {
s := testStore(t)
if s.dialect.Driver() != DriverPostgres {
t.Skip("the READ COMMITTED window this pins exists on Postgres only (make test-pg)")
}
ws := createTestWorkspace(t, s, "Recovery race")
coll, err := s.CreateCollection(ws.ID, models.CollectionCreate{Name: "Notes", Schema: `{"fields":[]}`})
if err != nil {
t.Fatal(err)
}
it, err := s.CreateItem(ws.ID, coll.ID, models.ItemCreate{Title: "Racy", Content: "stale", Fields: `{}`})
if err != nil {
t.Fatal(err)
}
cursor, err := s.AppendYjsUpdate(it.ID, []byte{0x00, 0x02, 0x05, 0x01, 0x01, 0x00, 0x7F, 0x00}, "1")
if err != nil {
t.Fatal(err)
}
var late int64
materializeFlushAfterCheckHook = func(itemID string) {
materializeFlushAfterCheckHook = nil
var aerr error
late, aerr = s.AppendYjsUpdate(itemID, []byte{0x00, 0x02, 0x05, 0x01, 0x02, 0x00, 0x7F, 0x00}, "1")
if aerr != nil {
t.Errorf("late append: %v", aerr)
}
}
t.Cleanup(func() { materializeFlushAfterCheckHook = nil })
outcome, _, err := s.MaterializeFlush(it.ID, cursor, "recovered")
if err != nil || outcome != MaterializeApplied {
t.Fatalf("MaterializeFlush = %q, %v", outcome, err)
}
if late <= cursor {
t.Fatalf("the late row %d did not land after the cursor %d", late, cursor)
}
wm, _, err := s.GetItemContentFlushedOpLogID(it.ID)
if err != nil {
t.Fatal(err)
}
if wm == late {
t.Fatalf("watermark %d covers the late row the recovered body never saw", wm)
}
got, _ := s.GetItem(it.ID)
if got.ContentState != models.ContentStatePendingFlush {
t.Fatalf("content_state %q, want still pending for the late row", got.ContentState)
}
}
+346
View File
@@ -0,0 +1,346 @@
package store_test
import (
"errors"
"fmt"
"testing"
"time"
"github.com/PerpetualSoftware/pad/internal/diff"
"github.com/PerpetualSoftware/pad/internal/models"
"github.com/PerpetualSoftware/pad/internal/store"
)
// TASK-2198 U4, store half, on both dialects: the op-log recovery write
// (MaterializeFlush) and the reads that build its job. Postgres legs skip
// unless PAD_TEST_POSTGRES_URL is set (make test-pg).
// recoveryFrame is a content-bearing y-protocols sync Update frame, distinct
// per n (the store never parses past the envelope; an identical frame would be
// classified as a re-send and not count).
func recoveryFrame(n byte) []byte {
return []byte{0x00, 0x02, 0x05, 0x01, n, 0x00, 0x7F, 0x00}
}
// syncStep1Frame is a SyncStep1 (an empty state vector): persisted, but never
// content-bearing (BUG-3124).
var syncStep1Frame = []byte{0x00, 0x00, 0x01, 0x00}
func appendFrames(t *testing.T, s *store.Store, itemID string, frames ...[]byte) int64 {
t.Helper()
var last int64
for _, f := range frames {
id, err := s.AppendYjsUpdate(itemID, f, "1")
if err != nil {
t.Fatalf("AppendYjsUpdate: %v", err)
}
last = id
}
return last
}
func versionCount(t *testing.T, s *store.Store, itemID string) int {
t.Helper()
vs, err := s.ListItemVersions(itemID)
if err != nil {
t.Fatal(err)
}
return len(vs)
}
func TestMaterializeFlushAppliesWhenCaughtUpAndPending(t *testing.T) {
eachBackend(t, func(t *testing.T, s *store.Store) {
_, _, item := seedStaleItem(t, s)
cursor := appendFrames(t, s, item.ID, recoveryFrame(1), recoveryFrame(2))
before, err := s.GetItem(item.ID)
if err != nil {
t.Fatal(err)
}
if before.ContentState != models.ContentStatePendingFlush {
t.Fatalf("precondition: content_state %q, want %q", before.ContentState, models.ContentStatePendingFlush)
}
versionsBefore := versionCount(t, s, item.ID)
outcome, updated, err := s.MaterializeFlush(item.ID, cursor, "recovered body")
if err != nil || outcome != store.MaterializeApplied || updated == nil {
t.Fatalf("MaterializeFlush = %q, %v, %v; want applied", outcome, updated, err)
}
got, _ := s.GetItem(item.ID)
if got.Content != "recovered body" {
t.Fatalf("content %q, want the recovered body", got.Content)
}
if got.ContentState != "" {
t.Fatalf("content_state %q after recovery, want clean", got.ContentState)
}
if got.Seq <= before.Seq {
t.Fatalf("seq %d not bumped past %d", got.Seq, before.Seq)
}
if got.LastModifiedBy != before.LastModifiedBy {
t.Fatalf("last_modified_by %q changed from %q; recovery attributes only the version row", got.LastModifiedBy, before.LastModifiedBy)
}
wm, ok, err := s.GetItemContentFlushedOpLogID(item.ID)
if err != nil || !ok || wm != cursor {
t.Fatalf("watermark = %d (ok %v, err %v), want the cursor %d", wm, ok, err, cursor)
}
// Exactly one version row, attributed to the system.
vs, err := s.ListItemVersions(item.ID)
if err != nil {
t.Fatal(err)
}
if len(vs) != versionsBefore+1 {
t.Fatalf("versions %d, want %d", len(vs), versionsBefore+1)
}
v := vs[0]
if v.CreatedBy != models.VersionCreatedBySystem || v.Source != models.VersionSourceRecovery || v.ChangeSummary != models.RecoveryChangeSummary {
t.Fatalf("version attribution = (%q, %q, %q), want (%q, %q, %q)", v.CreatedBy, v.Source, v.ChangeSummary,
models.VersionCreatedBySystem, models.VersionSourceRecovery, models.RecoveryChangeSummary)
}
// PLAN-2348 U2 (#1691): a recovery row is a system row, so no user_id;
// it is an update row, not a create; and it records its change's line
// counts like any other update, since it goes through the same insert.
wantAdded, wantRemoved := diff.LineCounts(before.Content, "recovered body")
if v.UserID != "" || v.IsCreate || v.LinesAdded == nil || v.LinesRemoved == nil ||
*v.LinesAdded != wantAdded || *v.LinesRemoved != wantRemoved {
t.Fatalf("recovery version history data: user %q create %v lines %v/%v, want \"\" false %d/%d",
v.UserID, v.IsCreate, v.LinesAdded, v.LinesRemoved, wantAdded, wantRemoved)
}
// The op-log is NOT pruned: the next tab replays it.
if ops, _ := s.LoadYjsUpdatesSince(item.ID, 0); len(ops) != 2 {
t.Fatalf("op-log rows after recovery = %d, want 2 (never pruned)", len(ops))
}
// A second recovery moments later still writes its own version row:
// the per-(actor, source) throttle must not let a recovery move the
// body with no version bracketing it.
cursor2 := appendFrames(t, s, item.ID, recoveryFrame(3))
if outcome, _, err := s.MaterializeFlush(item.ID, cursor2, "recovered again"); err != nil || outcome != store.MaterializeApplied {
t.Fatalf("second MaterializeFlush = %q, %v", outcome, err)
}
if n := versionCount(t, s, item.ID); n != versionsBefore+2 {
t.Fatalf("versions after two recoveries %d, want %d", n, versionsBefore+2)
}
before, _ = s.GetItem(item.ID)
before.Seq = got.Seq // the token a caller read after the FIRST recovery
// A writer holding the pre-recovery seq is refused, not allowed to
// overwrite the recovered text (BUG-3037 / BUG-3133).
stale := before.Seq
body := "caller's stale body"
_, err = s.UpdateItem(item.ID, models.ItemUpdate{Content: &body, ExpectedSeq: &stale})
var conflict *store.UpdateConflictError
if !errors.As(err, &conflict) {
t.Fatalf("stale expected_seq update err = %v, want UpdateConflictError", err)
}
})
}
func TestMaterializeFlushAbortsWhenTheCursorMoved(t *testing.T) {
eachBackend(t, func(t *testing.T, s *store.Store) {
_, _, item := seedStaleItem(t, s)
appendFrames(t, s, item.ID, recoveryFrame(1))
in, err := s.LoadMaterializeInput(item.ID)
if err != nil || in == nil {
t.Fatalf("LoadMaterializeInput: %v, %v", in, err)
}
// A row lands between job build and write.
appendFrames(t, s, item.ID, recoveryFrame(2))
before, _ := s.GetItem(item.ID)
nv := versionCount(t, s, item.ID)
outcome, updated, err := s.MaterializeFlush(item.ID, in.Cursor, "recovered body")
if err != nil || outcome != store.MaterializeCursorMoved || updated != nil {
t.Fatalf("MaterializeFlush = %q, %v, %v; want cursor_moved", outcome, updated, err)
}
got, _ := s.GetItem(item.ID)
if got.Content != "stored body" || got.Seq != before.Seq || versionCount(t, s, item.ID) != nv {
t.Fatalf("an aborted recovery wrote: content %q seq %d->%d", got.Content, before.Seq, got.Seq)
}
if got.ContentState != models.ContentStatePendingFlush {
t.Fatalf("content_state %q, want still pending", got.ContentState)
}
})
}
func TestMaterializeFlushAbortsWhenNothingIsPending(t *testing.T) {
eachBackend(t, func(t *testing.T, s *store.Store) {
_, _, item := seedStaleItem(t, s)
cursor := appendFrames(t, s, item.ID, recoveryFrame(1))
if err := s.SetItemContentFlushedOpLogIDForTesting(item.ID, cursor); err != nil {
t.Fatal(err)
}
before, _ := s.GetItem(item.ID)
outcome, _, err := s.MaterializeFlush(item.ID, cursor, "recovered body")
if err != nil || outcome != store.MaterializeNothingPending {
t.Fatalf("MaterializeFlush = %q, %v; want nothing_pending", outcome, err)
}
got, _ := s.GetItem(item.ID)
if got.Content != "stored body" || got.Seq != before.Seq {
t.Fatalf("wrote with nothing pending: content %q seq %d->%d", got.Content, before.Seq, got.Seq)
}
})
}
func TestMaterializeFlushLeavesSetAsideRowsAlone(t *testing.T) {
eachBackend(t, func(t *testing.T, s *store.Store) {
_, _, item := seedStaleItem(t, s)
setAsideOne(t, s, item.ID)
cursor := appendFrames(t, s, item.ID, recoveryFrame(9))
outcome, _, err := s.MaterializeFlush(item.ID, cursor, "recovered body")
if err != nil || outcome != store.MaterializeSetAside {
t.Fatalf("MaterializeFlush = %q, %v; want set_aside", outcome, err)
}
// Same answer when the body already matches (the stamp-only path).
outcome, _, err = s.MaterializeFlush(item.ID, cursor, "stored body")
if err != nil || outcome != store.MaterializeSetAside {
t.Fatalf("MaterializeFlush (equal body) = %q, %v; want set_aside", outcome, err)
}
rows, err := s.ListYjsSetAside(item.ID)
if err != nil || len(rows) != 1 {
t.Fatalf("set-aside rows = %d (%v), want 1, untouched", len(rows), err)
}
got, _ := s.GetItem(item.ID)
if got.Content != "stored body" || got.ContentState != models.ContentStateSetAside {
t.Fatalf("content %q state %q", got.Content, got.ContentState)
}
})
}
// A document that serializes to the stored body moves the watermark only: no
// seq bump, no version row.
func TestMaterializeFlushStampsWhenTheBodyAlreadyMatches(t *testing.T) {
eachBackend(t, func(t *testing.T, s *store.Store) {
_, _, item := seedStaleItem(t, s)
cursor := appendFrames(t, s, item.ID, recoveryFrame(1))
before, _ := s.GetItem(item.ID)
nv := versionCount(t, s, item.ID)
outcome, _, err := s.MaterializeFlush(item.ID, cursor, "stored body")
if err != nil || outcome != store.MaterializeStamped {
t.Fatalf("MaterializeFlush = %q, %v; want stamped", outcome, err)
}
got, _ := s.GetItem(item.ID)
if got.Seq != before.Seq || versionCount(t, s, item.ID) != nv || got.ContentState != "" {
t.Fatalf("stamp: seq %d->%d versions %d->%d state %q", before.Seq, got.Seq, nv, versionCount(t, s, item.ID), got.ContentState)
}
})
}
// The job covers the WHOLE op-log's content-bearing rows (not only those above
// the watermark), and the cursor is MAX over every row.
func TestLoadMaterializeInputReadsTheWholeOpLog(t *testing.T) {
eachBackend(t, func(t *testing.T, s *store.Store) {
_, _, item := seedStaleItem(t, s)
first := appendFrames(t, s, item.ID, recoveryFrame(1))
if err := s.SetItemContentFlushedOpLogIDForTesting(item.ID, first); err != nil {
t.Fatal(err)
}
appendFrames(t, s, item.ID, recoveryFrame(2))
last := appendFrames(t, s, item.ID, syncStep1Frame)
in, err := s.LoadMaterializeInput(item.ID)
if err != nil || in == nil {
t.Fatalf("LoadMaterializeInput: %v, %v", in, err)
}
if in.Cursor != last {
t.Fatalf("cursor %d, want MAX over all rows %d", in.Cursor, last)
}
if len(in.Rows) != 2 || string(in.Rows[0]) != string(recoveryFrame(1)) || string(in.Rows[1]) != string(recoveryFrame(2)) {
t.Fatalf("rows = %d, want both content-bearing rows in id order", len(in.Rows))
}
if in.Pending != 1 || in.SetAside != 0 {
t.Fatalf("pending %d set-aside %d, want 1, 0", in.Pending, in.SetAside)
}
if len(in.SchemaVersions) != 1 || in.SchemaVersions[0] != "1" {
t.Fatalf("schema versions %v", in.SchemaVersions)
}
})
}
func TestListMaterializeCandidates(t *testing.T) {
eachBackend(t, func(t *testing.T, s *store.Store) {
wsID, collID, pending := seedStaleItem(t, s)
appendFrames(t, s, pending.ID, recoveryFrame(1))
flushed, err := s.CreateItem(wsID, collID, models.ItemCreate{Title: "Flushed", Content: "x"})
if err != nil {
t.Fatal(err)
}
c := appendFrames(t, s, flushed.ID, recoveryFrame(2))
if err := s.SetItemContentFlushedOpLogIDForTesting(flushed.ID, c); err != nil {
t.Fatal(err)
}
aside, err := s.CreateItem(wsID, collID, models.ItemCreate{Title: "Aside", Content: "y"})
if err != nil {
t.Fatal(err)
}
setAsideOne(t, s, aside.ID)
appendFrames(t, s, aside.ID, recoveryFrame(3))
future := time.Now().Add(time.Hour)
ids, err := s.ListMaterializeCandidates(future, store.MaterializeCursor{}, 10)
if err != nil {
t.Fatal(err)
}
if len(ids) != 1 || ids[0].ItemID != pending.ID {
t.Fatalf("candidates = %v, want only the pending item %s", ids, pending.ID)
}
// Not dormant yet: the newest row is younger than the cutoff.
ids, err = s.ListMaterializeCandidates(time.Now().Add(-time.Hour), store.MaterializeCursor{}, 10)
if err != nil || len(ids) != 0 {
t.Fatalf("candidates before the dormancy cutoff = %v (%v), want none", ids, err)
}
has, err := s.ItemHasPendingContent(pending.ID)
if err != nil || !has {
t.Fatalf("ItemHasPendingContent(pending) = %v, %v", has, err)
}
has, err = s.ItemHasPendingContent(flushed.ID)
if err != nil || has {
t.Fatalf("ItemHasPendingContent(flushed) = %v, %v", has, err)
}
})
}
// The sweep's read is keyset-paged: walking it one row at a time visits every
// candidate exactly once, in order, and then comes back empty.
func TestListMaterializeCandidatesPagesByKeyset(t *testing.T) {
eachBackend(t, func(t *testing.T, s *store.Store) {
wsID, collID, first := seedStaleItem(t, s)
want := map[string]bool{first.ID: true}
appendFrames(t, s, first.ID, recoveryFrame(1))
for i := byte(2); i <= 5; i++ {
it, err := s.CreateItem(wsID, collID, models.ItemCreate{Title: fmt.Sprintf("P%d", i), Content: "x"})
if err != nil {
t.Fatal(err)
}
appendFrames(t, s, it.ID, recoveryFrame(i))
want[it.ID] = true
}
future := time.Now().Add(time.Hour)
all, err := s.ListMaterializeCandidates(future, store.MaterializeCursor{}, 100)
if err != nil || len(all) != 5 {
t.Fatalf("full read = %d (%v), want 5", len(all), err)
}
var cur store.MaterializeCursor
var walked []string
for i := 0; i < 10; i++ {
page, err := s.ListMaterializeCandidates(future, cur, 1)
if err != nil {
t.Fatal(err)
}
if len(page) == 0 {
break
}
walked = append(walked, page[0].ItemID)
cur = store.MaterializeCursor{LastAt: page[0].LastAt, ItemID: page[0].ItemID}
}
if len(walked) != 5 {
t.Fatalf("walked %d candidates, want 5", len(walked))
}
for i, c := range all {
if walked[i] != c.ItemID || !want[c.ItemID] {
t.Fatalf("page walk order %v differs from the full read", walked)
}
}
})
}
@@ -0,0 +1,91 @@
// TASK-2198 U4, leg (i) / (ii): Y.Docs holding internal links, for the Go
// tests that run the op-log recovery with the server's WHOLE-workspace link
// index (internal/server/materialize_recovery_links_test.go).
//
// Each case is a markdown body a tab would hold in its editor AFTER loading:
// the stored [[REF|text]] form has already been rendered by wikiLinksToMarkdown
// into a [text](/owner/ws/collection/REF) link, which is what the Y.Doc
// stores. The case's document is built with the headless editor (the live
// editor's schema, materializerSchema.svelte.test.ts) and encoded as ONE
// y-protocols Update frame, with a fixed client id so regeneration is
// byte-stable.
//
// `raw` is the serializer's output BEFORE the flush pipeline (no link index):
// the Go test uses it as "what the document itself contains", so any title in
// the recovered body that is not in `raw` came from the index.
//
// Regenerate with PAD_GEN_MATERIALIZE_LINK_FIXTURE=1. Without it the test
// checks the committed fixture still decodes to the same `raw`.
import { describe, it, expect } from 'vitest';
import fs from 'node:fs';
import path from 'node:path';
import * as Y from 'yjs';
import { prosemirrorToYXmlFragment } from '@tiptap/y-tiptap';
import { createHeadlessEditor, materialize } from './entry';
import { updateFrame } from './replay';
const FIXTURE = path.resolve(__dirname, '../../../../../internal/server/testdata/materialize_links.json');
// X is SECR-1, "Secret Launch Codename", slug secret-launch-codename, in a
// collection the Go test restricts. Its title appears in a document ONLY in
// case `text-equals-current-title`.
const CASES: Array<{ name: string; markdown: string }> = [
{ name: 'ref-link-other-text', markdown: 'See [shown text](/alice/ws/secrets/SECR-1) here.' },
{ name: 'ref-link-old-title', markdown: 'See [Old Codename](/alice/ws/secrets/SECR-1) here.' },
{ name: 'slug-link-slug-text', markdown: 'See [the plan](/alice/ws/secrets/secret-launch-codename) here.' },
{ name: 'text-equals-current-title', markdown: 'See [Secret Launch Codename](/alice/ws/secrets/SECR-1) here.' },
{ name: 'broken-link', markdown: 'See [gone text](broken) here.' },
{ name: 'cross-workspace', markdown: 'See [xw text](/-/r/other/SECR-1) here.' },
{ name: 'typed-wiki-literal', markdown: 'Typed \\[\\[SECR-1\\]\\] literal.' },
{ name: 'public-ref-link', markdown: 'See [the public plan](/alice/ws/public/PUB-2) here.' },
// BUG-3315 U2: a tab loads a link that follows its target's title with the
// follows-title marker in the link's title attribute, and the marker lives
// in the Y.Doc. Recovery must strip it exactly as a tab's save does.
{ name: 'marker-follows-title', markdown: 'See [Secret Launch Codename](/alice/ws/secrets/SECR-1 "pad-follows-title:Secret Launch Codename") here.' },
{ name: 'marker-renamed', markdown: 'See [Old Codename](/alice/ws/secrets/SECR-1 "pad-follows-title:Old Codename") here.' },
{ name: 'marker-external-href', markdown: 'See [Old Codename](https://example.com/x "pad-follows-title:Old Codename") here.' },
];
interface FixtureCase {
name: string;
markdown: string;
rows: string[];
raw: string;
}
function build(): FixtureCase[] {
const ed = createHeadlessEditor();
try {
return CASES.map((c) => {
ed.commands.setContent(c.markdown);
// A marker case is only a marker case if the document holds it.
if (c.name.startsWith('marker-')) {
expect((ed.storage as unknown as { markdown: { getMarkdown(): string } }).markdown.getMarkdown()).toContain('pad-follows-title');
}
const doc = new Y.Doc();
doc.clientID = 2198;
prosemirrorToYXmlFragment(ed.state.doc, doc.getXmlFragment('default'));
const frame = updateFrame(Y.encodeStateAsUpdate(doc));
doc.destroy();
const rows = [Buffer.from(frame).toString('base64')];
const raw = materialize(JSON.stringify({ rows, link_index: [], workspace_slug: 'ws' }));
return { name: c.name, markdown: c.markdown, rows, raw };
});
} finally {
ed.destroy();
}
}
describe('materializer link fixture (TASK-2198 U4)', () => {
it('is current', () => {
const cases = build();
if (process.env.PAD_GEN_MATERIALIZE_LINK_FIXTURE === '1') {
fs.writeFileSync(FIXTURE, JSON.stringify(cases, null, '\t') + '\n');
}
const committed = JSON.parse(fs.readFileSync(FIXTURE, 'utf8')) as FixtureCase[];
expect(committed).toEqual(cases);
// Each document holds a link (or the literal), so the fixture is about
// what it claims to be about.
for (const c of cases) expect(c.raw).toMatch(/SECR-1|secret-launch-codename|gone text|PUB-2|example\.com/);
});
});
@@ -188,8 +188,19 @@
}
}
// 'system' is the op-log recovery's version row (TASK-2198 U4): the server
// rebuilt the body from an editor session that closed without saving, and
// the op-log records no author, so it must not read as a user's edit.
function actorLabel(actor: string): string {
return actor === 'agent' ? 'Agent' : 'User';
if (actor === 'agent') return 'Agent';
if (actor === 'system') return 'System';
return 'User';
}
function actorColor(actor: string): string {
if (actor === 'agent') return 'var(--accent-purple)';
if (actor === 'system') return 'var(--text-secondary)';
return 'var(--status-blue)';
}
function sourceLabel(source: string): string {
@@ -197,7 +208,8 @@
cli: 'CLI',
web: 'Web',
skill: 'Skill',
'collab-snapshot': 'Autosave'
'collab-snapshot': 'Autosave',
recovery: 'Recovered'
};
return labels[source] ?? source;
}
@@ -215,7 +227,7 @@
<div class="badges">
<Chip
size="sm"
color={version.created_by === 'agent' ? 'var(--accent-purple)' : 'var(--status-blue)'}
color={actorColor(version.created_by)}
>
{actorLabel(version.created_by)}
</Chip>
@@ -150,3 +150,30 @@ describe('TimelineVersionCard flush-before-restore (BUG-2271)', () => {
});
});
});
// TASK-2198 U4: the op-log recovery's version row is the system's, and must
// not read as a user's edit.
describe('TimelineVersionCard recovery attribution (TASK-2198 U4)', () => {
it('labels a recovery version System / Recovered, and a user version User / Web', () => {
const labels = (v: Version) => {
const root = target();
const inst = mount(TimelineVersionCard, {
target: root,
props: { version: v, wsSlug: 'ws', itemSlug: 'ITEM-1', currentContent: 'now', onRestore: () => {}, frozen: false },
});
flushSync();
const text = root.querySelector('.badges')?.textContent?.replace(/\s+/g, ' ').trim();
unmount(inst);
root.remove();
return text;
};
const recovered = {
...version,
created_by: 'system',
source: 'recovery',
change_summary: 'recovered from an unsaved editor session',
} as unknown as Version;
expect(labels(recovered)).toBe('System Recovered');
expect(labels(version)).toBe('User Web');
});
});