From 1e95113464539f287013c4234d917ca4844f896e Mon Sep 17 00:00:00 2001 From: Keaton Hoskins Date: Fri, 9 Oct 2026 21:04:16 -0700 Subject: [PATCH 1/2] feat(sync): prune seen txns below a per-origin local_seq floor, refuse loudly below it (T13318) Proof first (code reading on the stack top, tests below): an origin's transactions are FIRST delivered in local_seq order on every path. The sealer mints local_seq rising (never reused, T13033); buildSegment packs the lowest sealed txns of its own replica in order and stops at a partial group; push uploads lowest replicaSeq first, resending the same bytes; the server and every puller refuse a replicaSeq gap or replay. A raced or crashed genesis cut returns folded txns to sealed only while genesis_pending blocks push. Rebind and copy never re-send inherited txn ids (spec H3); revival replays from the local inbox and a re-emit is a new txn. A retired replica's late segment is still contiguous in its own replicaSeq. - _sync_seen_floor (new local-only journal table): per stream and origin the highest local_seq staged, pruned_upto, and the seen row/byte counts; the migration seeds it from rows already staged. - A txn at or below its origin's floor with no seen row refuses its segment (refusedKind 'below-floor', naming pruned vs out of order); the pull stops before it and never skips it as seen. - A segment carrying a txn that is not its replica's is refused. - pruneSeenTxns drops each origin's rows below its floor (keeping the floor's own row); the cloud pull prunes after every pull. pullStream's unused stream-seq pruneSeenUpTo is replaced by pruneSeen. - seenTxnReport sums the per-origin counts: O(origins), not O(rows) (review-hotfix LOW on #1972). - Vault bundles clear _sync_seen_floor with _sync_seen_txn and the cursor. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01VjjTKbQXyCFdkLBSeuyKLt --- .changeset/t13318-seen-floor.md | 21 ++ packages/contracts/src/nexus-cloud.ts | 6 +- .../migration.sql | 41 +++ .../cloud/__tests__/cloud-sync-status.test.ts | 25 +- .../src/cloud/__tests__/nexus-attach.test.ts | 1 + packages/core/src/cloud/nexus-vault.ts | 15 +- .../table-classification-gate.test.ts | 3 + .../src/store/sync/__tests__/pull.test.ts | 176 ++++++++++- packages/core/src/store/sync/pull.ts | 298 +++++++++++++----- .../core/src/store/table-classification.ts | 12 + scripts/lint-no-raw-table-writes.mjs | 7 +- 11 files changed, 496 insertions(+), 109 deletions(-) create mode 100644 .changeset/t13318-seen-floor.md create mode 100644 packages/core/migrations/sync-journal/20261009120000_t13318-seen-floor/migration.sql diff --git a/.changeset/t13318-seen-floor.md b/.changeset/t13318-seen-floor.md new file mode 100644 index 0000000000..f8d95026d0 --- /dev/null +++ b/.changeset/t13318-seen-floor.md @@ -0,0 +1,21 @@ +--- +id: t13318-seen-floor +tasks: [T13318] +kind: feat +summary: "The cloud pull prunes its seen-transaction ledger below a per-origin local_seq floor, and refuses loudly below it" +--- + +A transaction id is `:`, and every origin's transactions are first delivered in `local_seq` order: +the sealer mints `local_seq` rising, a segment packs the lowest sealed transactions in order, uploads go out lowest +`replicaSeq` first, and every puller refuses a `replicaSeq` gap or replay. The pull now relies on that. + +- A new local-only journal table, `_sync_seen_floor`, keeps per stream and origin the highest `local_seq` staged, how far + its seen rows were pruned (`pruned_upto`), and the row and byte counts. Its migration seeds it from rows already staged. +- A transaction at or below its origin's floor with no seen row is refused with `refusedKind: 'below-floor'`, and the pull + stops before its segment. The message says whether its seen row was pruned or it arrived out of order. It is never + skipped as seen. +- A segment carrying a transaction that is not its replica's is refused (`refusedKind: 'segment'`). +- The cloud pull prunes each origin's seen rows below its floor after every pull, keeping the floor's own row. + `pullStream`'s `pruneSeenUpTo` (a stream-seq floor no caller used) is replaced by `pruneSeen`. +- `cleo cloud status` sums the per-origin counts instead of walking the whole ledger. +- Vault bundles clear `_sync_seen_floor` together with `_sync_seen_txn` and `_sync_cursor`. diff --git a/packages/contracts/src/nexus-cloud.ts b/packages/contracts/src/nexus-cloud.ts index 5cb81e5f62..4650826ad2 100644 --- a/packages/contracts/src/nexus-cloud.ts +++ b/packages/contracts/src/nexus-cloud.ts @@ -615,9 +615,9 @@ export interface CloudStatusUndoBudget { } /** - * The pull's seen-transaction ledger (`_sync_seen_txn`, T13317). Nothing - * prunes it yet: a seen txn can return in a new segment above any checkpoint, - * so pruning waits for a proven refusal floor (T13318). + * The pull's seen-transaction ledger (`_sync_seen_txn`, T13317). The cloud + * pull prunes each origin's rows below its staged `local_seq` floor (T13318); + * a pruned transaction delivered again is refused, never applied twice. */ export interface CloudStatusSeenTxns { /** Rows held, every stream. */ diff --git a/packages/core/migrations/sync-journal/20261009120000_t13318-seen-floor/migration.sql b/packages/core/migrations/sync-journal/20261009120000_t13318-seen-floor/migration.sql new file mode 100644 index 0000000000..3b106ac09b --- /dev/null +++ b/packages/core/migrations/sync-journal/20261009120000_t13318-seen-floor/migration.sql @@ -0,0 +1,41 @@ +-- T13318 (journal spec §3.1 "Receive") — a per-origin floor for the +-- seen-transaction ledger. +-- +-- A transaction id is `:`, and every origin's +-- transactions are FIRST delivered in local_seq order (the sealer mints +-- local_seq rising, a segment packs the lowest sealed ones in order, and +-- pullers refuse a replicaSeq gap or replay). So per (stream, origin): +-- +-- staged_upto: the highest local_seq staged. A transaction at or below it +-- that has no _sync_seen_txn row was either pruned or arrived out of order; +-- the pull refuses it loudly, never skips it as seen. +-- pruned_upto: seen rows at or below it were dropped (0: none), so a refusal +-- can say which of the two it is. +-- seen_rows, seen_bytes: the origin's _sync_seen_txn rows and their payload +-- bytes, kept with every insert and prune, so `cleo cloud status` sums this +-- table instead of walking the ledger. +-- +-- The backfill seeds the floor from rows staged before this table existed. +-- +-- Local-only. +-- +-- Do not edit this file once applied: its hash is recorded per store. +CREATE TABLE IF NOT EXISTS _sync_seen_floor ( + stream TEXT NOT NULL, + origin TEXT NOT NULL, + staged_upto INTEGER NOT NULL, + pruned_upto INTEGER NOT NULL DEFAULT 0, + seen_rows INTEGER NOT NULL DEFAULT 0, + seen_bytes INTEGER NOT NULL DEFAULT 0, + PRIMARY KEY (stream, origin) +) WITHOUT ROWID; +INSERT OR IGNORE INTO _sync_seen_floor (stream, origin, staged_upto, pruned_upto, seen_rows, seen_bytes) +SELECT stream, + substr(txn, 1, instr(txn, ':') - 1), + max(CAST(substr(txn, instr(txn, ':') + 1) AS INTEGER)), + 0, + count(*), + sum(octet_length(stream) + octet_length(txn) + 8) + FROM _sync_seen_txn + WHERE instr(txn, ':') > 1 + GROUP BY stream, substr(txn, 1, instr(txn, ':') - 1); diff --git a/packages/core/src/cloud/__tests__/cloud-sync-status.test.ts b/packages/core/src/cloud/__tests__/cloud-sync-status.test.ts index fc0e759025..a2a4ef9638 100644 --- a/packages/core/src/cloud/__tests__/cloud-sync-status.test.ts +++ b/packages/core/src/cloud/__tests__/cloud-sync-status.test.ts @@ -170,19 +170,26 @@ describe('cloud status sync block (T12998)', () => { expect(project?.quarantined).toEqual({ tasks_tasks: 1 }); }); - it('reports the seen-txn ledger per stream with its estimated bytes; nothing prunes it (T13317)', async () => { + it('reports the seen-txn ledger per stream from the per-origin counts the pull keeps (T13317, T13318)', async () => { const db = await store(true); - const ins = db.prepare('INSERT INTO _sync_seen_txn (stream, txn, seq) VALUES (?, ?, ?)'); - ins.run('project:p1', 'r1:1', 1); - ins.run('project:p1', 'r1:2', 2); - ins.run('home:u1', 'r2:1', 1); + // stream + txn text + an 8-byte seq per row, as the pull counts them. + const row = (stream: string, txn: string) => stream.length + txn.length + 8; + const floor = db.prepare( + 'INSERT INTO _sync_seen_floor (stream, origin, staged_upto, seen_rows, seen_bytes) VALUES (?, ?, ?, ?, ?)', + ); + floor.run('project:p1', 'r1', 2, 2, row('project:p1', 'r1:1') + row('project:p1', 'r1:2')); + floor.run('project:p1', 'r3', 7, 0, 0); // fully pruned: no rows, no stream entry of its own + floor.run('home:u1', 'r2', 1, 1, row('home:u1', 'r2:1')); + // The ledger itself is never walked: a stray row with no count is not reported. + db.prepare('INSERT INTO _sync_seen_txn (stream, txn, seq) VALUES (?, ?, ?)').run( + 'project:p9', + 'r9:1', + 1, + ); const { project } = await projectBlock(); - // stream + txn text + an 8-byte seq per row. - const bytes = - 2 * ('project:p1'.length + 'r1:1'.length + 8) + ('home:u1'.length + 'r2:1'.length + 8); expect(project?.seenTxns).toEqual({ rows: 3, - bytes, + bytes: 2 * row('project:p1', 'r1:1') + row('home:u1', 'r2:1'), byStream: { 'home:u1': 1, 'project:p1': 2 }, }); // A read never prunes: the next read sees the same rows. diff --git a/packages/core/src/cloud/__tests__/nexus-attach.test.ts b/packages/core/src/cloud/__tests__/nexus-attach.test.ts index 780a40a5e1..2e9e3d49c6 100644 --- a/packages/core/src/cloud/__tests__/nexus-attach.test.ts +++ b/packages/core/src/cloud/__tests__/nexus-attach.test.ts @@ -319,6 +319,7 @@ describe('presence and binding hold their guarantees (review M3)', () => { '_sync_replica', '_sync_row_meta', '_sync_row_undo', + '_sync_seen_floor', '_sync_seen_txn', '_sync_segment', '_sync_segment_txn', diff --git a/packages/core/src/cloud/nexus-vault.ts b/packages/core/src/cloud/nexus-vault.ts index 49a28e85f6..a3d996683a 100644 --- a/packages/core/src/cloud/nexus-vault.ts +++ b/packages/core/src/cloud/nexus-vault.ts @@ -826,6 +826,9 @@ const VAULT_CLEARED_JOURNAL_TABLES = [ // from the checkpoint it restored, never from another device's position. '_sync_cursor', '_sync_seen_txn', + // Its floor goes with it: a floor without its seen rows would refuse + // every transaction below it (T13318). + '_sync_seen_floor', ] as const; /** @@ -1981,18 +1984,16 @@ async function pullWithSession(session: StreamSession): Promise { const page = await journal.pull( { diff --git a/packages/core/src/store/__tests__/table-classification-gate.test.ts b/packages/core/src/store/__tests__/table-classification-gate.test.ts index 47f983574e..bc4d6500cc 100644 --- a/packages/core/src/store/__tests__/table-classification-gate.test.ts +++ b/packages/core/src/store/__tests__/table-classification-gate.test.ts @@ -201,6 +201,8 @@ const SYNC_SEGMENT_DDL = 'packages/core/migrations/sync-journal/20261007120000_t12343-segment-outbox/migration.sql'; const SYNC_PULL_DDL = 'packages/core/migrations/sync-journal/20261008120000_t12343-pull-cursor/migration.sql'; +const SYNC_SEEN_FLOOR_DDL = + 'packages/core/migrations/sync-journal/20261009120000_t13318-seen-floor/migration.sql'; const SYNC_JOURNAL_TABLES = { _sync_capture: { class: 'local-only', ddl: SYNC_CAPTURE_DDL }, _sync_frame: { class: 'local-only', ddl: SYNC_CAPTURE_DDL }, @@ -221,6 +223,7 @@ const SYNC_JOURNAL_TABLES = { _sync_segment_txn: { class: 'local-only', ddl: SYNC_SEGMENT_DDL }, _sync_cursor: { class: 'local-only', ddl: SYNC_PULL_DDL }, _sync_seen_txn: { class: 'local-only', ddl: SYNC_PULL_DDL }, + _sync_seen_floor: { class: 'local-only', ddl: SYNC_SEEN_FLOOR_DDL }, _sync_clock: { class: 'local-only', ddl: SYNC_JOURNAL_DDL }, _sync_meta: { class: 'local-only', ddl: SYNC_JOURNAL_DDL }, _sync_replica: { class: 'local-only', ddl: SYNC_JOURNAL_DDL }, diff --git a/packages/core/src/store/sync/__tests__/pull.test.ts b/packages/core/src/store/sync/__tests__/pull.test.ts index ebca26a7c7..fd8f66390b 100644 --- a/packages/core/src/store/sync/__tests__/pull.test.ts +++ b/packages/core/src/store/sync/__tests__/pull.test.ts @@ -8,7 +8,7 @@ * @task T12343 */ -import { mkdirSync, mkdtempSync, rmSync } from 'node:fs'; +import { cpSync, mkdirSync, mkdtempSync, readdirSync, rmSync } from 'node:fs'; import { tmpdir } from 'node:os'; import { join, resolve } from 'node:path'; import type { DatabaseSync } from 'node:sqlite'; @@ -28,6 +28,7 @@ import { readStreamCursor, type StreamCursor, } from '../pull.js'; +import { ensureSyncSchema } from '../schema.js'; import { sealPending } from '../sealer.js'; import { buildSegment } from '../segments.js'; import { firstBadTxnSignature, signTxn } from '../txn-signing.js'; @@ -36,6 +37,9 @@ const SYNC_SCHEMA = resolve(import.meta.dirname, '../../../../migrations/sync-jo const STREAM = 'project:0192ffff-7f00-7000-8000-00000000000f'; const RA = '0192aaaa-7f00-7000-8000-00000000000a'; const RB = '0192bbbb-7f00-7000-8000-00000000000b'; +const RC = '0192cccc-7f00-7000-8000-00000000000c'; +/** A's successor after a rebind: a new replica id, the store's local_seq counter continues. */ +const RA2 = '0192aaaa-7f00-7000-8000-0000000000a2'; const DEV_A = 'dev-a'; const KEY_A = generateEd25519(); let clock = Date.now(); @@ -76,23 +80,23 @@ const seal = (db: DatabaseSync, replica: string) => () => { }); }; -function write(db: DatabaseSync, sql: string): void { +function write(db: DatabaseSync, sql: string, replica = RA): void { db.exec('BEGIN IMMEDIATE'); const frame = openCaptureFrame(db, 'write', null); db.exec(sql); finishCaptureFrame(db, frame); db.exec('COMMIT'); - seal(db, RA)(); + seal(db, replica)(); } const addTask = (id: string) => `INSERT INTO tasks_tasks (id, title, type, status, priority, uid, birth_fp) VALUES ('${id}', 'title ${id}', 'task', 'pending', 'medium', 'uid-${id}', 'fp-${id}')`; /** A's next segment, signed by A's device, as the stream serves it (the sealer is identity here). */ -function authorSegment(a: DatabaseSync): Uint8Array { +function authorSegment(a: DatabaseSync, replica = RA): Uint8Array { const seg = buildSegment(a, { stream: STREAM, - replica: RA, + replica, scope: 'project', project: null, sealer: (_seq, plaintext) => Buffer.from(plaintext), @@ -246,30 +250,170 @@ describe('pullStream (S5-1)', () => { expect(n(b, "SELECT count(*) AS n FROM tasks_tasks WHERE id = 'T1'")).toBe(1); }); - it('seen-txn rows at or below the prune floor are dropped, later ones kept', async () => { + it('refuses with sync.pull off: nothing pulled, nothing staged', async () => { const a = await store('a'); const b = await store('b'); const stream = fakeStream(); write(a, addTask('T1')); stream.append(authorSegment(a), 0); + setSyncFlag(b, 'sync.pull', false, { schemaRoot: SYNC_SCHEMA }); + const r = await pullStream(b, pullOpts(b, stream)); + expect(r).toMatchObject({ refused: 'sync.pull is off', segments: 0, apply: null }); + expect(stream.pages).toBe(0); + expect(n(b, 'SELECT count(*) AS n FROM _sync_inbox')).toBe(0); + }); +}); + +describe('per-origin seen floor (T13318)', () => { + const floorOf = (db: DatabaseSync, origin: string) => + db + .prepare( + 'SELECT staged_upto AS stagedUpto, pruned_upto AS prunedUpto, seen_rows AS rows FROM _sync_seen_floor WHERE stream = ? AND origin = ?', + ) + .get(STREAM, origin); + + it('a re-delivered txn below the floor is skipped while its seen row exists, and refused loudly once pruned', async () => { + const a = await store('a'); + const b = await store('b'); + const stream = fakeStream(); + write(a, addTask('T1')); + const first = authorSegment(a); // RA:1 + stream.append(first, 0); write(a, addTask('T2')); - stream.append(authorSegment(a), 1); + stream.append(authorSegment(a), 1); // RA:2 + write(a, addTask('T3')); + const third = authorSegment(a); // RA:3 + stream.append(third, 2); await pullStream(b, pullOpts(b, stream)); - expect(n(b, 'SELECT count(*) AS n FROM _sync_seen_txn')).toBe(2); - await pullStream(b, { ...pullOpts(b, stream), pruneSeenUpTo: 1 }); - expect(b.prepare('SELECT seq FROM _sync_seen_txn').all()).toEqual([{ seq: 2 }]); + expect(floorOf(b, RA)).toEqual({ stagedUpto: 3, prunedUpto: 0, rows: 3 }); + + // Below the floor, row present: skipped as seen. + stream.append(first, 3); + const seen = await pullStream(b, { ...pullOpts(b, stream), pruneSeen: true }); + expect(seen).toMatchObject({ refused: null, staged: 0, redelivered: 1, after: 4 }); + // The prune drops every row below the floor and keeps the floor's own. + expect(floorOf(b, RA)).toEqual({ stagedUpto: 3, prunedUpto: 2, rows: 1 }); + expect(b.prepare('SELECT txn FROM _sync_seen_txn').all()).toEqual([{ txn: `${RA}:3` }]); + + // The latest txn is still skipped quietly; a pruned one is refused, never staged. + stream.append(third, 4); + stream.append(first, 5); + const r = await pullStream(b, pullOpts(b, stream)); + expect(r.refusedKind).toBe('below-floor'); + expect(r.refused).toMatch( + new RegExp(`segment 6 of replica ${RA}: transaction ${RA}:1 .*pruned \\(pruned_upto 2\\)`), + ); + expect(r).toMatchObject({ segments: 1, staged: 0, redelivered: 1, after: 5 }); + expect(readStreamCursor(b, STREAM)).toMatchObject({ after: 5 }); + expect(n(b, 'SELECT count(*) AS n FROM _sync_inbox')).toBe(3); + expect(n(b, "SELECT count(*) AS n FROM tasks_tasks WHERE id = 'T1'")).toBe(1); + // The refusal holds on retry: the stream stops there until someone looks. + const retry = await pullStream(b, pullOpts(b, stream)); + expect(retry).toMatchObject({ refusedKind: 'below-floor', segments: 0, after: 5 }); }); - it('refuses with sync.pull off: nothing pulled, nothing staged', async () => { + it('a txn below the floor that was never staged is refused as out of order, never skipped', async () => { const a = await store('a'); const b = await store('b'); const stream = fakeStream(); write(a, addTask('T1')); - stream.append(authorSegment(a), 0); - setSyncFlag(b, 'sync.pull', false, { schemaRoot: SYNC_SCHEMA }); + const early = authorSegment(a); // RA:1, held back + write(a, addTask('T2')); + stream.append(authorSegment(a), 0); // RA:2 first + stream.append(early, 1); const r = await pullStream(b, pullOpts(b, stream)); - expect(r).toMatchObject({ refused: 'sync.pull is off', segments: 0, apply: null }); - expect(stream.pages).toBe(0); - expect(n(b, 'SELECT count(*) AS n FROM _sync_inbox')).toBe(0); + expect(r.refusedKind).toBe('below-floor'); + expect(r.refused).toMatch(/transaction .*:1 is at or below .*floor 2.*out of local_seq order/); + expect(r).toMatchObject({ segments: 1, staged: 1, after: 1 }); + expect(n(b, "SELECT count(*) AS n FROM tasks_tasks WHERE id = 'T1'")).toBe(0); + expect(n(b, 'SELECT count(*) AS n FROM _sync_seen_txn')).toBe(1); + }); + + it("the migration seeds each origin's floor and counts from rows staged before it existed", async () => { + // A journal at the folder before the floor, with seen rows already staged. + const older = join(dir, 'older-schema'); + for (const f of readdirSync(SYNC_SCHEMA).filter((f) => f < '20261009120000')) { + cpSync(join(SYNC_SCHEMA, f), join(older, f), { recursive: true }); + } + mkdirSync(join(dir, 'old', '.cleo'), { recursive: true }); + const db = getDualScopeNativeDb( + await openDualScopeDbAtPath('project', join(dir, 'old', '.cleo', 'cleo.db')), + ); + ensureSyncSchema(db, { root: older }); + const ins = db.prepare('INSERT INTO _sync_seen_txn (stream, txn, seq) VALUES (?, ?, ?)'); + ins.run(STREAM, `${RA}:1`, 1); + ins.run(STREAM, `${RA}:9`, 2); + ins.run(STREAM, `${RC}:4`, 3); + ensureSyncSchema(db, { root: SYNC_SCHEMA }); + const row = (txn: string) => STREAM.length + txn.length + 8; + expect(floorOf(db, RA)).toEqual({ stagedUpto: 9, prunedUpto: 0, rows: 2 }); + expect(floorOf(db, RC)).toEqual({ stagedUpto: 4, prunedUpto: 0, rows: 1 }); + expect(db.prepare('SELECT sum(seen_bytes) AS b FROM _sync_seen_floor').get()).toEqual({ + b: row(`${RA}:1`) + row(`${RA}:9`) + row(`${RC}:4`), + }); + }); + + it("a transaction that is not the segment replica's is refused", async () => { + const a = await store('a'); + const b = await store('b'); + const stream = fakeStream(); + write(a, addTask('T1')); + stream.append(authorSegment(a), 0, RC); // RA's txn served as replica RC's + const r = await pullStream(b, pullOpts(b, stream)); + expect(r.refusedKind).toBe('segment'); + expect(r.refused).toMatch(new RegExp(`transaction ${RA}:1 is not one of its transactions`)); + expect(r).toMatchObject({ staged: 0, after: 0 }); + }); + + it("rebind proof case: a retired replica's late segment and its successor keep separate floors; nothing is refused", async () => { + const a = await store('a'); + const b = await store('b'); + const stream = fakeStream(); + write(a, addTask('T1')); + const old0 = authorSegment(a); // RA:1 + write(a, addTask('T2')); + const late = authorSegment(a); // RA:2: pushed before the rebind, sequenced after + // The rebind: a new replica id; the store's local_seq counter continues. + write(a, addTask('T3'), RA2); + const succ = authorSegment(a, RA2); + stream.append(old0, 0, RA); + stream.append(succ, 0, RA2); + stream.append(late, 1, RA); + const r = await pullStream(b, pullOpts(b, stream)); + expect(r).toMatchObject({ refused: null, staged: 3, after: 3 }); + expect(floorOf(b, RA)).toMatchObject({ stagedUpto: 2 }); + expect(floorOf(b, RA2)).toMatchObject({ stagedUpto: 3 }); + expect(n(b, "SELECT count(*) AS n FROM tasks_tasks WHERE id IN ('T1', 'T2', 'T3')")).toBe(3); + }); + + it("revive proof case: a voided txn stays seen; the author's later txn is staged above the floor; re-delivery of the void is skipped", async () => { + const a = await store('a'); + const b = await store('b'); + const c = await store('c'); + const stream = fakeStream(); + write(a, addTask('T1')); // RA:1 + stream.append(authorSegment(a), 0); + await pullStream(c, { ...pullOpts(c, stream), replica: RC, seal: seal(c, RC) }); + write(a, "DELETE FROM tasks_tasks WHERE id = 'T1'"); // RA:2 + stream.append(authorSegment(a), 1); + await new Promise((r) => setTimeout(r, 5)); // C's edit is newer than A's delete + write(c, "UPDATE tasks_tasks SET priority = 'high' WHERE id = 'T1'", RC); // RC:1 + const voided = authorSegment(c, RC); + stream.append(voided, 0, RC); + await pullStream(b, pullOpts(b, stream)); + expect(b.prepare('SELECT status FROM _sync_inbox WHERE replica_id = ?').all(RC)).toEqual([ + { status: 'void' }, + ]); + expect(floorOf(b, RC)).toMatchObject({ stagedUpto: 1 }); + // The author writes again (a re-emit is a new txn, `reemitOf`, spec §2.11 §6). + write(c, addTask('T4'), RC); // RC:2 + stream.append(authorSegment(c, RC), 1, RC); + stream.append(voided, 2, RC); + const r = await pullStream(b, { ...pullOpts(b, stream), pruneSeen: true }); + expect(r).toMatchObject({ refused: null, staged: 1, redelivered: 1 }); + expect(floorOf(b, RC)).toMatchObject({ stagedUpto: 2, prunedUpto: 1 }); + expect(b.prepare('SELECT count(*) AS n FROM _sync_inbox WHERE replica_id = ?').get(RC)).toEqual( + { n: 2 }, + ); }); }); diff --git a/packages/core/src/store/sync/pull.ts b/packages/core/src/store/sync/pull.ts index 3f5f538664..ae7ab3364a 100644 --- a/packages/core/src/store/sync/pull.ts +++ b/packages/core/src/store/sync/pull.ts @@ -17,6 +17,11 @@ * - skips every transaction id already staged from this stream * (`_sync_seen_txn`): a re-delivered transaction is never staged twice, so * a counter delta is never applied twice; + * - refuses a transaction at or below its origin's floor that has no seen + * row (`_sync_seen_floor`, T13318): an origin's transactions are first + * delivered in `local_seq` order, so such a transaction was either pruned + * from the seen ledger or arrived out of order; the pull stops there, loud, + * and never skips it as seen; * - stages the rest (`stageTxns`), schema-ahead ones included (the applier * holds them `refused-schema` and replays them after an upgrade); * - advances the cursor. @@ -39,7 +44,7 @@ import { type ApplyReport, type ApplyStagedOptions, applyStagedTxns } from './ap import { withImmediateTransaction } from './clock-store.js'; import { isSyncFlagOn } from './flags.js'; import { stageTxns } from './inbox.js'; -import { hasTable } from './schema.js'; +import { ensureSyncSchema, hasTable } from './schema.js'; /** Where a pull stands (the journal client's `PullCursor`). */ export interface StreamCursor { @@ -95,13 +100,25 @@ export interface PullStreamOptions { /** Environment for the `sync.pull` kill switch. @defaultValue process.env */ readonly env?: NodeJS.ProcessEnv; /** - * Drop seen-txn rows first staged at or below this stream seq, or null to - * keep every row. Only a floor the applier refuses below is safe - * ({@link pruneSeenTxns}); the cloud pull passes none yet (T13256). + * After staging, drop each origin's seen-txn rows below its staged floor + * ({@link pruneSeenTxns}, T13318). A pruned transaction delivered again is + * refused (`below-floor`), never applied twice. @defaultValue false */ - readonly pruneSeenUpTo?: number | null; + readonly pruneSeen?: boolean; } +/** + * Why {@link pullStream} stopped short, as a kind callers branch on (the + * message in `refused` is for people and may be reworded): + * - `pull-off`: `sync.pull` is off, nothing was pulled; + * - `segment`: a segment is malformed, carries a transaction its device did + * not sign, or a transaction that is not its replica's; + * - `below-floor`: a segment carries a transaction at or below its origin's + * staged floor with no seen row: pruned from the seen ledger, or delivered + * out of `local_seq` order (T13318). + */ +export type PullRefusal = 'pull-off' | 'segment' | 'below-floor'; + /** What {@link pullStream} did. */ export interface PullStreamReport { readonly stream: string; @@ -112,12 +129,8 @@ export interface PullStreamReport { * segment was staged and applied, and the cursor stops just before it. */ readonly refused: string | null; - /** - * The kind of `refused` (callers branch on this; the message may be - * reworded): `pull-off`, or `segment` for a refused segment. Null when - * nothing was refused. - */ - readonly refusedKind: 'pull-off' | 'segment' | null; + /** The kind of `refused` ({@link PullRefusal}), or null when nothing was refused. */ + readonly refusedKind: PullRefusal | null; /** Segments received, vault deltas passed over, transactions staged, re-deliveries skipped. */ readonly segments: number; readonly vaultDeltas: number; @@ -218,10 +231,93 @@ function refusalOf( if (bad !== null) { return `segment ${seg.seq} of replica ${seg.replicaId}: transaction ${bad} is not signed by device ${seg.deviceId}`; } + // A replica packs only its own transactions (`buildSegment`); the floor + // is kept per origin, so a foreign id is refused, never staged. + const foreign = txns.find((t) => localSeqOf(t.txn, seg.replicaId) === null); + if (foreign !== undefined) { + return `segment ${seg.seq} of replica ${seg.replicaId}: transaction ${foreign.txn} is not one of its transactions`; + } } return { txns }; } +/** + * The `local_seq` of `txn` when it is `replica`'s (`:`, + * minted by the sealer from 1), else null. + */ +function localSeqOf(txn: string, replica: string): number | null { + const prefix = `${replica}:`; + if (!txn.startsWith(prefix)) return null; + const rest = txn.slice(prefix.length); + if (!/^[1-9][0-9]*$/.test(rest)) return null; + const n = Number(rest); + return Number.isSafeInteger(n) ? n : null; +} + +/** One origin's floor on a stream (`_sync_seen_floor`). */ +interface SeenFloor { + readonly stagedUpto: number; + readonly prunedUpto: number; +} + +function readFloor(db: DatabaseSync, stream: string, origin: string): SeenFloor { + const row = db + .prepare( + 'SELECT staged_upto, pruned_upto FROM _sync_seen_floor WHERE stream = ? AND origin = ?', + ) + .get(stream, origin) as { staged_upto: number; pruned_upto: number } | undefined; + return row + ? { stagedUpto: Number(row.staged_upto), prunedUpto: Number(row.pruned_upto) } + : { stagedUpto: 0, prunedUpto: 0 }; +} + +/** What a segment brings past the seen ledger, or why it is refused. */ +type FreshPlan = + | { readonly fresh: LedgerTxn[]; readonly redelivered: number; readonly stagedUpto: number } + | { readonly refused: string }; + +/** + * Split a verified segment's transactions into re-deliveries (a seen row + * exists) and fresh ones, against the origin's floor (T13318). Read-only. A + * transaction with no seen row at or below the floor refuses the whole + * segment: its seen row was pruned, or it arrived out of `local_seq` order, + * and either way skipping it as seen could lose it. + */ +function planFresh( + db: DatabaseSync, + stream: string, + seg: PulledStreamSegment, + txns: readonly LedgerTxn[], +): FreshPlan { + const floor = readFloor(db, stream, seg.replicaId); + const isSeen = db.prepare('SELECT 1 FROM _sync_seen_txn WHERE stream = ? AND txn = ?'); + const inSegment = new Set(); + const fresh: LedgerTxn[] = []; + let redelivered = 0; + let upto = floor.stagedUpto; + for (const t of txns) { + if (inSegment.has(t.txn) || isSeen.get(stream, t.txn) !== undefined) { + redelivered += 1; + continue; + } + // refusalOf proved every id is the segment replica's. + const seq = localSeqOf(t.txn, seg.replicaId) ?? 0; + if (seq <= upto) { + const why = + seq <= floor.prunedUpto + ? `its seen row was pruned (pruned_upto ${floor.prunedUpto})` + : 'it arrived out of local_seq order'; + return { + refused: `segment ${seg.seq} of replica ${seg.replicaId}: transaction ${t.txn} is at or below the origin's staged floor ${upto} on ${stream} with no seen row: ${why}; it is not staged`, + }; + } + inSegment.add(t.txn); + fresh.push(t); + upto = seq; + } + return { fresh, redelivered, stagedUpto: upto }; +} + /** `cursor` advanced over `segs` (a page cut short by a refused segment). */ function advance(cursor: StreamCursor, segs: readonly PulledStreamSegment[]): StreamCursor { const replicas = { ...cursor.replicas }; @@ -236,54 +332,96 @@ function advance(cursor: StreamCursor, segs: readonly PulledStreamSegment[]): St /** Bytes counted for a seen-txn row's `seq` integer. */ const SEEN_SEQ_BYTES = 8; +/** Estimated payload bytes of one seen-txn row (stream and txn text plus the seq integer). */ +const seenBytes = (stream: string, txn: string): number => + Buffer.byteLength(stream, 'utf8') + Buffer.byteLength(txn, 'utf8') + SEEN_SEQ_BYTES; + /** * The size of `_sync_seen_txn` (T13317), for `cleo cloud status`: rows per * stream and the estimated payload bytes (stream and txn text plus the seq - * integer, without SQLite's page overhead). Read-only. + * integer, without SQLite's page overhead). Read-only, and O(origins): it + * sums the per-origin counts `_sync_seen_floor` keeps with every insert and + * prune (T13318), never the ledger itself. * * @param db - The store. - * @returns Zeroes when the table is not installed. + * @returns Zeroes when the tables are not installed. */ export function seenTxnReport(db: DatabaseSync): { rows: number; bytes: number; byStream: Record; } { - if (!hasTable(db, '_sync_seen_txn')) return { rows: 0, bytes: 0, byStream: {} }; + if (!hasTable(db, '_sync_seen_floor')) return { rows: 0, bytes: 0, byStream: {} }; const byStream: Record = {}; let rows = 0; let bytes = 0; for (const r of db .prepare( - `SELECT stream, count(*) AS n, sum(octet_length(stream) + octet_length(txn) + ${SEEN_SEQ_BYTES}) AS b - FROM _sync_seen_txn GROUP BY stream ORDER BY stream`, + `SELECT stream, sum(seen_rows) AS n, sum(seen_bytes) AS b + FROM _sync_seen_floor GROUP BY stream HAVING sum(seen_rows) > 0 ORDER BY stream`, ) .all() as Array<{ stream: string; n: number; b: number | null }>) { - byStream[r.stream] = r.n; - rows += r.n; - bytes += r.b ?? 0; + byStream[r.stream] = Number(r.n); + rows += Number(r.n); + bytes += Number(r.b ?? 0); } return { rows, bytes, byStream }; } /** - * Drop the seen-txn rows of `stream` first staged at or below `seq`, keeping - * `_sync_seen_txn` bounded (§3.1). Only safe below a floor the applier - * refuses any txn under: a seen txn can return in a NEW segment above a - * checkpoint (a retired replica's late segment, a rebind's re-push), so a - * checkpoint's coversSeq alone is not such a floor. No caller prunes until - * that floor exists (T13256). + * Drop each origin's seen-txn rows of `stream` below its staged floor, + * keeping `_sync_seen_txn` bounded (§3.1, T13318), and record how far each + * origin was pruned (`pruned_upto`). Safe because an origin's transactions + * are first delivered in `local_seq` order and the pull refuses any + * transaction at or below the floor that has no seen row (`below-floor`): + * a pruned transaction delivered again is refused loudly, never applied + * twice. The row at the floor itself is kept, so re-delivery of the latest + * transaction is still skipped quietly. * - * @param db - The store. + * @param db - The store, outside a transaction. * @param stream - The stream. - * @param seq - The floor, inclusive. * @returns How many rows were dropped. */ -export function pruneSeenTxns(db: DatabaseSync, stream: string, seq: number): number { - if (!hasTable(db, '_sync_seen_txn')) return 0; - return Number( - db.prepare('DELETE FROM _sync_seen_txn WHERE stream = ? AND seq <= ?').run(stream, seq).changes, - ); +export function pruneSeenTxns(db: DatabaseSync, stream: string): number { + if (!hasTable(db, '_sync_seen_floor')) return 0; + return withImmediateTransaction(db, () => { + const floors = db + .prepare( + 'SELECT origin, staged_upto FROM _sync_seen_floor WHERE stream = ? AND staged_upto - 1 > pruned_upto', + ) + .all(stream) as Array<{ origin: string; staged_upto: number }>; + // The origin's ids sort between ':' and ';' (':' + 1), + // a range of the (stream, txn) primary key. + const rowsOf = db.prepare( + `SELECT txn FROM _sync_seen_txn + WHERE stream = ? AND txn >= ? AND txn < ? AND CAST(substr(txn, ?) AS INTEGER) <= ?`, + ); + const drop = db.prepare('DELETE FROM _sync_seen_txn WHERE stream = ? AND txn = ?'); + const mark = db.prepare( + `UPDATE _sync_seen_floor + SET pruned_upto = ?, seen_rows = max(seen_rows - ?, 0), seen_bytes = max(seen_bytes - ?, 0) + WHERE stream = ? AND origin = ?`, + ); + let dropped = 0; + for (const f of floors) { + const upto = Number(f.staged_upto) - 1; + const victims = rowsOf.all( + stream, + `${f.origin}:`, + `${f.origin};`, + f.origin.length + 2, + upto, + ) as Array<{ txn: string }>; + let bytes = 0; + for (const v of victims) { + drop.run(stream, v.txn); + bytes += seenBytes(stream, v.txn); + } + mark.run(upto, victims.length, bytes, stream, f.origin); + dropped += victims.length; + } + return dropped; + }); } /** @@ -316,17 +454,16 @@ export async function pullStream( apply: null, }; } + // A store whose journal predates the floor gains it before the first page. + if (!hasTable(db, '_sync_seen_floor')) ensureSyncSchema(db); let cursor = readStreamCursor(db, o.stream) ?? o.initialCursor; let segments = 0; let vaultDeltas = 0; let staged = 0; let redelivered = 0; let head = cursor.after; - let refusedSegment: string | null = null; - // Seen-txn rows at or below the floor can no longer be re-delivered. - if (o.pruneSeenUpTo !== undefined && o.pruneSeenUpTo !== null) { - pruneSeenTxns(db, o.stream, o.pruneSeenUpTo); - } + let refused: string | null = null; + let refusedKind: PullRefusal | null = null; for (;;) { const page = await o.pull(cursor); head = page.head; @@ -338,52 +475,67 @@ export async function pullStream( for (const seg of page.segments) { const why = refusalOf(seg, o.verify); if (typeof why === 'string') { - refusedSegment = why; + refused = why; + refusedKind = 'segment'; break; } decoded.push({ seg, txns: why.txns }); } - const pageCursor = - refusedSegment === null - ? page.cursor - : advance( - cursor, - decoded.map((d) => d.seg), - ); - withImmediateTransaction(db, () => { - const seen = db.prepare( - 'INSERT INTO _sync_seen_txn (stream, txn, seq) VALUES (?, ?, ?) ON CONFLICT (stream, txn) DO NOTHING', + cursor = withImmediateTransaction(db, () => { + const seen = db.prepare('INSERT INTO _sync_seen_txn (stream, txn, seq) VALUES (?, ?, ?)'); + const raise = db.prepare( + `INSERT INTO _sync_seen_floor (stream, origin, staged_upto, seen_rows, seen_bytes) + VALUES (?, ?, ?, ?, ?) + ON CONFLICT (stream, origin) DO UPDATE SET + staged_upto = excluded.staged_upto, + seen_rows = seen_rows + excluded.seen_rows, + seen_bytes = seen_bytes + excluded.seen_bytes`, ); + const done: PulledStreamSegment[] = []; for (const { seg, txns } of decoded) { - segments += 1; - if (txns === null) { + if (txns !== null) { + const plan = planFresh(db, o.stream, seg, txns); + if ('refused' in plan) { + refused = plan.refused; + refusedKind = 'below-floor'; + break; + } + redelivered += plan.redelivered; + if (plan.fresh.length > 0) { + let bytes = 0; + for (const t of plan.fresh) { + seen.run(o.stream, t.txn, seg.seq); + bytes += seenBytes(o.stream, t.txn); + } + raise.run(o.stream, seg.replicaId, plan.stagedUpto, plan.fresh.length, bytes); + staged += stageTxns( + db, + o.stream, + { + seq: seg.seq, + replicaId: seg.replicaId, + replicaSeq: seg.replicaSeq, + deviceId: seg.deviceId, + schemaVersion: seg.schemaVersion, + txns: plan.fresh, + }, + nowIso, + ); + } + } else { vaultDeltas += 1; - continue; } - // A transaction already staged under an earlier segment is re-delivered: skip it. - const fresh = txns.filter((t) => Number(seen.run(o.stream, t.txn, seg.seq).changes) === 1); - redelivered += txns.length - fresh.length; - if (fresh.length === 0) continue; - staged += stageTxns( - db, - o.stream, - { - seq: seg.seq, - replicaId: seg.replicaId, - replicaSeq: seg.replicaSeq, - deviceId: seg.deviceId, - schemaVersion: seg.schemaVersion, - txns: fresh, - }, - nowIso, - ); + segments += 1; + done.push(seg); } - writeStreamCursor(db, o.stream, pageCursor, nowIso); + const next = refused === null ? page.cursor : advance(cursor, done); + writeStreamCursor(db, o.stream, next, nowIso); + return next; }); - cursor = pageCursor; - if (refusedSegment !== null) break; + if (refused !== null) break; if (page.segments.length === 0 || cursor.after >= page.head) break; } + if (o.pruneSeen === true) pruneSeenTxns(db, o.stream); const apply = applyStagedTxns(db, { scope: o.scope, stream: o.stream, @@ -394,8 +546,8 @@ export async function pullStream( }); return { stream: o.stream, - refused: refusedSegment, - refusedKind: refusedSegment === null ? null : 'segment', + refused, + refusedKind, segments, vaultDeltas, staged, diff --git a/packages/core/src/store/table-classification.ts b/packages/core/src/store/table-classification.ts index 96861a961c..b918e0d5df 100644 --- a/packages/core/src/store/table-classification.ts +++ b/packages/core/src/store/table-classification.ts @@ -208,6 +208,12 @@ const PROJECT_TABLES: Readonly> = { source: 'journal spec t12342-t12343-journal-design §3.1 Receive (T12343 S5-1)', note: 'every transaction id staged from a stream, so a re-delivered transaction is never staged twice (a counter delta would double). This device only. Created lazily by store/sync/schema.ts when a sync.* flag is first enabled (T12343)', }, + _sync_seen_floor: { + class: 'local-only', + status: 'optional-transient', + source: 'journal spec t12342-t12343-journal-design §3.1 Receive (T13318)', + note: 'per stream and origin replica: the highest local_seq staged, how far its seen-txn rows were pruned, and their row and byte counts. A txn at or below the floor with no seen row is refused. This device only. Created lazily by store/sync/schema.ts when a sync.* flag is first enabled (T13318)', + }, _sync_row_meta: { class: 'local-only', status: 'optional-transient', @@ -1454,6 +1460,12 @@ const GLOBAL_TABLES: Readonly> = { source: 'journal spec t12342-t12343-journal-design §3.1 Receive (T12343 S5-1)', note: 'every transaction id staged from a stream, so a re-delivered transaction is never staged twice (a counter delta would double). This device only. Created lazily by store/sync/schema.ts when a sync.* flag is first enabled (T12343)', }, + _sync_seen_floor: { + class: 'local-only', + status: 'optional-transient', + source: 'journal spec t12342-t12343-journal-design §3.1 Receive (T13318)', + note: 'per stream and origin replica: the highest local_seq staged, how far its seen-txn rows were pruned, and their row and byte counts. A txn at or below the floor with no seen row is refused. This device only. Created lazily by store/sync/schema.ts when a sync.* flag is first enabled (T13318)', + }, _sync_row_meta: { class: 'local-only', status: 'optional-transient', diff --git a/scripts/lint-no-raw-table-writes.mjs b/scripts/lint-no-raw-table-writes.mjs index 969adc2b7c..1b17e85adc 100644 --- a/scripts/lint-no-raw-table-writes.mjs +++ b/scripts/lint-no-raw-table-writes.mjs @@ -405,7 +405,12 @@ export const EXEMPT = { _sync_seen_txn: { count: 2, reason: - 'staged transaction ids (local-only): a re-delivered transaction is never staged twice, and rows past the latest verified checkpoint are pruned (§3.1, T12343 S5-1)', + "staged transaction ids (local-only): a re-delivered transaction is never staged twice, and rows below their origin's staged floor are pruned (§3.1, T12343 S5-1, T13318)", + }, + _sync_seen_floor: { + count: 2, + reason: + 'per-origin staged local_seq floor and seen-ledger counts (local-only): raised with every staged txn, marked with every prune (§3.1, T13318)', }, }, 'packages/core/src/store/sync/push.ts': { From ce966ac54a1db2cb2c5d3df5eb77a625a4557163 Mon Sep 17 00:00:00 2001 From: Keaton Hoskins Date: Fri, 9 Oct 2026 21:45:13 -0700 Subject: [PATCH 2/2] fix(sync): a lost seen-floor table is re-created on pull; status walks the ledger meanwhile (T13318, #1981 LOW) A store whose journal records the floor folder but whose _sync_seen_floor table is gone got nothing from ensureSyncSchema (the folder reads as applied), so pull threw "no such table" and cloud status read 0. The pull now heals the table (healSyncSchema re-runs the applied folders, which re-seeds the counts from the seen rows), and seenTxnReport walks the ledger until then. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01VjjTKbQXyCFdkLBSeuyKLt --- .../src/store/sync/__tests__/pull.test.ts | 21 ++++++++ packages/core/src/store/sync/pull.ts | 49 +++++++++++++++++-- 2 files changed, 65 insertions(+), 5 deletions(-) diff --git a/packages/core/src/store/sync/__tests__/pull.test.ts b/packages/core/src/store/sync/__tests__/pull.test.ts index fd8f66390b..0dc5bd65b7 100644 --- a/packages/core/src/store/sync/__tests__/pull.test.ts +++ b/packages/core/src/store/sync/__tests__/pull.test.ts @@ -27,6 +27,7 @@ import { pullStream, readStreamCursor, type StreamCursor, + seenTxnReport, } from '../pull.js'; import { ensureSyncSchema } from '../schema.js'; import { sealPending } from '../sealer.js'; @@ -353,6 +354,26 @@ describe('per-origin seen floor (T13318)', () => { }); }); + it('a store whose journal records the floor folder but lost the table gets it back on pull, counts re-seeded', async () => { + const a = await store('a'); + const b = await store('b'); + const stream = fakeStream(); + write(a, addTask('T1')); + stream.append(authorSegment(a), 0); // RA:1 + write(a, addTask('T2')); + stream.append(authorSegment(a), 1); // RA:2 + await pullStream(b, pullOpts(b, stream)); + b.exec('DROP TABLE _sync_seen_floor'); + // Until the next pull, status still reports the ledger, never zero. + expect(seenTxnReport(b)).toMatchObject({ rows: 2, byStream: { [STREAM]: 2 } }); + write(a, addTask('T3')); + stream.append(authorSegment(a), 2); // RA:3 + const r = await pullStream(b, pullOpts(b, stream)); + expect(r).toMatchObject({ refused: null, staged: 1, after: 3 }); + expect(floorOf(b, RA)).toEqual({ stagedUpto: 3, prunedUpto: 0, rows: 3 }); + expect(seenTxnReport(b)).toMatchObject({ rows: 3, byStream: { [STREAM]: 3 } }); + }); + it("a transaction that is not the segment replica's is refused", async () => { const a = await store('a'); const b = await store('b'); diff --git a/packages/core/src/store/sync/pull.ts b/packages/core/src/store/sync/pull.ts index ae7ab3364a..be7bdb7c9a 100644 --- a/packages/core/src/store/sync/pull.ts +++ b/packages/core/src/store/sync/pull.ts @@ -44,7 +44,7 @@ import { type ApplyReport, type ApplyStagedOptions, applyStagedTxns } from './ap import { withImmediateTransaction } from './clock-store.js'; import { isSyncFlagOn } from './flags.js'; import { stageTxns } from './inbox.js'; -import { ensureSyncSchema, hasTable } from './schema.js'; +import { ensureSyncSchema, hasTable, healSyncSchema } from './schema.js'; /** Where a pull stands (the journal client's `PullCursor`). */ export interface StreamCursor { @@ -336,6 +336,45 @@ const SEEN_SEQ_BYTES = 8; const seenBytes = (stream: string, txn: string): number => Buffer.byteLength(stream, 'utf8') + Buffer.byteLength(txn, 'utf8') + SEEN_SEQ_BYTES; +/** + * Install the floor table before a pull: a store whose journal predates it + * gains it, and a store whose journal records the folder but lost the table + * gets it re-created (§2.3a rule 9), its counts re-seeded from the seen rows. + * + * @param db - The store, outside a transaction. + */ +function ensureSeenFloor(db: DatabaseSync): void { + if (hasTable(db, '_sync_seen_floor')) return; + ensureSyncSchema(db); + healSyncSchema(db, ['_sync_seen_floor']); +} + +/** + * {@link seenTxnReport} for a store with seen rows but no floor table yet + * (its next pull installs it): the ledger walked once, O(rows). + */ +function seenLedgerWalk(db: DatabaseSync): { + rows: number; + bytes: number; + byStream: Record; +} { + const byStream: Record = {}; + let rows = 0; + let bytes = 0; + if (!hasTable(db, '_sync_seen_txn')) return { rows, bytes, byStream }; + for (const r of db + .prepare( + `SELECT stream, count(*) AS n, sum(octet_length(stream) + octet_length(txn) + ${SEEN_SEQ_BYTES}) AS b + FROM _sync_seen_txn GROUP BY stream ORDER BY stream`, + ) + .all() as Array<{ stream: string; n: number; b: number | null }>) { + byStream[r.stream] = Number(r.n); + rows += Number(r.n); + bytes += Number(r.b ?? 0); + } + return { rows, bytes, byStream }; +} + /** * The size of `_sync_seen_txn` (T13317), for `cleo cloud status`: rows per * stream and the estimated payload bytes (stream and txn text plus the seq @@ -344,14 +383,15 @@ const seenBytes = (stream: string, txn: string): number => * prune (T13318), never the ledger itself. * * @param db - The store. - * @returns Zeroes when the tables are not installed. + * @returns Zeroes when the tables are not installed; the ledger walked when + * only the floor table is missing. */ export function seenTxnReport(db: DatabaseSync): { rows: number; bytes: number; byStream: Record; } { - if (!hasTable(db, '_sync_seen_floor')) return { rows: 0, bytes: 0, byStream: {} }; + if (!hasTable(db, '_sync_seen_floor')) return seenLedgerWalk(db); const byStream: Record = {}; let rows = 0; let bytes = 0; @@ -454,8 +494,7 @@ export async function pullStream( apply: null, }; } - // A store whose journal predates the floor gains it before the first page. - if (!hasTable(db, '_sync_seen_floor')) ensureSyncSchema(db); + ensureSeenFloor(db); let cursor = readStreamCursor(db, o.stream) ?? o.initialCursor; let segments = 0; let vaultDeltas = 0;