From b319c4859be57a79277f861f8795538aaab1eab2 Mon Sep 17 00:00:00 2001 From: Keaton Hoskins Date: Tue, 6 Oct 2026 20:31:07 -0700 Subject: [PATCH 1/3] feat(sync): a restored store joins a journaled stream; two-device sync end to end (T13312, T12999, T12996) A store restored from another device's journal checkpoint had no row meta and no join state, so enable push refused it and a second device could not enter the stream. - Restore keeps _sync_row_meta and _sync_field_leave on a journal checkpoint (carryMachineState snapshotCarried). - joinStream: refuses unless every row matches its restored meta, there are no waiting captures and the triggers are present, then cuts at the capture position, baselines the ledgers, turns on push/pull/undo and seeds the pull cursor from the checkpoint. Sealing is enabled only after the check. - enableSyncPush joins (status 'joined') when the head journal checkpoint is the one this store synced and its manifest matches; a changed store is refused with the restore --force remedy, never folded; any other store gets E_NEXUS_SYNC_STREAM_JOURNALED. - Tests: the two-device headline (A and B each write, sync, see the other's write; a concurrent done/cancel converges by the typed rules), pull-only joiner reports synced with the push leg skipped, the three refusals, the T13306 remedy followed through to a join. All 7 mutations killed. Co-Authored-By: Claude Opus 5.5 --- .changeset/t13312-journal-join.md | 28 +++ packages/contracts/src/nexus-vault.ts | 6 +- .../src/cloud/__tests__/nexus-vault.test.ts | 216 +++++++++++++++++- packages/core/src/cloud/nexus-vault.ts | 78 ++++++- packages/core/src/store/sync/genesis.ts | 112 ++++++++- packages/core/src/store/sync/pull.ts | 27 ++- packages/core/src/store/vault-manifest.ts | 11 + scripts/lint-no-raw-table-writes.mjs | 4 +- 8 files changed, 455 insertions(+), 27 deletions(-) create mode 100644 .changeset/t13312-journal-join.md diff --git a/.changeset/t13312-journal-join.md b/.changeset/t13312-journal-join.md new file mode 100644 index 0000000000..dce06f66ed --- /dev/null +++ b/.changeset/t13312-journal-join.md @@ -0,0 +1,28 @@ +--- +id: t13312-journal-join +tasks: [T13312, T12999, T12996] +kind: feat +summary: "`cleo sync enable push` joins a stream another device journaled, from the journal checkpoint this store restored" +--- + +A second device can now enter a stream whose change journal another device already started (T13312, T12999). Before this, a +store restored from the journal checkpoint carried no row meta and no join state, so `cleo sync enable push` refused it and +there was no way in. + +- **Restore keeps the journal's row state.** On a journal checkpoint, the restore keeps the snapshot's `_sync_row_meta` and + `_sync_field_leave` rows instead of clearing them as machine-local. +- **The join.** On a stream whose head is a journal checkpoint this store restored, `cleo sync enable push` returns status + `joined`: + - the store's cut is placed at its current capture position; + - the ledgers are baselined; + - push, pull and undo are turned on; + - pulling resumes right after the restored checkpoint. + No second genesis is pushed. +- **A changed store is refused, never folded.** The join is refused with `E_NEXUS_SYNC_REFUSED` when: + - any table's rows differ from the checkpoint's manifest; + - any row no longer matches its restored row meta; + - a capture is still waiting to seal. + The remedy is to restore the checkpoint again (`cleo cloud restore --force`), then join. A refused join changes nothing. +- **A store that never restored the journal checkpoint** gets `E_NEXUS_SYNC_STREAM_JOURNALED`, naming the restore-then-join + path. The pull refusal for a store behind the journal genesis (T13306) names the same path. +- `sync.push` and `sync.pull` stay unreleased. diff --git a/packages/contracts/src/nexus-vault.ts b/packages/contracts/src/nexus-vault.ts index ae78c7dfc6..21bae1ada3 100644 --- a/packages/contracts/src/nexus-vault.ts +++ b/packages/contracts/src/nexus-vault.ts @@ -186,9 +186,11 @@ export interface CloudSyncPushEnableResult { streamId: string; /** * `enabled`: the cut was recorded and its genesis checkpoint stored; `resumed`: a cut recorded - * by an earlier run whose checkpoint push failed was pushed now; `already`: push was already on. + * by an earlier run whose checkpoint push failed was pushed now; `already`: push was already on; + * `joined`: this store holds the stream's journal checkpoint another device cut, and now pushes its + * own writes and pulls the stream's after it, with no genesis of its own (T13312). */ - status: 'enabled' | 'resumed' | 'already'; + status: 'enabled' | 'resumed' | 'already' | 'joined'; /** The genesis cut: the highest capture seq the checkpoint carries. */ cut: number; /** Transactions sealed before the cut, folded into the checkpoint, and rows given genesis row meta. */ diff --git a/packages/core/src/cloud/__tests__/nexus-vault.test.ts b/packages/core/src/cloud/__tests__/nexus-vault.test.ts index 7cae0f777a..09b27973f2 100644 --- a/packages/core/src/cloud/__tests__/nexus-vault.test.ts +++ b/packages/core/src/cloud/__tests__/nexus-vault.test.ts @@ -67,6 +67,7 @@ import { import { setCaptureEnabled } from '../../store/sync/capture.js'; import { isSyncFlagOn, setSyncFlag } from '../../store/sync/flags.js'; import { cutGenesis, genesisCutOf, genesisPending } from '../../store/sync/genesis.js'; +import { readStreamCursor } from '../../store/sync/pull.js'; import { ensureProjectReplica, storeHwm } from '../../store/sync/replica.js'; import { ensureSyncSchema } from '../../store/sync/schema.js'; import { @@ -4861,9 +4862,15 @@ describe('cloud project link stores a new project key with its registration (onb describe('sync enable push: the genesis checkpoint (T12343 S4-1b)', () => { const SYNC_JOURNAL = path.resolve(import.meta.dirname, '../../../migrations/sync-journal'); - /** A machine whose project store is a real cleo.db with capture and seal on, bound and linked. */ - async function journalMachine(): Promise<{ m: Machine; dbPath: string }> { - const m = await machine('a', DEVICE_A, REPLICA_A); + /** + * A machine whose project store is a real cleo.db with capture and seal on, bound and linked. + * Device A by default; device B attaches to the project A registered. + */ + async function journalMachine(device: 'a' | 'b' = 'a'): Promise<{ m: Machine; dbPath: string }> { + const m = + device === 'a' + ? await machine('a', DEVICE_A, REPLICA_A) + : await machine('b', DEVICE_B, REPLICA_B); const cleo = path.join(m.root, '.cleo'); fs.mkdirSync(cleo, { recursive: true }); fs.writeFileSync(path.join(cleo, 'project-id'), `${LOCAL_PROJECT}\n`); @@ -4887,7 +4894,8 @@ describe('sync enable push: the genesis checkpoint (T12343 S4-1b)', () => { return ensureProjectReplica(db, { dbPath, mode: 'live' }).replicaId; }); m.replicaId = replicaId; - fake.addProject(REMOTE_PROJECT, { [replicaId]: DEVICE_A }); + if (device === 'a') fake.addProject(REMOTE_PROJECT, { [replicaId]: DEVICE_A }); + else fake.replicas.get(REMOTE_PROJECT)?.set(replicaId, DEVICE_B); link(m); return { m, dbPath }; } @@ -4895,6 +4903,25 @@ describe('sync enable push: the genesis checkpoint (T12343 S4-1b)', () => { const storeOf = async (dbPath: string) => getDualScopeNativeDb(await openDualScopeDbAtPath('project', dbPath)); + /** + * Device B with A's journal checkpoint restored, attached under its own bound + * replica, and not yet joined. + */ + async function restoredJoiner(): Promise<{ b: Machine; bDb: string }> { + 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 () => ensureProjectReplica(await storeOf(bDb), { dbPath: bDb, mode: 'live' }).replicaId, + ); + b.replicaId = bound; + fake.replicas.get(REMOTE_PROJECT)?.set(bound, DEVICE_B); + link(b); + return { b, bDb }; + } + it('cuts the store and pushes a checkpoint/v3 genesis the server accepts; push turns on and the vault steps aside', async () => { const { m, dbPath } = await journalMachine(); const r = await on(m, () => enableSyncPush(vopts(m, { allowUnreleased: true }))); @@ -4925,13 +4952,22 @@ describe('sync enable push: the genesis checkpoint (T12343 S4-1b)', () => { const vault = await failure(on(m, () => pushNexusVault(vopts(m)))); expect(vault.code).toBe('E_NEXUS_VAULT_STREAM_UPGRADED'); expect(s.checkpoints).toHaveLength(1); - // A store that never cut this stream (its cut forgotten) meets a journal head: it joins by - // pulling, never with a second genesis. + // A store that holds the journal checkpoint but forgot its cut joins it, never with a + // second genesis (T13312). await on(m, async () => { (await storeOf(dbPath)).exec("DELETE FROM _sync_meta WHERE key LIKE 'genesis_%'"); }); - const second = await failure(on(m, () => enableSyncPush(vopts(m, { allowUnreleased: true })))); + expect((await on(m, () => enableSyncPush(vopts(m, { allowUnreleased: true })))).status).toBe( + 'joined', + ); + expect(s.checkpoints).toHaveLength(1); + // A store that never restored that checkpoint is told to restore it, then join. + const { m: other } = await journalMachine('b'); + const second = await failure( + on(other, () => enableSyncPush(vopts(other, { allowUnreleased: true }))), + ); expect(second.code).toBe('E_NEXUS_SYNC_STREAM_JOURNALED'); + expect(second.fix).toContain('cleo cloud restore'); expect(s.checkpoints).toHaveLength(1); }); @@ -5272,6 +5308,172 @@ describe('sync enable push: the genesis checkpoint (T12343 S4-1b)', () => { 0, ); }); + // The remedy works: B restores the journal checkpoint, joins, and holds A's folded change. + const restored = await on(b, () => restoreNexusVault(vopts(b, { mode: 'pull', force: true }))); + expect(restored.status).toBe('restored'); + expect((await on(b, () => enableSyncPush(vopts(b, { allowUnreleased: true })))).status).toBe( + 'joined', + ); + expect( + await on(b, async () => + (await storeOf(bDb)).prepare("SELECT title FROM tasks_tasks WHERE id = 'G1'").get(), + ), + ).toEqual({ title: 'title G1' }); + expect((await on(b, () => pullSyncStream(vopts(b)))).refused).toBeNull(); + }); + + it("two devices (T13312, T12996): B restores A's genesis and joins; cloud sync carries each device's writes to the other and converges a concurrent status change", async () => { + const { m: a, dbPath: aDb } = await journalMachine(); + await on(a, () => enableSyncPush(vopts(a, { allowUnreleased: true }))); + await on(a, async () => { + setSyncFlag(await storeOf(aDb), 'sync.pull', true, { allowUnreleased: true }); + }); + const { b, bDb } = await restoredJoiner(); + const joined = await on(b, () => enableSyncPush(vopts(b, { allowUnreleased: true }))); + expect(joined.status).toBe('joined'); + const journalCps = fake + .stream(STREAM) + .checkpoints.filter((c) => manifestVersion(c.manifest) === 3); + expect(journalCps).toHaveLength(1); + // B pulls on from the checkpoint it restored. + const genesisCp = journalCps[0]; + if (!genesisCp) throw new Error('no journal checkpoint'); + expect(await on(b, async () => readStreamCursor(await storeOf(bDb), STREAM))).toEqual( + cursorFromCheckpoint(genesisCp), + ); + + const sync = async (m: Machine) => { + const r = await on(m, () => cloudSync(vopts(m, { scope: 'project', allowUnreleased: true }))); + expect(r.streams[0]?.refused ?? null, `${m.name}: ${r.streams[0]?.refused}`).toBeNull(); + return r.streams[0]; + }; + const value = async (m: Machine, dbFile: string, sql: string) => + on(m, async () => (await storeOf(dbFile)).prepare(sql).get()); + + // Each device edits; both sync; each sees the other's write. + await on(a, async () => { + (await storeOf(aDb)).exec("UPDATE tasks_tasks SET title = 'from A' WHERE id = 'T1'"); + }); + await on(b, async () => { + (await storeOf(bDb)).exec( + "INSERT INTO tasks_tasks (id, title, type, status, priority, uid, birth_fp) VALUES ('T9', 'from B', 'task', 'pending', 'medium', 'uid-T9', 'fp-T9')", + ); + }); + await sync(a); + await sync(b); + await sync(a); + expect(await value(b, bDb, "SELECT title FROM tasks_tasks WHERE id = 'T1'")).toEqual({ + title: 'from A', + }); + expect(await value(a, aDb, "SELECT title FROM tasks_tasks WHERE id = 'T9'")).toEqual({ + title: 'from B', + }); + + // A concurrent status change of one task: the typed rules decide it the same way on both. + await on(a, async () => { + (await storeOf(aDb)).exec( + "UPDATE tasks_tasks SET status = 'done', pipeline_stage = 'contribution', completed_at = '2026-10-07T00:00:00.000Z' WHERE id = 'T2'", + ); + }); + await on(b, async () => { + (await storeOf(bDb)).exec( + "UPDATE tasks_tasks SET status = 'cancelled', pipeline_stage = 'cancelled', cancelled_at = '2026-10-07T00:00:01.000Z', cancellation_reason = 'dup' WHERE id = 'T2'", + ); + }); + await sync(a); + await sync(b); + await sync(a); + await sync(b); + const statusSql = + "SELECT status, pipeline_stage, completed_at, cancelled_at, cancellation_reason FROM tasks_tasks WHERE id = 'T2'"; + const onA = await value(a, aDb, statusSql); + const onB = await value(b, bDb, statusSql); + expect(onB).toEqual(onA); + // Both are absorbing, so the later write (B's cancel) wins its LWW; the + // stamps move with it as one group and the stage follows the status. + expect(onA).toEqual({ + status: 'cancelled', + pipeline_stage: 'cancelled', + completed_at: null, + cancelled_at: '2026-10-07T00:00:01.000Z', + cancellation_reason: 'dup', + }); + }); + + it('a joined store with push off syncs pull-only: synced, the push leg skipped (T13312)', async () => { + const { m: a, dbPath: aDb } = await journalMachine(); + await on(a, () => enableSyncPush(vopts(a, { allowUnreleased: true }))); + const { b, bDb } = await restoredJoiner(); + expect((await on(b, () => enableSyncPush(vopts(b, { allowUnreleased: true })))).status).toBe( + 'joined', + ); + await on(b, async () => { + setSyncFlag(await storeOf(bDb), 'sync.push', false, { allowUnreleased: true }); + }); + await on(a, async () => { + (await storeOf(aDb)).exec("UPDATE tasks_tasks SET title = 'from A' WHERE id = 'T1'"); + }); + await on(a, () => pushSyncStream(vopts(a, { allowUnreleased: true }))); + const r = await on(b, () => cloudSync(vopts(b, { scope: 'project', allowUnreleased: true }))); + expect(r.streams[0]).toMatchObject({ status: 'synced', skipped: ['sync.push is off'] }); + expect( + await on(b, async () => + (await storeOf(bDb)).prepare("SELECT title FROM tasks_tasks WHERE id = 'T1'").get(), + ), + ).toEqual({ title: 'from A' }); + }); + + it('a join is refused, never folded, when the store changed since its restore (T13312)', async () => { + const { m: a } = await journalMachine(); + await on(a, () => enableSyncPush(vopts(a, { allowUnreleased: true }))); + const { b, bDb } = await restoredJoiner(); + await on(b, async () => { + (await storeOf(bDb)).exec("UPDATE tasks_tasks SET title = 'edited on B' WHERE id = 'T1'"); + }); + const err = await failure(on(b, () => enableSyncPush(vopts(b, { allowUnreleased: true })))); + expect(err.code).toBe('E_NEXUS_SYNC_REFUSED'); + expect(err.message).toContain('tasks_tasks'); + expect(err.fix).toContain('cleo cloud restore --force'); + await on(b, async () => { + const db = await storeOf(bDb); + expect(genesisCutOf(db, STREAM)).toBeUndefined(); + expect(isSyncFlagOn(db, 'sync.push', {})).toBe(false); + }); + expect(fake.stream(STREAM).segments).toHaveLength(0); + }); + + it('a join is refused, never sealed past, while a capture waits to seal (T13312)', async () => { + const { m: a } = await journalMachine(); + await on(a, () => enableSyncPush(vopts(a, { allowUnreleased: true }))); + const { b, bDb } = await restoredJoiner(); + // A write captured but not sealed: joining would seal it as if the checkpoint authored it. + await on(b, async () => { + (await storeOf(bDb)).exec( + "INSERT INTO _sync_capture (tbl, op, rk, uid, img, at_ms) VALUES ('tasks_tasks', 'U', 'T1', 'uid-T1', '{}', 0)", + ); + }); + const err = await failure(on(b, () => enableSyncPush(vopts(b, { allowUnreleased: true })))); + expect(err.code).toBe('E_NEXUS_SYNC_REFUSED'); + expect(err.message).toContain('captures still waiting to seal'); + }); + + it('a join is refused when the restored row meta does not match the rows (T13312)', async () => { + const { m: a } = await journalMachine(); + await on(a, () => enableSyncPush(vopts(a, { allowUnreleased: true }))); + const { b, bDb } = await restoredJoiner(); + await on(b, async () => { + (await storeOf(bDb)).exec( + "DELETE FROM _sync_row_meta WHERE tbl = 'tasks_tasks' AND uid = (SELECT uid FROM _sync_row_meta WHERE tbl = 'tasks_tasks' ORDER BY uid LIMIT 1)", + ); + }); + const err = await failure(on(b, () => enableSyncPush(vopts(b, { allowUnreleased: true })))); + expect(err.code).toBe('E_NEXUS_SYNC_REFUSED'); + expect(err.message).toContain('tasks_tasks'); + await on(b, async () => { + const db = await storeOf(bDb); + expect(genesisCutOf(db, STREAM)).toBeUndefined(); + expect(isSyncFlagOn(db, 'sync.seal', {})).toBe(false); + }); }); it('S4-2: a device clock ahead of the server pauses push', async () => { diff --git a/packages/core/src/cloud/nexus-vault.ts b/packages/core/src/cloud/nexus-vault.ts index 32754733e5..e6bd37a637 100644 --- a/packages/core/src/cloud/nexus-vault.ts +++ b/packages/core/src/cloud/nexus-vault.ts @@ -112,6 +112,7 @@ import { GenesisRacedError, genesisCutOf, genesisPending, + joinStream, readGenesisCut, } from '../store/sync/genesis.js'; import { type PullStreamReport, pullStream, readStreamCursor } from '../store/sync/pull.js'; @@ -797,6 +798,17 @@ function compareWithSynced( * hold values the snapshot strips, and a restore takes these tables from the * live store anyway. */ +/** + * The merge state a journal checkpoint carries for the rows it holds (journal + * spec §2.10: row meta travels inside checkpoint bundles): a restore keeps + * the snapshot's, never this machine's, so the restored rows keep their HLCs + * and a JOIN can push and pull them (T13312). + */ +const JOURNAL_SNAPSHOT_CARRIED: ReadonlySet = new Set([ + '_sync_row_meta', + '_sync_field_leave', +]); + const VAULT_CLEARED_JOURNAL_TABLES = [ '_sync_capture', '_sync_undo', @@ -1565,16 +1577,60 @@ async function enableSyncPushImpl( }; } } - // Another device already started this stream's journal: this store joins it - // by pulling (S5), never with a second genesis. - if (isJournalSnapshot(parent)) { - 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; 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); + // Another device already started this stream's journal: never a second + // genesis. A store that holds exactly that journal checkpoint JOINS it + // (T13312); any other store restores it first. + if (parent !== null && isJournalSnapshot(parent)) { + if (synced?.lastCheckpointId !== parent.checkpointId) { + throw vaultError( + 'E_NEXUS_SYNC_STREAM_JOURNALED', + `${t.streamId} already carries the change journal: its head snapshot ${parent.checkpointId} is a journal checkpoint this store has not restored`, + 'nothing was cut; restore the journal checkpoint (`cleo cloud restore`), then join the stream with `cleo sync enable push`', + ); + } + // Unchanged since the restore: the rows match the checkpoint's manifest + // (row data) here, and joinStream checks every row against its meta. + const git = gitTracking(t); + const local = await localManifest(t, git, untrackedComparison(t, git, synced)); + const changed = + local === null + ? ['(no store)'] + : compareVaultManifests(local.manifest, comparable(parent.manifest, git)) + .filter( + (r) => r.table !== VAULT_FILES_KEY && !isVaultAnnotationKey(r.table) && !r.match, + ) + .map((r) => r.table); + if (changed.length > 0) { + throw vaultError( + 'E_NEXUS_SYNC_REFUSED', + `this store changed since it restored ${parent.checkpointId} (${changed.slice(0, 8).join(', ')}); joining would lose those changes`, + 'nothing was joined; restore the journal checkpoint again (`cleo cloud restore --force`, a safety backup is taken first), then `cleo sync enable push`', + ); + } + const joined = joinStream(db, { + scope: tableScopeOf(t), + stream: t.streamId, + cursor: cursorFromCheckpoint(parent), + ...(opts.allowUnreleased ? { allowUnreleased: true } : {}), + }); + if (joined.refused !== null) { + throw vaultError('E_NEXUS_SYNC_REFUSED', `the join was refused: ${joined.refused}`); + } + warnings.push(...conn.state.drainWarnings()); + return { + ...base, + status: 'joined', + cut: joined.cut ?? 0, + sealed: 0, + folded: 0, + baselined: {}, + snapshot: snapshotOf(parent, await deviceNames(conn)), + deltaSegmentSeq: null, + replicaSeqFloor: null, + warnings, + }; + } if (head.headCheckpointId !== null && head.headCheckpointId !== synced?.lastCheckpointId) { throw vaultError( 'E_NEXUS_VAULT_BEHIND', @@ -2594,6 +2650,10 @@ async function restoreNexusVaultImpl(opts: NexusVaultRestoreOptions): Promise = new Set([ + 'table does not exist', + 'not in the sync set', + 'table has no uid column', +]); + +/** + * Join a stream another device already started (T13312; journal spec + * §2.11 §10, §3.5 Rule 2): the store holds exactly the journal checkpoint it + * restored, with that checkpoint's row meta, and from here on pushes its own + * writes and pulls the stream's after it. No genesis, no fold: the checkpoint + * is the store's base. + * + * Every sync-set row must carry meta that matches it (the restore carried + * the checkpoint's merge state, and nothing changed since); a row without + * meta, or one whose content left its meta behind, refuses the join. Then, + * capture and seal turned on, in ONE `BEGIN IMMEDIATE`: the restore's + * suspect marks are cleared (the check just proved the rows clean), the + * ledgers set, `genesis_cut:` records the join point (the stream has + * started here: T13217's meta-less-row rule applies from now), `undo_enabled`, + * `sync.push` and `sync.pull` are set, and the pull position is seeded from + * the checkpoint. + * + * @param db - The store, outside a transaction, bound to its own replica. + * @param o - Scope, stream, the checkpoint's pull position, and the unreleased opt-in. + * @returns The join point, or why not. + */ +export function joinStream( + db: DatabaseSync, + o: { + readonly scope: TableScope; + readonly stream: string; + readonly cursor: StreamCursor; + readonly now?: () => number; + readonly allowUnreleased?: boolean; + }, +): JoinStreamReport { + if (db.isTransaction) { + // @sync-invariant none:local-only programming-error guard: the join opens its own transaction + throw new Error('joinStream must run outside a transaction'); + } + const already = genesisCutOf(db, o.stream); + if (already !== undefined) return { stream: o.stream, refused: null, cut: already }; + const replica = activeReplica(db, o.scope)?.replicaId; + if (!replica) return { stream: o.stream, refused: 'no bound replica', cut: null }; + const now = o.now ?? Date.now; + // Capture installs the triggers the row check needs; sealing waits for the + // check, so a refused join seals nothing. + setCaptureEnabled(db, o.scope, true); + // Every row must still match the meta the checkpoint carried. + const view = sealerRowView(db, o.scope); + for (const table of syncSetTables(o.scope)) { + const plan = planRepair(db, o.scope, table, view); + if (plan.skipped !== null && !JOIN_BENIGN_SKIPS.has(plan.skipped)) { + return { stream: o.stream, refused: `${table}: ${plan.skipped}`, cut: null }; + } + if (plan.skipped !== null) continue; + const off = + plan.inserts.length + plan.updates.length + plan.deletes.length + plan.unbaselined.length; + if (off > 0) { + return { + stream: o.stream, + refused: `${table}: ${off} row(s) without matching row meta; the store is not the journal checkpoint it restored (restore it again)`, + cut: null, + }; + } + } + setSyncFlag(db, 'sync.seal', true, o.allowUnreleased ? { allowUnreleased: true } : {}); + const at = now(); + const atIso = new Date(at).toISOString(); + db.exec('BEGIN IMMEDIATE'); + try { + db.prepare("DELETE FROM _sync_meta WHERE key LIKE 'suspect:%'").run(); + // Ledgers (count + held) and baseline keys; every row already has meta. + for (const table of syncSetTables(o.scope)) baselineRowMeta(db, o.scope, table, replica, at); + const cut = capturePosition(db); + setMeta(db, `${GENESIS_CUT_KEY_PREFIX}${o.stream}`, String(cut), atIso); + setMeta(db, `${GENESIS_SOURCE_SEQ_KEY_PREFIX}${o.stream}`, String(cut), atIso); + setMeta(db, UNDO_ENABLED_KEY, '1', atIso); + raiseMinWriterVersion(db); + const flag = { now: new Date(at), ...(o.allowUnreleased ? { allowUnreleased: true } : {}) }; + setSyncFlag(db, 'sync.push', true, flag); + setSyncFlag(db, 'sync.pull', true, flag); + writeStreamCursor(db, o.stream, o.cursor, atIso); + db.exec('COMMIT'); + return { stream: o.stream, refused: null, cut }; + } catch (err) { + if (db.isTransaction) db.exec('ROLLBACK'); + throw err; + } +} diff --git a/packages/core/src/store/sync/pull.ts b/packages/core/src/store/sync/pull.ts index cd66eb154f..7e0357d1dd 100644 --- a/packages/core/src/store/sync/pull.ts +++ b/packages/core/src/store/sync/pull.ts @@ -137,6 +137,28 @@ export function readStreamCursor(db: DatabaseSync, stream: string): StreamCursor return row ? (JSON.parse(row.cursor_json) as StreamCursor) : null; } +/** + * Persist the store's pull position for `stream`, in the caller's + * transaction (the pull's page, or a JOIN seeding it from the checkpoint it + * restored). + * + * @param db - The store. + * @param stream - The stream. + * @param cursor - The position. + * @param nowIso - The time. + */ +export function writeStreamCursor( + db: DatabaseSync, + stream: string, + cursor: StreamCursor, + nowIso: string, +): void { + 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(stream, JSON.stringify(cursor), nowIso); +} + /** Decode a segment's transactions, or `null` for a vault delta segment. */ function decode(seg: PulledStreamSegment): LedgerTxn[] | null { let text: string; @@ -312,10 +334,7 @@ export async function pullStream( 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(pageCursor), nowIso); + writeStreamCursor(db, o.stream, pageCursor, nowIso); }); cursor = pageCursor; if (refusedSegment !== null) break; diff --git a/packages/core/src/store/vault-manifest.ts b/packages/core/src/store/vault-manifest.ts index 81573379e6..9fdb597443 100644 --- a/packages/core/src/store/vault-manifest.ts +++ b/packages/core/src/store/vault-manifest.ts @@ -638,6 +638,13 @@ export interface CarryMachineStateOptions { * the relocation rewrites them. `null` (a global store): no path is relocated. */ snapshotRoot: string | null; + /** + * Local-only tables whose rows come from the SNAPSHOT, not this machine: a + * journal checkpoint carries the merge state of the rows it holds (row + * meta, field frontiers; journal spec §2.10), which describes those rows + * and nothing on this machine (T13312). + */ + snapshotCarried?: ReadonlySet; } type Cell = string | number | bigint | null | Uint8Array; @@ -832,6 +839,10 @@ export function carryMachineState( out.preserved.push(t); }; + if (c.class === 'local-only' && opts.snapshotCarried?.has(t)) { + out.skipped.push(t); // left as the snapshot had it + continue; + } if (c.class === 'local-only') { // Machine state never comes from another machine: a table this store // does not have yet is emptied, not filled with the pusher's rows. diff --git a/scripts/lint-no-raw-table-writes.mjs b/scripts/lint-no-raw-table-writes.mjs index eb9a9820f5..969adc2b7c 100644 --- a/scripts/lint-no-raw-table-writes.mjs +++ b/scripts/lint-no-raw-table-writes.mjs @@ -377,8 +377,8 @@ export const EXEMPT = { }, 'packages/core/src/store/sync/genesis.ts': { _sync_meta: { - count: 3, - reason: `${SYNC_BOOKKEEPING}; plus dropping a stream's genesis keys when a failed or raced cut is undone (T13296), and clearing genesis_pending once the genesis checkpoint is stored (T13300)`, + count: 4, + reason: `${SYNC_BOOKKEEPING}; plus dropping a stream's genesis keys when a failed or raced cut is undone (T13296), clearing genesis_pending once the genesis checkpoint is stored (T13300), and clearing the restored suspect marks when a store joins a journal it restored (T13312)`, }, _sync_txn: { count: 2, From 7c6889b8043c2209392f1a5b0c28f14172639f3e Mon Sep 17 00:00:00 2001 From: Keaton Hoskins Date: Tue, 6 Oct 2026 20:51:46 -0700 Subject: [PATCH 2/3] fix(sync): a write between the join's row check and its cut refuses the join; a refused join restores capture (T13312 review) MED (#1967): planRepair runs outside the cut's transaction, so a write in that gap was captured live at seq <= cut and baselined as checkpoint state. Capture is on before the check, so any such write is a live capture: inside the cut's BEGIN IMMEDIATE the join now refuses on one (the store changed since its restore). Sealing turns on inside that transaction too. LOW: a refused join puts capture back as it found it, so a store that had it off does not pile up captures that refuse every retry. Tests (genesis.test.ts): a join of a meta-complete store; a write in the gap (via the now() hook) refuses, no cut, push off, the row unbaselined; a refused join leaves capture off and no capture triggers. Both mutations killed. Co-Authored-By: Claude Opus 5.5 --- .changeset/t13312-journal-join.md | 6 +- .../src/store/sync/__tests__/genesis.test.ts | 66 ++++++++++++++++++- packages/core/src/store/sync/genesis.ts | 41 ++++++++---- 3 files changed, 98 insertions(+), 15 deletions(-) diff --git a/.changeset/t13312-journal-join.md b/.changeset/t13312-journal-join.md index dce06f66ed..9cd01f5539 100644 --- a/.changeset/t13312-journal-join.md +++ b/.changeset/t13312-journal-join.md @@ -21,8 +21,10 @@ there was no way in. - **A changed store is refused, never folded.** The join is refused with `E_NEXUS_SYNC_REFUSED` when: - any table's rows differ from the checkpoint's manifest; - any row no longer matches its restored row meta; - - a capture is still waiting to seal. - The remedy is to restore the checkpoint again (`cleo cloud restore --force`), then join. A refused join changes nothing. + - a capture is still waiting to seal; + - a write landed after the check, which is caught inside the cut's own transaction, so it is never baselined as + checkpoint state. + The remedy is to restore the checkpoint again (`cleo cloud restore --force`), then join. A refused join changes nothing: capture is left as it was found, and sealing stays off. - **A store that never restored the journal checkpoint** gets `E_NEXUS_SYNC_STREAM_JOURNALED`, naming the restore-then-join path. The pull refusal for a store behind the journal genesis (T13306) names the same path. - `sync.push` and `sync.pull` stay unreleased. diff --git a/packages/core/src/store/sync/__tests__/genesis.test.ts b/packages/core/src/store/sync/__tests__/genesis.test.ts index 7daac48f44..e5169f68e1 100644 --- a/packages/core/src/store/sync/__tests__/genesis.test.ts +++ b/packages/core/src/store/sync/__tests__/genesis.test.ts @@ -25,7 +25,7 @@ import { ROW_IDENTITY_RECIPE_KEY, } from '../../row-identity.js'; import { finishCaptureFrame, openCaptureFrame, setCaptureEnabled } from '../capture.js'; -import { isSyncFlagOn, setSyncFlag } from '../flags.js'; +import { isSyncFlagOn, readSyncFlags, setSyncFlag } from '../flags.js'; import { cutGenesis, cutGenesisWithSnapshot, @@ -35,6 +35,7 @@ import { GenesisRacedError, genesisCutOf, genesisPending, + joinStream, UNDO_ENABLED_KEY, } from '../genesis.js'; import { streamStarted } from '../repair.js'; @@ -529,3 +530,66 @@ describe('a cut that crashed before its snapshot (T13301), and per-cut folds (LO expect(isSyncFlagOn(db, 'sync.push', {})).toBe(true); }); }); + +describe('joining a journaled stream (T13312)', () => { + /** A store whose rows all carry row meta (a cut), with the cut forgotten: what a restore leaves to join. */ + async function restoredLike(): Promise { + const { db } = await store(); + expect(cut(db).refused).toBeNull(); + db.exec( + "DELETE FROM _sync_meta WHERE key LIKE 'genesis_%' OR key IN ('undo_enabled', 'sync.push', 'sync.pull')", + ); + expect(genesisCutOf(db, STREAM)).toBeUndefined(); + return db; + } + const join = (db: DatabaseSync, now: () => number = () => ++clock) => + joinStream(db, { + scope: 'project', + stream: STREAM, + cursor: { after: 0, knowsAllReplicas: true, replicas: {} }, + now, + allowUnreleased: true, + }); + + it('joins a store whose rows match their meta: cut at the capture position, push and pull on', async () => { + const db = await restoredLike(); + const r = join(db); + expect(r.refused).toBeNull(); + expect(genesisCutOf(db, STREAM)).toBe(r.cut); + expect(isSyncFlagOn(db, 'sync.push', {})).toBe(true); + expect(isSyncFlagOn(db, 'sync.pull', {})).toBe(true); + }); + + it('a write between the row check and the cut refuses the join; its row is never baselined as checkpoint state', async () => { + const db = await restoredLike(); + let raced = false; + const r = join(db, () => { + // The last step before the cut's transaction opens. + if (!raced) { + raced = true; + db.exec(addTask('T5')); + } + return ++clock; + }); + expect(raced).toBe(true); + expect(r.refused).toMatch(/a write landed while the join checked the store/); + expect(genesisCutOf(db, STREAM)).toBeUndefined(); + expect(isSyncFlagOn(db, 'sync.push', {})).toBe(false); + expect(n(db, "SELECT count(*) AS n FROM _sync_row_meta WHERE uid = 'uid-T5'")).toBe(0); + }); + + it('a refused join leaves capture off when it was off', async () => { + const db = await restoredLike(); + setCaptureEnabled(db, 'project', false, { schemaRoot: SYNC_SCHEMA }); + db.exec("UPDATE tasks_tasks SET title = 'changed' WHERE id = 'T0'"); // rows no longer match meta + const r = join(db); + expect(r.refused).toMatch(/without matching row meta/); + expect(readSyncFlags(db)['sync.capture']).toBe(false); + expect( + n( + db, + "SELECT count(*) AS n FROM sqlite_master WHERE type = 'trigger' AND name LIKE '_sync_cap%'", + ), + ).toBe(0); + }); +}); diff --git a/packages/core/src/store/sync/genesis.ts b/packages/core/src/store/sync/genesis.ts index 46bdfabb0e..34fb56ddd3 100644 --- a/packages/core/src/store/sync/genesis.ts +++ b/packages/core/src/store/sync/genesis.ts @@ -37,7 +37,7 @@ import { UID_COLUMN, } from '../row-identity.js'; import { captureTriggerDrift, setCaptureEnabled, syncSetTables } from './capture.js'; -import { isSyncFlagOn, setSyncFlag } from './flags.js'; +import { isSyncFlagOn, readSyncFlags, setSyncFlag } from './flags.js'; import { type StreamCursor, writeStreamCursor } from './pull.js'; import { baselineRowMeta, planRepair } from './repair.js'; import { activeReplica, persistStoreSeq } from './replica.js'; @@ -594,8 +594,10 @@ const JOIN_BENIGN_SKIPS: ReadonlySet = new Set([ * * Every sync-set row must carry meta that matches it (the restore carried * the checkpoint's merge state, and nothing changed since); a row without - * meta, or one whose content left its meta behind, refuses the join. Then, - * capture and seal turned on, in ONE `BEGIN IMMEDIATE`: the restore's + * meta, or one whose content left its meta behind, refuses the join, and so + * does a write captured after the check (seen inside the cut's transaction). + * A refusal leaves capture as it found it. Then, in ONE `BEGIN IMMEDIATE` + * that also turns sealing on: the restore's * suspect marks are cleared (the check just proved the rows clean), the * ledgers set, `genesis_cut:` records the join point (the stream has * started here: T13217's meta-less-row rule applies from now), `undo_enabled`, @@ -625,32 +627,47 @@ export function joinStream( const replica = activeReplica(db, o.scope)?.replicaId; if (!replica) return { stream: o.stream, refused: 'no bound replica', cut: null }; const now = o.now ?? Date.now; - // Capture installs the triggers the row check needs; sealing waits for the - // check, so a refused join seals nothing. + // Capture installs the triggers the row check needs, and from then on + // every write is a live capture, which the cut below checks for. A refusal + // puts capture back as it was, so a store that had it off does not pile up + // captures that would refuse every retry. + const captureWasOn = readSyncFlags(db)['sync.capture']; setCaptureEnabled(db, o.scope, true); + const refuse = (why: string): JoinStreamReport => { + if (!captureWasOn) setCaptureEnabled(db, o.scope, false); + return { stream: o.stream, refused: why, cut: null }; + }; // Every row must still match the meta the checkpoint carried. const view = sealerRowView(db, o.scope); for (const table of syncSetTables(o.scope)) { const plan = planRepair(db, o.scope, table, view); if (plan.skipped !== null && !JOIN_BENIGN_SKIPS.has(plan.skipped)) { - return { stream: o.stream, refused: `${table}: ${plan.skipped}`, cut: null }; + return refuse(`${table}: ${plan.skipped}`); } if (plan.skipped !== null) continue; const off = plan.inserts.length + plan.updates.length + plan.deletes.length + plan.unbaselined.length; if (off > 0) { - return { - stream: o.stream, - refused: `${table}: ${off} row(s) without matching row meta; the store is not the journal checkpoint it restored (restore it again)`, - cut: null, - }; + return refuse( + `${table}: ${off} row(s) without matching row meta; the store is not the journal checkpoint it restored (restore it again)`, + ); } } - setSyncFlag(db, 'sync.seal', true, o.allowUnreleased ? { allowUnreleased: true } : {}); const at = now(); const atIso = new Date(at).toISOString(); db.exec('BEGIN IMMEDIATE'); try { + // The check ran outside this transaction: a write since then (another + // process, or this one) is a live capture, and the cut would baseline its + // row as the checkpoint's. The store changed since its restore: refuse. + if (db.prepare("SELECT 1 FROM _sync_capture WHERE state = 'live' LIMIT 1").get()) { + db.exec('ROLLBACK'); + return refuse( + 'a write landed while the join checked the store; it is no longer the journal checkpoint it restored (restore it again)', + ); + } + // Sealing starts with the join, never before the check. + setSyncFlag(db, 'sync.seal', true, o.allowUnreleased ? { allowUnreleased: true } : {}); db.prepare("DELETE FROM _sync_meta WHERE key LIKE 'suspect:%'").run(); // Ledgers (count + held) and baseline keys; every row already has meta. for (const table of syncSetTables(o.scope)) baselineRowMeta(db, o.scope, table, replica, at); From dafeaa7949f59f6874458224b89f367f0555a77b Mon Sep 17 00:00:00 2001 From: Keaton Hoskins Date: Wed, 7 Oct 2026 10:00:18 -0700 Subject: [PATCH 3/3] test(sync): relink B after the rebinding restore in the T13306 remedy test Main's T13109 binds a new replica when a vault restore places a store; the CLI's relink attaches it, which this helper machine does by hand. Co-Authored-By: Claude Opus 5.5 --- packages/core/src/cloud/__tests__/nexus-vault.test.ts | 7 +++++++ 1 file changed, 7 insertions(+) diff --git a/packages/core/src/cloud/__tests__/nexus-vault.test.ts b/packages/core/src/cloud/__tests__/nexus-vault.test.ts index bb15dfd3dd..c39b8ee73c 100644 --- a/packages/core/src/cloud/__tests__/nexus-vault.test.ts +++ b/packages/core/src/cloud/__tests__/nexus-vault.test.ts @@ -5653,6 +5653,13 @@ describe('sync enable push: the genesis checkpoint (T12343 S4-1b)', () => { // The remedy works: B restores the journal checkpoint, joins, and holds A's folded change. const restored = await on(b, () => restoreNexusVault(vopts(b, { mode: 'pull', force: true }))); expect(restored.status).toBe('restored'); + // The placed store is a new instance bound to a new replica (T13109); `cleo cloud pull` + // relinks it (the CLI's relink), which this machine helper does by hand. + const rebound = restored.replica?.current; + expect(rebound).toBeDefined(); + b.replicaId = rebound ?? ''; + fake.replicas.get(REMOTE_PROJECT)?.set(b.replicaId, DEVICE_B); + link(b); expect((await on(b, () => enableSyncPush(vopts(b, { allowUnreleased: true })))).status).toBe( 'joined', );