diff --git a/.changeset/t13193-pages.md b/.changeset/t13193-pages.md new file mode 100644 index 0000000000..5cf726c460 --- /dev/null +++ b/.changeset/t13193-pages.md @@ -0,0 +1,26 @@ +--- +id: t13193-pages +tasks: [T13193, T13274] +kind: feat +summary: Sync apply runs in pages - one frame and one scoped rebase per page of staged transactions +--- + +This is the second part of the fourth slice of the scoped rebase (T13193 R-4c), against journal spec §3.5 Rule 3 (R6-4). + +- **Pages.** Staged transactions now apply in pages. Each page is one apply frame with one scoped rebase: + 1. rewind the page's scope once; + 2. apply its transactions in stream order, each in its own Gate C savepoint, so a post-apply void rolls back only that transaction; + 3. replay once. + + Before this, every transaction rewound and replayed the whole local scope again. +- **Page bounds.** A page holds up to `REBASE_PAGE_OPS` (2,000) ops and stops taking transactions after `REBASE_PAGE_MS` (50 ms). + Both are overridable through `pageOps` and `pageMs`. + - A page breaks only between transactions. A split transaction is one staged unit, so its part-set stays whole. + - A page also breaks on an actor change, because one apply frame records one actor. + - A transaction larger than a page is a page alone (the declared exemption). +- **Own echoes in a page.** A rewound local transaction whose echo is in the page is applied at its stream position and sequenced there. + Every other rewound local transaction is replayed at the end of the page. The own-echo fast path is judged once per page. +- **Natural twins.** A local insert that the stream also made is now rewound to the stream's row instead of being deleted, so a later + stream update of that row applies (D2). +- **Composite UNIQUE keys (T13274).** An update that sets part of a composite UNIQUE key fills the rest of the key from the current row + for its footprint pseudo-row. diff --git a/packages/core/src/store/sync/apply/__tests__/footprints.test.ts b/packages/core/src/store/sync/apply/__tests__/footprints.test.ts index 26e7a21e1e..8783afb794 100644 --- a/packages/core/src/store/sync/apply/__tests__/footprints.test.ts +++ b/packages/core/src/store/sync/apply/__tests__/footprints.test.ts @@ -142,4 +142,51 @@ describe('declared footprints (R7-6)', () => { keys({ t: 'tasks_tasks', u: 'a', o: 'I', h, a: { id: 'T1', idempotency_key: null } }), ).toEqual([]); }); + + it('a U setting part of a composite UNIQUE key takes the rest from the row (T13274)', async () => { + const db = await store(); + db.exec( + "INSERT INTO tasks_evidence_ac_bindings (id, evidence_atom_id, ac_id, binding_type, uid) VALUES ('B1', 'E1', 'AC9', 'direct', 'b1')", + ); + const defs = (t: string) => captureTableDef(db, 'project', t) ?? null; + const h = '1791000000000-000000-0192aaaa-7f00-7000-8000-00000000000a'; + const keys = (op: LedgerOp) => { + const out: string[] = []; + widenFootprint(db, [op], defs, (t, u) => { + if (u.startsWith('#')) out.push(`${t}:${u}`); + }); + return out; + }; + expect( + keys({ + t: 'tasks_evidence_ac_bindings', + u: 'b1', + o: 'U', + h, + a: { binding_type: 'satisfies' }, + }), + ).toEqual([ + 'tasks_evidence_ac_bindings:#evidence_atom_id,ac_id,binding_type=["E1","AC9","satisfies"]', + ]); + // A reference part of the key travels as its target's uid, as an op carries it. + db.exec( + "INSERT INTO tasks_tasks (id, title, type, status, priority, uid, birth_fp) VALUES ('TX', 'x', 'task', 'pending', 'medium', 'tx', 'fp-tx')", + ); + db.exec( + "INSERT INTO tasks_task_acceptance_criteria (id, task_id, ordinal, text, uid, birth_fp) VALUES ('AC1', 'TX', 1, 'c', 'acu', 'fp-acu')", + ); + expect( + keys({ t: 'tasks_task_acceptance_criteria', u: 'acu', o: 'U', h, a: { ordinal: 2 } }), + ).toContain('tasks_task_acceptance_criteria:#task_id,ordinal=["tx",2]'); + // An update that does not move the key adds no pseudo-row. + expect( + keys({ + t: 'tasks_evidence_ac_bindings', + u: 'b1', + o: 'U', + h, + a: { created_at: '2026-10-05' }, + }), + ).toEqual([]); + }); }); diff --git a/packages/core/src/store/sync/apply/__tests__/rebase.test.ts b/packages/core/src/store/sync/apply/__tests__/rebase.test.ts index c4bc57bea8..9d8613b1bc 100644 --- a/packages/core/src/store/sync/apply/__tests__/rebase.test.ts +++ b/packages/core/src/store/sync/apply/__tests__/rebase.test.ts @@ -14,7 +14,7 @@ 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 { LedgerOp, type LedgerTxn } from '@cleocode/contracts/ledger'; +import { LedgerActor, LedgerOp, type LedgerTxn } from '@cleocode/contracts/ledger'; import { SYNC_SCHEMA_VERSION } from '@cleocode/contracts/sync-schema.js'; import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; import { showTask } from '../../../../tasks/show.js'; @@ -39,7 +39,7 @@ import { FOREIGN_TOUCH_INCOMPLETE_KEY, FOREIGN_TOUCH_MAX, } from '../../sequencing.js'; -import { type ApplyReport, applyStagedTxns } from '../applier.js'; +import { type ApplyReport, type ApplyStagedOptions, applyStagedTxns } from '../applier.js'; const SYNC_SCHEMA = resolve(import.meta.dirname, '../../../../../migrations/sync-journal'); const STREAM = 'project:t13193-rebase'; @@ -105,9 +105,9 @@ function seal(r: Replica): void { } /** One local write, through a capture frame, sealed; returns its txn id. */ -function write(r: Replica, sql: string): string { +function write(r: Replica, sql: string, actor: string | null = null): string { r.db.exec('BEGIN IMMEDIATE'); - const frame = openCaptureFrame(r.db, 'write', null); + const frame = openCaptureFrame(r.db, 'write', actor); r.db.exec(sql); finishCaptureFrame(r.db, frame); r.db.exec('COMMIT'); @@ -122,11 +122,14 @@ function write(r: Replica, sql: string): string { /** Publish these sealed transactions of a replica as one stream segment. */ function publish(r: Replica, ...ids: string[]): void { const txns = ids.map((id): LedgerTxn => { - const t = r.db.prepare('SELECT txn, hlc, via, kind FROM _sync_txn WHERE txn = ?').get(id) as { + const t = r.db + .prepare('SELECT txn, hlc, via, kind, actor FROM _sync_txn WHERE txn = ?') + .get(id) as { txn: string; hlc: string; via: LedgerTxn['via']; kind: LedgerTxn['kind']; + actor: string | null; }; return { v: 1, @@ -136,7 +139,7 @@ function publish(r: Replica, ...ids: string[]): void { scope: 'project', via: t.via, kind: t.kind, - actor: null, + actor: t.actor?.startsWith('{') ? LedgerActor.parse(JSON.parse(t.actor)) : null, ops: ( r.db.prepare('SELECT body FROM _sync_op WHERE txn = ? ORDER BY idx').all(t.txn) as Array<{ body: string; @@ -149,7 +152,11 @@ function publish(r: Replica, ...ids: string[]): void { } /** Stage every segment this replica has not seen (up to `upTo`), then apply. */ -function sync(r: Replica, upTo = published.length): ApplyReport { +function sync( + r: Replica, + upTo = published.length, + extra: Partial> = {}, +): ApplyReport { for (; r.cursor < upTo; r.cursor++) { const s = published[r.cursor] as { replicaId: string; txns: LedgerTxn[] }; stageTxns( @@ -172,6 +179,7 @@ function sync(r: Replica, upTo = published.length): ApplyReport { replica: r.id, now: () => Date.now(), seal: () => seal(r), + ...extra, }); expect(n(r.db, 'SELECT count(*) AS n FROM cleo_trigger_suspend'), 'triggers left suspended').toBe( 0, @@ -822,6 +830,117 @@ describe('natural-key rows (T13273)', () => { }); }); +describe('natural twins (D2)', () => { + it('a local insert the stream also made is rewound to the stream row, not deleted', async () => { + const [a, b, c] = await threeReplicas(); + publish(a, write(a, `${addTask('TA', 'ta')}; ${addTask('TB', 'tb')}`)); + for (const r of [a, b, c]) sync(r); + const rel = (reason: string) => + `INSERT INTO tasks_task_relations (task_id, related_to, relation_type, reason, uid) VALUES ('TA', 'TB', 'blocks', '${reason}', 'rel-ab')`; + // Relations journal under a stored uid; both replicas give the twin the same one. + const la = write(a, rel('from A')); + publish(b, write(b, rel('from B'))); + sync(a); + publish(b, write(b, "UPDATE tasks_task_relations SET reason = 'B later' WHERE task_id = 'TA'")); + expect(sync(a), 'the update of the twin row waits forever').toMatchObject({ + applied: 1, + pending: 0, + }); + const reason = (r: Replica) => + r.db.prepare("SELECT reason FROM tasks_task_relations WHERE task_id = 'TA'").get(); + expect(reason(a)).toEqual({ reason: 'B later' }); + publish(a, la); + for (const r of [a, b, c]) sync(r); + expect(reason(b)).toEqual(reason(a)); + expect(reason(c)).toEqual(reason(a)); + }); +}); + +describe('pages (§3.5 Rule 3)', () => { + const frames = (r: Replica) => + ( + r.db + .prepare('SELECT seq, status, applied_frame AS f FROM _sync_inbox ORDER BY seq, txn_idx') + .all() as Array<{ + seq: number; + status: string; + f: string | null; + }> + ).map((x) => ({ seq: Number(x.seq), status: x.status, f: x.f })); + + it('a replica applying one transaction per page converges with one applying a page', async () => { + const [a, b, c] = await threeReplicas(); + const la = write(a, "UPDATE tasks_tasks SET priority = 'high' WHERE uid = 'x'"); + const lb = write(b, "UPDATE tasks_tasks SET title = 'from B' WHERE uid = 'x'"); + const lc = write(c, "UPDATE tasks_tasks SET title = 'from C' WHERE uid = 'y'"); + publish(b, lb); + publish(c, lc); + publish(a, la); + expect(sync(a, published.length, { pageOps: 1 })).toMatchObject({ applied: 3, void: 0 }); + sync(b); + sync(c, published.length, { pageOps: 1 }); + converged([a, b, c]); + const fa = frames(a).slice(-3); + expect(new Set(fa.map((x) => x.f)).size, 'one transaction per page').toBe(3); + const fb = frames(b).slice(-3); + expect(new Set(fb.map((x) => x.f)).size, 'one page').toBe(1); + }); + + it('a post-apply void inside a page rolls back that transaction only', async () => { + const [a, b, c] = await threeReplicas(); + publish(a, write(a, "UPDATE tasks_tasks SET type = 'epic' WHERE uid = 'x'")); + for (const r of [a, b, c]) sync(r); + // C puts a task under X; B, not yet seeing it, retypes X; B edits Y. + publish(c, write(c, addTask('TK', 'k', 'X'))); + const retype = write(b, "UPDATE tasks_tasks SET type = 'task' WHERE uid = 'x'"); + const title = write(b, "UPDATE tasks_tasks SET title = 'after' WHERE uid = 'y'"); + publish(b, retype, title); + expect(sync(a)).toMatchObject({ applied: 2, void: 1 }); + const page = frames(a).slice(-2); + expect(page.map((x) => x.status)).toEqual(['void', 'applied']); + expect(page[0]?.f, 'one page, one frame').toBe(page[1]?.f); + expect(row(a, 'y')).toMatchObject({ title: 'after' }); + expect(a.db.prepare("SELECT type FROM tasks_tasks WHERE uid = 'x'").get()).toEqual({ + type: 'epic', + }); + for (const r of [a, b, c]) sync(r); + converged([a, b, c]); + }); + + it('the time bound ends a page between transactions', async () => { + const [a, b] = await threeReplicas(); + publish(b, write(b, "UPDATE tasks_tasks SET title = 't1' WHERE uid = 'x'")); + publish(b, write(b, "UPDATE tasks_tasks SET title = 't2' WHERE uid = 'y'")); + let t = Date.now(); + expect(sync(a, published.length, { pageMs: 0, now: () => (t += 5) })).toMatchObject({ + applied: 2, + }); + const last2 = frames(a).slice(-2); + expect(last2[0]?.f).not.toBe(last2[1]?.f); + }); + + it('a page breaks on an actor change and keeps a larger transaction whole', async () => { + const [a, b] = await threeReplicas(); + publish( + b, + write( + b, + "UPDATE tasks_tasks SET title = 'one' WHERE uid = 'x'", + JSON.stringify({ op: 'tasks.update' }), + ), + ); + publish(b, write(b, "UPDATE tasks_tasks SET title = 'two' WHERE uid = 'y'")); + expect(sync(a)).toMatchObject({ applied: 2 }); + const last2 = frames(a).slice(-2); + expect(last2[0]?.f, 'two actors shared one frame').not.toBe(last2[1]?.f); + // A two-op transaction under a one-op page is a page alone, never split. + publish(b, write(b, "UPDATE tasks_tasks SET priority = 'low' WHERE uid IN ('x', 'y')")); + expect(sync(a, published.length, { pageOps: 1 })).toMatchObject({ applied: 1 }); + expect(row(a, 'x')).toMatchObject({ priority: 'low' }); + expect(row(a, 'y')).toMatchObject({ priority: 'low' }); + }); +}); + describe('foreign-touch index bounds (#1912 follow-ups)', () => { it('a restarted capture counter marks the index incomplete, so the fast path declines', async () => { const a = await replica(RA); @@ -831,7 +950,8 @@ describe('foreign-touch index bounds (#1912 follow-ups)', () => { a.db.prepare("DELETE FROM sqlite_sequence WHERE name = '_sync_capture'").run(); publish(b, write(b, addTask('Q', 'q'))); publish(a, la); - expect(sync(a), 'the fast path trusted a restarted counter').toMatchObject({ rebased: 1 }); + // One page: B's insert and A's echo, rebased together. + expect(sync(a), 'the fast path trusted a restarted counter').toMatchObject({ rebased: 2 }); expect( n(a.db, 'SELECT count(*) AS n FROM _sync_meta WHERE key = ?', FOREIGN_TOUCH_INCOMPLETE_KEY), ).toBe(0); diff --git a/packages/core/src/store/sync/apply/__tests__/sequencing.test.ts b/packages/core/src/store/sync/apply/__tests__/sequencing.test.ts index 0e6ca08a31..44d1be22d6 100644 --- a/packages/core/src/store/sync/apply/__tests__/sequencing.test.ts +++ b/packages/core/src/store/sync/apply/__tests__/sequencing.test.ts @@ -261,7 +261,11 @@ describe('own-echo fast path (§3.5 Rule 3)', () => { const db = await store(); write(db, addTask('A', 'a')); seal(db); - const l1 = lastTxn(db); + stage(db, LOCAL, own(db, [lastTxn(db)])); + apply(db); + // L1 edits A; a foreign txn then edits A too (a stream that knows A). + write(db, "UPDATE tasks_tasks SET priority = 'high' WHERE uid = 'a'"); + seal(db); stage(db, R2, [foreignTitle('a', 'x')]); apply(db); expect(n(db, 'SELECT count(*) AS n FROM _sync_foreign_touch')).toBe(1); diff --git a/packages/core/src/store/sync/apply/applier.ts b/packages/core/src/store/sync/apply/applier.ts index f0117817f4..47aaf3d80a 100644 --- a/packages/core/src/store/sync/apply/applier.ts +++ b/packages/core/src/store/sync/apply/applier.ts @@ -124,6 +124,10 @@ export interface ApplyStagedOptions { readonly seal?: () => void; /** At most this many passes; defaults to 16. */ readonly maxPasses?: number; + /** Ops per page; defaults to {@link REBASE_PAGE_OPS}. */ + readonly pageOps?: number; + /** A page stops taking transactions after this many ms; defaults to {@link REBASE_PAGE_MS}. */ + readonly pageMs?: number; } /** What one {@link applyStagedTxns} call left in the inbox, per status. */ @@ -848,37 +852,30 @@ function echoInPlace( st: StagedTxn, defs: (table: string) => CaptureTableDef | null, local: LocalTxn, + unsequenced: readonly UnsequencedTxn[], ): boolean { // A held op sits rewound: only a rebase can apply it and restore what it kept. if (txnHasHold(db, local.txn)) return false; const fp = footprintOf(db, st.txn.ops, defs); if (!ownEchoFastPath(db, local, [...fp.values()])) return false; - const locals = unsequencedLocalTxns(db); - const later = locals.slice(locals.findIndex((l) => l.txn === local.txn) + 1); + const later = unsequenced.slice(unsequenced.findIndex((l) => l.txn === local.txn) + 1); return later.every((l) => ![...footprintOf(db, l.ops, defs).keys()].some((k) => fp.has(k))); } /** - * Plan the scoped rebase around an incoming transaction, or null when none is - * needed: the unsequenced local transactions whose footprint meets the - * incoming one's, to a fixed point (Rule 3). An own echo's transaction is - * rewound and not replayed: the echo applies it at its stream position. A - * transaction without row undo for every op cannot be rewound, so the plan - * declines (the transaction applies in place, as before undo existed). + * Grow `scope` over `locals` to a fixed point (Rule 3): a local transaction + * whose footprint meets `reach` joins, and its footprint joins `reach`. The + * rewind set, in local commit order, or null when it is empty or a member + * has an op without row undo (it cannot be rewound: apply in place, as + * before undo existed). */ -function planRebase( +function scopeRewind( db: DatabaseSync, - st: StagedTxn, + locals: readonly UnsequencedTxn[], defs: (table: string) => CaptureTableDef | null, - localReplica: string, -): RebasePlan | null { - const locals = unsequencedLocalTxns(db); - if (locals.length === 0) return null; - const own = - st.replicaId === localReplica ? (locals.find((l) => l.txn === st.txn.txn) ?? null) : null; - const scope = new Set(own ? [own.txn] : []); - const reach = new Set(footprintOf(db, st.txn.ops, defs).keys()); - if (own) for (const k of footprintOf(db, own.ops, defs).keys()) reach.add(k); + reach: Set, + scope: Set, +): UnsequencedTxn[] | null { for (let grown = true; grown; ) { grown = false; for (const l of locals) { @@ -896,7 +893,64 @@ function planRebase( for (const l of rewind) { for (let i = 0; i < l.ops.length; i++) if (!readRowUndo(db, l.txn, i)) return null; } - return { rewind, replay: rewind.filter((l) => l !== own), own }; + return rewind; +} + +/** + * Plan the scoped rebase around one own echo the post-apply checks refused + * while it sat in place (Rule 6): its transaction and the local ones its + * footprint reaches, leaving out those the page already rewound. + */ +function planOwnRebase( + db: DatabaseSync, + st: StagedTxn, + defs: (table: string) => CaptureTableDef | null, + exclude: ReadonlySet, +): RebasePlan | null { + const locals = unsequencedLocalTxns(db).filter((l) => !exclude.has(l.txn)); + const own = locals.find((l) => l.txn === st.txn.txn) ?? null; + if (!own) return null; + const reach = new Set(footprintOf(db, st.txn.ops, defs).keys()); + for (const k of footprintOf(db, own.ops, defs).keys()) reach.add(k); + const rewind = scopeRewind(db, locals, defs, reach, new Set([own.txn])); + return rewind ? { rewind, replay: rewind.filter((l) => l !== own), own } : null; +} + +/** + * Plan a page's scoped rebase (Rule 3): the unsequenced local transactions + * whose footprint meets a foreign transaction of the page, or an own echo + * of the page that fails the fast path, to a fixed point. Null when none. + */ +function planPageRebase( + db: DatabaseSync, + page: readonly StagedTxn[], + defs: (table: string) => CaptureTableDef | null, + localReplica: string, +): { readonly plan: RebasePlan | null; readonly inPlace: ReadonlySet } { + const inPlace = new Set(); + const locals = unsequencedLocalTxns(db); + if (locals.length === 0) return { plan: null, inPlace }; + const reach = new Set(); + const scope = new Set(); + for (const st of page) { + if (st.replicaId !== localReplica) { + for (const k of footprintOf(db, st.txn.ops, defs).keys()) reach.add(k); + continue; + } + const own = locals.find((l) => l.txn === st.txn.txn); + if (!own) continue; + // Judged once per page: a page write meeting it puts it in scope instead. + if (echoInPlace(db, st, defs, own, locals)) { + inPlace.add(own.txn); + continue; + } + scope.add(own.txn); + for (const k of footprintOf(db, st.txn.ops, defs).keys()) reach.add(k); + for (const k of footprintOf(db, own.ops, defs).keys()) reach.add(k); + } + const rewind = scopeRewind(db, locals, defs, reach, scope); + for (const l of rewind ?? []) inPlace.delete(l.txn); + return { plan: rewind ? { rewind, replay: rewind, own: null } : null, inPlace }; } /** Wire values with references turned into local keys (unresolvable ones drop to NULL). */ @@ -997,7 +1051,12 @@ function rewindTxns(c: OpContext, plan: RebasePlan): Rewound { const before = undo.values ?? op.b ?? {}; // A replay that found its row already gone snapshotted it as absent. const absent = undo.values !== null && Object.keys(undo.values).length === 0; - if (op.o === 'I' && exists(op.u)) { + if (op.o === 'I' && exists(op.u) && undo.values && Object.keys(undo.values).length > 0) { + // Its last replay sat on a row the stream had inserted too (a natural + // twin): rewinding restores that row, never deletes it (D2). + const local = toLocal(c.db, def, undo.values); + if (Object.keys(local).length > 0) c.api.writeFields(op.t, op.u, local); + } else if (op.o === 'I' && exists(op.u)) { kept.set(rowKey(op.t, op.u), { local: c.api.readLocalOnly(op.t, op.u), descendants: c.api.readLocalDescendants(op.t, op.u), @@ -1099,6 +1158,195 @@ function replayTxns(c: OpContext, plan: RebasePlan, rw: Rewound): void { }); } +/** A page holds at most this many ops (§3.5 Rule 3), unless one transaction is larger. */ +export const REBASE_PAGE_OPS = 2_000; +/** A page stops taking transactions after this long (§3.5 Rule 3). */ +export const REBASE_PAGE_MS = 50; + +/** A staged transaction of a page, with the actor its apply frame records. */ +interface PageTxn { + readonly st: StagedTxn; + readonly actor: string | null; +} + +/** + * The next page from `staged[from]`: consecutive transactions with the same + * actor (one apply frame records one), up to `maxOps` ops. A + * page breaks only between transactions, and a transaction larger than a + * page is a page alone (the declared exemption). + */ +function takePage(staged: readonly StagedTxn[], from: number, maxOps: number): PageTxn[] { + const out: PageTxn[] = []; + let ops = 0; + for (let i = from; i < staged.length; i++) { + const st = staged[i] as StagedTxn; + const actor = st.txn.actor ? JSON.stringify(st.txn.actor) : null; + const first = out[0]; + if (first && (actor !== first.actor || ops + st.txn.ops.length > maxOps)) break; + out.push({ st, actor }); + ops += st.txn.ops.length; + } + return out; +} + +/** What {@link applyInPage} needs from its page. */ +interface PageContext { + readonly db: DatabaseSync; + readonly api: ApplyApi; + readonly defs: (table: string) => CaptureTableDef | null; + readonly opts: ApplyStagedOptions; + readonly sequencingOn: boolean; + /** Local transactions the page rewound. */ + readonly rewound: ReadonlySet; + /** Local transactions whose echo in this page takes the fast path. */ + readonly inPlace: ReadonlySet; + readonly rw: Rewound; + readonly nowMs: number; + readonly nowIso: string; + readonly held: ReadonlySet; + readonly heldReplicas: ReadonlySet; +} + +/** How one staged transaction of a page ended. */ +interface InPageResult { + readonly status: InboxStatus; + /** Rows a pending transaction holds. */ + readonly holds: readonly string[]; + /** Conflicts recorded. */ + readonly n: number; + /** Whether it went through apply (not refused or held before it). */ + readonly applied: boolean; + /** The rewound local transaction its echo decided, when it was one. */ + readonly decided: string | null; +} + +/** + * Apply one staged transaction inside its page's frame: the pre-checks, + * planning, the clock, its ops in one Gate C savepoint, the post-apply + * checks, sequencing and the inbox mark. The page's rewind is already done + * and its replay follows the page. + */ +function applyInPage(x: PageContext, st: StagedTxn): InPageResult { + const { db, api, defs, opts, nowIso } = x; + const stop = (status: InboxStatus, reason: string | null, holds: readonly string[] = []) => { + markTxns(db, st.parts, status, { reason, nowIso }); + return { status, holds, n: 0, applied: false, decided: null }; + }; + const refusal = checkSchemaVersion(st.schemaVersion, st.txn.v); + if (refusal) return stop('refused-schema', `${refusal.code}: ${refusal.message}`); + if (x.heldReplicas.has(st.replicaId)) { + return stop('held-skew', `behind a held transaction of replica ${st.replicaId}`); + } + if (!isWithinSkew(parseHlc(st.txn.hlc), x.nowMs)) { + return stop('held-skew', `HLC ${st.txn.hlc} is beyond the skew bound`); + } + const waits = st.txn.ops.find((o) => x.held.has(rowKey(o.t, o.u))); + if (waits) { + return stop( + 'pending', + `waits on a pending transaction writing ${waits.t}/${waits.u}`, + st.txn.ops.map((o) => rowKey(o.t, o.u)), + ); + } + const plan = planTxn(db, api, st, defs, opts.replica); + if (plan.kind !== 'apply') { + return stop(plan.kind, plan.reason, plan.kind === 'pending' ? plan.holds : []); + } + if (api.clockReceive(opts.replica, st.txn.hlc, x.nowMs).held) { + return stop('held-skew', 'clock refused the HLC'); + } + // §3.5 Rule 3 (T13193): where this txn sits in the capture order. + const touchPos = x.sequencingOn ? capturePosition(db) : 0; + const local = + x.sequencingOn && st.replicaId === opts.replica ? unsequencedLocalTxn(db, st.txn.txn) : null; + // An own echo whose transaction the page rewound applies it at its stream position. + const rewoundEcho = local !== null && x.rewound.has(local.txn); + const c: OpContext = { db, api, st, defs, replica: opts.replica, nowIso }; + let n = 0; + // Gate C (§3.6): the whole transaction is one savepoint, so a broken + // multi-row invariant rolls all of it back (the page's rewind stays). + db.exec('SAVEPOINT apply_txn'); + const results = applyOrder(st.txn.ops, defs).map((i) => { + const op = st.txn.ops[i] as LedgerOp; + const r = applyOne(c, i, op); + n += r.conflicts; + // A rewound own insert applies again here: put back what it kept. + if (rewoundEcho && r.result === 'applied' && op.o === 'I') { + withTriggersSuspended(db, ['capture', 'side-effect'], 'forward', () => { + restoreKept(c, op, r.result, x.rw.kept); + }); + } + return r.result; + }); + // Every per-transaction post-apply check, by name (the registry's runtime gates). + const violations = [...checkTaskTreeShape(db, treeShapePage(st.txn.ops))]; + if (violations.length > 0) { + db.exec('ROLLBACK TO apply_txn'); + db.exec('RELEASE apply_txn'); + recordConflicts( + db, + { ...st.key, opIdx: -1 }, + violations.map((v) => ({ + kind: 'post-apply' as const, + table: v.table, + uid: v.uid, + columns: [], + rule: `${v.check}: ${v.message}`.slice(0, 200), + resolution: 'op-voided' as const, + opHlc: st.txn.hlc, + })), + st.replicaId, + nowIso, + ); + // A refused own echo stays rewound and keeps its undo (Rule 6). Rewound + // by the page, the rollback left it so; in place, rewind it now. + const own = local && !rewoundEcho ? planOwnRebase(db, st, defs, x.rewound) : null; + if (own) replayTxns(c, own, rewindTxns(c, own)); + if (local && (rewoundEcho || own)) { + markSequenced(db, local, { stream: st.key.stream, seq: st.key.seq, nowIso, outcome: 'void' }); + } + markTxns(db, st.parts, 'void', { + frame: api.frame, + reason: `post-apply: ${violations.map((v) => v.check).join(', ')}`, + nowIso, + }); + return { + status: 'void', + holds: [], + n: violations.length, + applied: true, + decided: rewoundEcho && local ? local.txn : null, + }; + } + db.exec('RELEASE apply_txn'); + const status = txnStatus(results, n); + if (x.sequencingOn) { + if (st.replicaId !== opts.replica) { + // The rows it wrote: an echo widens its own footprint by what its + // guards read, so a write to any of those rows meets it (R7-6). + recordForeignTouches(db, touchedRows(st.txn.ops), touchPos); + } else if (local && rewoundEcho) { + // Rebased at its stream position: the stream decided it. A voided + // echo stays rewound and keeps its undo (Rule 6). + markSequenced(db, local, { + stream: st.key.stream, + seq: st.key.seq, + nowIso, + outcome: results.includes('void') ? 'void' : 'applied', + }); + } else if (local && !results.includes('void') && x.inPlace.has(local.txn)) { + // Own echo in place: the stream order agrees with ours. + markSequenced(db, local, { stream: st.key.stream, seq: st.key.seq, nowIso }); + } + } + markTxns(db, st.parts, status, { + frame: api.frame, + reason: n > 0 ? `${n} conflict(s) recorded` : null, + nowIso, + }); + return { status, holds: [], n, applied: true, decided: rewoundEcho && local ? local.txn : null }; +} + /** * Apply a stream's staged transactions (see the module doc). * @@ -1149,172 +1397,82 @@ export function applyStagedTxns(db: DatabaseSync, opts: ApplyStagedOptions): App passes += 1; const heldReplicas = new Set(); const held = new Set(); // rows written by a pending transaction - for (const st of stagedTxns(db, opts.stream)) { - const nowMs = now(); - const nowIso = new Date(nowMs).toISOString(); - const id = keyText(st.key); - const mark = (status: InboxStatus, reason: string | null): void => { - markTxns(db, st.parts, status, { reason, nowIso }); - last.set(id, status); - }; - const refusal = checkSchemaVersion(st.schemaVersion, st.txn.v); - if (refusal) { - mark('refused-schema', `${refusal.code}: ${refusal.message}`); - continue; - } - if (heldReplicas.has(st.replicaId)) { - mark('held-skew', `behind a held transaction of replica ${st.replicaId}`); - continue; - } - if (!isWithinSkew(parseHlc(st.txn.hlc), nowMs)) { - heldReplicas.add(st.replicaId); - mark('held-skew', `HLC ${st.txn.hlc} is beyond the skew bound`); - continue; - } - const waits = st.txn.ops.find((o) => held.has(rowKey(o.t, o.u))); - if (waits) { - for (const o of st.txn.ops) held.add(rowKey(o.t, o.u)); - mark('pending', `waits on a pending transaction writing ${waits.t}/${waits.u}`); - continue; - } - const actor = st.txn.actor ? JSON.stringify(st.txn.actor) : null; - const result = withApplyFrame(db, opts.scope, actor, (api) => { - adoptNaturalRows(db, api, defs, st.txn.ops); - const plan = planTxn(db, api, st, defs, opts.replica); - if (plan.kind !== 'apply') { - markTxns(db, st.parts, plan.kind, { reason: plan.reason, nowIso }); - return { - status: plan.kind, - holds: plan.kind === 'pending' ? plan.holds : [], - n: 0, - rebased: false, - }; - } - if (api.clockReceive(opts.replica, st.txn.hlc, nowMs).held) { - markTxns(db, st.parts, 'held-skew', { reason: 'clock refused the HLC', nowIso }); - return { status: 'held-skew' as const, holds: [], n: 0, rebased: false }; - } - // §3.5 Rule 3 (T13193): where this txn sits in the capture order. - const touchPos = sequencingOn ? capturePosition(db) : 0; - // A scoped rebase when unsequenced local txns meet this one: around a - // foreign txn always, around an own echo only when the fast path fails. - let rebase: RebasePlan | null = null; - if (sequencingOn) { - if (st.replicaId !== opts.replica) { - rebase = planRebase(db, st, defs, opts.replica); - } else { - const local = unsequencedLocalTxn(db, st.txn.txn); - if (local && !echoInPlace(db, st, defs, local)) { - rebase = planRebase(db, st, defs, opts.replica); - } - } - } - const c: OpContext = { db, api, st, defs, replica: opts.replica, nowIso }; - let n = 0; - // Gate C (§3.6): the whole transaction is one savepoint, so a broken - // multi-row invariant rolls all of it back. - db.exec('SAVEPOINT apply_txn'); - // Inside the savepoint: a post-apply void rolls the rewind back too. - const rw: Rewound = rebase ? rewindTxns(c, rebase) : { kept: new Map(), after: new Map() }; - const results = applyOrder(st.txn.ops, defs).map((i) => { - const op = st.txn.ops[i] as LedgerOp; - const r = applyOne(c, i, op); - n += r.conflicts; - // A rewound own insert applies again here: put back what it kept. - if (rebase?.own && r.result === 'applied' && op.o === 'I') { - withTriggersSuspended(db, ['capture', 'side-effect'], 'forward', () => { - restoreKept(c, op, r.result, rw.kept); - }); - } - return r.result; - }); - // Every per-transaction post-apply check, by name (the registry's runtime gates). - const violations = [...checkTaskTreeShape(db, treeShapePage(st.txn.ops))]; - if (violations.length > 0) { - db.exec('ROLLBACK TO apply_txn'); - db.exec('RELEASE apply_txn'); - recordConflicts( - db, - { ...st.key, opIdx: -1 }, - violations.map((v) => ({ - kind: 'post-apply' as const, - table: v.table, - uid: v.uid, - columns: [], - rule: `${v.check}: ${v.message}`.slice(0, 200), - resolution: 'op-voided' as const, - opHlc: st.txn.hlc, - })), - st.replicaId, - nowIso, - ); - // A refused own echo stays rewound and keeps its undo (Rule 6): the - // rollback put back whatever of it the store held, so rewind it. - const own = - sequencingOn && st.replicaId === opts.replica - ? planRebase(db, st, defs, opts.replica) - : null; - if (own?.own) { - replayTxns(c, own, rewindTxns(c, own)); - markSequenced(db, own.own, { - stream: st.key.stream, - seq: st.key.seq, + const staged = [...stagedTxns(db, opts.stream)]; + // Each page is one frame and one scoped rebase (§3.5 Rule 3): rewind the + // page's scope once, apply its transactions in stream order, replay once. + for (let at = 0; at < staged.length; ) { + const page = takePage(staged, at, opts.pageOps ?? REBASE_PAGE_OPS); + const pageStart = now(); + const done = withApplyFrame(db, opts.scope, page[0]?.actor ?? null, (api) => { + const pageIso = new Date(pageStart).toISOString(); + for (const p of page) adoptNaturalRows(db, api, defs, p.st.txn.ops); + const planned = sequencingOn + ? planPageRebase( + db, + page.map((p) => p.st), + defs, + opts.replica, + ) + : { plan: null, inPlace: new Set() }; + const rebase = planned.plan; + const c0: OpContext = { + db, + api, + st: (page[0] as PageTxn).st, + defs, + replica: opts.replica, + nowIso: pageIso, + }; + const rw: Rewound = rebase ? rewindTxns(c0, rebase) : { kept: new Map(), after: new Map() }; + const rewound = new Set(rebase?.rewind.map((l) => l.txn) ?? []); + const decided = new Set(); // rewound locals whose echo this page applied + let taken = 0; + for (const { st } of page) { + // The time bound cuts a page between transactions (never inside one). + if (taken > 0 && now() - pageStart > (opts.pageMs ?? REBASE_PAGE_MS)) break; + taken += 1; + const nowMs = now(); + const nowIso = new Date(nowMs).toISOString(); + const id = keyText(st.key); + const result = applyInPage( + { + db, + api, + defs, + opts, + sequencingOn, + rewound, + inPlace: planned.inPlace, + rw, + nowMs, nowIso, - outcome: 'void', - }); + held, + heldReplicas, + }, + st, + ); + if (result.decided) decided.add(result.decided); + if (result.status !== null) last.set(id, result.status); + conflicts += result.n; + if (rebase && result.applied) rebased += 1; + if (result.status === 'held-skew') heldReplicas.add(st.replicaId); + if (result.status === 'pending') for (const k of result.holds) held.add(k); + if ( + result.status === 'applied' || + result.status === 'conflict' || + result.status === 'void' + ) { + progress = true; } - markTxns(db, st.parts, 'void', { - frame: api.frame, - reason: `post-apply: ${violations.map((v) => v.check).join(', ')}`, - nowIso, - }); - return { - status: 'void' as const, - holds: [], - n: violations.length, - rebased: own?.own != null, - }; } - if (rebase) replayTxns(c, rebase, rw); - db.exec('RELEASE apply_txn'); - const status = txnStatus(results, n); - if (sequencingOn) { - if (st.replicaId !== opts.replica) { - // The rows it wrote: an echo widens its own footprint by what its - // guards read, so a write to any of those rows meets it (R7-6). - recordForeignTouches(db, touchedRows(st.txn.ops), touchPos); - } else if (rebase?.own) { - // Rebased at its stream position: the stream decided it. A voided - // echo stays rewound and keeps its undo (Rule 6). - markSequenced(db, rebase.own, { - stream: st.key.stream, - seq: st.key.seq, - nowIso, - outcome: results.includes('void') ? 'void' : 'applied', - }); - } else { - // Own echo: sequence it when the stream order agrees with ours. - const local = unsequencedLocalTxn(db, st.txn.txn); - if (local && !results.includes('void') && echoInPlace(db, st, defs, local)) { - markSequenced(db, local, { stream: st.key.stream, seq: st.key.seq, nowIso }); - } - } + // Replay what the page rewound and did not decide by its echo. + if (rebase) { + const replay = rebase.rewind.filter((l) => !decided.has(l.txn)); + replayTxns(c0, { rewind: rebase.rewind, replay, own: null }, rw); } - markTxns(db, st.parts, status, { - frame: api.frame, - reason: n > 0 ? `${n} conflict(s) recorded` : null, - nowIso, - }); - return { status, holds: [], n, rebased: rebase !== null }; + return taken; }); - if (result.rebased) rebased += 1; - last.set(id, result.status); - conflicts += result.n; - if (result.status === 'held-skew') heldReplicas.add(st.replicaId); - if (result.status === 'pending') for (const k of result.holds) held.add(k); - if (result.status === 'applied' || result.status === 'conflict' || result.status === 'void') { - progress = true; - } + at += done; } } const count = (s: InboxStatus): number => [...last.values()].filter((v) => v === s).length; diff --git a/packages/core/src/store/sync/apply/footprints.ts b/packages/core/src/store/sync/apply/footprints.ts index f511f5eab1..55fcc6e0a9 100644 --- a/packages/core/src/store/sync/apply/footprints.ts +++ b/packages/core/src/store/sync/apply/footprints.ts @@ -20,10 +20,11 @@ */ import type { DatabaseSync } from 'node:sqlite'; -import type { LedgerOp } from '@cleocode/contracts/ledger'; +import type { LedgerOp, LedgerWireValue } from '@cleocode/contracts/ledger'; import { UID_COLUMN } from '../../row-identity-registry.js'; -import type { CaptureTableDef } from '../capture.js'; -import { resolveRef } from './refs.js'; +import { type CaptureTableDef, enc } from '../capture.js'; +import { decodeEnc } from '../sealer-values.js'; +import { resolveRef, uidOfKey } from './refs.js'; /** * What a guard or a check reads beyond the op's own row: @@ -232,8 +233,42 @@ function uniqueIndexes(db: DatabaseSync, def: CaptureTableDef): string[][] { function uniqueKeyFootprint(db: DatabaseSync, def: CaptureTableDef, op: LedgerOp, add: Add): void { if (op.o !== 'I' && op.o !== 'U') return; for (const cols of uniqueIndexes(db, def)) { - const values = cols.map((c) => op.a?.[c]); + // A U that sets part of the key: the rest is the row's current value (T13274). + const missing = op.o === 'U' ? cols.filter((c) => !(op.a !== undefined && c in op.a)) : []; + if (missing.length === cols.length) continue; // the key does not move + const current = missing.length > 0 ? currentWire(db, def, op.u, missing) : {}; + const values = cols.map((c) => (op.a !== undefined && c in op.a ? op.a[c] : current[c])); if (values.some((v) => v === undefined || v === null)) continue; add(def.table, `#${cols.join(',')}=${JSON.stringify(values)}`); } } + +/** + * `cols` of the row with `uid` as wire values, as an op carries them (stored + * values through `enc()`, references as their target's uid). + */ +function currentWire( + db: DatabaseSync, + def: CaptureTableDef, + uid: string, + cols: readonly string[], +): Record { + const row = db + .prepare( + `SELECT ${cols.map((c) => `${enc(ident(c))} AS ${ident(c)}`).join(', ')} FROM main.${ident(def.table)} WHERE ${ident(UID_COLUMN)} = ?`, + ) + .get(uid) as Record | undefined; + const out: Record = {}; + if (!row) return out; + for (const c of cols) { + const v = decodeEnc(row[c] as string); + const target = def.refs.get(c); + out[c] = target && v !== null ? (uidOfKey(db, target, v) ?? v) : v; + } + return out; +} + +/** Quote an SQL identifier. */ +function ident(name: string): string { + return `"${name.replaceAll('"', '""')}"`; +} diff --git a/packages/core/src/store/sync/apply/write-api.ts b/packages/core/src/store/sync/apply/write-api.ts index ac3a689e16..c9feaf0ef0 100644 --- a/packages/core/src/store/sync/apply/write-api.ts +++ b/packages/core/src/store/sync/apply/write-api.ts @@ -499,7 +499,8 @@ export function createApplyWriteApi( assertActive(); const def = defOf(table); if (rowExists(table, uid) || def.key.some((k) => !(k in localKey))) return false; - // The uid is an identity column: capture never fires on it. + // The uid is an identity column: capture never fires on it. The local key + // is UNIQUE, so 0 or 1 row matches. return ( Number( db