From 9f1fd32e916ca58239db43ac95998a9d626f202c Mon Sep 17 00:00:00 2001 From: Keaton Hoskins Date: Tue, 6 Oct 2026 18:15:48 -0700 Subject: [PATCH 1/4] feat(sync): pull a stream's journal - verify, stage once, apply (T12343 S5-1) pullStream (store/sync/pull.ts) pages through the stream after the store's persisted cursor and, per page in one transaction, decodes each segment (a cleo-vault-delta/v1 segment carries no ops and is passed over), verifies every transaction's device signature before writing anything, skips transaction ids already staged from the stream (_sync_seen_txn: a re-delivered transaction is never staged or applied twice), stages the rest and advances _sync_cursor; then the applier applies, own echoes included. pullSyncStream (cloud/nexus-vault.ts) wires it to the journal client and the trusted device keys, refuses a store bound to another replica than the link, and starts a store with no position after the checkpoint it last synced. New local-only migration t12343-pull-cursor; both tables are cleared from vault bundles. Co-Authored-By: Claude Opus 5.5 --- .changeset/t12343-pull.md | 25 ++ .../migration.sql | 28 ++ .../src/cloud/__tests__/nexus-attach.test.ts | 2 + .../src/cloud/__tests__/nexus-vault.test.ts | 26 ++ packages/core/src/cloud/nexus-vault.ts | 115 +++++++- .../table-classification-gate.test.ts | 4 + .../src/store/sync/__tests__/pull.test.ts | 231 ++++++++++++++++ packages/core/src/store/sync/pull.ts | 250 ++++++++++++++++++ .../core/src/store/table-classification.ts | 24 ++ scripts/lint-no-raw-table-writes.mjs | 12 + 10 files changed, 715 insertions(+), 2 deletions(-) create mode 100644 .changeset/t12343-pull.md create mode 100644 packages/core/migrations/sync-journal/20261008120000_t12343-pull-cursor/migration.sql create mode 100644 packages/core/src/store/sync/__tests__/pull.test.ts create mode 100644 packages/core/src/store/sync/pull.ts diff --git a/.changeset/t12343-pull.md b/.changeset/t12343-pull.md new file mode 100644 index 0000000000..f5af7912e9 --- /dev/null +++ b/.changeset/t12343-pull.md @@ -0,0 +1,25 @@ +--- +id: t12343-pull +tasks: [T12343] +kind: feat +summary: Sync pull - stage and apply a stream's verified segments, never a transaction twice (S5-1) +--- + +A store can now pull its stream's journal (journal spec §3.1, §2.8; T12343 S5-1). + +- **`pullStream(db, …)`** (`store/sync/pull.ts`) pages through the stream after the store's persisted cursor. For each page, in one + local transaction: + - **Decode:** `deflate-raw` canonical JSON of `LedgerTxn[]`, parsed with the ledger schema. A vault count-delta segment + (`cleo-vault-delta/v1`) carries no ops and is passed over. + - **Verify:** every transaction's signature is checked against the segment's device key, before anything of the page is written. + A bad signature or a malformed body refuses the segment (`SegmentRefusedError`). + - **Deduplicate:** every transaction id already staged from the stream is skipped (`_sync_seen_txn`), so a re-delivered + transaction is never staged or applied twice and a counter delta never doubles. + - **Stage and advance:** the rest are staged, and the cursor advances (`_sync_cursor`). + + Then the applier applies, in stream order, own echoes included (they feed the fast path and the scoped rebase). +- **`pullSyncStream(opts)`** (`cloud/nexus-vault.ts`) wires `pullStream` to the journal client and the trusted device keys. It refuses + a store bound to another replica than the link. A store with no pull position starts after the checkpoint it last synced (its + genesis, or the snapshot it restored), seeded from that checkpoint's signed replica map. +- **New sync-journal migration** `t12343-pull-cursor`, local-only: `_sync_cursor` and `_sync_seen_txn`. Both are cleared from vault + bundles, so a restored store resumes from its own checkpoint. diff --git a/packages/core/migrations/sync-journal/20261008120000_t12343-pull-cursor/migration.sql b/packages/core/migrations/sync-journal/20261008120000_t12343-pull-cursor/migration.sql new file mode 100644 index 0000000000..f2a7e131ff --- /dev/null +++ b/packages/core/migrations/sync-journal/20261008120000_t12343-pull-cursor/migration.sql @@ -0,0 +1,28 @@ +-- T12343 S5-1 (journal spec §3.1 "Receive") — where a store's pull stands. +-- +-- _sync_cursor: per stream, the PullCursor the journal client verifies the +-- next page against (the last applied stream seq, and per replica its pinned +-- device and last replicaSeq). It is advanced in the SAME local transaction +-- that stages the segments it covers (§3.1 step 3), so a crash re-pulls +-- nothing twice and loses nothing. +-- +-- _sync_seen_txn: every transaction id this store has staged from a stream. +-- A transaction re-delivered under another segment (a replayed or re-sealed +-- segment, a retired replica's late segment carrying a txn already seen) is +-- skipped, never staged twice: a counter delta ($inc) applied twice would +-- double (T12343 S5-1). +-- +-- Local-only. +-- +-- Do not edit this file once applied: its hash is recorded per store. +CREATE TABLE IF NOT EXISTS _sync_cursor ( + stream TEXT PRIMARY KEY, + cursor_json TEXT NOT NULL, + updated_at TEXT NOT NULL +) WITHOUT ROWID; +CREATE TABLE IF NOT EXISTS _sync_seen_txn ( + stream TEXT NOT NULL, + txn TEXT NOT NULL, + seq INTEGER NOT NULL, + PRIMARY KEY (stream, txn) +) WITHOUT ROWID; diff --git a/packages/core/src/cloud/__tests__/nexus-attach.test.ts b/packages/core/src/cloud/__tests__/nexus-attach.test.ts index e180eb624d..780a40a5e1 100644 --- a/packages/core/src/cloud/__tests__/nexus-attach.test.ts +++ b/packages/core/src/cloud/__tests__/nexus-attach.test.ts @@ -307,6 +307,7 @@ describe('presence and binding hold their guarantees (review M3)', () => { '_sync_capture', '_sync_clock', '_sync_conflict', + '_sync_cursor', '_sync_field_leave', '_sync_foreign_touch', '_sync_frame', @@ -318,6 +319,7 @@ describe('presence and binding hold their guarantees (review M3)', () => { '_sync_replica', '_sync_row_meta', '_sync_row_undo', + '_sync_seen_txn', '_sync_segment', '_sync_segment_txn', '_sync_sequenced', diff --git a/packages/core/src/cloud/__tests__/nexus-vault.test.ts b/packages/core/src/cloud/__tests__/nexus-vault.test.ts index 52b734f633..f95aa138dd 100644 --- a/packages/core/src/cloud/__tests__/nexus-vault.test.ts +++ b/packages/core/src/cloud/__tests__/nexus-vault.test.ts @@ -120,6 +120,7 @@ import { listNexusNamedProjects, resolveNexusProjectRef } from '../nexus-project import { enableSyncPush, nexusVaultStatus, + pullSyncStream, pushNexusVault, pushSyncStream, releaseNexusVaultLease, @@ -5127,6 +5128,31 @@ describe('sync enable push: the genesis checkpoint (T12343 S4-1b)', () => { }); }); + it('S5-1: pull starts after the genesis checkpoint, verifies and stages the own echo, and sequences it', async () => { + const { m, dbPath } = await journalMachine(); + await on(m, () => enableSyncPush(vopts(m, { allowUnreleased: true }))); + await on(m, async () => { + (await storeOf(dbPath)).exec( + "INSERT INTO tasks_tasks (id, title, type, status, priority, uid, birth_fp) VALUES ('T8', 'title T8', 'task', 'pending', 'medium', 'uid-T8', 'fp-T8')", + ); + }); + const pushed = await on(m, () => pushSyncStream(vopts(m, { allowUnreleased: true }))); + expect(pushed.pushed).toBe(1); + const pulled = await on(m, () => pullSyncStream(vopts(m))); + expect(pulled).toMatchObject({ segments: 1, staged: 1, redelivered: 0, vaultDeltas: 0 }); + expect(pulled.after).toBe(pushed.lastServerSeq); + expect(pulled.apply.applied).toBe(1); + await on(m, async () => { + const db = await storeOf(dbPath); + // The own echo was sequenced (§3.5): its transaction is no longer unsequenced. + expect( + (db.prepare('SELECT count(*) AS n FROM _sync_sequenced').get() as { n: number }).n, + ).toBe(1); + }); + const again = await on(m, () => pullSyncStream(vopts(m))); + expect(again).toMatchObject({ segments: 0, staged: 0 }); + }); + it('S4-2: a device clock ahead of the server pauses push', async () => { const { m, dbPath } = await journalMachine(); await on(m, () => enableSyncPush(vopts(m, { allowUnreleased: true }))); diff --git a/packages/core/src/cloud/nexus-vault.ts b/packages/core/src/cloud/nexus-vault.ts index 4024f02c1d..52f156a309 100644 --- a/packages/core/src/cloud/nexus-vault.ts +++ b/packages/core/src/cloud/nexus-vault.ts @@ -112,11 +112,13 @@ import { genesisPending, readGenesisCut, } from '../store/sync/genesis.js'; +import { type PullStreamReport, pullStream, readStreamCursor } from '../store/sync/pull.js'; import { type PushStreamReport, pushStream } from '../store/sync/push.js'; import { replayPinOf } from '../store/sync/replay-pin.js'; import { activeReplica, readActiveReplicaId } from '../store/sync/replica.js'; import { ReplicaRegistry } from '../store/sync/replica-registry.js'; -import { signTxn } from '../store/sync/txn-signing.js'; +import { sealPending } from '../store/sync/sealer.js'; +import { firstBadTxnSignature, signTxn } from '../store/sync/txn-signing.js'; import { buildVaultManifest, type CarriedMachineState, @@ -145,7 +147,7 @@ import { foreignWriterLeases, storeOpenElsewhere } from '../store/writer-lease.j import { deriveKey, uuidv7 } from './crypto.js'; import { NexusError } from './http.js'; import { cursorFromCheckpoint, initialPullCursor, Journal, type PullCursor } from './journal.js'; -import type { TrustedSigners } from './keys.js'; +import { signerKeys, type TrustedSigners } from './keys.js'; import { manifestVersion } from './manifest-check.js'; import { canonicalGlobalReplicaBinder } from './nexus-attach.js'; import { NexusAccountError } from './nexus-auth.js'; @@ -798,6 +800,10 @@ const VAULT_CLEARED_JOURNAL_TABLES = [ '_sync_frame', '_sync_quarantine', '_sync_apply_intent', + // Where this device's pull stands (T12343 S5-1): a restored store resumes + // from the checkpoint it restored, never from another device's position. + '_sync_cursor', + '_sync_seen_txn', ] as const; /** @@ -1814,6 +1820,95 @@ async function pushSyncStreamImpl( } } +async function pullSyncStreamImpl(opts: NexusVaultCommandOptions = {}): Promise { + const conn = await connectNexusVault(opts); + const key = await unlockNexusAccountKey(conn); + const t = await resolveTarget(conn, key.masterKey, opts, 'push'); + const replicaId = t.replicaId; + if (!replicaId) { + throw vaultError( + 'E_NEXUS_VAULT_NOT_LINKED', + 'this copy of the project is not attached from this device', + 'run `cleo project link`', + ); + } + const journal = journalFor(conn, t); + const { openDualScopeDbAtPath, getDualScopeNativeDb } = await import('../store/dual-scope-db.js'); + const db = getDualScopeNativeDb( + t.scope === 'global' + ? await openDualScopeDbAtPath('global', t.dbPath) + : await openDualScopeDbAtPath('project', t.dbPath), + ); + // Own echoes are recognised by the bound replica: it must be the stream's (T13304). + const bound = activeReplica(db, tableScopeOf(t))?.replicaId ?? null; + if (bound !== replicaId) { + throw vaultError( + 'E_NEXUS_SYNC_REFUSED', + `this store is bound to replica ${bound ?? '(none)'}, but ${t.streamId} knows it as ${replicaId}`, + 'nothing was pulled; relink the project (`cleo project link`) so both name one replica', + ); + } + // A store with no pull position starts after the checkpoint it last synced + // (the genesis it cut, or the snapshot it restored), seeded from its signed + // replica map. + let initialCursor = readStreamCursor(db, t.streamId); + if (initialCursor === null) { + const head = await streamHead(conn, t.streamId); + const checkpoints = head.headCheckpointId ? await listCheckpoints(conn, t.streamId) : []; + const { state: synced } = syncedState(conn, t, checkpoints, head.headCheckpointId); + const from = checkpoints.find((c) => c.checkpointId === synced?.lastCheckpointId) ?? null; + if (from === null) { + throw vaultError( + 'E_NEXUS_SYNC_REFUSED', + `this store has synced no checkpoint of ${t.streamId}, so it has no position to pull from`, + 'restore the stream (`cleo cloud restore`), or enable push (`cleo sync enable push`), first', + ); + } + journal.verifyCheckpoint(from, key.signers); + initialCursor = cursorFromCheckpoint(from); + } + return pullStream(db, { + scope: tableScopeOf(t), + stream: t.streamId, + replica: replicaId, + initialCursor, + pull: async (cursor) => { + const page = await journal.pull( + { + after: cursor.after, + knowsAllReplicas: cursor.knowsAllReplicas, + replicas: { ...cursor.replicas }, + }, + key.signers, + ); + return { + segments: page.segments.map((s) => ({ + seq: s.seq, + replicaId: s.replicaId, + replicaSeq: s.replicaSeq, + deviceId: s.deviceId, + plaintext: s.plaintext, + schemaVersion: s.meta.schemaVersion, + })), + cursor: page.cursor, + head: page.head, + }; + }, + verify: (deviceId, txns) => { + const keys = signerKeys(key.signers, deviceId); + const first = keys[0]; + if (first === undefined) return 0; + for (const k of keys) { + if (firstBadTxnSignature(k.publicKey, t.streamId, txns) === null) return null; + } + return firstBadTxnSignature(first.publicKey, t.streamId, txns); + }, + seal: () => { + sealPending(db, { scope: tableScopeOf(t), replica: replicaId }); + }, + }); +} + /** The keyed hash of an empty table (a table the parent lists that this store no longer has). */ function buildEmptyHash(t: VaultTarget, table: string): string { return emptyVaultTableHash(hashKeyOf(t.dataKey), table); @@ -2667,6 +2762,22 @@ export function pushSyncStream( return mapped(() => pushSyncStreamImpl(opts)); } +/** + * Pull this store's stream journal (journal spec §3.1; T12343 S5-1): verified + * segments after the store's cursor are decoded, signature-checked, staged + * (a re-delivered transaction never twice, a vault delta passed over) and + * applied. A store with no pull position starts after the checkpoint it last + * synced. `cleo cloud sync` (T12996) runs it with push. + * + * @param opts - Scope and overrides. + * @returns What was received, staged and applied. + * @throws {NexusAccountError} `E_NEXUS_SYNC_REFUSED` (replica mismatch, no position), + * `E_NEXUS_VAULT_NOT_LINKED`, or a mapped API error. + */ +export function pullSyncStream(opts: NexusVaultCommandOptions = {}): Promise { + return mapped(() => pullSyncStreamImpl(opts)); +} + /** * `cleo cloud pull` / `cleo cloud restore`: verify and activate a snapshot. * 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 7a9a7c1627..47f983574e 100644 --- a/packages/core/src/store/__tests__/table-classification-gate.test.ts +++ b/packages/core/src/store/__tests__/table-classification-gate.test.ts @@ -199,6 +199,8 @@ const SYNC_REBASE_DDL = 'packages/core/migrations/sync-journal/20261006120000_t13193-rebase-undo/migration.sql'; 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_JOURNAL_TABLES = { _sync_capture: { class: 'local-only', ddl: SYNC_CAPTURE_DDL }, _sync_frame: { class: 'local-only', ddl: SYNC_CAPTURE_DDL }, @@ -217,6 +219,8 @@ const SYNC_JOURNAL_TABLES = { _sync_sequenced: { class: 'local-only', ddl: SYNC_REBASE_DDL }, _sync_segment: { class: 'local-only', ddl: SYNC_SEGMENT_DDL }, _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_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 new file mode 100644 index 0000000000..a33e1cdfbb --- /dev/null +++ b/packages/core/src/store/sync/__tests__/pull.test.ts @@ -0,0 +1,231 @@ +/** + * Pulling a stream's journal (T12343 S5-1; journal spec §3.1, §2.8). + * + * Two real stores: A authors and persists segments with the real sealer, + * segment builder and transaction signatures; a fake stream serves them; + * B pulls, stages and applies. + * + * @task T12343 + */ + +import { mkdirSync, mkdtempSync, rmSync } from 'node:fs'; +import { tmpdir } from 'node:os'; +import { join, resolve } from 'node:path'; +import type { DatabaseSync } from 'node:sqlite'; +import { SYNC_SCHEMA_VERSION } from '@cleocode/contracts/sync-schema.js'; +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; +import { generateEd25519 } from '../../../cloud/crypto.js'; +import { + _resetDualScopeDbCache, + getDualScopeNativeDb, + openDualScopeDbAtPath, +} from '../../dual-scope-db.js'; +import { finishCaptureFrame, openCaptureFrame, setCaptureEnabled } from '../capture.js'; +import { setSyncFlag } from '../flags.js'; +import { + type PulledStreamSegment, + pullStream, + readStreamCursor, + SegmentRefusedError, + type StreamCursor, +} from '../pull.js'; +import { sealPending } from '../sealer.js'; +import { buildSegment } from '../segments.js'; +import { firstBadTxnSignature, signTxn } from '../txn-signing.js'; + +const SYNC_SCHEMA = resolve(import.meta.dirname, '../../../../migrations/sync-journal'); +const STREAM = 'project:0192ffff-7f00-7000-8000-00000000000f'; +const RA = '0192aaaa-7f00-7000-8000-00000000000a'; +const RB = '0192bbbb-7f00-7000-8000-00000000000b'; +const DEV_A = 'dev-a'; +const KEY_A = generateEd25519(); +let clock = Date.now(); +let dir: string; + +beforeEach(() => { + dir = mkdtempSync(join(tmpdir(), 'cleo-pull-')); + mkdirSync(join(dir, 'cleo'), { recursive: true }); + vi.stubEnv('CLEO_HOME', join(dir, 'cleo')); + vi.stubEnv('CLEO_ROOT', undefined); + vi.stubEnv('CLEO_DIR', undefined); +}); + +afterEach(() => { + _resetDualScopeDbCache(); + vi.unstubAllEnvs(); + rmSync(dir, { recursive: true, force: true }); +}); + +async function store(name: string): Promise { + mkdirSync(join(dir, name, '.cleo'), { recursive: true }); + const db = getDualScopeNativeDb( + await openDualScopeDbAtPath('project', join(dir, name, '.cleo', 'cleo.db')), + ); + setCaptureEnabled(db, 'project', true, { schemaRoot: SYNC_SCHEMA }); + setSyncFlag(db, 'sync.seal', true, { schemaRoot: SYNC_SCHEMA, allowUnreleased: true }); + return db; +} + +const seal = (db: DatabaseSync, replica: string) => () => { + sealPending(db, { + scope: 'project', + replica, + now: () => ++clock, + env: {}, + allowUnreleased: true, + }); +}; + +function write(db: DatabaseSync, sql: string): void { + db.exec('BEGIN IMMEDIATE'); + const frame = openCaptureFrame(db, 'write', null); + db.exec(sql); + finishCaptureFrame(db, frame); + db.exec('COMMIT'); + seal(db, RA)(); +} + +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 { + const seg = buildSegment(a, { + stream: STREAM, + replica: RA, + scope: 'project', + project: null, + sealer: (_seq, plaintext) => Buffer.from(plaintext), + signTxn: (stream, txn) => signTxn(KEY_A, stream, txn), + nowIso: new Date(++clock).toISOString(), + }); + if (!seg) throw new Error('nothing to pack'); + return seg.sealed; +} + +/** A fake stream: segments in seq order, served after the cursor, one page at a time. */ +function fakeStream() { + const segments: PulledStreamSegment[] = []; + let pages = 0; + return { + segments, + get pages() { + return pages; + }, + append(plaintext: Uint8Array, replicaSeq: number, replicaId = RA, deviceId = DEV_A): void { + segments.push({ + seq: segments.length + 1, + replicaId, + replicaSeq, + deviceId, + plaintext, + schemaVersion: SYNC_SCHEMA_VERSION, + }); + }, + pull: async (cursor: StreamCursor) => { + pages += 1; + const page = segments.filter((s) => s.seq > cursor.after).slice(0, 2); + const replicas = { ...cursor.replicas }; + for (const s of page) + replicas[s.replicaId] = { deviceId: s.deviceId, replicaSeq: s.replicaSeq }; + return { + segments: page, + cursor: { after: page.at(-1)?.seq ?? cursor.after, knowsAllReplicas: true, replicas }, + head: segments.length, + }; + }, + }; +} + +const START: StreamCursor = { after: 0, knowsAllReplicas: true, replicas: {} }; + +const pullOpts = (b: DatabaseSync, stream: ReturnType) => ({ + scope: 'project' as const, + stream: STREAM, + replica: RB, + pull: stream.pull, + verify: (deviceId: string, txns: Parameters[2]) => + deviceId === DEV_A ? firstBadTxnSignature(KEY_A.publicKey, STREAM, txns) : 0, + initialCursor: START, + now: () => ++clock, + seal: seal(b, RB), +}); + +const n = (db: DatabaseSync, sql: string): number => + Number((db.prepare(sql).get() as { n: number }).n); + +describe('pullStream (S5-1)', () => { + it("stages and applies another replica's segments in stream order, persisting the cursor; a rerun pulls nothing", async () => { + const a = await store('a'); + const b = await store('b'); + const stream = fakeStream(); + write(a, addTask('T1')); + stream.append(authorSegment(a), 0); + write(a, addTask('T2')); + stream.append(authorSegment(a), 1); + write(a, "UPDATE tasks_tasks SET priority = 'high' WHERE id = 'T1'"); + stream.append(authorSegment(a), 2); + const r = await pullStream(b, pullOpts(b, stream)); + expect(r).toMatchObject({ segments: 3, staged: 3, redelivered: 0, after: 3, head: 3 }); + expect(r.apply.applied).toBe(3); + expect(b.prepare("SELECT priority FROM tasks_tasks WHERE id = 'T1'").get()).toEqual({ + priority: 'high', + }); + expect(readStreamCursor(b, STREAM)).toMatchObject({ after: 3 }); + const again = await pullStream(b, pullOpts(b, stream)); + expect(again).toMatchObject({ segments: 0, staged: 0, after: 3 }); + }); + + it('a re-delivered transaction is never staged or applied twice', async () => { + const a = await store('a'); + const b = await store('b'); + const stream = fakeStream(); + write(a, addTask('T1')); + const first = authorSegment(a); + stream.append(first, 0); + await pullStream(b, pullOpts(b, stream)); + // The same transaction arrives again in a later segment (a replayed or re-sealed segment). + stream.append(first, 1); + const r = await pullStream(b, pullOpts(b, stream)); + expect(r).toMatchObject({ segments: 1, staged: 0, redelivered: 1 }); + expect(n(b, 'SELECT count(*) AS n FROM _sync_inbox')).toBe(1); + expect(n(b, "SELECT count(*) AS n FROM tasks_tasks WHERE id = 'T1'")).toBe(1); + expect(n(b, 'SELECT count(*) AS n FROM _sync_conflict')).toBe(0); + }); + + it('a vault delta segment carries no ops and is passed over', async () => { + const a = await store('a'); + const b = await store('b'); + const stream = fakeStream(); + stream.append( + Buffer.from( + JSON.stringify({ kind: 'cleo-vault-delta/v1', base: 'cp-1', tables: { tasks_tasks: 2 } }), + ), + 0, + ); + write(a, addTask('T1')); + stream.append(authorSegment(a), 1); + const r = await pullStream(b, pullOpts(b, stream)); + expect(r).toMatchObject({ segments: 2, vaultDeltas: 1, staged: 1, after: 2 }); + expect(n(b, "SELECT count(*) AS n FROM tasks_tasks WHERE id = 'T1'")).toBe(1); + }); + + it('a transaction its device did not sign refuses the segment: nothing of its page is staged', async () => { + const a = await store('a'); + const b = await store('b'); + const stream = fakeStream(); + write(a, addTask('T1')); + stream.append(authorSegment(a), 0, RA, 'dev-mallory'); // served under another device's pin + await expect(pullStream(b, pullOpts(b, stream))).rejects.toBeInstanceOf(SegmentRefusedError); + expect(n(b, 'SELECT count(*) AS n FROM _sync_inbox')).toBe(0); + expect(readStreamCursor(b, STREAM)).toBeNull(); + }); + + it('a body that is neither a ledger segment nor a vault delta is refused', async () => { + const b = await store('b'); + const stream = fakeStream(); + stream.append(Buffer.from('not a segment'), 0); + await expect(pullStream(b, pullOpts(b, stream))).rejects.toThrow(/not a ledger segment/); + expect(readStreamCursor(b, STREAM)).toBeNull(); + }); +}); diff --git a/packages/core/src/store/sync/pull.ts b/packages/core/src/store/sync/pull.ts new file mode 100644 index 0000000000..ab57f1e084 --- /dev/null +++ b/packages/core/src/store/sync/pull.ts @@ -0,0 +1,250 @@ +/** + * Pull one stream's journal into the store (journal spec §3.1 "Receive", + * §2.9; T12343 S5-1). + * + * {@link pullStream} pages through the stream after the store's persisted + * `_sync_cursor` (the journal client verifies order, gaps, the replica-to- + * device pin, trust, hashes and decryption before anything is returned), + * and for each segment, in ONE local transaction per page: + * - decodes it: `deflate-raw(canonical JSON of LedgerTxn[])`, parsed with + * the ledger schema; a vault delta segment (`cleo-vault-delta/v1`, a count + * correction a vault push appended before the stream's genesis) carries no + * ops and is passed over; + * - checks every transaction's signature against the segment's device key + * (§2.8); one bad signature refuses the segment, and the pull stops before + * staging anything of it; + * - 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; + * - stages the rest (`stageTxns`), schema-ahead ones included (the applier + * holds them `refused-schema` and replays them after an upgrade); + * - advances the cursor. + * + * Then the applier applies what is staged, in stream order, own echoes + * included (they feed the fast path and the scoped rebase, §3.5). + * + * The page source is a port ({@link SegmentPuller}): the journal client in + * production (`cloud/nexus-vault.ts`), a fake in tests. + * + * @task T12343 + * @module store/sync/pull + */ + +import type { DatabaseSync } from 'node:sqlite'; +import { inflateRawSync } from 'node:zlib'; +import type { TableScope } from '@cleocode/contracts'; +import { LedgerTxn } from '@cleocode/contracts/ledger'; +import { type ApplyReport, type ApplyStagedOptions, applyStagedTxns } from './apply/applier.js'; +import { withImmediateTransaction } from './clock-store.js'; +import { stageTxns } from './inbox.js'; +import { hasTable } from './schema.js'; + +/** Where a pull stands (the journal client's `PullCursor`). */ +export interface StreamCursor { + /** The last stream seq staged. */ + readonly after: number; + /** Whether `replicas` lists every replica with a segment at or below `after`. */ + readonly knowsAllReplicas: boolean; + /** Per replica: the device pinned to it and its last replicaSeq staged. */ + readonly replicas: Readonly< + Record + >; +} + +/** One verified, decrypted segment, as the journal client returns it. */ +export interface PulledStreamSegment { + readonly seq: number; + readonly replicaId: string; + readonly replicaSeq: number; + readonly deviceId: string; + readonly plaintext: Uint8Array; + readonly schemaVersion: number; +} + +/** Pulls the next verified page after `cursor`. */ +export type SegmentPuller = (cursor: StreamCursor) => Promise<{ + readonly segments: readonly PulledStreamSegment[]; + readonly cursor: StreamCursor; + readonly head: number; +}>; + +/** The index of a segment's first transaction its device did not sign, or null when every one is signed. */ +export type TxnVerifier = (deviceId: string, txns: readonly LedgerTxn[]) => number | null; + +/** The plaintext kind of a vault count-delta segment (`cloud/nexus-vault.ts`). */ +export const VAULT_DELTA_KIND = 'cleo-vault-delta/v1'; + +/** Options for {@link pullStream}. */ +export interface PullStreamOptions { + readonly scope: TableScope; + readonly stream: string; + /** This store's bound replica (own echoes are recognised by it). */ + readonly replica: string; + readonly pull: SegmentPuller; + readonly verify: TxnVerifier; + /** The cursor to start from when the store has none persisted (the checkpoint it was cut from or restored). */ + readonly initialCursor: StreamCursor; + /** Wall clock. @defaultValue Date.now */ + readonly now?: () => number; + /** Seals pending local captures before a rebase (the applier's `seal`). */ + readonly seal: ApplyStagedOptions['seal']; + /** Applier options passed through (pages, budget). */ + readonly apply?: Partial>; +} + +/** What {@link pullStream} did. */ +export interface PullStreamReport { + readonly stream: string; + /** Segments received, vault deltas passed over, transactions staged, re-deliveries skipped. */ + readonly segments: number; + readonly vaultDeltas: number; + readonly staged: number; + readonly redelivered: number; + /** The stream position the store has staged up to, and the server's head. */ + readonly after: number; + readonly head: number; + readonly apply: ApplyReport; +} + +/** A segment the store refuses to stage (a malformed body or a bad transaction signature). */ +export class SegmentRefusedError extends Error { + readonly code = 'E_SYNC_SEGMENT_REFUSED'; +} + +/** The store's persisted cursor for `stream`, or null. */ +export function readStreamCursor(db: DatabaseSync, stream: string): StreamCursor | null { + if (!hasTable(db, '_sync_cursor')) return null; + const row = db.prepare('SELECT cursor_json FROM _sync_cursor WHERE stream = ?').get(stream) as + | { cursor_json: string } + | undefined; + return row ? (JSON.parse(row.cursor_json) as StreamCursor) : null; +} + +/** Decode a segment's transactions, or `null` for a vault delta segment. */ +function decode(seg: PulledStreamSegment): LedgerTxn[] | null { + let text: string; + try { + text = inflateRawSync(seg.plaintext).toString('utf8'); + } catch { + // Not deflated: a vault delta segment is plain JSON. + try { + const raw: unknown = JSON.parse(Buffer.from(seg.plaintext).toString('utf8')); + if ( + typeof raw === 'object' && + raw !== null && + 'kind' in raw && + raw.kind === VAULT_DELTA_KIND + ) { + return null; + } + } catch { + // fall through to the refusal + } + // @sync-invariant none:input-shape a segment body that is neither a ledger segment nor a vault delta is never staged + throw new SegmentRefusedError( + `segment ${seg.seq} of replica ${seg.replicaId}: not a ledger segment`, + ); + } + const parsed = LedgerTxn.array().safeParse(JSON.parse(text)); + if (!parsed.success) { + // @sync-invariant none:input-shape a ledger segment that does not parse is never staged + throw new SegmentRefusedError( + `segment ${seg.seq} of replica ${seg.replicaId}: transactions do not parse`, + ); + } + return parsed.data; +} + +/** + * Pull and stage a stream's new segments, then apply (module docs). Must run + * outside a transaction. A refused segment stops the pull with + * {@link SegmentRefusedError} after the pages before it were staged. + * + * @param db - The store. + * @param o - {@link PullStreamOptions}. + * @returns What was received, staged and applied. + */ +export async function pullStream( + db: DatabaseSync, + o: PullStreamOptions, +): Promise { + const now = o.now ?? Date.now; + let cursor = readStreamCursor(db, o.stream) ?? o.initialCursor; + let segments = 0; + let vaultDeltas = 0; + let staged = 0; + let redelivered = 0; + let head = cursor.after; + for (;;) { + const page = await o.pull(cursor); + head = page.head; + const nowIso = new Date(now()).toISOString(); + // Decode and verify the whole page before writing anything of it. + const decoded = page.segments.map((seg) => { + const txns = decode(seg); + if (txns !== null) { + const bad = o.verify(seg.deviceId, txns); + if (bad !== null) { + // @sync-invariant none:input-shape a transaction its device did not sign is never staged (§2.8) + throw new SegmentRefusedError( + `segment ${seg.seq} of replica ${seg.replicaId}: transaction ${bad} is not signed by device ${seg.deviceId}`, + ); + } + } + return { seg, txns }; + }); + withImmediateTransaction(db, () => { + const seen = db.prepare( + 'INSERT INTO _sync_seen_txn (stream, txn, seq) VALUES (?, ?, ?) ON CONFLICT (stream, txn) DO NOTHING', + ); + for (const { seg, txns } of decoded) { + segments += 1; + if (txns === null) { + 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, + ); + } + db.prepare( + 'INSERT INTO _sync_cursor (stream, cursor_json, updated_at) VALUES (?, ?, ?) ' + + 'ON CONFLICT (stream) DO UPDATE SET cursor_json = excluded.cursor_json, updated_at = excluded.updated_at', + ).run(o.stream, JSON.stringify(page.cursor), nowIso); + }); + cursor = page.cursor; + if (page.segments.length === 0 || cursor.after >= page.head) break; + } + const apply = applyStagedTxns(db, { + scope: o.scope, + stream: o.stream, + replica: o.replica, + now, + seal: o.seal, + ...(o.apply ?? {}), + }); + return { + stream: o.stream, + segments, + vaultDeltas, + staged, + redelivered, + after: cursor.after, + head, + apply, + }; +} diff --git a/packages/core/src/store/table-classification.ts b/packages/core/src/store/table-classification.ts index 4bdc8ac176..96861a961c 100644 --- a/packages/core/src/store/table-classification.ts +++ b/packages/core/src/store/table-classification.ts @@ -196,6 +196,18 @@ const PROJECT_TABLES: Readonly> = { source: 'journal spec t12342-t12343-journal-design §2.8 (T12343)', note: 'the transactions each persisted segment carries, in order. This device only. Created lazily by store/sync/schema.ts when a sync.* flag is first enabled (T12343)', }, + _sync_cursor: { + class: 'local-only', + status: 'optional-transient', + source: 'journal spec t12342-t12343-journal-design §3.1 Receive (T12343)', + note: "where this store's pull stands per stream: the PullCursor the journal client verifies the next page against, advanced in the transaction that stages what it covers. This device only. Created lazily by store/sync/schema.ts when a sync.* flag is first enabled (T12343)", + }, + _sync_seen_txn: { + class: 'local-only', + status: 'optional-transient', + 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_row_meta: { class: 'local-only', status: 'optional-transient', @@ -1430,6 +1442,18 @@ const GLOBAL_TABLES: Readonly> = { source: 'journal spec t12342-t12343-journal-design §2.8 (T12343)', note: 'the transactions each persisted segment carries, in order. This device only. Created lazily by store/sync/schema.ts when a sync.* flag is first enabled (T12343)', }, + _sync_cursor: { + class: 'local-only', + status: 'optional-transient', + source: 'journal spec t12342-t12343-journal-design §3.1 Receive (T12343)', + note: "where this store's pull stands per stream: the PullCursor the journal client verifies the next page against, advanced in the transaction that stages what it covers. This device only. Created lazily by store/sync/schema.ts when a sync.* flag is first enabled (T12343)", + }, + _sync_seen_txn: { + class: 'local-only', + status: 'optional-transient', + 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_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 2fb3fcbf73..32872db65f 100644 --- a/scripts/lint-no-raw-table-writes.mjs +++ b/scripts/lint-no-raw-table-writes.mjs @@ -396,6 +396,18 @@ export const EXEMPT = { 'capture undo (local-only): dropped when a failed or raced genesis cut is undone and no stream is cut, so undo is off again (T13296)', }, }, + 'packages/core/src/store/sync/pull.ts': { + _sync_cursor: { + count: 1, + reason: + 'pull cursor (local-only): advanced in the transaction that stages the segments it covers (§3.1 step 3, T12343)', + }, + _sync_seen_txn: { + count: 1, + reason: + 'staged transaction ids (local-only): a re-delivered transaction is never staged twice (§3.1, T12343 S5-1)', + }, + }, 'packages/core/src/store/sync/push.ts': { _sync_meta: { count: 2, From 0dbc899e662735a80a4320c966b3e627b18f7fc2 Mon Sep 17 00:00:00 2001 From: Keaton Hoskins Date: Tue, 6 Oct 2026 18:40:15 -0700 Subject: [PATCH 2/4] fix(sync): pull refuses while sync.pull is off (T12343 S5-1) pullStream ran regardless of the sync.pull flag; it now refuses (refused: 'sync.pull is off', nothing pulled or staged) unless the flag is on, as push does for sync.push. PullStreamReport gains `refused` and its apply report is null on a refusal. Co-Authored-By: Claude Opus 5.5 --- .changeset/t12343-pull.md | 3 ++- .../src/cloud/__tests__/nexus-vault.test.ts | 5 ++++- .../src/store/sync/__tests__/pull.test.ts | 17 +++++++++++++- packages/core/src/store/sync/pull.ts | 22 ++++++++++++++++++- 4 files changed, 43 insertions(+), 4 deletions(-) diff --git a/.changeset/t12343-pull.md b/.changeset/t12343-pull.md index f5af7912e9..5a4531ccc5 100644 --- a/.changeset/t12343-pull.md +++ b/.changeset/t12343-pull.md @@ -7,7 +7,8 @@ summary: Sync pull - stage and apply a stream's verified segments, never a trans A store can now pull its stream's journal (journal spec §3.1, §2.8; T12343 S5-1). -- **`pullStream(db, …)`** (`store/sync/pull.ts`) pages through the stream after the store's persisted cursor. For each page, in one +- **`pullStream(db, …)`** (`store/sync/pull.ts`) refuses while `sync.pull` is off. Otherwise it pages through the stream after the + store's persisted cursor. For each page, in one local transaction: - **Decode:** `deflate-raw` canonical JSON of `LedgerTxn[]`, parsed with the ledger schema. A vault count-delta segment (`cleo-vault-delta/v1`) carries no ops and is passed over. diff --git a/packages/core/src/cloud/__tests__/nexus-vault.test.ts b/packages/core/src/cloud/__tests__/nexus-vault.test.ts index f95aa138dd..9de695d88f 100644 --- a/packages/core/src/cloud/__tests__/nexus-vault.test.ts +++ b/packages/core/src/cloud/__tests__/nexus-vault.test.ts @@ -5138,10 +5138,13 @@ describe('sync enable push: the genesis checkpoint (T12343 S4-1b)', () => { }); const pushed = await on(m, () => pushSyncStream(vopts(m, { allowUnreleased: true }))); expect(pushed.pushed).toBe(1); + await on(m, async () => { + setSyncFlag(await storeOf(dbPath), 'sync.pull', true, { allowUnreleased: true }); + }); const pulled = await on(m, () => pullSyncStream(vopts(m))); expect(pulled).toMatchObject({ segments: 1, staged: 1, redelivered: 0, vaultDeltas: 0 }); expect(pulled.after).toBe(pushed.lastServerSeq); - expect(pulled.apply.applied).toBe(1); + expect(pulled.apply?.applied).toBe(1); await on(m, async () => { const db = await storeOf(dbPath); // The own echo was sequenced (§3.5): its transaction is no longer unsequenced. diff --git a/packages/core/src/store/sync/__tests__/pull.test.ts b/packages/core/src/store/sync/__tests__/pull.test.ts index a33e1cdfbb..d6e74a6f91 100644 --- a/packages/core/src/store/sync/__tests__/pull.test.ts +++ b/packages/core/src/store/sync/__tests__/pull.test.ts @@ -63,6 +63,7 @@ async function store(name: string): Promise { ); setCaptureEnabled(db, 'project', true, { schemaRoot: SYNC_SCHEMA }); setSyncFlag(db, 'sync.seal', true, { schemaRoot: SYNC_SCHEMA, allowUnreleased: true }); + setSyncFlag(db, 'sync.pull', true, { schemaRoot: SYNC_SCHEMA, allowUnreleased: true }); return db; } @@ -148,6 +149,7 @@ const pullOpts = (b: DatabaseSync, stream: ReturnType) => ({ deviceId === DEV_A ? firstBadTxnSignature(KEY_A.publicKey, STREAM, txns) : 0, initialCursor: START, now: () => ++clock, + env: {}, seal: seal(b, RB), }); @@ -167,7 +169,7 @@ describe('pullStream (S5-1)', () => { stream.append(authorSegment(a), 2); const r = await pullStream(b, pullOpts(b, stream)); expect(r).toMatchObject({ segments: 3, staged: 3, redelivered: 0, after: 3, head: 3 }); - expect(r.apply.applied).toBe(3); + expect(r.apply?.applied).toBe(3); expect(b.prepare("SELECT priority FROM tasks_tasks WHERE id = 'T1'").get()).toEqual({ priority: 'high', }); @@ -228,4 +230,17 @@ describe('pullStream (S5-1)', () => { await expect(pullStream(b, pullOpts(b, stream))).rejects.toThrow(/not a ledger segment/); expect(readStreamCursor(b, STREAM)).toBeNull(); }); + + 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); + }); }); diff --git a/packages/core/src/store/sync/pull.ts b/packages/core/src/store/sync/pull.ts index ab57f1e084..2f19fa7f8f 100644 --- a/packages/core/src/store/sync/pull.ts +++ b/packages/core/src/store/sync/pull.ts @@ -36,6 +36,7 @@ import type { TableScope } from '@cleocode/contracts'; import { LedgerTxn } from '@cleocode/contracts/ledger'; import { type ApplyReport, type ApplyStagedOptions, applyStagedTxns } from './apply/applier.js'; import { withImmediateTransaction } from './clock-store.js'; +import { isSyncFlagOn } from './flags.js'; import { stageTxns } from './inbox.js'; import { hasTable } from './schema.js'; @@ -90,11 +91,15 @@ export interface PullStreamOptions { readonly seal: ApplyStagedOptions['seal']; /** Applier options passed through (pages, budget). */ readonly apply?: Partial>; + /** Environment for the `sync.pull` kill switch. @defaultValue process.env */ + readonly env?: NodeJS.ProcessEnv; } /** What {@link pullStream} did. */ export interface PullStreamReport { readonly stream: string; + /** Why nothing was pulled (`sync.pull` off), or null. */ + readonly refused: string | null; /** Segments received, vault deltas passed over, transactions staged, re-deliveries skipped. */ readonly segments: number; readonly vaultDeltas: number; @@ -103,7 +108,8 @@ export interface PullStreamReport { /** The stream position the store has staged up to, and the server's head. */ readonly after: number; readonly head: number; - readonly apply: ApplyReport; + /** What the applier did, or null when the pull was refused. */ + readonly apply: ApplyReport | null; } /** A segment the store refuses to stage (a malformed body or a bad transaction signature). */ @@ -169,6 +175,19 @@ export async function pullStream( o: PullStreamOptions, ): Promise { const now = o.now ?? Date.now; + if (!isSyncFlagOn(db, 'sync.pull', o.env ?? process.env)) { + return { + stream: o.stream, + refused: 'sync.pull is off', + segments: 0, + vaultDeltas: 0, + staged: 0, + redelivered: 0, + after: readStreamCursor(db, o.stream)?.after ?? o.initialCursor.after, + head: 0, + apply: null, + }; + } let cursor = readStreamCursor(db, o.stream) ?? o.initialCursor; let segments = 0; let vaultDeltas = 0; @@ -239,6 +258,7 @@ export async function pullStream( }); return { stream: o.stream, + refused: null, segments, vaultDeltas, staged, From 4bd20cd28b436d8cc366352015a51236c7e60799 Mon Sep 17 00:00:00 2001 From: Keaton Hoskins Date: Tue, 6 Oct 2026 19:43:00 -0700 Subject: [PATCH 3/4] fix(sync): pull refuses to cross a journal genesis and applies what precedes a refused segment (T13306, T13307) review-hotfix on #1959: - T13306: a store whose last synced checkpoint is a vault snapshot from before the stream's journal genesis was seeded from it and pulled across the genesis, missing everything folded into it: silent divergence. It is now refused, nothing pulled, with the remedy to restore the journal checkpoint and join with `cleo sync enable push` (T12999); #1955's journaled-stream remedy says the same instead of "join by pulling". - T13307: a refused segment threw before apply, stranding earlier staged pages and sticking every retry. The pull now stops at the refused segment: what precedes it (in the same page too) is staged and applied, the cursor stops just before it, and the report's `refused` names its seq, replica and device. - Pruning: _sync_seen_txn rows first staged at or below the latest verified checkpoint's coversSeq (capped at the store's own position) are dropped at the start of each pull. - Nit: with sync.pull off, pullSyncStream refuses before any network call. Co-Authored-By: Claude Opus 5.5 --- .changeset/t12343-pull.md | 12 +- packages/contracts/src/nexus-account.ts | 5 +- .../src/cloud/__tests__/nexus-vault.test.ts | 35 ++++++ packages/core/src/cloud/nexus-vault.ts | 40 +++++- .../src/store/sync/__tests__/pull.test.ts | 49 ++++++-- packages/core/src/store/sync/pull.ts | 115 ++++++++++++++---- scripts/lint-no-raw-table-writes.mjs | 4 +- 7 files changed, 219 insertions(+), 41 deletions(-) diff --git a/.changeset/t12343-pull.md b/.changeset/t12343-pull.md index 5a4531ccc5..a1e2b9cff7 100644 --- a/.changeset/t12343-pull.md +++ b/.changeset/t12343-pull.md @@ -1,6 +1,6 @@ --- id: t12343-pull -tasks: [T12343] +tasks: [T12343, T13306, T13307] kind: feat summary: Sync pull - stage and apply a stream's verified segments, never a transaction twice (S5-1) --- @@ -13,14 +13,20 @@ A store can now pull its stream's journal (journal spec §3.1, §2.8; T12343 S5- - **Decode:** `deflate-raw` canonical JSON of `LedgerTxn[]`, parsed with the ledger schema. A vault count-delta segment (`cleo-vault-delta/v1`) carries no ops and is passed over. - **Verify:** every transaction's signature is checked against the segment's device key, before anything of the page is written. - A bad signature or a malformed body refuses the segment (`SegmentRefusedError`). + A bad signature or a malformed body refuses the segment. The pull stops there: what precedes it is staged and applied, the + cursor stops just before it, and the report's `refused` names its seq, replica and device (T13307). - **Deduplicate:** every transaction id already staged from the stream is skipped (`_sync_seen_txn`), so a re-delivered transaction is never staged or applied twice and a counter delta never doubles. - **Stage and advance:** the rest are staged, and the cursor advances (`_sync_cursor`). Then the applier applies, in stream order, own echoes included (they feed the fast path and the scoped rebase). +- **`_sync_seen_txn` stays bounded.** Rows first staged at or below the latest verified checkpoint's coversSeq (capped at the + store's own position) are pruned at the start of each pull (`pruneSeenTxns`). - **`pullSyncStream(opts)`** (`cloud/nexus-vault.ts`) wires `pullStream` to the journal client and the trusted device keys. It refuses - a store bound to another replica than the link. A store with no pull position starts after the checkpoint it last synced (its + a store bound to another replica than the link. With `sync.pull` off it refuses before any network call. A store whose last synced checkpoint is a vault snapshot from + before the stream's journal genesis is refused, with nothing pulled (T13306). The changes folded into that genesis never travel + as segments, so pulling across it would silently diverge. The remedy is to restore the journal checkpoint, then join with + `cleo sync enable push`. A store with no pull position starts after the checkpoint it last synced (its genesis, or the snapshot it restored), seeded from that checkpoint's signed replica map. - **New sync-journal migration** `t12343-pull-cursor`, local-only: `_sync_cursor` and `_sync_seen_txn`. Both are cleared from vault bundles, so a restored store resumes from its own checkpoint. diff --git a/packages/contracts/src/nexus-account.ts b/packages/contracts/src/nexus-account.ts index f5ef38f2e8..443e9c7ffa 100644 --- a/packages/contracts/src/nexus-account.ts +++ b/packages/contracts/src/nexus-account.ts @@ -104,7 +104,10 @@ export const NEXUS_ACCOUNT_ERROR_CODES = [ // Change-journal push (`cleo sync enable push`, T12343 S4-1b). /** The store is not fit to reach genesis (a precondition failed); nothing was cut. */ 'E_NEXUS_SYNC_REFUSED', - /** Another device already started this stream's change journal: this store joins it by pulling. */ + /** + * Another device already started this stream's change journal: this store restores its journal + * checkpoint, then joins the stream (T12999), never writing a second genesis. + */ 'E_NEXUS_SYNC_STREAM_JOURNALED', // Projects by name (`cleo cloud restore `, T13102). /** No project of the account has that name, label or id. */ diff --git a/packages/core/src/cloud/__tests__/nexus-vault.test.ts b/packages/core/src/cloud/__tests__/nexus-vault.test.ts index 9de695d88f..5d16ca7140 100644 --- a/packages/core/src/cloud/__tests__/nexus-vault.test.ts +++ b/packages/core/src/cloud/__tests__/nexus-vault.test.ts @@ -5156,6 +5156,41 @@ describe('sync enable push: the genesis checkpoint (T12343 S4-1b)', () => { expect(again).toMatchObject({ segments: 0, staged: 0 }); }); + it('S5-1: a store that synced a vault snapshot from before the journal genesis refuses to pull across it (T13306)', async () => { + const { m: a, dbPath } = await journalMachine(); + await on(a, () => pushNexusVault(vopts(a))); // a vault snapshot, before any journal + // B restores that vault snapshot and is attached under its own bound replica. + const b = await machine('b', DEVICE_B, REPLICA_B); + fake.replicas.get(REMOTE_PROJECT)?.set(REPLICA_B, DEVICE_B); + await restoreOntoB(b); + const bDb = path.join(b.root, '.cleo', 'cleo.db'); + const bound = await on(b, async () => { + const db = await storeOf(bDb); + setSyncFlag(db, 'sync.pull', true, { allowUnreleased: true }); + return ensureProjectReplica(db, { dbPath: bDb, mode: 'live' }).replicaId; + }); + b.replicaId = bound; + fake.replicas.get(REMOTE_PROJECT)?.set(bound, DEVICE_B); + link(b); + // A changes the store, then starts the journal: that change is folded into the genesis, never a segment. + await on(a, async () => { + (await storeOf(dbPath)).exec( + "INSERT INTO tasks_tasks (id, title, type, status, priority, uid, birth_fp) VALUES ('G1', 'title G1', 'task', 'pending', 'medium', 'uid-G1', 'fp-G1')", + ); + }); + await on(a, () => enableSyncPush(vopts(a, { allowUnreleased: true }))); + const refused = await failure(on(b, () => pullSyncStream(vopts(b)))); + expect(refused.code).toBe('E_NEXUS_SYNC_REFUSED'); + expect(refused.message).toMatch(/started its change journal .* would silently diverge/); + expect(refused.fix).toMatch(/restore the journal checkpoint first .*cleo sync enable push/); + await on(b, async () => { + const db = await storeOf(bDb); + expect((db.prepare('SELECT count(*) AS n FROM _sync_inbox').get() as { n: number }).n).toBe( + 0, + ); + }); + }); + it('S4-2: a device clock ahead of the server pauses push', async () => { const { m, dbPath } = await journalMachine(); await on(m, () => enableSyncPush(vopts(m, { allowUnreleased: true }))); diff --git a/packages/core/src/cloud/nexus-vault.ts b/packages/core/src/cloud/nexus-vault.ts index 52f156a309..99617058ac 100644 --- a/packages/core/src/cloud/nexus-vault.ts +++ b/packages/core/src/cloud/nexus-vault.ts @@ -103,7 +103,7 @@ import { sha256File, } from '../store/portable-bundle-scan.js'; import { FIRST_OPEN_LOCK_SUFFIX } from '../store/sqlite.js'; -import { UNRELEASED_FLAGS } from '../store/sync/flags.js'; +import { isSyncFlagOn, UNRELEASED_FLAGS } from '../store/sync/flags.js'; import { completeGenesis, cutGenesisWithSnapshot, @@ -1568,7 +1568,7 @@ async function enableSyncPushImpl( throw vaultError( 'E_NEXUS_SYNC_STREAM_JOURNALED', `${t.streamId} already carries the change journal: its head snapshot ${parent?.checkpointId} is a journal checkpoint`, - 'nothing was cut; this store joins the journal by pulling it (`cleo cloud sync`, once pull ships)', + 'nothing was cut; restore the journal checkpoint (`cleo cloud restore`), then join the stream with `cleo sync enable push` (T12999)', ); } const { state: synced } = syncedState(conn, t, parent ? [parent] : [], head.headCheckpointId); @@ -1848,13 +1848,25 @@ async function pullSyncStreamImpl(opts: NexusVaultCommandOptions = {}): Promise< 'nothing was pulled; relink the project (`cleo project link`) so both name one replica', ); } + // With sync.pull off, pullStream refuses at once: no network round trip. + if (!isSyncFlagOn(db, 'sync.pull')) { + return pullStream(db, { + scope: tableScopeOf(t), + stream: t.streamId, + replica: replicaId, + initialCursor: initialPullCursor(), + pull: async () => ({ segments: [], cursor: initialPullCursor(), head: 0 }), + verify: () => 0, + seal: () => {}, + }); + } + const head = await streamHead(conn, t.streamId); + const checkpoints = head.headCheckpointId ? await listCheckpoints(conn, t.streamId) : []; // A store with no pull position starts after the checkpoint it last synced // (the genesis it cut, or the snapshot it restored), seeded from its signed // replica map. let initialCursor = readStreamCursor(db, t.streamId); if (initialCursor === null) { - const head = await streamHead(conn, t.streamId); - const checkpoints = head.headCheckpointId ? await listCheckpoints(conn, t.streamId) : []; const { state: synced } = syncedState(conn, t, checkpoints, head.headCheckpointId); const from = checkpoints.find((c) => c.checkpointId === synced?.lastCheckpointId) ?? null; if (from === null) { @@ -1864,14 +1876,34 @@ async function pullSyncStreamImpl(opts: NexusVaultCommandOptions = {}): Promise< 'restore the stream (`cleo cloud restore`), or enable push (`cleo sync enable push`), first', ); } + // T13306: a vault snapshot from before the stream's journal genesis. What + // the genesis device changed between that snapshot and its cut was folded + // into the genesis checkpoint, never sent as segments: pulling across it + // would silently diverge. + const genesis = isJournalSnapshot(from) + ? undefined + : checkpoints.find((c) => isJournalSnapshot(c) && c.coversSeq >= from.coversSeq); + if (genesis) { + throw vaultError( + 'E_NEXUS_SYNC_REFUSED', + `${t.streamId} started its change journal at checkpoint ${genesis.checkpointId}, after the vault snapshot ${from.checkpointId} this store last synced: the changes folded into that journal checkpoint never travel as segments, so pulling from here would silently diverge`, + 'nothing was pulled; restore the journal checkpoint first (`cleo cloud restore`), then join the stream with `cleo sync enable push` (T12999)', + ); + } journal.verifyCheckpoint(from, key.signers); initialCursor = cursorFromCheckpoint(from); } + // Seen-txn rows past the latest verified checkpoint can no longer be re-delivered. + const headCheckpoint = checkpoints.find((c) => c.checkpointId === head.headCheckpointId) ?? null; + if (headCheckpoint) journal.verifyCheckpoint(headCheckpoint, key.signers); + const pruneSeenUpTo = + headCheckpoint === null ? null : Math.min(headCheckpoint.coversSeq, initialCursor.after); return pullStream(db, { scope: tableScopeOf(t), stream: t.streamId, replica: replicaId, initialCursor, + pruneSeenUpTo, pull: async (cursor) => { const page = await journal.pull( { diff --git a/packages/core/src/store/sync/__tests__/pull.test.ts b/packages/core/src/store/sync/__tests__/pull.test.ts index d6e74a6f91..ebca26a7c7 100644 --- a/packages/core/src/store/sync/__tests__/pull.test.ts +++ b/packages/core/src/store/sync/__tests__/pull.test.ts @@ -26,7 +26,6 @@ import { type PulledStreamSegment, pullStream, readStreamCursor, - SegmentRefusedError, type StreamCursor, } from '../pull.js'; import { sealPending } from '../sealer.js'; @@ -212,23 +211,53 @@ describe('pullStream (S5-1)', () => { expect(n(b, "SELECT count(*) AS n FROM tasks_tasks WHERE id = 'T1'")).toBe(1); }); - it('a transaction its device did not sign refuses the segment: nothing of its page is staged', async () => { + it('a refused segment stops the pull there: what precedes it is applied, the cursor stops before it, a retry re-stages nothing (T13307)', async () => { const a = await store('a'); const b = await store('b'); const stream = fakeStream(); write(a, addTask('T1')); - stream.append(authorSegment(a), 0, RA, 'dev-mallory'); // served under another device's pin - await expect(pullStream(b, pullOpts(b, stream))).rejects.toBeInstanceOf(SegmentRefusedError); - expect(n(b, 'SELECT count(*) AS n FROM _sync_inbox')).toBe(0); - expect(readStreamCursor(b, STREAM)).toBeNull(); + stream.append(authorSegment(a), 0); + write(a, addTask('T2')); + stream.append(authorSegment(a), 1); + write(a, addTask('T3')); + stream.append(authorSegment(a), 2, RA, 'dev-mallory'); // page 2: served under another device's pin + const r = await pullStream(b, pullOpts(b, stream)); + expect(r.refused).toMatch(/segment 3 of replica .*not signed by device dev-mallory/); + expect(r).toMatchObject({ segments: 2, staged: 2, after: 2 }); + expect(r.apply?.applied).toBe(2); + expect(n(b, "SELECT count(*) AS n FROM tasks_tasks WHERE id IN ('T1', 'T2')")).toBe(2); + expect(n(b, "SELECT count(*) AS n FROM tasks_tasks WHERE id = 'T3'")).toBe(0); + expect(readStreamCursor(b, STREAM)).toMatchObject({ after: 2 }); + const retry = await pullStream(b, pullOpts(b, stream)); + expect(retry).toMatchObject({ segments: 0, staged: 0, after: 2 }); + expect(retry.refused).toMatch(/segment 3/); + }); + + it('a refused segment mid-page: the segments before it in that page are staged and applied', async () => { + const a = await store('a'); + const b = await store('b'); + const stream = fakeStream(); + write(a, addTask('T1')); + stream.append(authorSegment(a), 0); + stream.append(Buffer.from('not a segment'), 1); // same page as seq 1 + const r = await pullStream(b, pullOpts(b, stream)); + expect(r.refused).toMatch(/segment 2 of replica .*not a ledger segment/); + expect(r).toMatchObject({ segments: 1, staged: 1, after: 1 }); + expect(n(b, "SELECT count(*) AS n FROM tasks_tasks WHERE id = 'T1'")).toBe(1); }); - it('a body that is neither a ledger segment nor a vault delta is refused', async () => { + it('seen-txn rows at or below the prune floor are dropped, later ones kept', async () => { + const a = await store('a'); const b = await store('b'); const stream = fakeStream(); - stream.append(Buffer.from('not a segment'), 0); - await expect(pullStream(b, pullOpts(b, stream))).rejects.toThrow(/not a ledger segment/); - expect(readStreamCursor(b, STREAM)).toBeNull(); + write(a, addTask('T1')); + stream.append(authorSegment(a), 0); + write(a, addTask('T2')); + stream.append(authorSegment(a), 1); + 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 }]); }); it('refuses with sync.pull off: nothing pulled, nothing staged', async () => { diff --git a/packages/core/src/store/sync/pull.ts b/packages/core/src/store/sync/pull.ts index 2f19fa7f8f..cd66eb154f 100644 --- a/packages/core/src/store/sync/pull.ts +++ b/packages/core/src/store/sync/pull.ts @@ -11,8 +11,9 @@ * correction a vault push appended before the stream's genesis) carries no * ops and is passed over; * - checks every transaction's signature against the segment's device key - * (§2.8); one bad signature refuses the segment, and the pull stops before - * staging anything of it; + * (§2.8); one bad signature refuses the segment, and the pull stops there: + * the segments before it are staged and applied, the segment and what + * follows are not; * - 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; @@ -93,12 +94,22 @@ export interface PullStreamOptions { readonly apply?: Partial>; /** 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 (the latest + * verified checkpoint's coversSeq), or null to keep every row. + */ + readonly pruneSeenUpTo?: number | null; } /** What {@link pullStream} did. */ export interface PullStreamReport { readonly stream: string; - /** Why nothing was pulled (`sync.pull` off), or null. */ + /** + * Why the pull stopped short, or null: `sync.pull` off (nothing pulled), or + * a refused segment (a malformed body or a transaction its device did not + * sign), naming its seq, replica and device. Everything before a refused + * segment was staged and applied, and the cursor stops just before it. + */ readonly refused: string | null; /** Segments received, vault deltas passed over, transactions staged, re-deliveries skipped. */ readonly segments: number; @@ -161,10 +172,61 @@ function decode(seg: PulledStreamSegment): LedgerTxn[] | null { return parsed.data; } +/** A segment's transactions (null for a vault delta), or why it is refused. */ +function refusalOf( + seg: PulledStreamSegment, + verify: TxnVerifier, +): { readonly txns: LedgerTxn[] | null } | string { + let txns: LedgerTxn[] | null; + try { + txns = decode(seg); + } catch (err) { + if (err instanceof SegmentRefusedError) return err.message; + throw err; + } + if (txns !== null) { + const bad = verify(seg.deviceId, txns); + if (bad !== null) { + return `segment ${seg.seq} of replica ${seg.replicaId}: transaction ${bad} is not signed by device ${seg.deviceId}`; + } + } + return { txns }; +} + +/** `cursor` advanced over `segs` (a page cut short by a refused segment). */ +function advance(cursor: StreamCursor, segs: readonly PulledStreamSegment[]): StreamCursor { + const replicas = { ...cursor.replicas }; + for (const s of segs) replicas[s.replicaId] = { deviceId: s.deviceId, replicaSeq: s.replicaSeq }; + return { + after: segs.at(-1)?.seq ?? cursor.after, + knowsAllReplicas: cursor.knowsAllReplicas, + replicas, + }; +} + +/** + * Drop the seen-txn rows of `stream` first staged at or below `seq`: past a + * verified checkpoint's coversSeq no segment re-delivers them (§3.1). Keeps + * `_sync_seen_txn` bounded. + * + * @param db - The store. + * @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, + ); +} + /** * Pull and stage a stream's new segments, then apply (module docs). Must run - * outside a transaction. A refused segment stops the pull with - * {@link SegmentRefusedError} after the pages before it were staged. + * outside a transaction. A refused segment stops the pull: everything + * before it is staged and applied, the cursor stops just before it, and the + * report's `refused` names it (T13307), so the next pull reaches it again + * without re-staging anything. * * @param db - The store. * @param o - {@link PullStreamOptions}. @@ -194,24 +256,34 @@ export async function pullStream( 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); + } for (;;) { const page = await o.pull(cursor); head = page.head; const nowIso = new Date(now()).toISOString(); - // Decode and verify the whole page before writing anything of it. - const decoded = page.segments.map((seg) => { - const txns = decode(seg); - if (txns !== null) { - const bad = o.verify(seg.deviceId, txns); - if (bad !== null) { - // @sync-invariant none:input-shape a transaction its device did not sign is never staged (§2.8) - throw new SegmentRefusedError( - `segment ${seg.seq} of replica ${seg.replicaId}: transaction ${bad} is not signed by device ${seg.deviceId}`, - ); - } + // Decode and verify the page before writing anything of it, up to the + // first segment refused (T13307): what precedes it is staged and applied, + // the cursor stops just before it, and the refusal is reported. + const decoded: Array<{ seg: PulledStreamSegment; txns: LedgerTxn[] | null }> = []; + for (const seg of page.segments) { + const why = refusalOf(seg, o.verify); + if (typeof why === 'string') { + refusedSegment = why; + break; } - return { seg, txns }; - }); + 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', @@ -243,9 +315,10 @@ export async function pullStream( db.prepare( 'INSERT INTO _sync_cursor (stream, cursor_json, updated_at) VALUES (?, ?, ?) ' + 'ON CONFLICT (stream) DO UPDATE SET cursor_json = excluded.cursor_json, updated_at = excluded.updated_at', - ).run(o.stream, JSON.stringify(page.cursor), nowIso); + ).run(o.stream, JSON.stringify(pageCursor), nowIso); }); - cursor = page.cursor; + cursor = pageCursor; + if (refusedSegment !== null) break; if (page.segments.length === 0 || cursor.after >= page.head) break; } const apply = applyStagedTxns(db, { @@ -258,7 +331,7 @@ export async function pullStream( }); return { stream: o.stream, - refused: null, + refused: refusedSegment, segments, vaultDeltas, staged, diff --git a/scripts/lint-no-raw-table-writes.mjs b/scripts/lint-no-raw-table-writes.mjs index 32872db65f..eb9a9820f5 100644 --- a/scripts/lint-no-raw-table-writes.mjs +++ b/scripts/lint-no-raw-table-writes.mjs @@ -403,9 +403,9 @@ export const EXEMPT = { 'pull cursor (local-only): advanced in the transaction that stages the segments it covers (§3.1 step 3, T12343)', }, _sync_seen_txn: { - count: 1, + count: 2, reason: - 'staged transaction ids (local-only): a re-delivered transaction is never staged twice (§3.1, T12343 S5-1)', + '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)', }, }, 'packages/core/src/store/sync/push.ts': { From f1460b1c4a4f5e028942e9e4c31fd41434c5e45b Mon Sep 17 00:00:00 2001 From: Keaton Hoskins Date: Tue, 6 Oct 2026 20:46:33 -0700 Subject: [PATCH 4/4] fix(sync): the cloud pull does not prune seen-txn rows until a refusal floor exists (T13256) A seen txn can return in a NEW segment above any checkpoint (a retired replica's late segment, a rebind re-pushing a stored-but-unrecorded upload); with its first-delivery row pruned by checkpoint coversSeq it would stage and apply twice (an $inc column would double). pullSyncStream no longer passes pruneSeenUpTo; pruneSeenTxns stays, documented as safe only below a floor the applier refuses under (the receive watermark). Co-Authored-By: Claude Opus 5.5 --- .changeset/t12343-pull.md | 6 ++++-- packages/core/src/cloud/nexus-vault.ts | 13 +++++++------ packages/core/src/store/sync/pull.ts | 14 +++++++++----- 3 files changed, 20 insertions(+), 13 deletions(-) diff --git a/.changeset/t12343-pull.md b/.changeset/t12343-pull.md index a1e2b9cff7..912d961a62 100644 --- a/.changeset/t12343-pull.md +++ b/.changeset/t12343-pull.md @@ -20,8 +20,10 @@ A store can now pull its stream's journal (journal spec §3.1, §2.8; T12343 S5- - **Stage and advance:** the rest are staged, and the cursor advances (`_sync_cursor`). Then the applier applies, in stream order, own echoes included (they feed the fast path and the scoped rebase). -- **`_sync_seen_txn` stays bounded.** Rows first staged at or below the latest verified checkpoint's coversSeq (capped at the - store's own position) are pruned at the start of each pull (`pruneSeenTxns`). +- **`_sync_seen_txn` is not pruned yet.** A transaction already seen can come back in a new segment above any checkpoint, for + example a retired replica's late segment or a rebind's re-push. If its first-delivery row had been dropped, it would apply + twice. `pruneSeenTxns` exists, but the cloud pull calls it only once the receive watermark (T13256) provides a floor the + applier refuses to go below. - **`pullSyncStream(opts)`** (`cloud/nexus-vault.ts`) wires `pullStream` to the journal client and the trusted device keys. It refuses a store bound to another replica than the link. With `sync.pull` off it refuses before any network call. A store whose last synced checkpoint is a vault snapshot from before the stream's journal genesis is refused, with nothing pulled (T13306). The changes folded into that genesis never travel diff --git a/packages/core/src/cloud/nexus-vault.ts b/packages/core/src/cloud/nexus-vault.ts index 99617058ac..412760df5c 100644 --- a/packages/core/src/cloud/nexus-vault.ts +++ b/packages/core/src/cloud/nexus-vault.ts @@ -1893,17 +1893,18 @@ async function pullSyncStreamImpl(opts: NexusVaultCommandOptions = {}): Promise< journal.verifyCheckpoint(from, key.signers); initialCursor = cursorFromCheckpoint(from); } - // Seen-txn rows past the latest verified checkpoint can no longer be re-delivered. - const headCheckpoint = checkpoints.find((c) => c.checkpointId === head.headCheckpointId) ?? null; - if (headCheckpoint) journal.verifyCheckpoint(headCheckpoint, key.signers); - const pruneSeenUpTo = - headCheckpoint === null ? null : Math.min(headCheckpoint.coversSeq, initialCursor.after); + // Seen-txn rows are never pruned here yet: a txn already seen can come + // back in a NEW segment above any checkpoint (a retired replica's late + // segment, a rebind re-pushing an upload that was stored but never + // recorded), and with its first-delivery row gone it would apply twice. + // Pruning waits for a floor the applier refuses below anyway, such as the + // receive watermark (T13256), and then prunes by that floor, not by + // stream seq. return pullStream(db, { scope: tableScopeOf(t), stream: t.streamId, replica: replicaId, initialCursor, - pruneSeenUpTo, pull: async (cursor) => { const page = await journal.pull( { diff --git a/packages/core/src/store/sync/pull.ts b/packages/core/src/store/sync/pull.ts index cd66eb154f..7a8c394007 100644 --- a/packages/core/src/store/sync/pull.ts +++ b/packages/core/src/store/sync/pull.ts @@ -95,8 +95,9 @@ 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 (the latest - * verified checkpoint's coversSeq), or null to keep every row. + * 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). */ readonly pruneSeenUpTo?: number | null; } @@ -205,9 +206,12 @@ function advance(cursor: StreamCursor, segs: readonly PulledStreamSegment[]): St } /** - * Drop the seen-txn rows of `stream` first staged at or below `seq`: past a - * verified checkpoint's coversSeq no segment re-delivers them (§3.1). Keeps - * `_sync_seen_txn` bounded. + * 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). * * @param db - The store. * @param stream - The stream.