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 7fef03b3b9..6a2f07dd66 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..0dc5bd65b7 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'; @@ -27,7 +27,9 @@ import { pullStream, readStreamCursor, type StreamCursor, + seenTxnReport, } 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 +38,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 +81,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 +251,190 @@ 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 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'); + 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..be7bdb7c9a 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, healSyncSchema } 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,136 @@ 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; + +/** + * 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 - * 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; 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_txn')) return { rows: 0, bytes: 0, byStream: {} }; + if (!hasTable(db, '_sync_seen_floor')) return seenLedgerWalk(db); 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 +494,15 @@ export async function pullStream( apply: null, }; } + ensureSeenFloor(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 +514,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 +585,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 2fce3ff6ba..28cbdb70a5 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', @@ -1461,6 +1467,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 fb32543959..996ae5c326 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': {