amadan

eleven / messenger Public

web: full-history backfill into the local store — resumable, cancellable (#457) (#777)

store.js grows the backfill layer the search substrate needs (#586):

- runBackfill(threadId, fetchPage, {onPage, signal}) pages history backwards
  with the server's before cursor, persisting an oldest-fetched watermark per
  thread (kv thread:<tid>) so an interrupted walk RESUMES rather than
  restarts; an empty page marks the thread complete — never re-walked,
  forward sync keeps coverage total from then on. Transport is injected, so
  the driver holds no token and is E2E-testable by module import.
- oldestStoredId / getBackfillMark / backfillCursor (pure, Node-tested) +
  allMessages(threadId) — the search scan's input.
- clearThread already removes the watermark key; sign-out wipes everything.

Rows are stored, never folded — display paths are untouched. Known accepted
limit (commented): a gap ABOVE the walk's starting row (thread outran the
cached window between sessions) isn't detected; revisit with interval
tracking if it bites.

E2E (Playwright, real protocol, 450-message thread): cancel-mid-run persists
the watermark; resume continues exactly there (call trace pinned); complete
mark short-circuits future runs to zero fetches; all 450 rows land locally.
Co-authored-by: Claude Fable 5 <noreply@anthropic.com>
Paul Campbell 0dfc3c27e8c64b7575eb5377528b89888f4adf85 parent a0f064f
2 files changed, +111 −1
  • test/store.test.mjs +12 −1
  • web/static/store.js +99 −0
diff --git a/test/store.test.mjs b/test/store.test.mjs
index c1591fd..03ad5c8 100644
--- a/test/store.test.mjs
+++ b/test/store.test.mjs
@@ -4,7 +4,7 @@
// a fetched page translate to store writes, and which cached threads must be
// pruned against a live thread list.
import assert from "assert";
-import { SNAP_VERSION, snapshotUsable, partitionRows, staleThreadIds } from "../web/static/store.js";
+import { SNAP_VERSION, snapshotUsable, partitionRows, staleThreadIds, backfillCursor } from "../web/static/store.js";
function test(name, fn) {
try {
@@ -73,4 +73,15 @@ test("staleThreadIds never prunes on a malformed live list (an error body must n
assert.deepStrictEqual(staleThreadIds(["a"], { threads: [] }), ["a"]);
});
+test("backfillCursor: resumes from the persisted watermark, not from scratch", () => {
+ assert.strictEqual(backfillCursor({ oldest: 42, complete: false }, 900), 42, "interrupted walk continues at its mark");
+ assert.strictEqual(backfillCursor(null, 900), 900, "first walk starts at the oldest stored row");
+ assert.strictEqual(backfillCursor(null, 0), 0, "empty store → seed a recent page first");
+});
+
+test("backfillCursor: a completed thread is never re-walked", () => {
+ assert.strictEqual(backfillCursor({ oldest: 1, complete: true }, 900), -1);
+ assert.strictEqual(backfillCursor({ oldest: 0, complete: true }, 0), -1, "empty-thread completion holds too");
+});
+
console.log("\nALL PASSED");
diff --git a/web/static/store.js b/web/static/store.js
index 1802e8d..627080d 100644
--- a/web/static/store.js
+++ b/web/static/store.js
@@ -227,6 +227,105 @@ export function cachedThreadIds() {
}, []);
}
+// ---- full-history backfill (#457) -----------------------------------------
+// Page a thread's history backwards into the store with the server's `before`
+// cursor, persisting an oldest-fetched watermark per thread so an interrupted
+// run RESUMES rather than restarts. A thread that reaches its beginning (an
+// empty page) is marked complete and never re-walked — from then on the
+// ordinary forward sync keeps coverage total. Rows are stored, never folded:
+// this is the search substrate (#586), not a display path.
+//
+// Known limit, accepted for now: the watermark only ever descends from the
+// oldest row present when the walk started. A gap ABOVE that point (a thread
+// that outran the cached window between sessions) isn't detected — those
+// messages exist server-side but not locally until a fresh walk. Fine for
+// search-completeness at this product's scale; revisit with interval tracking
+// if it ever bites.
+
+// Where a thread's backfill has reached. null = never started.
+export function getBackfillMark(threadId) {
+ return tx(["kv"], "readonly", (t, done) => {
+ const req = t.objectStore("kv").get("thread:" + threadId);
+ req.onsuccess = () => done(req.result || null);
+ }, null);
+}
+
+function setBackfillMark(threadId, mark) {
+ return tx(["kv"], "readwrite", (t) => {
+ t.objectStore("kv").put(mark, "thread:" + threadId);
+ });
+}
+
+// The lowest stored id for a thread — where a first walk starts from.
+export function oldestStoredId(threadId) {
+ return tx(["msgs"], "readonly", (t, done) => {
+ const range = IDBKeyRange.bound([threadId, 0], [threadId, Infinity]);
+ const cur = t.objectStore("msgs").openKeyCursor(range, "next");
+ cur.onsuccess = () => done(cur.result ? cur.result.key[1] : 0);
+ }, 0);
+}
+
+// Pure: the `before` cursor a run should continue from. -1 = already complete;
+// 0 = nothing local yet (seed a recent page first).
+export function backfillCursor(mark, oldestStored) {
+ if (mark && mark.complete) return -1;
+ if (mark && mark.oldest) return mark.oldest;
+ return oldestStored || 0;
+}
+
+// Drive a thread's backfill to completion (or cancellation). Transport is
+// injected — fetchPage(before) resolves the raw rows of one page (before=0 ⇒
+// the most recent page) — so this stays free of tokens and testable end to
+// end by importing the module. onPage({oldest, oldestAt, count}) narrates
+// progress ("searching back to 2024…"); an AbortSignal cancels between pages.
+export async function runBackfill(threadId, fetchPage, { onPage, signal } = {}) {
+ let cursor = backfillCursor(await getBackfillMark(threadId), await oldestStoredId(threadId));
+ if (cursor === -1) return { complete: true };
+ if (!cursor) {
+ const rows = await fetchPage(0);
+ if (!rows.length) {
+ // A thread with no history at all is trivially complete.
+ await setBackfillMark(threadId, { oldest: 0, complete: true });
+ return { complete: true };
+ }
+ await putMessages(threadId, rows);
+ cursor = rows[0].id;
+ await setBackfillMark(threadId, { oldest: cursor, complete: false });
+ }
+ while (!(signal && signal.aborted)) {
+ const rows = await fetchPage(cursor);
+ if (!rows.length) {
+ await setBackfillMark(threadId, { oldest: cursor, complete: true });
+ return { complete: true };
+ }
+ await putMessages(threadId, rows);
+ cursor = rows[0].id; // pages come ascending; the first row is the oldest
+ await setBackfillMark(threadId, { oldest: cursor, complete: false });
+ if (onPage) onPage({ oldest: cursor, oldestAt: rows[0].created_at || 0, count: rows.length });
+ }
+ return { complete: false };
+}
+
+// Every stored row for a thread, ascending — the search scan's input. Bounded
+// by cap as a runaway guard; search streams per thread, so a giant thread
+// still yields in slices.
+export function allMessages(threadId, cap = 50000) {
+ return tx(["msgs"], "readonly", (t, done) => {
+ const rows = [];
+ const range = IDBKeyRange.bound([threadId, 0], [threadId, Infinity]);
+ const cur = t.objectStore("msgs").openCursor(range, "next");
+ cur.onsuccess = () => {
+ const c = cur.result;
+ if (c && rows.length < cap) {
+ rows.push(c.value);
+ c.continue();
+ } else {
+ done(rows);
+ }
+ };
+ }, []);
+}
+
// Sign-out: everything goes, snapshot included — the next visitor to this
// browser profile must find nothing.
export function wipeStore() {