diff --git a/CLAUDE.md b/CLAUDE.md index 691436b7b..ab72d4dd6 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -84,7 +84,7 @@ REST API at `/api/v1/`. Key endpoints: - `GET/PATCH/DELETE /workspaces/{ws}/items/{slug}` — item by slug - **Write responses (create/update) may carry `warnings.undeclared_fields`** (BUG-2850) — field keys stored in the item's `fields` blob that the collection's schema does not declare. They are ACCEPTED, not refused: a census found 168 live values under 14 such keys, and refusing them would break read-modify-write on items nobody edited wrongly. The element is additive and `omitempty`, so a clean write is byte-identical to before; system-written metadata (`implementation_notes`, `decision_log`, `github_pr`, `convention`) is excluded. The CLI prints the same list to **stderr**, never stdout, so `--format json` stays parseable - **Write responses may also carry `warnings.content_outcome`** (BUG-2995) — on a 200 it takes exactly one value, `applied_pending_flush`: the content went to the live collaborative document (a browser tab has the item open) rather than to `items.content`. It describes the WRITE, not the row's state at response time — a concurrent flush may already have updated the row, and the server does not re-read to find out. The response's `content` is then the markdown as **sent**, not as stored, and a read before a later `?source=collab-snapshot` flush lands answers with the PREVIOUS content — the lag, not a failed write. The stored form may also differ from the sent form: the markdown round-trips through the editor, so bullet markers, emphasis characters, list numbering and blank-line runs normalise, while already-canonical markdown survives unchanged (measured — `web/src/lib/collab/bug2995Roundtrip.svelte.test.ts`). No duration is claimed anywhere, and the flush is NOT guaranteed to happen at all (BUG-3000). The CLI prints this to **stderr**, like the other write warnings, and since TASK-3132 the LAST line of `pad item update`'s stdout also names it (`— body sent to the open editor … a following "pad item show" will return the OLD body`), so it survives `| tail -1` and a discarded stderr; that last line reads `body replaced, A → B bytes` / `body cleared` on every other content write - - **READ responses may carry `content_state`** (BUG-3000 / BUG-3033) — the read-side counterpart to `warnings.content_outcome` above, and it takes the same one value, `applied_pending_flush`. A write warning says what a REQUEST did; this says what the ROW is: the item's op-log holds CONTENT-BEARING updates above `items.content_flushed_op_log_id`, so the body being served is BEHIND the live collaborative document. Since BUG-3124 rows that provably cannot change the document — y-protocols SyncStep1 frames, empty updates, and byte-identical re-sends of an earlier row — are persisted but marked `content_bearing = false` and not counted; they used to keep an item "pending" forever after its tab closed (measured: 470 pending items → 189 on the live DB). The classifier is envelope-only (no Yjs parse) and strict: a frame that does not parse exactly stays content-bearing. Additive and `omitempty`, and emitted only where a real body is — a summary or metadata projection that serves no content never claims its absent body is stale. The window is **unbounded**: nothing server-side moves that content into `items.content` (the op-log GC's flush-coverage guard deliberately RETAINS unflushed rows rather than flushing them), so the row catches up only when a browser tab next opens the item and flushes, which may never happen. The content is not lost — the op-log is the durability layer — but the marker is a signal to RE-READ LATER, never to re-send. The predicate is broader than "an API write is pending flush" and deliberately so: a user typing in an open tab produces the same stale body, and the row does not record WHY the op-log is ahead. Carried by every door that serialises `models.Item`, and by the hand-built ones too — the playbook `run` response, bootstrap convention and bodies-mode include payloads, the exported artifact's `provenance` block, and a line above the body in the MCP item resource. The CLI prints it to **stderr** (`pad item show --format markdown`, `pad playbook show`, `pad playbook run`), like the write warnings — except `pad item show`'s default TABLE output, which since TASK-3132 opens with it on **stdout**; markdown stays stderr because its stdout is the body verbatim for `update --stdin` + - **READ responses may carry `content_state`** (BUG-3000 / BUG-3033) — the read-side counterpart to `warnings.content_outcome` above, and it takes the same one value, `applied_pending_flush`. A write warning says what a REQUEST did; this says what the ROW is: the item's op-log holds CONTENT-BEARING updates above `items.content_flushed_op_log_id`, so the body being served is BEHIND the live collaborative document. Since BUG-3124 rows that provably cannot change the document — y-protocols SyncStep1 frames, empty updates, and byte-identical re-sends of an earlier row — are persisted but marked `content_bearing = false` and not counted; they used to keep an item "pending" forever after its tab closed (measured: 470 pending items → 189 on the live DB). The classifier is envelope-only (no Yjs parse) and strict: a frame that does not parse exactly stays content-bearing. Additive and `omitempty`, and emitted only where a real body is — a summary or metadata projection that serves no content never claims its absent body is stale. The window is **unbounded**: nothing server-side moves that content into `items.content` (the op-log GC's flush-coverage guard deliberately RETAINS unflushed rows rather than flushing them), so the row catches up only when a browser tab next opens the item, which may never happen. Opening it is now enough even when nothing needs flushing (BUG-3124 unit B): a caught-up tab whose document still renders to the stored body calls `POST /workspaces/{ws}/items/{slug}/collab-watermark {op_log_cursor, content_sha256}`, and the server advances the flush watermark iff `cursor == MAX(op-log id)` and the sha256 matches `items.content`, atomically, writing no content, version, `seq` or `updated_at` — before this, a view that changed nothing was deduped and left the item "pending" forever. Web-client only; no CLI/MCP surface. The content is not lost — the op-log is the durability layer — but the marker is a signal to RE-READ LATER, never to re-send. The predicate is broader than "an API write is pending flush" and deliberately so: a user typing in an open tab produces the same stale body, and the row does not record WHY the op-log is ahead. Carried by every door that serialises `models.Item`, and by the hand-built ones too — the playbook `run` response, bootstrap convention and bodies-mode include payloads, the exported artifact's `provenance` block, and a line above the body in the MCP item resource. The CLI prints it to **stderr** (`pad item show --format markdown`, `pad playbook show`, `pad playbook run`), like the write warnings — except `pad item show`'s default TABLE output, which since TASK-3132 opens with it on **stdout**; markdown stays stderr because its stdout is the body verbatim for `update --stdin` - **Write responses may also carry `warnings.dropped_fields`** (TASK-2878, widened by BUG-3079) — schema-declared keys the write DISCARDED. The general case is an injected schema DEFAULT that fails the same `validateFieldType` a caller-supplied value takes: it used to be assigned and skipped past that check, so the SAME BYTES were refused when supplied and stored when injected, decided only by who put them there — a `status` retyped to `multi_select` whose scalar default survived answered 400 to `{"status":"open"}` and 201 to `{}`, storing `{"status":"open"}`. It is now dropped and named here. The relation case keeps its own later pass, because "is this a string" and "does this string name a live, visible item" are different questions and only the first is answerable in the DB-free validator. Dropped rather than refused because nobody in the request typed it — EXCEPT on a `required` field, where dropping leaves it absent and the write is refused as required, with a message naming the default as the cause. Additive and `omitempty` - `POST /workspaces/{ws}/items/{slug}/copy/preflight` — cross-workspace copy dry run: what would carry / drop / need a value, plus the full warning set. Read-only and safe to call repeatedly (PLAN-2357) - `POST /workspaces/{ws}/items/{slug}/copy` — cross-workspace copy; with `archive_source` it is the move. Same request shape as the preflight. **Never retry it automatically** — there is no idempotency key, so a retry duplicates the item diff --git a/internal/server/bug3124_watermark_stamp_test.go b/internal/server/bug3124_watermark_stamp_test.go new file mode 100644 index 000000000..f15c122ea --- /dev/null +++ b/internal/server/bug3124_watermark_stamp_test.go @@ -0,0 +1,191 @@ +package server + +import ( + "bytes" + "crypto/sha256" + "encoding/hex" + "encoding/json" + "net/http" + "net/http/httptest" + "testing" +) + +// BUG-3124 unit B: a caught-up tab stamps the flush watermark without writing +// content. The proof is cursor == MAX(op-log id) AND sha256(items.content) == +// the tab's hash, applied in one conditional UPDATE. + +func sha(s string) string { + sum := sha256.Sum256([]byte(s)) + return hex.EncodeToString(sum[:]) +} + +type stampFixture struct { + srv *Server + ws, slug string + itemID string + content string + lastOpLog int64 +} + +// newStampFixture makes the state the stamp exists for: an item whose op-log +// holds a row above its (NULL) watermark while items.content is current — what +// a tab's seed update leaves behind when the view dedupe skips the flush. +func newStampFixture(t *testing.T) *stampFixture { + t.Helper() + srv := testServer(t) + ws := createWSWithCollections(t, srv) + item := createTaskWithFields(t, srv, ws, "Stamp me", `{"status":"open"}`) + got, err := srv.store.GetItem(item.ID) + if err != nil { + t.Fatal(err) + } + id, err := srv.store.AppendYjsUpdate(item.ID, []byte{0x00, 0x02, 0x03, 0x01, 0x02, 0x03}, "1") + if err != nil { + t.Fatal(err) + } + return &stampFixture{srv: srv, ws: ws, slug: item.Slug, itemID: item.ID, content: got.Content, lastOpLog: id} +} + +func (f *stampFixture) stamp(t *testing.T, body map[string]any) (int, bool) { + t.Helper() + rr := doRequest(f.srv, "POST", "/api/v1/workspaces/"+f.ws+"/items/"+f.slug+"/collab-watermark", body) + var out struct { + Advanced bool `json:"advanced"` + } + _ = json.Unmarshal(rr.Body.Bytes(), &out) + return rr.Code, out.Advanced +} + +func (f *stampFixture) state(t *testing.T) string { + t.Helper() + it, err := f.srv.store.GetItem(f.itemID) + if err != nil { + t.Fatal(err) + } + return it.ContentState +} + +func TestWatermarkStamp_CaughtUpTabClearsPendingWithoutWriting(t *testing.T) { + f := newStampFixture(t) + if f.state(t) == "" { + t.Fatal("premise: an op-log row above a NULL watermark must read pending") + } + before, _ := f.srv.store.GetItem(f.itemID) + versionsBefore, err := f.srv.store.ListItemVersionsResolved(f.itemID, before.Content) + if err != nil { + t.Fatal(err) + } + + code, advanced := f.stamp(t, map[string]any{"op_log_cursor": f.lastOpLog, "content_sha256": sha(f.content)}) + if code != http.StatusOK || !advanced { + t.Fatalf("a caught-up tab with a matching body must stamp; got %d advanced=%v", code, advanced) + } + if s := f.state(t); s != "" { + t.Fatalf("after the stamp the item must read clean; got %q", s) + } + after, _ := f.srv.store.GetItem(f.itemID) + if after.Seq != before.Seq || !after.UpdatedAt.Equal(before.UpdatedAt) || after.Content != before.Content { + t.Fatalf("the stamp must not touch seq/updated_at/content: seq %d→%d updated_at %v→%v", + before.Seq, after.Seq, before.UpdatedAt, after.UpdatedAt) + } + versionsAfter, err := f.srv.store.ListItemVersionsResolved(f.itemID, after.Content) + if err != nil { + t.Fatal(err) + } + if len(versionsAfter) != len(versionsBefore) { + t.Fatalf("the stamp must not write a version row: %d → %d", len(versionsBefore), len(versionsAfter)) + } +} + +// The ruling's named case: a cursor behind MAX is a no-op. The tab has not +// applied every row, so it cannot vouch that the body covers them. +func TestWatermarkStamp_StaleCursorIsANoOp(t *testing.T) { + f := newStampFixture(t) + newer, err := f.srv.store.AppendYjsUpdate(f.itemID, []byte{0x00, 0x02, 0x03, 0x04, 0x05, 0x06}, "1") + if err != nil { + t.Fatal(err) + } + code, advanced := f.stamp(t, map[string]any{"op_log_cursor": f.lastOpLog, "content_sha256": sha(f.content)}) + if code != http.StatusOK || advanced { + t.Fatalf("a cursor behind MAX (%d < %d) must not stamp; got %d advanced=%v", f.lastOpLog, newer, code, advanced) + } + if f.state(t) == "" { + t.Fatal("a refused stamp must leave the item pending") + } + // CONTROL: the current cursor on the same item does stamp, so the refusal + // above is about the cursor and not about this item. + if _, adv := f.stamp(t, map[string]any{"op_log_cursor": newer, "content_sha256": sha(f.content)}); !adv { + t.Fatal("control: the current cursor must stamp") + } +} + +func TestWatermarkStamp_BodyMismatchIsANoOp(t *testing.T) { + f := newStampFixture(t) + if _, adv := f.stamp(t, map[string]any{"op_log_cursor": f.lastOpLog, "content_sha256": sha(f.content + "x")}); adv { + t.Fatal("a tab whose document renders to a different body must not stamp") + } + if f.state(t) == "" { + t.Fatal("the item must stay pending") + } +} + +// The body changed between the tab's load and its stamp (a CLI write): the +// tab's hash describes content the row no longer holds. +func TestWatermarkStamp_BodyChangedSinceTheTabLoadedIsANoOp(t *testing.T) { + f := newStampFixture(t) + rr := doRequest(f.srv, "PATCH", "/api/v1/workspaces/"+f.ws+"/items/"+f.slug, map[string]any{"content": "rewritten by the CLI"}) + if rr.Code != http.StatusOK { + t.Fatalf("PATCH: %d %s", rr.Code, rr.Body.String()) + } + // A server-driven content write stamps MAX itself; append a row above it so + // the item is pending again, then stamp with the tab's OLD body hash. + id, err := f.srv.store.AppendYjsUpdate(f.itemID, []byte{0x00, 0x02, 0x03, 0x07, 0x08, 0x09}, "1") + if err != nil { + t.Fatal(err) + } + if _, adv := f.stamp(t, map[string]any{"op_log_cursor": id, "content_sha256": sha(f.content)}); adv { + t.Fatal("a hash of the body the tab loaded must not stamp over a body that has since changed") + } +} + +func TestWatermarkStamp_MalformedInputIs400(t *testing.T) { + f := newStampFixture(t) + for name, body := range map[string]map[string]any{ + "cursor zero": {"op_log_cursor": 0, "content_sha256": sha(f.content)}, + "negative cursor": {"op_log_cursor": -3, "content_sha256": sha(f.content)}, + "short hash": {"op_log_cursor": f.lastOpLog, "content_sha256": "abc"}, + "non-hex hash": {"op_log_cursor": f.lastOpLog, "content_sha256": sha(f.content)[:62] + "zz"}, + "hash missing": {"op_log_cursor": f.lastOpLog}, + } { + if code, _ := f.stamp(t, body); code != http.StatusBadRequest { + t.Errorf("%s: want 400, got %d", name, code) + } + } +} + +// Moving the watermark is a write decision (it makes rows GC-eligible), so it +// takes edit permission like the flush it stands in for. +func TestWatermarkStamp_ViewerIsRefused(t *testing.T) { + f := newStampFixture(t) + ws, err := f.srv.store.GetWorkspaceBySlug(f.ws) + if err != nil || ws == nil { + t.Fatalf("GetWorkspaceBySlug: %v", err) + } + viewer, tok := loginTestUserAs(t, f.srv, "viewer@example.com", "Viewer", "correct-horse-battery") + if err := f.srv.store.AddWorkspaceMember(ws.ID, viewer.ID, "viewer"); err != nil { + t.Fatal(err) + } + payload, _ := json.Marshal(map[string]any{"op_log_cursor": f.lastOpLog, "content_sha256": sha(f.content)}) + req := httptest.NewRequest("POST", "/api/v1/workspaces/"+f.ws+"/items/"+f.slug+"/collab-watermark", bytes.NewReader(payload)) + req.Header.Set("Content-Type", "application/json") + req.Header.Set("Authorization", "Bearer "+tok) + req.RemoteAddr = "192.0.2.1:1234" + rr := httptest.NewRecorder() + f.srv.ServeHTTP(rr, req) + if rr.Code != http.StatusForbidden { + t.Fatalf("a viewer must be refused; got %d %s", rr.Code, rr.Body.String()) + } + if f.state(t) == "" { + t.Fatal("a refused stamp must leave the item pending") + } +} diff --git a/internal/server/handlers_collab_watermark.go b/internal/server/handlers_collab_watermark.go new file mode 100644 index 000000000..7ce9eb484 --- /dev/null +++ b/internal/server/handlers_collab_watermark.go @@ -0,0 +1,79 @@ +package server + +import ( + "errors" + "net/http" + + "github.com/PerpetualSoftware/pad/internal/store" + "github.com/go-chi/chi/v5" +) + +// handleStampCollabWatermark is POST /workspaces/{ws}/items/{itemSlug}/collab-watermark +// (BUG-3124 unit B): a caught-up browser tab whose document still renders to +// the stored body tells the server so, and the server advances the flush +// watermark without writing content. See store.StampContentWatermarkIfCaughtUp +// for the proof and for why this is not a PATCH. +// +// Edit permission, like the collab-snapshot flush it stands in for: moving the +// watermark makes op-log rows eligible for GC, which is a write decision even +// though the proof makes it safe. +// +// A stamp that does not apply is 200 {"advanced": false}, not an error: a +// cursor behind MAX or a body that changed means the tab is not caught up, and +// the tab's next flush or stamp is the retry. Only malformed input is a 400. +func (s *Server) handleStampCollabWatermark(w http.ResponseWriter, r *http.Request) { + workspaceID, ok := s.getWorkspaceID(w, r) + if !ok { + return + } + itemSlug := chi.URLParam(r, "itemSlug") + item, err := s.store.ResolveItem(workspaceID, itemSlug) + if err != nil { + writeInternalError(w, err) + return + } + if item == nil { + s.writeItemResolveError(w, r, workspaceID, itemSlug) + return + } + if !s.requireItemVisible(w, r, workspaceID, item) { + return + } + if !s.requireEditPermission(w, r, workspaceID, item.ID, item.CollectionID) { + return + } + + var input struct { + OpLogCursor int64 `json:"op_log_cursor"` + ContentSHA256 string `json:"content_sha256"` + } + if err := decodeJSON(r, &input); err != nil { + writeError(w, http.StatusBadRequest, "bad_request", err.Error()) + return + } + + var advanced bool + stamp := func() error { + var serr error + advanced, serr = s.store.StampContentWatermarkIfCaughtUp(item.ID, input.OpLogCursor, input.ContentSHA256) + return serr + } + // Under the per-item collab lock when collab is on, for the same reason the + // snapshot PATCH takes it: a prune must not land between the read and the + // conditional write. The UPDATE's predicate already refuses a moved MAX or + // body; the lock keeps this door ordered with the ones that move them. + if s.collab != nil { + err = s.collab.UnderItemLock(item.ID, stamp) + } else { + err = stamp() + } + if err != nil { + if errors.Is(err, store.ErrWatermarkStampBadInput) { + writeError(w, http.StatusBadRequest, "bad_request", err.Error()) + return + } + writeInternalError(w, err) + return + } + writeJSON(w, http.StatusOK, map[string]bool{"advanced": advanced}) +} diff --git a/internal/server/server.go b/internal/server/server.go index 8f061c369..2fcad00d2 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -1828,6 +1828,9 @@ func (s *Server) setupRouter() { r.Delete("/", s.handleDeleteItem) r.Post("/restore", s.handleRestoreItem) r.Post("/move", s.handleMoveItem) + // Content-free flush-watermark stamp from a caught-up + // tab (BUG-3124 unit B). Web-client only; no CLI/MCP. + r.Post("/collab-watermark", s.handleStampCollabWatermark) // Cross-workspace copy PREFLIGHT (PLAN-2357 / // TASK-2364). Reports what a copy into another // workspace would carry, drop and need, and diff --git a/internal/store/bug3124_watermark_stamp_test.go b/internal/store/bug3124_watermark_stamp_test.go new file mode 100644 index 000000000..fdb88be4a --- /dev/null +++ b/internal/store/bug3124_watermark_stamp_test.go @@ -0,0 +1,52 @@ +package store_test + +import ( + "crypto/sha256" + "encoding/hex" + "testing" + + "github.com/PerpetualSoftware/pad/internal/store" + "github.com/PerpetualSoftware/pad/internal/store/storetest" +) + +// BUG-3124 unit B, store half, on both dialects: the conditional UPDATE's +// parameter typing (`? = (SELECT COALESCE(MAX(id), 0) …)`) is the part most +// likely to differ between SQLite and Postgres. +func TestStampContentWatermarkIfCaughtUp(t *testing.T) { + for _, b := range []struct { + name string + open func(*testing.T) *store.Store + }{ + {"SQLite", storetest.NewSQLite}, + {"Postgres", storetest.NewPostgres}, // skips unless PAD_TEST_POSTGRES_URL is set + } { + t.Run(b.name, func(t *testing.T) { + s := b.open(t) + _, _, item := seedStaleItem(t, s) + first, err := s.AppendYjsUpdate(item.ID, []byte{0x00, 0x02, 0x02, 0x09, 0x09}, "1") + if err != nil { + t.Fatal(err) + } + second, err := s.AppendYjsUpdate(item.ID, []byte{0x00, 0x02, 0x02, 0x08, 0x08}, "1") + if err != nil { + t.Fatal(err) + } + sum := sha256.Sum256([]byte("stored body")) + h := hex.EncodeToString(sum[:]) + + if moved, err := s.StampContentWatermarkIfCaughtUp(item.ID, first, h); err != nil || moved { + t.Fatalf("a cursor behind MAX must not stamp: moved=%v err=%v", moved, err) + } + if moved, err := s.StampContentWatermarkIfCaughtUp(item.ID, second, h); err != nil || !moved { + t.Fatalf("cursor == MAX with a matching body must stamp: moved=%v err=%v", moved, err) + } + if got, _ := s.GetItem(item.ID); got.ContentState != "" { + t.Fatalf("after the stamp content_state = %q, want clean", got.ContentState) + } + // Never regresses, and a repeat is a no-op rather than an error. + if moved, err := s.StampContentWatermarkIfCaughtUp(item.ID, second, h); err != nil || moved { + t.Fatalf("a repeat stamp must be a no-op: moved=%v err=%v", moved, err) + } + }) + } +} diff --git a/internal/store/content_watermark_stamp.go b/internal/store/content_watermark_stamp.go new file mode 100644 index 000000000..15c212526 --- /dev/null +++ b/internal/store/content_watermark_stamp.go @@ -0,0 +1,86 @@ +package store + +import ( + "crypto/sha256" + "encoding/hex" + "errors" + "fmt" +) + +// ErrWatermarkStampBadInput is returned for a stamp request the server could +// never honour (a cursor below 1, a hash that is not 64 hex characters). +var ErrWatermarkStampBadInput = errors.New("watermark stamp: bad input") + +// StampContentWatermarkIfCaughtUp advances items.content_flushed_op_log_id to +// cursor WITHOUT writing content, when a browser tab proves that the stored body +// already is the live document (BUG-3124 unit B). +// +// Why it exists: a tab never re-flushes an unchanged document (the view dedupe, +// BUG-1899/1941), so op-log rows that arrived after the last flush, or that no +// flush ever covered, kept content_state "pending" forever, including after the +// tab that could clear it had opened and closed again. A PATCH is the wrong tool +// here: it writes a version row and bumps seq and updated_at on every idle view, +// which would make `pad item edit`'s expected_seq conflict with a tab that +// merely looked at the item (BUG-3035). +// +// THE PROOF is the TASK-1319 conditional flush's, applied to a flush whose +// content is identical to the row: +// - cursor == MAX(op-log id): the tab's Y.Doc has applied every persisted row; +// - sha256(items.content) == contentSHA256: the markdown that document +// renders to IS the stored body. +// +// Together they say the row covers every op, which is exactly what the +// watermark records. Both are checked in ONE conditional UPDATE, against the +// content value read in the same transaction, so a concurrent write (a CLI +// PATCH, a real flush, a prune) that changes either makes this a no-op rather +// than a stamp over content that no longer matches. +// +// cursor == MAX also covers both of the collab-snapshot PATCH's staleness gates: +// it is >= MIN(op-log id), and >= any restore boundary, because a restore +// records its boundary at or below the MAX its own applier op creates. The +// watermark never regresses (the `<` guard). +// +// Returns whether the watermark moved. Nothing else on the row is touched: no +// content, no seq, no updated_at, no version. +func (s *Store) StampContentWatermarkIfCaughtUp(itemID string, cursor int64, contentSHA256 string) (bool, error) { + if cursor < 1 { + return false, fmt.Errorf("%w: op_log_cursor must be a positive op-log id", ErrWatermarkStampBadInput) + } + if b, err := hex.DecodeString(contentSHA256); err != nil || len(b) != sha256.Size { + return false, fmt.Errorf("%w: content_sha256 must be 64 hex characters", ErrWatermarkStampBadInput) + } + + tx, err := s.db.Begin() + if err != nil { + return false, err + } + defer func() { _ = tx.Rollback() }() + + var content string + if err := tx.QueryRow(s.q(`SELECT content FROM items WHERE id = ? AND deleted_at IS NULL`), itemID).Scan(&content); err != nil { + return false, fmt.Errorf("watermark stamp: read item: %w", err) + } + sum := sha256.Sum256([]byte(content)) + if hex.EncodeToString(sum[:]) != contentSHA256 { + return false, nil + } + + res, err := tx.Exec(s.q(` + UPDATE items SET content_flushed_op_log_id = ? + WHERE id = ? + AND content = ? + AND COALESCE(content_flushed_op_log_id, 0) < ? + AND ? = (SELECT COALESCE(MAX(id), 0) FROM item_yjs_updates WHERE item_id = ?)`), + cursor, itemID, content, cursor, cursor, itemID) + if err != nil { + return false, fmt.Errorf("watermark stamp: update: %w", err) + } + n, err := res.RowsAffected() + if err != nil { + return false, err + } + if err := tx.Commit(); err != nil { + return false, err + } + return n == 1, nil +} diff --git a/web/e2e/bug-3124-view-clears-pending.spec.ts b/web/e2e/bug-3124-view-clears-pending.spec.ts new file mode 100644 index 000000000..c95881018 --- /dev/null +++ b/web/e2e/bug-3124-view-clears-pending.spec.ts @@ -0,0 +1,106 @@ +// BUG-3124 unit B: merely VIEWING an item must not leave it reading +// `content_state: applied_pending_flush` forever. +// +// Opening an item in a tab persists op-log rows (the tab's seed update, its +// sync frames) above the flush watermark, and the view dedupe (BUG-1899) means +// no flush ever covers them — so before this change the item read "pending" +// from the first view onward, and re-opening it could not clear it. A +// caught-up tab now STAMPS the watermark (POST …/collab-watermark) when its +// flush dedupes. +// +// Instruments are read server-side, through a second page, for the reason +// bug-3030-teardown-write-count.spec.ts records: a page that navigates away +// stops reporting its requests while its keepalive request still lands. +import { test, expect } from './fixtures'; +import { browserLogin, EDITOR_SELECTOR, SYNCED_BADGE_SELECTOR } from './lib/collab-helpers'; + +const SYNC_TIMEOUT = 20_000; +const BODY = 'Seed body for BUG-3124 — a paragraph a tab will seed from.'; + +async function seedWithBody(fixture, request): Promise { + const resp = await request.post(`/api/v1/workspaces/${fixture.workspaceSlug}/collections/docs/items`, { + headers: { Authorization: `Bearer ${fixture.apiToken}`, 'Content-Type': 'application/json' }, + data: { title: `BUG-3124 ${Date.now()}`, content: BODY }, + }); + if (!resp.ok()) throw new Error(`doc create failed (${resp.status()}): ${await resp.text()}`); + return ((await resp.json()) as { slug: string }).slug; +} + +async function readItem(page, fixture, slug: string) { + const res = await page.request.get(`/api/v1/workspaces/${fixture.workspaceSlug}/items/${slug}`); + expect(res.ok(), 'item read failed — the instrument, not the product').toBeTruthy(); + return (await res.json()) as { content_state?: string; content: string; seq: number }; +} + +async function readVersionCount(page, fixture, slug: string): Promise { + const res = await page.request.get(`/api/v1/workspaces/${fixture.workspaceSlug}/items/${slug}/versions`); + expect(res.ok(), 'version read failed — the instrument, not the product').toBeTruthy(); + const body = await res.json(); + return Array.isArray(body) ? body.length : (body.versions?.length ?? 0); +} + +async function openItem(page, fixture, slug: string) { + await browserLogin(page); + await page.goto(`/${fixture.adminUsername}/${fixture.workspaceSlug}/docs?item=${slug}`); + await expect(page.locator(EDITOR_SELECTOR)).toBeVisible({ timeout: SYNC_TIMEOUT }); + await expect(page.locator(SYNCED_BADGE_SELECTOR)).toBeVisible({ timeout: SYNC_TIMEOUT }); + await expect(page.locator(EDITOR_SELECTOR)).toContainText('Seed body for BUG-3124'); +} + +test('a view that changes nothing leaves the item clean, with no content write', async ({ page, browser, fixture, request }, testInfo) => { + test.skip(testInfo.project.name !== 'desktop-chromium', 'one project is enough'); + test.setTimeout(90_000); + const slug = await seedWithBody(fixture, request); + const probe = await browser.newPage(); + await browserLogin(probe); + const versionsBefore = await readVersionCount(probe, fixture, slug); + + await openItem(page, fixture, slug); + + // PREMISE: the view put rows above the watermark. Without this the clean + // read below could mean the view never touched the op-log at all, and the + // test would pass on a build with no stamp. + await expect + .poll(async () => (await readItem(probe, fixture, slug)).content_state ?? '', { + timeout: 5_000, + message: 'premise: opening the item must leave op-log rows above the watermark', + }) + .toBe('applied_pending_flush'); + + // THE CLAIM: while the tab stays open and idle, it stamps. + await expect + .poll(async () => (await readItem(probe, fixture, slug)).content_state ?? '', { + timeout: 20_000, + message: 'an idle, caught-up tab must stamp the watermark and clear the pending state', + }) + .toBe(''); + const after = await readItem(probe, fixture, slug); + expect(after.content).toBe(BODY); + expect(await readVersionCount(probe, fixture, slug), 'the stamp must not write a content version').toBe(versionsBefore); + await probe.close(); +}); + +test('navigating away before the idle flush still stamps, via pagehide', async ({ page, browser, fixture, request }, testInfo) => { + test.skip(testInfo.project.name !== 'desktop-chromium', 'one project is enough'); + test.setTimeout(90_000); + const slug = await seedWithBody(fixture, request); + await openItem(page, fixture, slug); + + const probe = await browser.newPage(); + await browserLogin(probe); + // PREMISE, as above: the view has put rows above the watermark. + await expect + .poll(async () => (await readItem(probe, fixture, slug)).content_state ?? '', { timeout: 5_000 }) + .toBe('applied_pending_flush'); + // Leave well inside the 5s idle window, so only the pagehide flushNow can + // have stamped. + await page.goto('about:blank'); + + await expect + .poll(async () => (await readItem(probe, fixture, slug)).content_state ?? '', { + timeout: 10_000, + message: 'the pagehide flush dedupes and must stamp with a keepalive request', + }) + .toBe(''); + await probe.close(); +}); diff --git a/web/src/lib/api/client.ts b/web/src/lib/api/client.ts index be1a66e20..f054d2efa 100644 --- a/web/src/lib/api/client.ts +++ b/web/src/lib/api/client.ts @@ -1439,6 +1439,28 @@ export const api = { }); }, + /** + * Content-free flush-watermark stamp (BUG-3124 unit B). Sent when this + * tab's Y.Doc is caught up (`opLogCursor` is the highest op-log id it + * has applied) AND still renders to the body the server holds, whose + * sha256 is `contentSHA256`. The server advances the watermark only if + * both still hold at commit time and writes nothing else — no content, + * no version, no seq bump — so an idle view no longer leaves the item + * reading "pending" forever. `{advanced: false}` is a normal answer. + */ + stampCollabWatermark: ( + ws: string, + slug: string, + opLogCursor: number, + contentSHA256: string, + opts?: { keepalive?: boolean }, + ) => + request<{ advanced: boolean }>(`/workspaces/${ws}/items/${slug}/collab-watermark`, { + method: 'POST', + body: JSON.stringify({ op_log_cursor: opLogCursor, content_sha256: contentSHA256 }), + keepalive: opts?.keepalive, + }), + delete: (ws: string, slug: string) => request(`/workspaces/${ws}/items/${slug}`, { method: 'DELETE' diff --git a/web/src/lib/collab/collabFlush.svelte.ts b/web/src/lib/collab/collabFlush.svelte.ts index 2048808d3..8dfab1ba4 100644 --- a/web/src/lib/collab/collabFlush.svelte.ts +++ b/web/src/lib/collab/collabFlush.svelte.ts @@ -100,6 +100,12 @@ export interface CollabFlusherConfig { * post-await force-refresh check (returning 'skipped' when it fires). * Resolves 'flushed' on success, 'failed' on PATCH error. */ save: (input: CollabSaveInput) => Promise<'flushed' | 'failed' | 'skipped'>; + /** Called when a flush DEDUPES — the server already holds this document's + * markdown — with the content the dedupe proved the server holds. The page + * uses it to stamp the flush watermark without a PATCH (BUG-3124 unit B): + * a deduped view used to leave op-log rows above the watermark forever. + * Optional; fire-and-forget. */ + stampWatermark?: (input: { ws: string; itemId: string; content: string; keepalive: boolean }) => void; } export interface CollabFlusher { @@ -117,6 +123,12 @@ export interface CollabFlusher { * teardown + beforeunload with keepalive=true to land the snapshot before * the provider tears down. */ flushNow(ctx: CollabFlushContext, keepalive: boolean): boolean; + /** Arm the idle debounce to run `flushNow(ctx, false)` — unless a flush is + * already pending, which then carries the settle too. Called when the op-log + * cursor advances without an editor update (a reconnect replay of frames + * that change nothing, or this tab's own trailing sync frames), so the tab + * still gets a chance to flush or, if nothing changed, stamp (BUG-3124). */ + settle(ctx: CollabFlushContext | null): void; /** Cancel the pending debounce without flushing. Leaves dedupe state intact. */ cancel(): void; /** Reset the per-item dedupe baseline (`lastFlushedContent`). Call on item @@ -189,6 +201,9 @@ export function createCollabFlusher(config: CollabFlusherConfig): CollabFlusher // its own doc comment for the revert-safety rationale the storage-space // compare below still owns. if (shouldDedupeEditorSpace(lastFlushedContent, ctx.seedMd, normalizedMarkdown)) { + // This arm fires only before the session's first flush, so the + // server holds the baseline. + config.stampWatermark?.({ ws: ctx.wsSlug, itemId: ctx.itemId, content: ctx.baseline, keepalive }); return 'deduped'; } // Serialize against the CAPTURED workspace (the item being flushed), not @@ -203,7 +218,10 @@ export function createCollabFlusher(config: CollabFlusherConfig): CollabFlusher // no-ops are suppressed. (Revert-safe: after a real flush // `lastFlushedContent` is non-null and takes precedence.) const serverContent = lastFlushedContent ?? ctx.baseline; - if (serverContent === toSave) return 'deduped'; + if (serverContent === toSave) { + config.stampWatermark?.({ ws: ctx.wsSlug, itemId: ctx.itemId, content: serverContent, keepalive }); + return 'deduped'; + } // Snapshot the reset generation across the PATCH so a resetDedup() that // lands while it's in flight invalidates the record below. @@ -230,6 +248,19 @@ export function createCollabFlusher(config: CollabFlusherConfig): CollabFlusher return result; } + function settle(ctx: CollabFlushContext | null): void { + // A pending edit flush already runs the full gauntlet (and stamps if it + // dedupes); re-arming here would postpone a real edit's flush on every + // cursor frame, so leave it alone. + if (timer !== undefined) return; + if (config.isRecovering()) return; + if (!ctx) return; + timer = setTimeout(() => { + timer = undefined; + flushNow(ctx, false); + }, idleMs); + } + function flushNow(ctx: CollabFlushContext, keepalive: boolean): boolean { cancel(); const md = config.readEditorMarkdown(); @@ -251,6 +282,7 @@ export function createCollabFlusher(config: CollabFlusherConfig): CollabFlusher schedule, flush, flushNow, + settle, cancel, resetDedup, get lastFlushed() { diff --git a/web/src/lib/collab/collabFlushStamp.test.ts b/web/src/lib/collab/collabFlushStamp.test.ts new file mode 100644 index 000000000..d329f655c --- /dev/null +++ b/web/src/lib/collab/collabFlushStamp.test.ts @@ -0,0 +1,112 @@ +import { describe, it, expect, vi, beforeEach, afterEach } from 'vitest'; +import { + createCollabFlusher, + type CollabFlushContext, + type CollabFlusherConfig, +} from './collabFlush.svelte'; + +// BUG-3124 unit B: a deduped flush proves the server holds this document's +// markdown, so it hands the page that content to stamp the watermark with; +// settle() gives a cursor advance with no editor update a flush to dedupe. + +function ctx(overrides: Partial = {}): CollabFlushContext { + return { wsSlug: 'ws', itemId: 'item-1', baseline: 'stored body', seedMd: null, ...overrides }; +} + +function makeFlusher(overrides: Partial = {}) { + const save = vi.fn(async () => 'flushed' as const); + const stampWatermark = vi.fn(); + const config: CollabFlusherConfig = { + idleMs: 5000, + isRecovering: () => false, + normalize: (m) => m, + serialize: (m) => m, + readEditorMarkdown: () => 'stored body', + isActiveItem: () => true, + save, + stampWatermark, + ...overrides, + }; + return { flusher: createCollabFlusher(config), save, stampWatermark }; +} + +describe('stampWatermark on dedupe', () => { + it('storage-space dedupe stamps with the content the server holds, and sends no PATCH', async () => { + const { flusher, save, stampWatermark } = makeFlusher(); + expect(await flusher.flush(ctx(), 'stored body', false)).toBe('deduped'); + expect(save).not.toHaveBeenCalled(); + expect(stampWatermark).toHaveBeenCalledTimes(1); + expect(stampWatermark).toHaveBeenCalledWith({ + ws: 'ws', + itemId: 'item-1', + content: 'stored body', + keepalive: false, + }); + }); + + it('editor-space dedupe stamps with the BASELINE (the server copy), not the editor-space markdown', async () => { + // serialize diverges from the seed, so only the editor-space arm can dedupe. + const { flusher, save, stampWatermark } = makeFlusher({ serialize: () => 'reserialized differently' }); + const c = ctx({ seedMd: 'editor view' }); + expect(await flusher.flush(c, 'editor view', true)).toBe('deduped'); + expect(save).not.toHaveBeenCalled(); + expect(stampWatermark).toHaveBeenCalledWith({ + ws: 'ws', + itemId: 'item-1', + content: 'stored body', + keepalive: true, + }); + }); + + it('after a real flush, a later dedupe stamps with what was FLUSHED, not the stale baseline', async () => { + const { flusher, stampWatermark } = makeFlusher(); + expect(await flusher.flush(ctx(), 'edited body', false)).toBe('flushed'); + expect(stampWatermark).not.toHaveBeenCalled(); + expect(await flusher.flush(ctx(), 'edited body', false)).toBe('deduped'); + expect(stampWatermark).toHaveBeenCalledWith(expect.objectContaining({ content: 'edited body' })); + }); + + it('a differing document PATCHes and does not stamp', async () => { + const { flusher, save, stampWatermark } = makeFlusher(); + expect(await flusher.flush(ctx(), 'a real edit', false)).toBe('flushed'); + expect(save).toHaveBeenCalledTimes(1); + expect(stampWatermark).not.toHaveBeenCalled(); + }); + + it('stamps synchronously, before flush() yields, so a pagehide keepalive request starts in time', () => { + const { flusher, stampWatermark } = makeFlusher(); + void flusher.flush(ctx(), 'stored body', true); + expect(stampWatermark).toHaveBeenCalledTimes(1); + }); +}); + +describe('settle()', () => { + beforeEach(() => vi.useFakeTimers()); + afterEach(() => vi.useRealTimers()); + + it('runs an idle flushNow of the live editor markdown, which dedupes and stamps', () => { + const { flusher, stampWatermark } = makeFlusher(); + flusher.settle(ctx()); + expect(stampWatermark).not.toHaveBeenCalled(); + vi.advanceTimersByTime(5000); + expect(stampWatermark).toHaveBeenCalledTimes(1); + }); + + it('does not postpone a pending edit flush', () => { + const { flusher, save } = makeFlusher(); + flusher.schedule(ctx(), 'an edit'); + vi.advanceTimersByTime(4000); + flusher.settle(ctx()); // must NOT re-arm the timer + vi.advanceTimersByTime(1000); + expect(save).toHaveBeenCalledTimes(1); + expect(save).toHaveBeenCalledWith(expect.objectContaining({ toSave: 'an edit' })); + }); + + it('is inert while force-refresh recovery is in flight', () => { + const { flusher, stampWatermark, save } = makeFlusher({ isRecovering: () => true }); + flusher.settle(ctx()); + vi.advanceTimersByTime(5000); + expect(stampWatermark).not.toHaveBeenCalled(); + expect(save).not.toHaveBeenCalled(); + }); +}); diff --git a/web/src/lib/collab/watermarkStamper.test.ts b/web/src/lib/collab/watermarkStamper.test.ts new file mode 100644 index 000000000..14762f578 --- /dev/null +++ b/web/src/lib/collab/watermarkStamper.test.ts @@ -0,0 +1,106 @@ +import { describe, expect, it, vi } from 'vitest'; +import { createWatermarkStamper } from './watermarkStamper'; +import { sha256Hex } from '$lib/utils/sha256'; + +// BUG-3124 unit B: the page-side rules for the content-free watermark stamp. +function make(cursor: number, opts: { recovering?: boolean; fail?: boolean } = {}) { + let current = cursor; + const send = vi.fn(async () => { + if (opts.fail) throw new Error('network'); + return { advanced: true }; + }); + const stamp = createWatermarkStamper({ + cursorFor: () => current, + isRecovering: () => !!opts.recovering, + send, + }); + return { stamp, send, setCursor: (c: number) => (current = c) }; +} +const input = { ws: 'ws', itemId: 'item-1', content: 'stored body', keepalive: false }; + +describe('createWatermarkStamper', () => { + it('sends the cursor and the sha256 of the content the server holds', () => { + const { stamp, send } = make(42); + stamp({ ...input, keepalive: true }); + expect(send).toHaveBeenCalledWith('ws', 'item-1', 42, sha256Hex('stored body'), true); + }); + + it('sends synchronously, so a pagehide keepalive request starts before unload', () => { + const { stamp, send } = make(42); + stamp(input); + expect(send).toHaveBeenCalledTimes(1); + }); + + it('does not re-send a cursor it already sent, and does send a newer one', () => { + const { stamp, send, setCursor } = make(42); + stamp(input); + stamp(input); + expect(send).toHaveBeenCalledTimes(1); + setCursor(43); + stamp(input); + expect(send).toHaveBeenCalledTimes(2); + }); + + it('sends nothing without a cursor the server issued', () => { + const { stamp, send } = make(0); + stamp(input); + expect(send).not.toHaveBeenCalled(); + }); + + it('sends nothing while force-refresh recovery is in flight', () => { + const { stamp, send } = make(42, { recovering: true }); + stamp(input); + expect(send).not.toHaveBeenCalled(); + }); + + it('forgets a cursor whose send failed, so a later settle can retry it', async () => { + const { stamp, send } = make(42, { fail: true }); + stamp(input); + await Promise.resolve(); + await Promise.resolve(); + stamp(input); + expect(send).toHaveBeenCalledTimes(2); + }); +}); + +describe('createWatermarkStamper — a send that throws synchronously', () => { + it('does not throw into its caller (the flusher dedupe arm), and can retry', () => { + const send = vi.fn(() => { + throw new TypeError('not a function'); + }); + const stamp = createWatermarkStamper({ cursorFor: () => 7, isRecovering: () => false, send }); + expect(() => stamp(input)).not.toThrow(); + stamp(input); + expect(send).toHaveBeenCalledTimes(2); + }); +}); + +describe('createWatermarkStamper — codex round 1', () => { + it('the keepalive (pagehide) path sends even for a cursor already sent', () => { + const { stamp, send } = make(42); + stamp(input); + stamp({ ...input, keepalive: true }); + expect(send).toHaveBeenCalledTimes(2); + expect(send).toHaveBeenLastCalledWith('ws', 'item-1', 42, sha256Hex('stored body'), true); + }); + + it('a 200 {advanced:false} forgets the cursor so a later flush retries it', async () => { + const send = vi.fn(async () => ({ advanced: false })); + const stamp = createWatermarkStamper({ cursorFor: () => 9, isRecovering: () => false, send }); + stamp(input); + await Promise.resolve(); + await Promise.resolve(); + stamp(input); + expect(send).toHaveBeenCalledTimes(2); + }); + + it('CONTROL: a 200 {advanced:true} keeps the cursor, so it is not re-sent', async () => { + const send = vi.fn(async () => ({ advanced: true })); + const stamp = createWatermarkStamper({ cursorFor: () => 9, isRecovering: () => false, send }); + stamp(input); + await Promise.resolve(); + await Promise.resolve(); + stamp(input); + expect(send).toHaveBeenCalledTimes(1); + }); +}); diff --git a/web/src/lib/collab/watermarkStamper.ts b/web/src/lib/collab/watermarkStamper.ts new file mode 100644 index 000000000..1db0786ff --- /dev/null +++ b/web/src/lib/collab/watermarkStamper.ts @@ -0,0 +1,75 @@ +// The page half of BUG-3124 unit B's content-free watermark stamp, kept out of +// ItemDetail so the component gains no new async continuation (its identity +// gate audits every one) and so the rules below are unit-testable. +// +// Called with the content a deduped flush proved the server holds. It sends +// the tab's op-log cursor and the sha256 of that content; the server stamps +// only if cursor == MAX(op-log id) and the hash matches items.content, both +// re-checked atomically, and writes nothing else. This side only decides +// whether sending is worth it. + +import { sha256Hex } from '$lib/utils/sha256'; + +export interface WatermarkStampInput { + ws: string; + itemId: string; + content: string; + keepalive: boolean; +} + +export interface WatermarkStamperDeps { + /** The op-log cursor of the provider for `itemId`, or 0 if none tracks it. */ + cursorFor: (itemId: string) => number; + /** True while force-refresh recovery is in flight: the Y.Doc is known stale. */ + isRecovering: () => boolean; + /** Resolves with the server's answer; `advanced: false` means the proof did + * not hold at commit time (cursor behind MAX, or the body changed). */ + send: ( + ws: string, + itemId: string, + cursor: number, + contentSHA256: string, + keepalive: boolean, + ) => Promise<{ advanced?: boolean } | void>; +} + +export function createWatermarkStamper(deps: WatermarkStamperDeps): (input: WatermarkStampInput) => void { + // Highest cursor already sent per item. Plain Map, not $state: a handler-only + // tracker (CONVE-1688). + const lastStamped = new Map(); + return ({ ws, itemId, content, keepalive }) => { + if (deps.isRecovering()) return; + const cursor = deps.cursorFor(itemId); + // A cursor the server never issued: nothing to prove. + if (cursor < 1) return; + // A cursor already sent is skipped — EXCEPT on the keepalive (pagehide) + // path, which always sends: an earlier ordinary request for the same + // cursor may still be in flight and be cancelled by the teardown, and + // nothing would trigger a retry after the page is gone (codex round 1). + // A duplicate stamp is a no-op on the server. + if (!keepalive && cursor <= (lastStamped.get(itemId) ?? 0)) return; + lastStamped.set(itemId, cursor); + // Synchronous up to the request (sha256Hex, not crypto.subtle), so the + // pagehide path starts its keepalive request before the page unloads. + const forget = () => { + // Forget the cursor so a later settle can retry. The stamp refines a + // signal; it is never a write the user owns, so a failure is silent. + if (lastStamped.get(itemId) === cursor) lastStamped.delete(itemId); + }; + // A SYNCHRONOUS throw is caught too: this runs inside the flusher's + // dedupe arm, and a throw there would reject the flush itself — whose + // 'deduped' result callers such as the rich→raw toggle await. + try { + deps + .send(ws, itemId, cursor, sha256Hex(content), keepalive) + .then((res) => { + // A 200 that did not advance is not a success for this cursor: + // the body or the op-log moved under the proof. Forget it so a + // later flush can try again (codex round 1). + if (res && res.advanced === false) forget(); + }, forget); + } catch { + forget(); + } + }; +} diff --git a/web/src/lib/collab/wsProvider.svelte.ts b/web/src/lib/collab/wsProvider.svelte.ts index fb1c1ff6e..dd0f5b7e9 100644 --- a/web/src/lib/collab/wsProvider.svelte.ts +++ b/web/src/lib/collab/wsProvider.svelte.ts @@ -210,6 +210,14 @@ export interface CollabProviderOptions { * leaving the editor on stale state. */ onForceRefresh?: ForceRefreshHandler; + /** + * Called after an op_log_cursor frame ADVANCES `lastOpLogID` (BUG-3124). + * Cursor frames arrive for every persisted op, this tab's own included, so + * this fires even when nothing in the document changed (a reconnect replay + * of frames it already had, its own trailing sync frames) — exactly the + * cases where no editor update fires and so no flush would ever run. + */ + onOpLogCursor?: (opLogID: number) => void; /** * The items.content generation (`item.seq`) this provider's Y.Doc was * SEEDED from (BUG-2264). Announced to the server as `?content_seq=` @@ -322,6 +330,7 @@ export class CollabProvider { private readonly WebSocketImpl: typeof WebSocket; private readonly onApplierRequest?: ApplierRequestHandler; private readonly onForceRefresh?: ForceRefreshHandler; + private readonly onOpLogCursor?: (opLogID: number) => void; /** * items.content generation the Y.Doc was seeded from, announced as * `?content_seq=` on every (re)connect (BUG-2264). Constant per provider. @@ -347,6 +356,7 @@ export class CollabProvider { this.WebSocketImpl = options.WebSocketImpl ?? globalThis.WebSocket; this.onApplierRequest = options.onApplierRequest; this.onForceRefresh = options.onForceRefresh; + this.onOpLogCursor = options.onOpLogCursor; this.contentSeq = options.contentSeq ?? 0; // Restore the per-tab cursor BEFORE the first connect. The // `since=` query string is appended in connect() each @@ -979,6 +989,12 @@ export class CollabProvider { if (msg.op_log_id <= this.lastOpLogID) return; this.lastOpLogID = msg.op_log_id; writeStoredCursor(this.itemID, msg.op_log_id); + // A page callback must not break the frame loop. + try { + this.onOpLogCursor?.(msg.op_log_id); + } catch (err) { + console.warn('collab: onOpLogCursor handler threw', err); + } return; } case 'force_refresh': { diff --git a/web/src/lib/collab/wsProviderCursor.svelte.test.ts b/web/src/lib/collab/wsProviderCursor.svelte.test.ts new file mode 100644 index 000000000..665039b6f --- /dev/null +++ b/web/src/lib/collab/wsProviderCursor.svelte.test.ts @@ -0,0 +1,68 @@ +import { describe, expect, it, vi } from 'vitest'; +import * as Y from 'yjs'; +import { CollabProvider } from './wsProvider.svelte'; + +// BUG-3124 unit B: the provider tells the page when the op-log cursor ADVANCES, +// because a cursor advance with no document change (a reconnect replaying ops +// this tab already has) fires no editor update and so would never reach the +// flusher. This pins the provider half of that wiring against the real frame +// handler, through a fake socket. + +class FakeSocket extends EventTarget { + static last: FakeSocket | null = null; + binaryType = 'arraybuffer'; + readyState = 1; + sent: unknown[] = []; + constructor(public url: string) { + super(); + FakeSocket.last = this; + } + send(data: unknown) { + this.sent.push(data); + } + close() { + this.readyState = 3; + } + control(msg: object) { + this.dispatchEvent(new MessageEvent('message', { data: JSON.stringify(msg) })); + } +} + +function makeProvider(onOpLogCursor: (id: number) => void) { + const provider = new CollabProvider('item-3124', new Y.Doc(), { + url: 'ws://test.invalid/api/v1/collab/item-3124', + WebSocketImpl: FakeSocket as unknown as typeof WebSocket, + onOpLogCursor, + }); + const sock = FakeSocket.last!; + sock.dispatchEvent(new Event('open')); + return { provider, sock }; +} + +describe('CollabProvider onOpLogCursor', () => { + it('fires on every cursor ADVANCE and not on a frame that does not advance', () => { + const seen: number[] = []; + const { provider, sock } = makeProvider((id) => seen.push(id)); + sock.control({ type: 'op_log_cursor', op_log_id: 5 }); + sock.control({ type: 'op_log_cursor', op_log_id: 9 }); + sock.control({ type: 'op_log_cursor', op_log_id: 9 }); // repeat: no advance + sock.control({ type: 'op_log_cursor', op_log_id: 7 }); // regression: never adopted + expect(seen).toEqual([5, 9]); + expect(provider.lastOpLogID).toBe(9); + provider.destroy(); + }); + + it('a throwing handler does not break the frame loop', () => { + const warn = vi.spyOn(console, 'warn').mockImplementation(() => {}); + const handler = vi.fn(() => { + throw new Error('page bug'); + }); + const { provider, sock } = makeProvider(handler); + sock.control({ type: 'op_log_cursor', op_log_id: 3 }); + sock.control({ type: 'op_log_cursor', op_log_id: 4 }); + expect(handler).toHaveBeenCalledTimes(2); + expect(provider.lastOpLogID).toBe(4); + provider.destroy(); + warn.mockRestore(); + }); +}); diff --git a/web/src/lib/components/items/ItemDetail.svelte b/web/src/lib/components/items/ItemDetail.svelte index 1eb0b4c74..f29919dcb 100644 --- a/web/src/lib/components/items/ItemDetail.svelte +++ b/web/src/lib/components/items/ItemDetail.svelte @@ -22,6 +22,7 @@ import { CollabProvider, type CollabConnectionState } from '$lib/collab/wsProvider.svelte'; import { userColor } from '$lib/collab/cursorColor'; import { createCollabFlusher, type CollabFlushContext } from '$lib/collab/collabFlush.svelte'; + import { createWatermarkStamper } from '$lib/collab/watermarkStamper'; import { createContentSaver } from '$lib/items/contentSaver.svelte'; import { authStore } from '$lib/stores/auth.svelte'; import FieldEditor from '$lib/components/fields/FieldEditor.svelte'; @@ -2571,6 +2572,12 @@ // the WS is already closing on the server side and the // $effect cleanup runs that destroy as part of swapping // in the new provider. + // BUG-3124: a cursor advance with no editor update (a reconnect + // replay of frames this tab already had, its own trailing sync + // frames) would otherwise never reach the flusher, so the item kept + // reading "pending" forever. Settle arms an idle flushNow, which + // either flushes a real difference or dedupes and stamps. + onOpLogCursor: () => settleCollabIfCurrent(ctx), onForceRefresh: () => { toastStore.show( 'Editor refreshed — rejoining with the latest content from the server.', @@ -3862,6 +3869,25 @@ // stay HERE in the injected `save` — the module owns none of it. This // mirrors contentSaver's split: the module owns timing/dedup; the page owns // the effectful bits it's coupled to. + // BUG-3124 unit B: see watermarkStamper.ts. + const stampCollabWatermark = createWatermarkStamper({ + cursorFor: (itemId) => + collabProvider && collabProvider.itemID === itemId ? collabProvider.lastOpLogID : 0, + isRecovering: () => forceRefreshInFlight, + send: (ws, itemId, cursor, contentSHA256, keepalive) => + api.items.stampCollabWatermark(ws, itemId, cursor, contentSHA256, { keepalive }), + }); + + // A cursor advance with no editor update never reaches the flusher (BUG-3124); + // settle it, for the context that is still current and still this user's. + // Component-level rather than inline in the provider's $effect, so the + // identity read is not lexically inside a reactive scope. + function settleCollabIfCurrent(ctx: NonNullable): void { + if (activeCollabContext === ctx && !ctx.retired && identityHeld(ctx.identityEpoch)) { + collabFlusher.settle(ctx); + } + } + const collabFlusher = createCollabFlusher({ idleMs: 5_000, // Force-refresh recovery gate: while in flight, any Y.Doc-derived @@ -3922,6 +3948,10 @@ // the item we flushed is still active, else a stale flush pollutes the // new page's dedupe state. isActiveItem: (itemId) => !!item && item.id === itemId, + // BUG-3124 unit B: a deduped flush proves this tab's document renders to + // the body the server holds; stamp the flush watermark (no PATCH, so no + // version row and no seq bump). Rules live in watermarkStamper.ts. + stampWatermark: (input) => stampCollabWatermark(input), // The actual PATCH + reactive bookkeeping. Owns saveStatus / // editorStore / toast / showSaved, the op-log-cursor read, and the // post-await force_refresh check (returns 'skipped' when it fires so diff --git a/web/src/lib/components/items/itemDetailIdentityFenceAst.test.ts b/web/src/lib/components/items/itemDetailIdentityFenceAst.test.ts index a0d4e6e10..333dbb5ae 100644 --- a/web/src/lib/components/items/itemDetailIdentityFenceAst.test.ts +++ b/web/src/lib/components/items/itemDetailIdentityFenceAst.test.ts @@ -92,7 +92,7 @@ const ASYNC_FUNCTIONS: Record = { jumpToSection: { reviewed: '17407aeb431b', why: 'switches this instance\'s tab and scrolls to an anchor', may: ['document.getElementById', 'document.getElementById(anchorId).scrollIntoView'] }, ensureGraphComp: { reviewed: 'c1565cfb8a18', why: 'lazy-loads a component module into this instance', may: ['ItemGraphComp', 'graphLoadError'] }, handleCopyRef: { reviewed: 'fb3adcaf167a', why: 'switchedAway before the copied flag' }, - loadData: { reviewed: 'd91fb3958cb7', why: 'IS the load: myGen against loadGeneration after every await' }, + loadData: { reviewed: '451e2cd3d6e8', why: 'IS the load: myGen against loadGeneration after every await' }, startEditTitle: { reviewed: '113d9b9da03f', why: 'focuses and sizes the input it opened synchronously', may: ['el', 'titleInputEl.focus', 'titleInputEl.setSelectionRange'] }, saveTitle: { reviewed: '8a6bedbc107a', why: 'gen against loadGeneration on both arms' }, updateField: { @@ -121,16 +121,16 @@ const ASYNC_FUNCTIONS: Record = { updateAssignedUser: { reviewed: '86f060522de1', why: 'gen against loadGeneration on both arms' }, updateAgentRole: { reviewed: 'ae4f320cdbab', why: 'gen against loadGeneration on both arms' }, flushRawIfPending: { - reviewed: 'd2494869c902', + reviewed: '39b3b4e91cbd', why: 'genAtFlush against loadGeneration after each PATCH; the re-entrancy waiter returns state; the finally clears this drain\'s own in-flight flag', may: ['rawFlushInFlight'], bareAwaits: ['await new Promise((r) => setTimeout(r, 50));'], }, refreshLinksPreservingOnFailure: { reviewed: '28ef88d9c682', why: 'returns a value; its callers fence' }, - flushCollabBeforeRestore: { reviewed: 'e536ec058fd5', why: 'identityHeld before its failure toast' }, + flushCollabBeforeRestore: { reviewed: '0d45557c8da9', why: 'identityHeld before its failure toast' }, closeCopyDialog: { reviewed: '1c925f081a26', why: 'restores focus after closing synchronously', may: ['paneMenuTrigger.focus'] }, closePushDialog: { reviewed: 'bda8c7529677', why: 'restores focus after closing synchronously', may: ['paneMenuTrigger.focus'] }, - flushContentBeforeCopy: { reviewed: '44ae74e1f4e9', why: 'returns a boolean to the dialog' }, + flushContentBeforeCopy: { reviewed: 'ea71ea2653cd', why: 'returns a boolean to the dialog' }, handleCopied: { reviewed: '66bc98609e3d', why: 'switchedAway before adopting the refreshed item' }, handleDelete: { reviewed: 'b66adcf84794', why: 'switchedAway on both arms' }, handleRestore: { reviewed: 'c3a1732554a4', why: 'switchedAway on every arm' }, @@ -169,7 +169,7 @@ const NESTED: SignedRow[] = [ /** Async functions in the markup. */ const MARKUP: SignedRow[] = [ { body: /startGen/, why: 'Rich toggle: startGen against loadGeneration after each await', reviewed: '63a9cce2ed93' }, - { body: /genAtToggle/, why: 'Markdown toggle: genAtToggle against loadGeneration after each await', reviewed: '19614ee2480c' }, + { body: /genAtToggle/, why: 'Markdown toggle: genAtToggle against loadGeneration after each await', reviewed: 'd651322b1528' }, ]; const collapse = (s: string) => s.replace(/\s+/g, ' '); @@ -336,7 +336,7 @@ const CONTINUATIONS: SignedRow[] = [ may: ['renameOverride'], }, { call: /^setTimeout\($/, body: /copied = false/, why: 'copy-flag reset: switchedAway', reviewed: '199047839886' }, - { call: /api\.items\.get\(wsSlug, itemSlug\)\.catch\($/, body: /./, why: 'loadData item fetch: sets a flag local to that load and re-throws', reviewed: '11bb17b34c22' }, + { call: /api\.items\.get\(wsSlug, itemSlug\)\.catch\($/, body: /./, why: 'loadData item fetch: sets a flag local to that load and re-throws', reviewed: '686654a9efa6' }, { call: /^setTimeout\($/, body: /staleConnecting = true/, @@ -345,14 +345,14 @@ const CONTINUATIONS: SignedRow[] = [ why: 'connection state of this instance\'s own provider', reviewed: 'c4d9f3230509', may: ['staleConnecting'], }, - { call: /\.get\(refreshCtx\.wsSlug, refreshCtx\.itemId\) \.then\($/, body: /./, why: 'force-refresh fetch: refreshGen against loadGeneration', reviewed: 'dad7156ff5b6' }, - { call: /forceRefreshNonce \+= 1; \}\) \.catch\($/, body: /./, why: 'force-refresh failure: refreshGen against loadGeneration', reviewed: 'a27963ab0e32' }, + { call: /\.get\(refreshCtx\.wsSlug, refreshCtx\.itemId\) \.then\($/, body: /./, why: 'force-refresh fetch: refreshGen against loadGeneration', reviewed: 'f446482ada25' }, + { call: /forceRefreshNonce \+= 1; \}\) \.catch\($/, body: /./, why: 'force-refresh failure: refreshGen against loadGeneration', reviewed: '9875afbc0f08' }, { call: /^setTimeout\($/, body: /teardownFlushed/, in: 'onBeforeUnload', code: '() => { teardownFlushed = false; }', - why: 're-arms the BUG-3005 teardown latch, itself identity-checked', reviewed: 'c25508ba8261', + why: 're-arms the BUG-3005 teardown latch, itself identity-checked', reviewed: 'db5ba014896a', may: ['teardownFlushed'], }, { call: /^queueMicrotask\($/, body: /./, why: 'collab lazy seed: refuses a retired or re-identified context first', reviewed: '5e2ef566eb67' }, @@ -370,16 +370,16 @@ const CONTINUATIONS: SignedRow[] = [ call: /^setTimeout\($/, body: /content: toSave \}\)\.then/, in: 'handleContentUpdate', - why: 'content debounce: loadData clears this timer before its first await, so the callback never runs across a load', reviewed: 'a770ebbad8d1', + why: 'content debounce: loadData clears this timer before its first await, so the callback never runs across a load', reviewed: '992da66e24a1', startSafe: true, pin: (src, unit) => clearsBeforeFirstAwait(src, 'loadData', 'contentDebounceTimer') ?? assignedTo(src, unit, 'contentDebounceTimer'), }, - { call: /\{ content: toSave \}\)\.then\($/, body: /^\(\) =>/, why: 'content save: switchedAway', reviewed: 'c6f5c89645db' }, - { call: /showSaved\(\); \}\)\.catch\($/, body: /./, why: 'content save failure: switchedAway', reviewed: 'e96323c8ebee' }, - { call: /\{ keepalive: true \}\) \.then\($/, body: /./, why: 'raw keepalive save: genAtSave against loadGeneration', reviewed: '0abbaff380e5' }, - { call: /localDirty = false; \} \}\) \.catch\($/, body: /^\(\) => \{\}$/, why: 'raw keepalive failure: empty', reviewed: 'e50ab6dcb406' }, - { call: /reqItemId, \{ content: toSave \}\)\.then\($/, body: /./, why: 'raw foreground save: genAtSave against loadGeneration', reviewed: '9911d0e1cfee' }, - { call: /content: item\.content \}\); \} \}\)\.catch\($/, body: /./, why: 'raw foreground failure: genAtSave against loadGeneration', reviewed: '083f740f76dc' }, + { call: /\{ content: toSave \}\)\.then\($/, body: /^\(\) =>/, why: 'content save: switchedAway', reviewed: '7654c1adabca' }, + { call: /showSaved\(\); \}\)\.catch\($/, body: /./, why: 'content save failure: switchedAway', reviewed: '18e80b72d9ac' }, + { call: /\{ keepalive: true \}\) \.then\($/, body: /./, why: 'raw keepalive save: genAtSave against loadGeneration', reviewed: '18c06bb62bcd' }, + { call: /localDirty = false; \} \}\) \.catch\($/, body: /^\(\) => \{\}$/, why: 'raw keepalive failure: empty', reviewed: 'ad9468ea6d29' }, + { call: /reqItemId, \{ content: toSave \}\)\.then\($/, body: /./, why: 'raw foreground save: genAtSave against loadGeneration', reviewed: '6f098a1b5fc0' }, + { call: /content: item\.content \}\); \} \}\)\.catch\($/, body: /./, why: 'raw foreground failure: genAtSave against loadGeneration', reviewed: 'a5f127faeed6' }, ]; /** @@ -398,7 +398,10 @@ const HELPERS: Record = { identityHeld: '1c5505d51f73', navigateToCollectionRoot: '64dce75693d3', refreshPrintMeta: '2312cc481ca5', - runTeardownFlush: '8c6faa52b52e', + runTeardownFlush: '60ff925f14db', + // BUG-3124 unit B: the cursor-advance settle. Synchronous; arms the flusher's + // single timer only for the current, identity-held context. + settleCollabIfCurrent: '0d243d24a373', showSaved: '3f1ef91fac07', switchedAway: '8ba85844f406', withInflightTags: '11523b4d849a', diff --git a/web/src/lib/utils/sha256.test.ts b/web/src/lib/utils/sha256.test.ts new file mode 100644 index 000000000..134719ce3 --- /dev/null +++ b/web/src/lib/utils/sha256.test.ts @@ -0,0 +1,25 @@ +import { describe, expect, it } from 'vitest'; +import { sha256Hex } from './sha256'; + +// BUG-3124: the server compares this against Go's sha256.Sum256 over the UTF-8 +// bytes of items.content, so every vector here is a reference digest computed +// outside this code (sha256sum / Python hashlib), not a value this function +// produced. The padding boundaries (55/56/63/64/119 bytes) are where a SHA-256 +// implementation most often goes wrong. +describe('sha256Hex', () => { + it.each([ + ['', 'e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855'], + ['abc', 'ba7816bf8f01cfea414140de5dae2223b00361a396177a9cb410ff61f20015ad'], + ['a'.repeat(55), '9f4390f8d30c2dd92ec9f095b65e2b9ae9b0a925a5258e241c9f1e910f734318'], + ['a'.repeat(56), 'b35439a4ac6f0948b6d6f9e3c6af0f5f590ce20f1bde7090ef7970686ec6738a'], + ['a'.repeat(63), '7d3e74a05d7db15bce4ad9ec0658ea98e3f06eeecf16b4c6fff2da457ddc2f34'], + ['a'.repeat(64), 'ffe054fe7ae0cb6dc65c3af9b61d5209f439851db43d0ba5997337df154668eb'], + ['a'.repeat(119), '31eba51c313a5c08226adf18d4a359cfdfd8d2e816b13f4af952f7ea6584dcfb'], + ['a'.repeat(1000), '41edece42d63e8d9bf515a9ba6932e1c20cbc9f5a5d134645adb5db1b9737ea3'], + // Multi-byte UTF-8 (2-, 3- and 4-byte sequences): hashing UTF-16 code + // units instead of UTF-8 bytes would disagree with Go here. + ['héllo — 世界 🌍', 'e5cba050fccd5c0bb055ebef890b44f96a744a16e886fe39ef5d66e6dbe84362'], + ])('matches the reference digest for %j', (input, want) => { + expect(sha256Hex(input)).toBe(want); + }); +}); diff --git a/web/src/lib/utils/sha256.ts b/web/src/lib/utils/sha256.ts new file mode 100644 index 000000000..932b936ef --- /dev/null +++ b/web/src/lib/utils/sha256.ts @@ -0,0 +1,72 @@ +// Synchronous SHA-256 over the UTF-8 bytes of a string, as lowercase hex. +// +// Why not crypto.subtle.digest: it exists only in a SECURE context, and a +// self-hosted Pad is routinely served over plain http on a LAN address, where +// `crypto.subtle` is undefined. It is also async, and the collab watermark +// stamp (BUG-3124) is sent from a pagehide handler, where a request must be +// started before the handler returns. The server compares against Go's +// sha256.Sum256([]byte(items.content)), so the input here must be the same +// UTF-8 bytes — TextEncoder produces exactly that. + +const K = new Uint32Array([ + 0x428a2f98, 0x71374491, 0xb5c0fbcf, 0xe9b5dba5, 0x3956c25b, 0x59f111f1, 0x923f82a4, 0xab1c5ed5, + 0xd807aa98, 0x12835b01, 0x243185be, 0x550c7dc3, 0x72be5d74, 0x80deb1fe, 0x9bdc06a7, 0xc19bf174, + 0xe49b69c1, 0xefbe4786, 0x0fc19dc6, 0x240ca1cc, 0x2de92c6f, 0x4a7484aa, 0x5cb0a9dc, 0x76f988da, + 0x983e5152, 0xa831c66d, 0xb00327c8, 0xbf597fc7, 0xc6e00bf3, 0xd5a79147, 0x06ca6351, 0x14292967, + 0x27b70a85, 0x2e1b2138, 0x4d2c6dfc, 0x53380d13, 0x650a7354, 0x766a0abb, 0x81c2c92e, 0x92722c85, + 0xa2bfe8a1, 0xa81a664b, 0xc24b8b70, 0xc76c51a3, 0xd192e819, 0xd6990624, 0xf40e3585, 0x106aa070, + 0x19a4c116, 0x1e376c08, 0x2748774c, 0x34b0bcb5, 0x391c0cb3, 0x4ed8aa4a, 0x5b9cca4f, 0x682e6ff3, + 0x748f82ee, 0x78a5636f, 0x84c87814, 0x8cc70208, 0x90befffa, 0xa4506ceb, 0xbef9a3f7, 0xc67178f2, +]); + +export function sha256Hex(input: string): string { + const data = new TextEncoder().encode(input); + const bitLen = data.length * 8; + // Pad: 0x80, zeros, then the 64-bit big-endian bit length, to a multiple of 64 bytes. + const padded = new Uint8Array(((data.length + 9 + 63) >> 6) << 6); + padded.set(data); + padded[data.length] = 0x80; + const view = new DataView(padded.buffer); + view.setUint32(padded.length - 8, Math.floor(bitLen / 0x100000000)); + view.setUint32(padded.length - 4, bitLen >>> 0); + + const h = new Uint32Array([ + 0x6a09e667, 0xbb67ae85, 0x3c6ef372, 0xa54ff53a, 0x510e527f, 0x9b05688c, 0x1f83d9ab, 0x5be0cd19, + ]); + const w = new Uint32Array(64); + const rotr = (x: number, n: number) => (x >>> n) | (x << (32 - n)); + for (let off = 0; off < padded.length; off += 64) { + for (let i = 0; i < 16; i++) w[i] = view.getUint32(off + i * 4); + for (let i = 16; i < 64; i++) { + const s0 = rotr(w[i - 15], 7) ^ rotr(w[i - 15], 18) ^ (w[i - 15] >>> 3); + const s1 = rotr(w[i - 2], 17) ^ rotr(w[i - 2], 19) ^ (w[i - 2] >>> 10); + w[i] = (w[i - 16] + s0 + w[i - 7] + s1) >>> 0; + } + let [a, b, c, d, e, f, g, hh] = h; + for (let i = 0; i < 64; i++) { + const S1 = rotr(e, 6) ^ rotr(e, 11) ^ rotr(e, 25); + const ch = (e & f) ^ (~e & g); + const t1 = (hh + S1 + ch + K[i] + w[i]) >>> 0; + const S0 = rotr(a, 2) ^ rotr(a, 13) ^ rotr(a, 22); + const maj = (a & b) ^ (a & c) ^ (b & c); + const t2 = (S0 + maj) >>> 0; + hh = g; + g = f; + f = e; + e = (d + t1) >>> 0; + d = c; + c = b; + b = a; + a = (t1 + t2) >>> 0; + } + h[0] = (h[0] + a) >>> 0; + h[1] = (h[1] + b) >>> 0; + h[2] = (h[2] + c) >>> 0; + h[3] = (h[3] + d) >>> 0; + h[4] = (h[4] + e) >>> 0; + h[5] = (h[5] + f) >>> 0; + h[6] = (h[6] + g) >>> 0; + h[7] = (h[7] + hh) >>> 0; + } + return Array.from(h, (x) => x.toString(16).padStart(8, '0')).join(''); +}