diff --git a/.changeset/adr-0119-d2-migration-journal.md b/.changeset/adr-0119-d2-migration-journal.md new file mode 100644 index 0000000000..c70563fdde --- /dev/null +++ b/.changeset/adr-0119-d2-migration-journal.md @@ -0,0 +1,78 @@ +--- +"@objectstack/spec": minor +"@objectstack/platform-objects": minor +"@objectstack/core": minor +"@objectstack/metadata-protocol": patch +--- + +feat(core,platform-objects,spec): the ADR-0119 D2 migration-journal runner — a migration killed mid-run is resumable to completion or compensable to clean, with journal rows proving which (#4617) + +**The gap D1 left open.** ADR-0119 D1 made `engine.transaction()` reachable +through the contract, which is the right answer for multi-write atomicity that +fits in one transaction. Migration-class work does not fit: a million-row +backfill cannot hold one write-lock for its duration, `driver-memory`'s +`beginTransaction` deep-clones the entire database (O(db) per begin), +`ObjectQL.transaction()` binds the **default driver only** so a multi-datasource +migration silently commits part of its work outside it, and a process **killed** +— as distinct from a thrown error — defeats in-process rollback entirely. So the +unit of atomicity is the *chunk*, and durability across chunks is a journal. + +Four consumers had each converged on the same four moves — dry-run preflight, +undo journal, LIFO compensation, re-entrant forward recovery (ADR-0105 D13 +promotion, ADR-0117 D8's ownership backfill, the org lifecycle transitions, and +D10 master-data distribution #4585). One copy is engineering; four is platform +debt, and the fourth author would have had to rediscover the invariant below +from scratch. + +**New: `runMigrationJournal` (`@objectstack/core`).** Preflight runs every +step's read-only validator before any step writes, so a plan that would fail at +step 3 has not written step 1. Rows are chunked per the `bulk-write.ts` +discipline; each chunk's writes run inside `engine.transaction()`. On failure, +committed chunks are compensated newest-first, each in its own transaction. On +restart, a rediscovered run resumes forward from the first chunk lacking +`chunk_done`, or unwinds, per the plan's `onCrash` policy. Forward and +compensate callbacks receive an `attempt` counter; `attempt > 1` means the prior +outcome is UNKNOWN and the callback must recheck by natural key before +re-writing — the same at-least-once contract `bulk-write.ts` already documents, +reused rather than re-derived. + +**The invariant that carries the design:** `chunk_done(i)` is written **inside** +the chunk's own transaction, so `done ⇔ committed` holds by construction; +`chunk_started(i)` is written autonomously **before** it. That asymmetry is what +gives `started ∧ ¬done` exactly one meaning — *the outcome is unknown* — which +is the only state a crash can leave and the only state recovery reasons about. +Making both writes symmetric would look tidier and would destroy recovery. + +**New: `sys_migration_journal` (`@objectstack/platform-objects`).** Rows keyed +`(run_id, seq)` under a unique index, so a resumed run that miscomputes its next +sequence fails loudly rather than double-recording an event. Registered +unconditionally alongside `sys_migration` because recovery must be discoverable +with **zero host wiring** — a journal some kernels compose and others do not is +a journal a boot scanner cannot rely on (ADR-0078). Distinct in grain from +`sys_migration`, which holds one durable verdict per named migration; this holds +many rows per *run*. Read-only over the API; writes go through the runner in +system context. + +**The runner refuses rather than degrades**, in four places: the runtime cannot +roll back; any preflight fails; the plan declares `onCrash: 'compensate'` but a +step cannot compensate; or a resume's plan hash disagrees with the journal +(resuming a changed plan would apply chunk boundaries the journal never +described). A compensation failure halts and is journalled — never swallowed — +and the run ends `failed`, not `compensated`, because a database in a state no +clean story covers must not be reported as a tidy rollback. + +**`engineCanRollBack` is now shared.** The two-level probe (engine method AND +default-driver `beginTransaction`) was the same condition written twice — here +and in `batchData`'s atomic gate. It now lives in `@objectstack/core` and +`@objectstack/metadata-protocol` imports it, as a type predicate so callers do +not each re-narrow the optional member by hand. Two copies of "can this runtime +actually roll back?" drift by one clause and leave one caller believing it has +atomicity it does not have. + +Boot reconciliation and `os migrate resume` land separately; `findInterruptedRuns` +is the discovery primitive they will consume, and is exported here. + +**Docs:** ADR-0118 (plugin-reachable transactions) is renumbered **ADR-0119**. +It merged one day after an unrelated ADR-0118 (非用户 actor 的平台契约) and the +earlier merge holds the number; citations of "ADR-0118 D1/D2/D3/D4" written +before 2026-08-03 mean the renumbered record. diff --git a/.changeset/adr-0118-plugin-reachable-transactions.md b/.changeset/adr-0119-plugin-reachable-transactions.md similarity index 96% rename from .changeset/adr-0118-plugin-reachable-transactions.md rename to .changeset/adr-0119-plugin-reachable-transactions.md index 1a18f5dbc8..eaeeff899a 100644 --- a/.changeset/adr-0118-plugin-reachable-transactions.md +++ b/.changeset/adr-0119-plugin-reachable-transactions.md @@ -3,7 +3,7 @@ "@objectstack/metadata-protocol": minor --- -feat(spec,metadata-protocol): `IObjectQLEngine.transaction` joins the slot contract, and `batchData`'s `atomic` flag becomes real — rollback or refusal, never silent best-effort (ADR-0118 D1/D4, #4612) +feat(spec,metadata-protocol): `IObjectQLEngine.transaction` joins the slot contract, and `batchData`'s `atomic` flag becomes real — rollback or refusal, never silent best-effort (ADR-0119 D1/D4, #4612) **D1 — the contract fix.** `ObjectQL.transaction()` — ADR-0034's ambient transaction, shipped since v8.0.0 — was reachable from plugin space only @@ -53,7 +53,7 @@ If you were passing `atomic: true` and relying on partial results surviving a failure, that was the bug — switch to `atomic: false` (or omit it) for best-effort semantics. -ADR-0118 also rules on two items landing separately: D2 specifies a +ADR-0119 also rules on two items landing separately: D2 specifies a framework-owned migration-journal runner for multi-step migrations too large for one transaction, and D3 retires the declared-but-unimplemented `IDataEngine.batch?`. diff --git a/content/docs/references/system/migration.mdx b/content/docs/references/system/migration.mdx index f5639d2a7f..a88030d7a8 100644 --- a/content/docs/references/system/migration.mdx +++ b/content/docs/references/system/migration.mdx @@ -34,8 +34,8 @@ irreversibly on migrated data gate on the flag instead of the version. ## TypeScript Usage ```typescript -import { AddFieldOperation, ChangeSetSchema, CreateObjectOperation, DataMigrationFlagSchema, DeleteObjectOperation, ExecuteSqlOperation, MigrationDependencySchema, MigrationOperationSchema, ModifyFieldOperation, RemoveFieldOperation, RenameObjectOperation } from '@objectstack/spec/system'; -import type { ChangeSet, DataMigrationFlag, MigrationOperation } from '@objectstack/spec/system'; +import { AddFieldOperation, ChangeSetSchema, CreateObjectOperation, DataMigrationFlagSchema, DeleteObjectOperation, ExecuteSqlOperation, MigrationDependencySchema, MigrationJournalEventSchema, MigrationOperationSchema, ModifyFieldOperation, RemoveFieldOperation, RenameObjectOperation } from '@objectstack/spec/system'; +import type { ChangeSet, DataMigrationFlag, MigrationJournalEvent, MigrationOperation } from '@objectstack/spec/system'; // Validate data const result = AddFieldOperation.parse(data); @@ -153,6 +153,27 @@ Dependency reference to another migration that must run first | **package** | `string` | optional | Package that owns the dependency migration | +--- + +## MigrationJournalEvent + +One event in a migration run journal — the durable trace that lets a killed run be resumed forward or compensated back, with rows proving which + +### Properties + +| Property | Type | Required | Description | +| :--- | :--- | :--- | :--- | +| **run_id** | `string` | ✅ | Identifies one run. Rows are keyed (run_id, seq) | +| **seq** | `integer` | ✅ | Monotonic per-run sequence. Ordering authority — wall-clock timestamps can tie or skew | +| **kind** | `Enum<'run_started' \| 'chunk_started' \| 'chunk_done' \| 'compensated' \| 'run_done' \| 'run_failed'>` | ✅ | Event kind | +| **migration_id** | `string` | optional | The named migration this run belongs to, when it has one — joins to sys_migration.id | +| **plan_hash** | `string` | optional | On run_started: hash of the plan shape. A resume whose plan hash differs REFUSES rather than resuming a changed plan against an old journal | +| **chunk_index** | `integer` | optional | On chunk_started / chunk_done / compensated: the run-global chunk index | +| **attempt** | `integer` | optional | Which attempt produced this event. attempt > 1 means a prior outcome was unknown and the callback was asked to recheck by natural key | +| **detail** | `string` | optional | JSON-encoded payload — the chunk plan on run_started, the error on run_failed / a failed compensation | +| **created_at** | `string` | optional | Wall-clock stamp, for humans. Never the ordering authority — that is seq | + + --- ## MigrationOperation diff --git a/docs/adr/0118-plugin-reachable-transactions-and-honest-atomic-batch.md b/docs/adr/0119-plugin-reachable-transactions-and-honest-atomic-batch.md similarity index 95% rename from docs/adr/0118-plugin-reachable-transactions-and-honest-atomic-batch.md rename to docs/adr/0119-plugin-reachable-transactions-and-honest-atomic-batch.md index 1b73187113..0d89e57c6d 100644 --- a/docs/adr/0118-plugin-reachable-transactions-and-honest-atomic-batch.md +++ b/docs/adr/0119-plugin-reachable-transactions-and-honest-atomic-batch.md @@ -1,6 +1,7 @@ -# ADR-0118: Multi-write atomicity is reachable through the contract, `atomic` means atomic or refuses, and migrations too big for one transaction get a journal runner +# ADR-0119: Multi-write atomicity is reachable through the contract, `atomic` means atomic or refuses, and migrations too big for one transaction get a journal runner -**Status**: Accepted (2026-08-02) — D1/D4 implemented in this PR; D2 tracked in [#4617](https://github.com/objectstack-ai/objectstack/issues/4617); D3 tracked in [#4618](https://github.com/objectstack-ai/objectstack/issues/4618) +**Status**: Accepted (2026-08-02) — D1/D4 implemented in [#4623](https://github.com/objectstack-ai/objectstack/pull/4623): D1 in `packages/spec/src/contracts/objectql-engine.ts` (test `packages/objectql/src/protocol-batch-atomic.test.ts`), D4 in `packages/metadata-protocol/src/protocol.ts` (test `packages/metadata-protocol/src/protocol.batch-atomic.test.ts`). D2 tracked in [#4617](https://github.com/objectstack-ai/objectstack/issues/4617); D3 tracked in [#4618](https://github.com/objectstack-ai/objectstack/issues/4618) — neither is implemented, so this record is *not* wholly "implemented". +**Renumbered**: published for one day as ADR-0118. Renumbered to 0119 because [ADR-0118 (非用户 actor 的平台契约)](./0118-non-user-actor-contract.md) merged first (10:37 vs 12:11 on 2026-08-02) and holds the number. Citations of "ADR-0118 D1/D2/D3/D4" written before 2026-08-03 mean this record. **Deciders**: ObjectStack Protocol Architects **Builds on**: [ADR-0034](./0034-transactional-writes-and-ambient-transaction.md) (the ambient `AsyncLocalStorage` transaction D1 declares — this ADR adds no mechanism to it), [ADR-0067](./0067-commit-history-and-rollback-for-ai-authoring.md) (D2 — the join-don't-nest rule that makes an outer transaction the sole owner of commit/rollback), [ADR-0049](./0049-no-unenforced-security-properties.md) (enforce-or-remove — the disposition method applied to `batch?` in D3 and to the `atomic` flag in D4), [ADR-0087](./0087-metadata-protocol-upgrade-contract.md) (D3's replayable migration chain — the metadata-side analogue of the data-side runner D2 specifies), [ADR-0008](./0008-metadata-repository-and-change-log.md) (the JSONL change log — the journal shape D2 deliberately does *not* reuse), [ADR-0060](./0060-conformance-ledger-platform-pattern.md) (framework-owned ledger pattern — the precedent for `sys_migration_journal`), [ADR-0117](./0117-owning-business-unit-record-stamp.md) (D8 — backfill plus a fail-closed enable gate, the migration posture D2 and D4 both inherit), [ADR-0078](./0078-no-silently-inert-metadata.md) (no silently inert declarations — why D2 rejects a pluggable journal store) **Consumers**: `@objectstack/spec` (`contracts/objectql-engine.ts`, `api/batch.zod.ts`), `@objectstack/metadata-protocol` (`protocol.ts`, `host-engine.ts`, `sys-metadata-repository.ts`), `@objectstack/objectql` (the implementation — unchanged by D1), `@objectstack/rest` (the `/batch` routes — unchanged), and for D2: `@objectstack/core`, `@objectstack/platform-objects` diff --git a/packages/core/src/index.ts b/packages/core/src/index.ts index a56fa6ab51..90318db1cd 100644 --- a/packages/core/src/index.ts +++ b/packages/core/src/index.ts @@ -30,6 +30,11 @@ export * from './utils/datetime.js'; // Export the shared batched-write helper (framework#2678) export * from './utils/bulk-write.js'; +// Export the migration-journal runner (ADR-0119 D2, #4617) — chunk-atomic +// migrations with durable recovery, plus the shared `engineCanRollBack` gate +// that `@objectstack/metadata-protocol`'s atomic `batchData` also uses. +export * from './utils/migration-journal.js'; + // Export the runtime filter-placeholder resolver (framework#3582) export * from './utils/filter-tokens.js'; diff --git a/packages/core/src/utils/migration-journal.test.ts b/packages/core/src/utils/migration-journal.test.ts new file mode 100644 index 0000000000..4c96e1e951 --- /dev/null +++ b/packages/core/src/utils/migration-journal.test.ts @@ -0,0 +1,472 @@ +// Copyright (c) 2026 ObjectStack. Licensed under the Apache-2.0 license. + +/** + * ADR-0119 D2 (#4617) conformance. + * + * The acceptance bar the ADR sets is not "the runner runs": it is that **a + * migration killed mid-run is either resumable to completion or compensable to + * clean, with journal rows proving which**. So the fake engine below implements + * REAL rollback — a failed transaction discards its writes, including the + * `chunk_done` row written inside it. Without that, every assertion here would + * pass against a runner that never opened a transaction at all, which is the + * exact class of bug ADR-0119 D4 was written to kill. + */ + +import { describe, it, expect, vi } from 'vitest'; +import { + runMigrationJournal, + resumeMigrationJournal, + findInterruptedRuns, + readRunJournal, + engineCanRollBack, + planChunks, + hashMigrationPlan, + MigrationJournalRefusal, + type MigrationPlan, + type MigrationPlanStep, +} from './migration-journal'; + +// ── a fake engine with real rollback ────────────────────────────────────── + +interface FakeRow { [k: string]: unknown } + +class FakeEngine { + tables = new Map(); + /** Set when a transaction is open — writes join it, and are discarded on throw. */ + private txDepth = 0; + private snapshot: Map | null = null; + /** Every context object handed to `insert`, so tests can prove tx binding. */ + insertContexts: unknown[] = []; + private driverHasTx: boolean; + + constructor(opts: { driverHasTx?: boolean } = {}) { + this.driverHasTx = opts.driverHasTx ?? true; + } + + private rows(name: string): FakeRow[] { + if (!this.tables.has(name)) this.tables.set(name, []); + return this.tables.get(name)!; + } + + async insert(objectName: string, data: FakeRow, options?: { context?: unknown }): Promise { + this.insertContexts.push(options?.context); + const row = { ...data }; + // The unique (run_id, seq) index is part of the object contract — model it, + // so a runner that miscomputes its next sequence fails here rather than + // silently double-recording an event. + if (objectName === 'sys_migration_journal') { + const dup = this.rows(objectName).find((r) => r.run_id === row.run_id && r.seq === row.seq); + if (dup) throw new Error(`duplicate journal key (${String(row.run_id)}, ${String(row.seq)})`); + } + this.rows(objectName).push(row); + return row; + } + + async find(objectName: string, query?: { where?: Record }): Promise { + const where = query?.where ?? {}; + return this.rows(objectName).filter((r) => Object.entries(where).every(([k, v]) => r[k] === v)); + } + + async findOne(objectName: string, query?: { where?: Record }): Promise { + return (await this.find(objectName, query))[0] ?? null; + } + + async update(): Promise { throw new Error('not used'); } + async delete(): Promise { throw new Error('not used'); } + async count(): Promise { return 0; } + async aggregate(): Promise { return []; } + getObject(name: string): unknown { return { name }; } + getDefaultDriverName(): string { return 'fake'; } + getDriverByName(): unknown { + return this.driverHasTx ? { beginTransaction: () => {}, commit: () => {}, rollback: () => {} } : {}; + } + + async transaction(cb: (trxCtx: unknown) => Promise, baseContext?: unknown): Promise { + // Nested calls JOIN (ADR-0067 D2) — the outermost owns commit/rollback. + if (this.txDepth > 0) return cb({ ...(baseContext as object), __tx: true }); + this.txDepth++; + this.snapshot = new Map([...this.tables].map(([k, v]) => [k, v.map((r) => ({ ...r }))])); + try { + const out = await cb({ ...(baseContext as object), __tx: true }); + this.snapshot = null; + return out; + } catch (err) { + this.tables = this.snapshot!; // real rollback + this.snapshot = null; + throw err; + } finally { + this.txDepth--; + } + } +} + +const asEngine = (e: FakeEngine) => e as unknown as Parameters[0]; + +/** A step over `n` synthetic rows that records what it wrote, per attempt. */ +function makeStep( + n: number, + overrides: Partial> = {}, +): MigrationPlanStep<{ i: number }> & { written: number[]; undone: number[]; attempts: number[] } { + const written: number[] = []; + const undone: number[] = []; + const attempts: number[] = []; + return { + name: 'step', + written, + undone, + attempts, + async load() { return Array.from({ length: n }, (_, i) => ({ i })); }, + async forward(rows, ctx) { + attempts.push(ctx.attempt); + // Idempotency by natural key, exactly as the header prescribes on a + // re-attempt whose prior outcome is unknown. + for (const r of rows) if (ctx.attempt === 1 || !written.includes(r.i)) written.push(r.i); + }, + async compensate(rows) { for (const r of rows) undone.push(r.i); }, + ...overrides, + }; +} + +const JOURNAL = 'sys_migration_journal'; +const kindsOf = (e: FakeEngine) => (e.tables.get(JOURNAL) ?? []).map((r) => r.kind); + +// ── capability gate ─────────────────────────────────────────────────────── + +describe('capability gate (ADR-0119 D2 item 7 / D4 probe)', () => { + it('refuses to start when the driver cannot begin a transaction', async () => { + const engine = new FakeEngine({ driverHasTx: false }); + const plan: MigrationPlan = { id: 'p', steps: [makeStep(3)] }; + await expect(runMigrationJournal(asEngine(engine), plan)).rejects.toThrow(MigrationJournalRefusal); + // Refused means NOTHING was written — not even a run_started row claiming + // a run that never legitimately began. + expect(engine.tables.get(JOURNAL) ?? []).toHaveLength(0); + }); + + it('engineCanRollBack is two-level: engine method AND driver capability', () => { + expect(engineCanRollBack(new FakeEngine())).toBe(true); + expect(engineCanRollBack(new FakeEngine({ driverHasTx: false }))).toBe(false); + expect(engineCanRollBack({})).toBe(false); + expect(engineCanRollBack(null)).toBe(false); + // A test double with no driver registry keeps the engine-level answer. + expect(engineCanRollBack({ transaction: () => {} })).toBe(true); + }); +}); + +// ── preflight ───────────────────────────────────────────────────────────── + +describe('preflight (ADR-0119 D2 item 1)', () => { + it('runs every validator before any write and refuses on failure', async () => { + const engine = new FakeEngine(); + const good = makeStep(2, { preflight: vi.fn(async () => {}) }); + const bad = makeStep(2, { preflight: async () => { throw new Error('column missing'); } }); + const plan: MigrationPlan = { id: 'p', steps: [good, bad] }; + + await expect(runMigrationJournal(asEngine(engine), plan)).rejects.toThrow(/preflight/i); + // The point of preflight: step 1 did not write because step 2 would fail. + expect(good.written).toEqual([]); + expect(engine.tables.get(JOURNAL) ?? []).toHaveLength(0); + }); + + it("refuses a plan declaring onCrash:'compensate' whose steps cannot compensate", async () => { + const engine = new FakeEngine(); + const step = makeStep(2, { compensate: undefined }); + const plan: MigrationPlan = { id: 'p', steps: [step], onCrash: 'compensate' }; + await expect(runMigrationJournal(asEngine(engine), plan)).rejects.toThrow(/NOT_COMPENSABLE|compensate/i); + }); +}); + +// ── the happy path and its journal shape ────────────────────────────────── + +describe('forward run', () => { + it('chunks per plan, commits each, and journals a complete trace', async () => { + const engine = new FakeEngine(); + const step = makeStep(5); + const plan: MigrationPlan = { id: 'p', steps: [step], chunkSize: 2 }; + + const result = await runMigrationJournal(asEngine(engine), plan); + + expect(result.status).toBe('completed'); + expect(result.chunksTotal).toBe(3); // ceil(5/2) + expect(step.written).toEqual([0, 1, 2, 3, 4]); + expect(kindsOf(engine)).toEqual([ + 'run_started', + 'chunk_started', 'chunk_done', + 'chunk_started', 'chunk_done', + 'chunk_started', 'chunk_done', + 'run_done', + ]); + }); + + it('writes chunk_done INSIDE the chunk transaction and chunk_started outside it', async () => { + const engine = new FakeEngine(); + const plan: MigrationPlan = { id: 'p', steps: [makeStep(1)], chunkSize: 1 }; + await runMigrationJournal(asEngine(engine), plan); + + // Context carries `__tx` only for writes that joined a transaction. The + // asymmetry IS the design: chunk_started must survive a crash, chunk_done + // must not survive a rollback. + const ctxFor = (kind: string) => { + const rows = engine.tables.get(JOURNAL)!; + const idx = rows.findIndex((r) => r.kind === kind); + return engine.insertContexts[idx] as { __tx?: boolean }; + }; + expect(ctxFor('chunk_started').__tx).toBeUndefined(); + expect(ctxFor('chunk_done').__tx).toBe(true); + expect(ctxFor('run_started').__tx).toBeUndefined(); + }); +}); + +// ── crash mid-chunk → resume completes exactly once ─────────────────────── + +describe('crash mid-chunk → resume (ADR-0119 D2 item 5, acceptance case 1)', () => { + it('rolls the killed chunk back, then resumes it exactly once on restart', async () => { + const engine = new FakeEngine(); + let boom = true; + const step = makeStep(4, { + async forward(rows, ctx) { + // Chunk 1 dies the first time it is attempted — after writing, so the + // rollback has something to undo. + (step as any).attempts.push(ctx.attempt); + for (const r of rows) if (ctx.attempt === 1 || !(step as any).written.includes(r.i)) (step as any).written.push(r.i); + if (ctx.chunkIndex === 1 && boom) { boom = false; throw new Error('killed mid-chunk'); } + }, + }); + const plan: MigrationPlan = { id: 'p', steps: [step], chunkSize: 2, onCrash: 'resume' }; + + // First run: chunk 0 commits, chunk 1 throws → LIFO compensation undoes 0. + const first = await runMigrationJournal(asEngine(engine), plan); + expect(first.status).toBe('compensated'); + + // The killed chunk left `chunk_started` with NO `chunk_done` — "outcome + // unknown" — and its own writes were rolled back with the transaction. + const events = await readRunJournal(asEngine(engine), first.runId); + const started = events.filter((e) => e.kind === 'chunk_started' && e.chunk_index === 1); + const done = events.filter((e) => e.kind === 'chunk_done' && e.chunk_index === 1); + expect(started).toHaveLength(1); + expect(done).toHaveLength(0); + }); + + it('resumes forward from the first chunk lacking chunk_done, skipping committed ones', async () => { + const engine = new FakeEngine(); + const step = makeStep(6); + const plan: MigrationPlan = { id: 'p', steps: [step], chunkSize: 2, onCrash: 'resume' }; + + // Simulate a crash by journalling a run that got through chunk 0 only. + const runId = 'run-partial'; + await engine.insert(JOURNAL, { + run_id: runId, seq: 0, kind: 'run_started', + plan_hash: hashMigrationPlan(plan, planChunks(plan, [6], 2)), + detail: JSON.stringify({ planId: 'p' }), + }); + await engine.insert(JOURNAL, { run_id: runId, seq: 1, kind: 'chunk_started', chunk_index: 0, attempt: 1 }); + await engine.insert(JOURNAL, { run_id: runId, seq: 2, kind: 'chunk_done', chunk_index: 0, attempt: 1 }); + + const result = await resumeMigrationJournal(asEngine(engine), plan, runId); + + expect(result.status).toBe('completed'); + // Chunk 0 was durable — it is NOT redone. Rows 0,1 are absent from this + // process's writes precisely because the journal says they already landed. + expect(step.written).toEqual([2, 3, 4, 5]); + const events = await readRunJournal(asEngine(engine), runId); + expect(events.filter((e) => e.kind === 'chunk_done').map((e) => e.chunk_index)).toEqual([0, 1, 2]); + expect(events.at(-1)!.kind).toBe('run_done'); + }); + + it('tells a re-attempted chunk that its prior outcome is unknown (attempt > 1)', async () => { + const engine = new FakeEngine(); + const step = makeStep(2); + const plan: MigrationPlan = { id: 'p', steps: [step], chunkSize: 2, onCrash: 'resume' }; + const runId = 'run-unknown'; + await engine.insert(JOURNAL, { + run_id: runId, seq: 0, kind: 'run_started', + plan_hash: hashMigrationPlan(plan, planChunks(plan, [2], 2)), + }); + // chunk_started with no chunk_done — the crash signature. + await engine.insert(JOURNAL, { run_id: runId, seq: 1, kind: 'chunk_started', chunk_index: 0, attempt: 1 }); + + await resumeMigrationJournal(asEngine(engine), plan, runId); + + // The callback is TOLD it is a re-attempt, which is what lets it recheck + // by natural key instead of blindly double-writing. + expect(step.attempts).toEqual([2]); + expect(step.written).toEqual([0, 1]); + }); +}); + +// ── plan-hash mismatch → refuse ─────────────────────────────────────────── + +describe('plan-hash mismatch on resume (acceptance case 3)', () => { + it('refuses to resume a changed plan against an old journal', async () => { + const engine = new FakeEngine(); + const plan: MigrationPlan = { id: 'p', steps: [makeStep(4)], chunkSize: 2 }; + const runId = 'run-stale'; + await engine.insert(JOURNAL, { + run_id: runId, seq: 0, kind: 'run_started', plan_hash: 'a-hash-from-a-different-plan', + }); + + await expect(resumeMigrationJournal(asEngine(engine), plan, runId)) + .rejects.toThrow(/PLAN_CHANGED|plan hash/i); + }); + + it('the hash tracks chunk boundaries, not just step names', () => { + const plan: MigrationPlan = { id: 'p', steps: [makeStep(4)] }; + const a = hashMigrationPlan(plan, planChunks(plan, [4], 2)); + const b = hashMigrationPlan(plan, planChunks(plan, [4], 4)); + // Same steps, same rows, different chunking — "chunk 1" means something + // different in each, so resuming across them must not be allowed. + expect(a).not.toBe(b); + }); +}); + +// ── LIFO compensation ───────────────────────────────────────────────────── + +describe('LIFO compensation (ADR-0119 D2 item 4)', () => { + it('undoes committed chunks newest-first', async () => { + const engine = new FakeEngine(); + const order: number[] = []; + const step = makeStep(6, { + async forward(rows, ctx) { if (ctx.chunkIndex === 2) throw new Error('fail late'); }, + async compensate(_rows, ctx) { order.push(ctx.chunkIndex); }, + }); + const plan: MigrationPlan = { id: 'p', steps: [step], chunkSize: 2 }; + + const result = await runMigrationJournal(asEngine(engine), plan); + + expect(result.status).toBe('compensated'); + expect(order).toEqual([1, 0]); // newest-first, not commit order + const events = await readRunJournal(asEngine(engine), result.runId); + expect(events.filter((e) => e.kind === 'compensated').map((e) => e.chunk_index)).toEqual([1, 0]); + expect(events.at(-1)!.kind).toBe('run_failed'); + }); + + it('halts loudly when a compensation fails — no silent partial (acceptance case 2)', async () => { + const engine = new FakeEngine(); + const step = makeStep(6, { + async forward(_rows, ctx) { if (ctx.chunkIndex === 2) throw new Error('fail late'); }, + async compensate(_rows, ctx) { if (ctx.chunkIndex === 1) throw new Error('compensation itself failed'); }, + }); + const plan: MigrationPlan = { id: 'p', steps: [step], chunkSize: 2 }; + + const result = await runMigrationJournal(asEngine(engine), plan); + + // NOT 'compensated' — the database is in a state no clean story covers, + // and the result says so rather than reporting a tidy rollback. + expect(result.status).toBe('failed'); + const events = await readRunJournal(asEngine(engine), result.runId); + const failure = events.filter((e) => e.kind === 'run_failed').at(-1)!; + expect(JSON.parse(failure.detail!)).toMatchObject({ phase: 'compensate' }); + expect(failure.chunk_index).toBe(1); + // It STOPPED at the failure: chunk 0 was not compensated behind its back. + expect(events.filter((e) => e.kind === 'compensated').map((e) => e.chunk_index)).toEqual([]); + }); + + it('halts when a committed chunk belongs to a step with no compensate()', async () => { + const engine = new FakeEngine(); + const step = makeStep(4, { + async forward(_rows, ctx) { if (ctx.chunkIndex === 1) throw new Error('boom'); }, + compensate: undefined, + }); + const plan: MigrationPlan = { id: 'p', steps: [step], chunkSize: 2 }; + + const result = await runMigrationJournal(asEngine(engine), plan); + + expect(result.status).toBe('failed'); + const events = await readRunJournal(asEngine(engine), result.runId); + expect(JSON.parse(events.at(-1)!.detail!)).toMatchObject({ reason: 'step declares no compensate()' }); + }); +}); + +// ── discovery ───────────────────────────────────────────────────────────── + +describe('findInterruptedRuns (the boot scanner input)', () => { + it('reports a run that started and never concluded, splitting known from unknown', async () => { + const engine = new FakeEngine(); + await engine.insert(JOURNAL, { run_id: 'r1', seq: 0, kind: 'run_started', plan_hash: 'h', detail: JSON.stringify({ planId: 'backfill' }) }); + await engine.insert(JOURNAL, { run_id: 'r1', seq: 1, kind: 'chunk_started', chunk_index: 0 }); + await engine.insert(JOURNAL, { run_id: 'r1', seq: 2, kind: 'chunk_done', chunk_index: 0 }); + await engine.insert(JOURNAL, { run_id: 'r1', seq: 3, kind: 'chunk_started', chunk_index: 1 }); + // …and the process died here. + + const found = await findInterruptedRuns(asEngine(engine)); + + expect(found).toHaveLength(1); + expect(found[0]).toMatchObject({ + runId: 'r1', planId: 'backfill', + committedChunks: [0], + unknownChunks: [1], // started, no done — the only state recovery reasons about + }); + }); + + it('does not report a completed run, nor a failed one that fully compensated', async () => { + const engine = new FakeEngine(); + await engine.insert(JOURNAL, { run_id: 'done', seq: 0, kind: 'run_started' }); + await engine.insert(JOURNAL, { run_id: 'done', seq: 1, kind: 'run_done' }); + + await engine.insert(JOURNAL, { run_id: 'undone', seq: 0, kind: 'run_started' }); + await engine.insert(JOURNAL, { run_id: 'undone', seq: 1, kind: 'chunk_done', chunk_index: 0 }); + await engine.insert(JOURNAL, { run_id: 'undone', seq: 2, kind: 'compensated', chunk_index: 0 }); + await engine.insert(JOURNAL, { run_id: 'undone', seq: 3, kind: 'run_failed' }); + + expect(await findInterruptedRuns(asEngine(engine))).toEqual([]); + }); + + it('reports a failed run whose compensation did NOT finish', async () => { + const engine = new FakeEngine(); + await engine.insert(JOURNAL, { run_id: 'stuck', seq: 0, kind: 'run_started' }); + await engine.insert(JOURNAL, { run_id: 'stuck', seq: 1, kind: 'chunk_done', chunk_index: 0 }); + await engine.insert(JOURNAL, { run_id: 'stuck', seq: 2, kind: 'chunk_done', chunk_index: 1 }); + await engine.insert(JOURNAL, { run_id: 'stuck', seq: 3, kind: 'compensated', chunk_index: 1 }); + await engine.insert(JOURNAL, { run_id: 'stuck', seq: 4, kind: 'run_failed' }); + + const found = await findInterruptedRuns(asEngine(engine)); + expect(found).toHaveLength(1); + expect(found[0].committedChunks).toEqual([0, 1]); + expect(found[0].compensatedChunks).toEqual([1]); // chunk 0 still owes an answer + }); +}); + +// ── onCrash: 'compensate' ───────────────────────────────────────────────── + +describe("onCrash: 'compensate' (ADR-0119 D2 item 5)", () => { + it('unwinds a rediscovered run instead of carrying it forward', async () => { + const engine = new FakeEngine(); + const step = makeStep(6); + const plan: MigrationPlan = { id: 'p', steps: [step], chunkSize: 2, onCrash: 'compensate' }; + const runId = 'run-unwind'; + await engine.insert(JOURNAL, { + run_id: runId, seq: 0, kind: 'run_started', + plan_hash: hashMigrationPlan(plan, planChunks(plan, [6], 2)), + }); + await engine.insert(JOURNAL, { run_id: runId, seq: 1, kind: 'chunk_done', chunk_index: 0 }); + await engine.insert(JOURNAL, { run_id: runId, seq: 2, kind: 'chunk_done', chunk_index: 1 }); + + const result = await resumeMigrationJournal(asEngine(engine), plan, runId); + + expect(result.status).toBe('compensated'); + expect(step.written).toEqual([]); // never went forward + expect(step.undone).toEqual([2, 3, 0, 1]); // chunk 1's rows, then chunk 0's + }); +}); + +// ── sequence integrity ──────────────────────────────────────────────────── + +describe('journal sequence', () => { + it('continues the sequence across a resume rather than restarting it', async () => { + const engine = new FakeEngine(); + const plan: MigrationPlan = { id: 'p', steps: [makeStep(4)], chunkSize: 2, onCrash: 'resume' }; + const runId = 'run-seq'; + await engine.insert(JOURNAL, { + run_id: runId, seq: 0, kind: 'run_started', + plan_hash: hashMigrationPlan(plan, planChunks(plan, [4], 2)), + }); + await engine.insert(JOURNAL, { run_id: runId, seq: 1, kind: 'chunk_started', chunk_index: 0 }); + await engine.insert(JOURNAL, { run_id: runId, seq: 2, kind: 'chunk_done', chunk_index: 0 }); + + // A restarted sequence would collide with the unique (run_id, seq) index + // the object declares — which the fake enforces, so this would throw. + await expect(resumeMigrationJournal(asEngine(engine), plan, runId)).resolves.toMatchObject({ status: 'completed' }); + + const events = await readRunJournal(asEngine(engine), runId); + expect(events.map((e) => e.seq)).toEqual([...events.map((_, i) => i)]); + }); +}); diff --git a/packages/core/src/utils/migration-journal.ts b/packages/core/src/utils/migration-journal.ts new file mode 100644 index 0000000000..1fc0362049 --- /dev/null +++ b/packages/core/src/utils/migration-journal.ts @@ -0,0 +1,670 @@ +// Copyright (c) 2026 ObjectStack. Licensed under the Apache-2.0 license. + +/** + * `runMigrationJournal` — the framework-owned runner for data migrations that + * are too big, too long, or too multi-step to live in one transaction + * (ADR-0119 D2, #4617). + * + * ## Why this is framework-owned rather than four hand-rolled copies + * + * Four migration-class consumers independently converged on the same four + * moves — dry-run preflight, an undo journal, LIFO compensation, re-entrant + * forward recovery: ADR-0105 D13 promotion, ADR-0117 D8's ownership backfill, + * the org lifecycle transitions, and ADR-0119 D10's master-data distribution + * (#4585). One copy is engineering. Four is platform debt, and the fourth + * author would have had to rediscover the `chunk_done`-inside-the-transaction + * subtlety below from scratch — or, far more likely, not rediscover it. + * + * ## Why a journal at all, given ADR-0034 gave us transactions + * + * ADR-0119 D1 made `engine.transaction()` reachable through the contract, but + * a transaction cannot be the whole answer here: + * + * - a million-row backfill cannot hold one write-lock for its duration; + * - `driver-memory`'s `beginTransaction` deep-clones the entire database, so + * "just wrap the whole thing" is O(db) per begin; + * - `ObjectQL.transaction()` binds the DEFAULT driver only, so a migration + * spanning datasources silently commits part of its work outside it; + * - a process KILLED — as distinct from a thrown error — defeats in-process + * rollback entirely, and that is the case operators actually hit. + * + * So the unit of atomicity is the CHUNK, and durability across chunks is the + * journal. Everything else in this file follows from that one sentence. + * + * ## The invariant that carries the whole design + * + * `chunk_done(i)` is written INSIDE the chunk's own transaction, so + * `done ⇔ committed` holds by construction rather than by luck. + * `chunk_started(i)` is written autonomously BEFORE it. A reader who "tidies" + * that asymmetry destroys recovery: it is what gives `started ∧ ¬done` exactly + * one meaning — **the outcome is unknown** — which is the only state a crash + * can leave and the only state recovery has to reason about. + * + * ## Delivery semantics: at-least-once, idempotency is the caller's job + * + * Inherited verbatim from `./bulk-write.ts` rather than re-derived, because a + * second delivery-semantics story in the same codebase is a second thing to + * get subtly wrong. Forward and compensate callbacks receive an `attempt` + * counter; `attempt > 1` means the previous outcome is UNKNOWN — the write may + * or may not have committed — and the callback must recheck by natural key + * before re-writing. That is the same contract the seed loader and import + * runner already honour on `attempt > 1`. + */ + +import { createHash, randomUUID } from 'node:crypto'; +import type { IObjectQLEngine } from '@objectstack/spec/contracts'; +import { + MIGRATION_JOURNAL_OBJECT, + type MigrationJournalEvent, + type MigrationJournalKind, + type MigrationOnCrashPolicy, +} from '@objectstack/spec/system'; + +/** Journal writes and recovery reads run as the platform, never as a user. */ +const SYSTEM_CTX = { isSystem: true } as const; + +/** Rows per chunk when a plan does not choose. Matches `bulk-write.ts`. */ +const DEFAULT_CHUNK_SIZE = 200; + +/** + * Can this runtime actually roll back? — the ADR-0119 D4 gate, shared. + * + * Exported from `@objectstack/core` and consumed by + * `@objectstack/metadata-protocol`'s `batchData` (which depends on core, so + * the direction is legal) so the two cannot drift. They were the same two-line + * condition written twice, which is precisely the shape that drifts by one + * clause and leaves one caller believing it has atomicity it does not have. + * + * TWO levels, both necessary. `engine.transaction()` exists but runs the + * callback with NO transaction and NO rollback when the default driver lacks + * `beginTransaction` — a declared caveat of the contract member (ADR-0119 D1), + * and one that turns "atomic" back into a lie precisely where it matters. So + * where the driver registry is inspectable the driver is checked too; where it + * is not (test doubles), the engine-level probe is all there is. + * + * A type predicate, not a bare boolean: every caller's next move is to CALL + * `transaction`, and on the host surfaces that declare it optionally + * (`MetadataHostEngine`) a boolean would leave each one re-narrowing by hand — + * which is the same restatement this helper exists to remove. + */ +export function engineCanRollBack(engine: T): engine is T & EngineWithTransaction { + const e = engine as { + transaction?: unknown; + getDefaultDriverName?: () => string | undefined; + getDriverByName?: (name: string) => unknown; + } | null | undefined; + if (typeof e?.transaction !== 'function') return false; + const defaultDriverName = e.getDefaultDriverName?.(); + const defaultDriver = defaultDriverName ? e.getDriverByName?.(defaultDriverName) : undefined; + return !defaultDriver || typeof (defaultDriver as { beginTransaction?: unknown }).beginTransaction === 'function'; +} + +/** What {@link engineCanRollBack} proves is present. Mirrors `IObjectQLEngine['transaction']`. */ +export interface EngineWithTransaction { + transaction(callback: (trxCtx: any) => Promise, baseContext?: any): Promise; +} + +/** What a forward/compensate callback is told about the chunk it is running. */ +export interface MigrationChunkContext { + readonly runId: string; + /** Run-global chunk index — the LIFO ordering key, stable across a resume. */ + readonly chunkIndex: number; + /** + * 1 on the first try. `> 1` means a previous attempt's outcome is UNKNOWN: + * recheck by natural key before re-writing (see this file's header). + */ + readonly attempt: number; + /** + * The transaction-bound execution context. Thread it to every engine call + * this callback makes — `engine.insert(obj, row, { context })` — so the + * write joins the chunk's transaction instead of committing beside it. + */ + readonly context: unknown; +} + +/** One step of a plan. Steps run in declaration order; each is chunked. */ +export interface MigrationPlanStep { + readonly name: string; + /** + * Read-only preflight. Throw to refuse the run. Runs for EVERY step before + * any step writes — a plan that would fail at step 3 must not have written + * step 1 (ADR-0117 D8's fail-closed enable gate, generalized). + */ + preflight?(engine: IObjectQLEngine): Promise; + /** The rows this step processes. Called once, before chunking. */ + load(engine: IObjectQLEngine): Promise; + /** Forward work for one chunk. Runs INSIDE the chunk's transaction. */ + forward(rows: TRow[], ctx: MigrationChunkContext, engine: IObjectQLEngine): Promise; + /** + * Undo one previously-committed chunk. Runs in its OWN transaction. + * A step without one makes the plan non-compensable — which the runner + * refuses up front rather than discovering at the worst possible moment + * (see {@link runMigrationJournal}'s preflight). + */ + compensate?(rows: TRow[], ctx: MigrationChunkContext, engine: IObjectQLEngine): Promise; +} + +export interface MigrationPlan { + /** Stable plan id. Part of the plan hash; identifies the plan across runs. */ + readonly id: string; + /** Optional join to `sys_migration.id` when this plan implements a named migration. */ + readonly migrationId?: string; + readonly steps: ReadonlyArray>; + readonly chunkSize?: number; + /** + * What a REDISCOVERED (crashed) run should do. Note this governs restart + * only — an in-run failure always compensates, because the runner is still + * alive to do it and a half-applied plan is nobody's intent. + */ + readonly onCrash?: MigrationOnCrashPolicy; +} + +/** One chunk in the run-global chunk plan. */ +export interface MigrationChunk { + /** Run-global index, 0-based, stable for a given plan hash. */ + readonly index: number; + readonly stepIndex: number; + readonly stepName: string; + readonly offset: number; + readonly length: number; +} + +export interface MigrationRunResult { + readonly runId: string; + /** + * `completed` — every chunk committed. + * `compensated` — a chunk failed and every committed chunk was undone. + * `failed` — a chunk failed AND compensation could not finish. The database + * is in a partial state that needs a human; the journal says exactly where. + */ + readonly status: 'completed' | 'compensated' | 'failed'; + readonly chunksTotal: number; + readonly chunksCommitted: number; + readonly chunksCompensated: number; + readonly planHash: string; + /** The failure that ended a non-`completed` run. */ + readonly error?: unknown; +} + +/** A run found by {@link findInterruptedRuns} — started, never concluded. */ +export interface InterruptedRun { + readonly runId: string; + readonly planId: string; + readonly planHash: string; + readonly migrationId?: string; + readonly startedAt?: string; + /** Chunks whose `chunk_done` is present — known committed. */ + readonly committedChunks: number[]; + /** Chunks with `chunk_started` and no `chunk_done` — outcome UNKNOWN. */ + readonly unknownChunks: number[]; + readonly compensatedChunks: number[]; +} + +/** Raised when the runner refuses to start or to resume. Never a partial run. */ +export class MigrationJournalRefusal extends Error { + readonly code: string; + constructor(code: string, message: string) { + super(message); + this.name = 'MigrationJournalRefusal'; + this.code = code; + } +} + +// ── plan shape ──────────────────────────────────────────────────────────── + +/** Flatten steps × rows into the run-global chunk list. */ +export function planChunks( + plan: MigrationPlan, + rowCounts: readonly number[], + chunkSize = plan.chunkSize ?? DEFAULT_CHUNK_SIZE, +): MigrationChunk[] { + const size = Math.max(1, chunkSize); + const chunks: MigrationChunk[] = []; + plan.steps.forEach((step, stepIndex) => { + const total = rowCounts[stepIndex] ?? 0; + for (let offset = 0; offset < total; offset += size) { + chunks.push({ + index: chunks.length, + stepIndex, + stepName: step.name, + offset, + length: Math.min(size, total - offset), + }); + } + }); + return chunks; +} + +/** + * Hash the plan SHAPE — id, step names, and the chunk boundaries. + * + * Resuming a changed plan against an old journal would apply chunk boundaries + * the journal never described: "chunk 7 done" would name a different range of + * different rows, and the resume would skip work it never did. So the hash + * covers exactly what a chunk index means, and a mismatch REFUSES. + */ +export function hashMigrationPlan(plan: MigrationPlan, chunks: readonly MigrationChunk[]): string { + const shape = JSON.stringify({ + id: plan.id, + steps: plan.steps.map((s) => s.name), + chunks: chunks.map((c) => [c.stepIndex, c.offset, c.length]), + }); + return createHash('sha256').update(shape, 'utf8').digest('hex').slice(0, 32); +} + +// ── journal I/O ─────────────────────────────────────────────────────────── + +/** + * Append one event. + * + * `execContext` is the transaction-bound context when the event must share a + * chunk's fate (`chunk_done`, `compensated`) and undefined when it must NOT + * (`chunk_started`, and every run-level event). Passing the wrong one is the + * single most consequential mistake available in this file — see the header. + */ +async function appendEvent( + engine: IObjectQLEngine, + event: MigrationJournalEvent, + execContext?: unknown, +): Promise { + await engine.insert( + MIGRATION_JOURNAL_OBJECT, + { ...event, created_at: event.created_at ?? new Date().toISOString() }, + { context: execContext ?? { ...SYSTEM_CTX } }, + ); +} + +/** + * Every event for a run, ordered by `seq`. + * + * Sorted in memory, deliberately. `seq` is the ordering authority (wall-clock + * stamps tie at coarse resolution and skew), and a run's journal is bounded by + * its chunk count, so this costs nothing and removes recovery's dependence on + * driver-side sort behaviour — which is not something a recovery path should + * be discovering the edges of. + */ +export async function readRunJournal( + engine: IObjectQLEngine, + runId: string, +): Promise { + const rows = (await engine.find( + MIGRATION_JOURNAL_OBJECT, + { where: { run_id: runId } }, + { context: { ...SYSTEM_CTX } }, + )) as MigrationJournalEvent[]; + return [...(rows ?? [])].sort((a, b) => Number(a.seq) - Number(b.seq)); +} + +/** Chunk indices carrying `kind`, as a set. */ +function chunkSetOf(events: readonly MigrationJournalEvent[], kind: MigrationJournalKind): Set { + const out = new Set(); + for (const e of events) { + if (e.kind === kind && typeof e.chunk_index === 'number') out.add(e.chunk_index); + } + return out; +} + +/** + * Runs that started and never concluded — the boot scanner's input. + * + * "Concluded" means `run_done` (finished forward) or `run_failed` with every + * committed chunk compensated (finished backward). Anything else is a run that + * stopped mid-flight and still owes the operator an answer. + */ +export async function findInterruptedRuns(engine: IObjectQLEngine): Promise { + const started = (await engine.find( + MIGRATION_JOURNAL_OBJECT, + { where: { kind: 'run_started' } }, + { context: { ...SYSTEM_CTX } }, + )) as MigrationJournalEvent[]; + + const out: InterruptedRun[] = []; + for (const start of started ?? []) { + const events = await readRunJournal(engine, start.run_id); + if (events.some((e) => e.kind === 'run_done')) continue; + + const committed = chunkSetOf(events, 'chunk_done'); + const compensated = chunkSetOf(events, 'compensated'); + const outstanding = [...committed].filter((i) => !compensated.has(i)); + // A failed run whose committed chunks were all undone is settled: it ended + // backward, on purpose, and its rows prove it. + if (events.some((e) => e.kind === 'run_failed') && outstanding.length === 0) continue; + + const unknown = [...chunkSetOf(events, 'chunk_started')].filter((i) => !committed.has(i)); + let planId = start.run_id; + try { + planId = start.detail ? (JSON.parse(start.detail).planId ?? start.run_id) : start.run_id; + } catch { + // A malformed detail payload must not hide an interrupted run — the run + // is still reported, just without its friendly plan id. + } + out.push({ + runId: start.run_id, + planId, + planHash: start.plan_hash ?? '', + migrationId: start.migration_id, + startedAt: start.created_at, + committedChunks: [...committed].sort((a, b) => a - b), + unknownChunks: unknown.sort((a, b) => a - b), + compensatedChunks: [...compensated].sort((a, b) => a - b), + }); + } + return out; +} + +// ── the runner ──────────────────────────────────────────────────────────── + +export interface RunMigrationJournalOptions { + /** Supply to resume an existing run; omit to start a new one. */ + readonly runId?: string; + readonly chunkSize?: number; + /** Injectable for deterministic tests. */ + readonly now?: () => string; +} + +interface LoadedPlan { + readonly chunks: MigrationChunk[]; + readonly planHash: string; + readonly rowsByStep: unknown[][]; +} + +/** Load every step's rows, derive the chunk plan, hash it. */ +async function loadPlan( + engine: IObjectQLEngine, + plan: MigrationPlan, + chunkSize?: number, +): Promise { + const rowsByStep: unknown[][] = []; + for (const step of plan.steps) rowsByStep.push((await step.load(engine)) ?? []); + const chunks = planChunks(plan, rowsByStep.map((r) => r.length), chunkSize ?? plan.chunkSize); + return { chunks, planHash: hashMigrationPlan(plan, chunks), rowsByStep }; +} + +/** + * Run `plan` under the journal, or resume a run left behind by a crash. + * + * Refuses (never partially runs) when: the runtime cannot roll back; any + * step's preflight fails; the plan declares `onCrash: 'compensate'` but some + * step cannot compensate; or a resume's plan hash disagrees with the journal. + */ +export async function runMigrationJournal( + engine: IObjectQLEngine, + plan: MigrationPlan, + options: RunMigrationJournalOptions = {}, +): Promise { + const now = options.now ?? (() => new Date().toISOString()); + + // ── capability gate ─────────────────────────────────────────────────── + // Refuse rather than degrade. A runner whose chunks are not actually + // atomic writes `chunk_done` rows that mean nothing, and a journal that + // cannot be trusted is worse than no journal — it will be believed. + if (!engineCanRollBack(engine)) { + throw new MigrationJournalRefusal( + 'NOT_IMPLEMENTED', + `Migration plan '${plan.id}' requires engine transaction support; this runtime cannot roll back. ` + + `The journal's chunk_done markers would not mean "committed", so the run is refused rather than started.`, + ); + } + + const { chunks, planHash, rowsByStep } = await loadPlan(engine, plan, options.chunkSize); + + // ── resume bookkeeping ──────────────────────────────────────────────── + const resuming = Boolean(options.runId); + const runId = options.runId ?? randomUUID(); + let events: MigrationJournalEvent[] = []; + let seq = 0; + let committed = new Set(); + let compensated = new Set(); + const attemptsByChunk = new Map(); + + if (resuming) { + events = await readRunJournal(engine, runId); + if (events.length === 0) { + throw new MigrationJournalRefusal('NO_SUCH_RUN', `No journal rows for run '${runId}'.`); + } + const start = events.find((e) => e.kind === 'run_started'); + if (start?.plan_hash && start.plan_hash !== planHash) { + // The plan changed under a journal that describes the old one. Chunk 7 + // in the journal and chunk 7 in this plan are different rows; resuming + // would skip work that was never done. + throw new MigrationJournalRefusal( + 'PLAN_CHANGED', + `Refusing to resume run '${runId}': plan hash ${planHash} does not match the journal's ${start.plan_hash}. ` + + `The chunk boundaries recorded in the journal describe a different plan.`, + ); + } + if (events.some((e) => e.kind === 'run_done')) { + return { + runId, status: 'completed', chunksTotal: chunks.length, + chunksCommitted: chunkSetOf(events, 'chunk_done').size, + chunksCompensated: chunkSetOf(events, 'compensated').size, planHash, + }; + } + seq = events.reduce((m, e) => Math.max(m, Number(e.seq) + 1), 0); + committed = chunkSetOf(events, 'chunk_done'); + compensated = chunkSetOf(events, 'compensated'); + for (const e of events) { + if (e.kind === 'chunk_started' && typeof e.chunk_index === 'number') { + attemptsByChunk.set(e.chunk_index, (attemptsByChunk.get(e.chunk_index) ?? 0) + 1); + } + } + } + + // ── preflight ───────────────────────────────────────────────────────── + // Every validator runs before any write, so a plan that would fail at step 3 + // has not written step 1. On a resume this re-runs too: the world moved + // while the process was dead, and the reason to refuse may have appeared + // since. + for (const step of plan.steps) { + if (!step.preflight) continue; + try { + await step.preflight(engine); + } catch (err) { + throw new MigrationJournalRefusal( + 'PREFLIGHT_FAILED', + `Migration plan '${plan.id}' refused: preflight for step '${step.name}' failed: ${errText(err)}`, + ); + } + } + + // A plan that says "undo me on crash" must be able to. Discovering that it + // cannot at compensation time means discovering it with rows already + // written and no way back. + if (plan.onCrash === 'compensate') { + const missing = plan.steps.filter((s) => !s.compensate).map((s) => s.name); + if (missing.length > 0) { + throw new MigrationJournalRefusal( + 'NOT_COMPENSABLE', + `Migration plan '${plan.id}' declares onCrash: 'compensate' but step(s) ${missing.join(', ')} declare no compensate().`, + ); + } + } + + const rowsOf = (c: MigrationChunk): unknown[] => rowsByStep[c.stepIndex].slice(c.offset, c.offset + c.length); + const next = (): number => seq++; + + if (!resuming) { + await appendEvent(engine, { + run_id: runId, seq: next(), kind: 'run_started', plan_hash: planHash, + migration_id: plan.migrationId, created_at: now(), + detail: JSON.stringify({ + planId: plan.id, + onCrash: plan.onCrash ?? 'resume', + chunks: chunks.map((c) => ({ i: c.index, step: c.stepName, offset: c.offset, length: c.length })), + }), + }); + } + + // A rediscovered run whose policy is 'compensate' does not go forward at + // all — it unwinds what it already did and stops. + if (resuming && plan.onCrash === 'compensate') { + return await unwind(engine, plan, { + runId, planHash, chunks, rowsOf, next, now, + committed, compensated, chunksTotal: chunks.length, + cause: new Error(`run '${runId}' rediscovered after interruption; plan policy is compensate`), + }); + } + + // ── forward ─────────────────────────────────────────────────────────── + for (const chunk of chunks) { + if (committed.has(chunk.index)) continue; // already durable — skip, do not redo + const attempt = (attemptsByChunk.get(chunk.index) ?? 0) + 1; + attemptsByChunk.set(chunk.index, attempt); + const step = plan.steps[chunk.stepIndex]; + const rows = rowsOf(chunk); + + // Autonomous, BEFORE the transaction: this is what makes an interrupted + // chunk visible as "started, outcome unknown" rather than invisible. + await appendEvent(engine, { + run_id: runId, seq: next(), kind: 'chunk_started', + chunk_index: chunk.index, attempt, migration_id: plan.migrationId, created_at: now(), + }); + + try { + await engine.transaction(async (trxCtx: unknown) => { + await step.forward(rows, { runId, chunkIndex: chunk.index, attempt, context: trxCtx }, engine); + // INSIDE the transaction — `done ⇔ committed`, not a race. + await appendEvent( + engine, + { + run_id: runId, seq: next(), kind: 'chunk_done', + chunk_index: chunk.index, attempt, migration_id: plan.migrationId, created_at: now(), + }, + trxCtx, + ); + }, { ...SYSTEM_CTX }); + committed.add(chunk.index); + } catch (err) { + // The chunk rolled back, so nothing of it is on disk — including its + // `chunk_done`. Unwind what earlier chunks committed. + return await unwind(engine, plan, { + runId, planHash, chunks, rowsOf, next, now, + committed, compensated, chunksTotal: chunks.length, cause: err, + }); + } + } + + await appendEvent(engine, { + run_id: runId, seq: next(), kind: 'run_done', migration_id: plan.migrationId, created_at: now(), + }); + return { + runId, status: 'completed', chunksTotal: chunks.length, + chunksCommitted: committed.size, chunksCompensated: compensated.size, planHash, + }; +} + +interface UnwindArgs { + runId: string; + planHash: string; + chunks: readonly MigrationChunk[]; + rowsOf: (c: MigrationChunk) => unknown[]; + next: () => number; + now: () => string; + committed: Set; + compensated: Set; + chunksTotal: number; + cause: unknown; +} + +/** + * LIFO compensation over committed chunks. + * + * Newest-first because later chunks may depend on earlier ones; undoing in + * commit order can hit a state the compensator was never written for. + * + * A compensation failure HALTS and is journalled — never swallowed, never + * "best effort, carry on". Continuing past it would produce a database whose + * state no journal describes, which is the one outcome this whole file exists + * to prevent. The run ends `failed`, and the rows say exactly which chunk + * resisted. + */ +async function unwind( + engine: IObjectQLEngine, + plan: MigrationPlan, + a: UnwindArgs, +): Promise { + const order = [...a.committed].sort((x, y) => y - x); // newest-first + for (const index of order) { + if (a.compensated.has(index)) continue; + const chunk = a.chunks[index]; + const step = plan.steps[chunk.stepIndex]; + + if (!step.compensate) { + // Nothing to undo this with. Say so loudly and stop — a silent skip + // would leave the row written and the journal claiming a clean unwind. + await appendEvent(engine, { + run_id: a.runId, seq: a.next(), kind: 'run_failed', + chunk_index: index, migration_id: plan.migrationId, created_at: a.now(), + detail: JSON.stringify({ + phase: 'compensate', reason: 'step declares no compensate()', + step: step.name, cause: errText(a.cause), + }), + }); + return { + runId: a.runId, status: 'failed', chunksTotal: a.chunksTotal, + chunksCommitted: a.committed.size, chunksCompensated: a.compensated.size, + planHash: a.planHash, error: a.cause, + }; + } + + const attempt = 1; + try { + await engine.transaction(async (trxCtx: unknown) => { + await step.compensate!(a.rowsOf(chunk), { runId: a.runId, chunkIndex: index, attempt, context: trxCtx }, engine); + await appendEvent( + engine, + { + run_id: a.runId, seq: a.next(), kind: 'compensated', + chunk_index: index, attempt, migration_id: plan.migrationId, created_at: a.now(), + }, + trxCtx, + ); + }, { ...SYSTEM_CTX }); + a.compensated.add(index); + } catch (err) { + await appendEvent(engine, { + run_id: a.runId, seq: a.next(), kind: 'run_failed', + chunk_index: index, migration_id: plan.migrationId, created_at: a.now(), + detail: JSON.stringify({ + phase: 'compensate', step: step.name, + error: errText(err), cause: errText(a.cause), + }), + }); + return { + runId: a.runId, status: 'failed', chunksTotal: a.chunksTotal, + chunksCommitted: a.committed.size, chunksCompensated: a.compensated.size, + planHash: a.planHash, error: err, + }; + } + } + + await appendEvent(engine, { + run_id: a.runId, seq: a.next(), kind: 'run_failed', + migration_id: plan.migrationId, created_at: a.now(), + detail: JSON.stringify({ phase: 'forward', error: errText(a.cause), compensated: [...a.compensated].sort((x, y) => x - y) }), + }); + return { + runId: a.runId, status: 'compensated', chunksTotal: a.chunksTotal, + chunksCommitted: a.committed.size, chunksCompensated: a.compensated.size, + planHash: a.planHash, error: a.cause, + }; +} + +/** Resume a run the journal says was interrupted. Thin alias for intent at call sites. */ +export async function resumeMigrationJournal( + engine: IObjectQLEngine, + plan: MigrationPlan, + runId: string, + options: Omit = {}, +): Promise { + return runMigrationJournal(engine, plan, { ...options, runId }); +} + +function errText(err: unknown): string { + if (err instanceof Error) return err.message; + try { + return String(err); + } catch { + return ''; + } +} diff --git a/packages/metadata-protocol/src/host-engine.ts b/packages/metadata-protocol/src/host-engine.ts index 5952e660e0..06f4ecae8f 100644 --- a/packages/metadata-protocol/src/host-engine.ts +++ b/packages/metadata-protocol/src/host-engine.ts @@ -21,7 +21,7 @@ export interface MetadataHostEngine extends IDataEngine { dropObjectSchema(...args: any[]): Promise; /** * ObjectQL's ambient transaction (ADR-0034), typed off the `objectql` slot - * contract (ADR-0118 D1) so this narrow host surface cannot drift from the + * contract (ADR-0119 D1) so this narrow host surface cannot drift from the * real signature. Declared explicitly because an explicit member beats the * index signature below — structurally `[key: string]: any` would type it * `any` and hide exactly the mistakes this contract exists to catch. @@ -29,7 +29,7 @@ export interface MetadataHostEngine extends IDataEngine { * Optional HERE, unlike on `IObjectQLEngine` where it is required: a host may * be a test double or a metadata-only store. Callers keep their runtime * probes and must say what degrading means at their seam — a caller that - * cannot lose atomicity silently fails closed (see `batchData`, ADR-0118 D4). + * cannot lose atomicity silently fails closed (see `batchData`, ADR-0119 D4). */ transaction?: IObjectQLEngine['transaction']; // Protocol accesses additional engine members structurally; keep it permissive diff --git a/packages/metadata-protocol/src/protocol.batch-atomic.test.ts b/packages/metadata-protocol/src/protocol.batch-atomic.test.ts index aa217fb8f4..011509fcbe 100644 --- a/packages/metadata-protocol/src/protocol.batch-atomic.test.ts +++ b/packages/metadata-protocol/src/protocol.batch-atomic.test.ts @@ -1,6 +1,6 @@ // Copyright (c) 2026 ObjectStack. Licensed under the Apache-2.0 license. // -// ADR-0118 D4 (#4612) — `batchData`'s `atomic` flag is REAL or REFUSED. +// ADR-0119 D4 (#4612) — `batchData`'s `atomic` flag is REAL or REFUSED. // // It used to be neither. The flag advertised "rollback entire batch on any // failure (transaction mode)" and opened no transaction at all: it `break`-ed @@ -71,7 +71,7 @@ function makeTransactionalEngine(opts: { driverCanTransact?: boolean } = {}) { return { engine, insert, update, findOne, del, commits, rollbacks, handle }; } -describe('batchData atomic — rollback is real and the response admits it (ADR-0118 D4)', () => { +describe('batchData atomic — rollback is real and the response admits it (ADR-0119 D4)', () => { it('rolls back the whole batch on the first failure and reports ZERO successes', async () => { const t = makeTransactionalEngine(); const p = new ObjectStackProtocolImplementation(t.engine); @@ -162,7 +162,7 @@ describe('batchData atomic — rollback is real and the response admits it (ADR- }); }); -describe('batchData atomic — refuses rather than degrading (ADR-0118 D4)', () => { +describe('batchData atomic — refuses rather than degrading (ADR-0119 D4)', () => { it('refuses with 501 when the engine has no transaction(), attempting NO writes', async () => { const t = makeTransactionalEngine(); delete t.engine.transaction; @@ -180,7 +180,7 @@ describe('batchData atomic — refuses rather than degrading (ADR-0118 D4)', () it('refuses when the default driver cannot begin a transaction, even though the engine exposes transaction()', async () => { // The subtle case: `engine.transaction()` silently runs the callback // with NO transaction when the driver lacks `beginTransaction` - // (ADR-0118 D1's declared caveat). Probing only the engine would let + // (ADR-0119 D1's declared caveat). Probing only the engine would let // "atomic" go back to meaning best-effort precisely here. const t = makeTransactionalEngine({ driverCanTransact: false }); const p = new ObjectStackProtocolImplementation(t.engine); @@ -195,7 +195,7 @@ describe('batchData atomic — refuses rather than degrading (ADR-0118 D4)', () }); }); -describe('batchData atomic — precedence and opt-in (ADR-0118 D4)', () => { +describe('batchData atomic — precedence and opt-in (ADR-0119 D4)', () => { it('atomic outranks continueOnError: the batch aborts and rolls back instead of continuing', async () => { const t = makeTransactionalEngine(); const p = new ObjectStackProtocolImplementation(t.engine); @@ -236,7 +236,7 @@ describe('batchData atomic — precedence and opt-in (ADR-0118 D4)', () => { }); }); -describe('batchData non-atomic — unchanged (ADR-0118 D4 regression net)', () => { +describe('batchData non-atomic — unchanged (ADR-0119 D4 regression net)', () => { it('opens no transaction and keeps prior successes when atomic is absent', async () => { const t = makeTransactionalEngine(); const p = new ObjectStackProtocolImplementation(t.engine); diff --git a/packages/metadata-protocol/src/protocol.ts b/packages/metadata-protocol/src/protocol.ts index a3db2303f1..ef69f1b65a 100644 --- a/packages/metadata-protocol/src/protocol.ts +++ b/packages/metadata-protocol/src/protocol.ts @@ -3,7 +3,7 @@ import type { DataProtocol, MetadataProtocol, PackageProtocol, } from '@objectstack/spec/api'; -import { IDataEngine } from '@objectstack/core'; +import { IDataEngine, engineCanRollBack } from '@objectstack/core'; import { readEnvWithDeprecation } from '@objectstack/types'; import type { MetadataHostEngine } from './host-engine.js'; import { SysMetadataRepository, type SysMetadataEngine } from './sys-metadata-repository.js'; @@ -815,11 +815,11 @@ function mergeDroppedFieldEvents(events: DroppedFieldsEvent[]): DroppedFieldsEve * has always emitted (`error: string`, `record`), which diverges from * `BatchOperationResultSchema`'s `errors: ApiError[]` / `data` — reconciling * the two is a wire-visible change that must not ride along on a bug fix - * (ADR-0118 D4; tracked separately). + * (ADR-0119 D4; tracked separately). */ type BatchDataRowResult = { id?: string; success: boolean; error?: string; record?: any; droppedFields?: DroppedFieldsEvent[] }; -/** What one pass of the `batchData` record loop produced (ADR-0118 D4). */ +/** What one pass of the `batchData` record loop produced (ADR-0119 D4). */ type BatchDataLoopOutcome = { results: BatchDataRowResult[]; succeeded: number; failed: number }; /** @@ -2298,7 +2298,7 @@ export class ObjectStackProtocolImplementation implements // honour a transaction, so `declared === enforced` (Prime Directive // #10). The rest-server producer ANDs this with `api.enableBatch` so // a server that doesn't mount the route reports `false` at its layer. - // (ADR-0118 D1: `transaction` is contract-declared, so this probe + // (ADR-0119 D1: `transaction` is contract-declared, so this probe // no longer needs a structural cast to ask the question.) transactionalBatch: typeof this.engine?.transaction === 'function', }; @@ -5233,7 +5233,7 @@ export class ObjectStackProtocolImplementation implements // strip context to `undefined`, treating every batch create as non-system). const batchSchema = this.engine.registry?.getObject(object); - // ADR-0118 D4 — `atomic` is REAL or REFUSED, never silent best-effort. + // ADR-0119 D4 — `atomic` is REAL or REFUSED, never silent best-effort. // This flag used to only `break` the loop: every write before the // failure stayed COMMITTED while the response called itself atomic and // reported those rows `success: true`. Same class as #4346 — a @@ -5255,7 +5255,7 @@ export class ObjectStackProtocolImplementation implements } /** - * The atomic arm of {@link batchData} (ADR-0118 D4): the whole batch runs + * The atomic arm of {@link batchData} (ADR-0119 D4): the whole batch runs * inside ONE `engine.transaction()`, so the first failure rolls back every * prior write — and the response says so, rather than reporting rows that * no longer exist as successes. @@ -5270,21 +5270,24 @@ export class ObjectStackProtocolImplementation implements }): Promise { const { object, operation, records, options, batchSchema, context } = args; - const engineTx = typeof this.engine?.transaction === 'function' - ? this.engine.transaction.bind(this.engine) - : undefined; - // Two-level probe. `engine.transaction()` runs the callback with NO - // transaction and NO rollback when the default driver lacks - // `beginTransaction` — a declared caveat of the contract member - // (ADR-0118 D1), and one that would turn "atomic" back into a lie + // Two-level probe, shared with the ADR-0119 D2 migration-journal runner + // as `engineCanRollBack` (#4617). `engine.transaction()` runs the + // callback with NO transaction and NO rollback when the default driver + // lacks `beginTransaction` — a declared caveat of the contract member + // (ADR-0119 D1), and one that would turn "atomic" back into a lie // precisely where it matters. So where the driver registry is // inspectable, the driver is checked too; where it is not (test // doubles), the engine-level probe is all there is. - const defaultDriverName = this.engine.getDefaultDriverName?.(); - const defaultDriver = defaultDriverName ? this.engine.getDriverByName?.(defaultDriverName) : undefined; - const driverCanTransact = !defaultDriver || typeof (defaultDriver as any).beginTransaction === 'function'; + // + // Shared rather than restated: this and the runner's gate were the same + // condition written twice, and two copies of "can this runtime actually + // roll back?" drift by one clause and leave one caller believing it has + // atomicity it does not have. + const engineTx = engineCanRollBack(this.engine) + ? this.engine.transaction.bind(this.engine) + : undefined; - if (!engineTx || !driverCanTransact) { + if (!engineTx) { // REFUSE, do not degrade. A caller that asked for atomicity is // exactly the caller who must not silently receive best-effort — // silent degradation is how this flag came to lie in the first @@ -5322,7 +5325,7 @@ export class ObjectStackProtocolImplementation implements } /** - * The per-record loop, shared by both arms of {@link batchData} (ADR-0118 + * The per-record loop, shared by both arms of {@link batchData} (ADR-0119 * D4) so atomic and non-atomic cannot drift apart. `atomic` changes exactly * two things: it aborts on the first failure regardless of * `continueOnError` (whose own contract text already scopes it to @@ -5387,7 +5390,7 @@ export class ObjectStackProtocolImplementation implements results.push({ id: created.id, success: true, record: created }); } } catch (err) { - // ADR-0118 D4 — no blind fallback inside a + // ADR-0119 D4 — no blind fallback inside a // transaction: once the failing statement has // aborted it, this insert can only fail with a // secondary error ("current transaction is @@ -5454,7 +5457,7 @@ export class ObjectStackProtocolImplementation implements } /** - * The response for an atomic batch that rolled back (ADR-0118 D4). + * The response for an atomic batch that rolled back (ADR-0119 D4). * * Nothing persisted, so nothing may report success — the old code's real * damage was not the missing transaction alone but telling the caller that @@ -7720,7 +7723,7 @@ export class ObjectStackProtocolImplementation implements const promoted: PromotedDraft[] = []; // (assigned inside the transaction closure — keep the wide type) let commit = null as { commitId: string } | null; - // ADR-0118 D1 — `transaction` is contract-declared, so this reaches it + // ADR-0119 D1 — `transaction` is contract-declared, so this reaches it // by name instead of through a structural cast. Bound once up front: // the probe and the call must agree on one resolved function. const engineTx = typeof this.engine?.transaction === 'function' diff --git a/packages/metadata-protocol/src/sys-metadata-repository.ts b/packages/metadata-protocol/src/sys-metadata-repository.ts index 20bcbacddf..d5b8c4201f 100644 --- a/packages/metadata-protocol/src/sys-metadata-repository.ts +++ b/packages/metadata-protocol/src/sys-metadata-repository.ts @@ -128,7 +128,7 @@ export interface SysMetadataEngine { * `ObjectQL.transaction` semantics). Repository code must not rely on * rollback for correctness against in-memory drivers. * - * Typed off the `objectql` slot contract (ADR-0118 D1) rather than restated + * Typed off the `objectql` slot contract (ADR-0119 D1) rather than restated * by hand, so this stub surface cannot drift from `ObjectQL.transaction`. */ transaction?: IObjectQLEngine['transaction']; diff --git a/packages/objectql/src/protocol-batch-atomic.test.ts b/packages/objectql/src/protocol-batch-atomic.test.ts index d1cfcf6a49..5afe986ae9 100644 --- a/packages/objectql/src/protocol-batch-atomic.test.ts +++ b/packages/objectql/src/protocol-batch-atomic.test.ts @@ -1,6 +1,6 @@ // Copyright (c) 2026 ObjectStack. Licensed under the Apache-2.0 license. // -// ADR-0118 (#4612), end-to-end: the real `ObjectQL` engine + the real metadata +// ADR-0119 (#4612), end-to-end: the real `ObjectQL` engine + the real metadata // protocol + a driver whose transactions actually roll back. The unit pins in // `metadata-protocol/src/protocol.batch-atomic.test.ts` prove `batchData` asks // for the right things; these prove the stack delivers them — that the rows are @@ -102,7 +102,7 @@ function makeSnapshotDriver() { return { driver, seen, rowsOf: (o: string) => Array.from(storeFor(o).values()) }; } -describe('atomic batchData over the real engine (ADR-0118 D4 / ADR-0034)', () => { +describe('atomic batchData over the real engine (ADR-0119 D4 / ADR-0034)', () => { let engine: ObjectQL; let protocol: ObjectStackProtocolImplementation; let d: ReturnType; @@ -216,7 +216,7 @@ describe('atomic batchData over the real engine (ADR-0118 D4 / ADR-0034)', () => }); }); -describe('ADR-0118 D1 — transaction is reachable through the contract', () => { +describe('ADR-0119 D1 — transaction is reachable through the contract', () => { it('calls transaction() on an engine typed as IObjectQLEngine, with no cast', async () => { const engine = new ObjectQL(); const d = makeSnapshotDriver(); @@ -224,7 +224,7 @@ describe('ADR-0118 D1 — transaction is reachable through the contract', () => await engine.init(); engine.registry.registerObject({ name: 'invoice', fields: { title: { type: 'text' } } } as any); - // The point of the pin is the TYPE, not the runtime: before ADR-0118 D1 + // The point of the pin is the TYPE, not the runtime: before ADR-0119 D1 // this line could not compile — `transaction` was absent from the // contract, so every cross-package consumer reached it through // `as unknown as { transaction: ... }`. diff --git a/packages/platform-objects/src/plugin.test.ts b/packages/platform-objects/src/plugin.test.ts index 0fb409f0da..932f044bea 100644 --- a/packages/platform-objects/src/plugin.test.ts +++ b/packages/platform-objects/src/plugin.test.ts @@ -2,7 +2,7 @@ import { describe, it, expect } from 'vitest'; import { PlatformObjectsPlugin } from './plugin.js'; -import { SysMigration, SysSecret } from './system/index.js'; +import { SysMigration, SysMigrationJournal, SysSecret } from './system/index.js'; /** * Hand-rolled fake PluginContext (mirrors the service-plugin tests) — this @@ -34,7 +34,7 @@ function makeCtx() { } describe('PlatformObjectsPlugin: platform-infrastructure object registration (#4243, #4270)', () => { - it('registers SysMigration and SysSecret through the manifest service', async () => { + it('registers SysMigration, SysMigrationJournal and SysSecret through the manifest service', async () => { const ctx = makeCtx(); const manifests: any[] = []; ctx.registerService('manifest', { register: (m: any) => manifests.push(m) }); @@ -44,8 +44,16 @@ describe('PlatformObjectsPlugin: platform-infrastructure object registration (#4 expect(manifests).toHaveLength(1); expect(manifests[0].id).toBe('com.objectstack.platform-objects'); expect(manifests[0].scope).toBe('system'); - expect(manifests[0].objects).toEqual([SysMigration, SysSecret]); - expect(manifests[0].objects.map((o: any) => o.name)).toEqual(['sys_migration', 'sys_secret']); + expect(manifests[0].objects).toEqual([SysMigration, SysMigrationJournal, SysSecret]); + expect(manifests[0].objects.map((o: any) => o.name)).toEqual([ + 'sys_migration', + // ADR-0119 D2 (#4617). Registered UNCONDITIONALLY, alongside the flag + // ledger rather than by the migration CLI that writes it: recovery has + // to be discoverable with zero host wiring, and a journal some kernels + // compose and others do not is a journal a boot scanner cannot rely on. + 'sys_migration_journal', + 'sys_secret', + ]); }); /** diff --git a/packages/platform-objects/src/plugin.ts b/packages/platform-objects/src/plugin.ts index f4d595d4aa..7b671aabcc 100644 --- a/packages/platform-objects/src/plugin.ts +++ b/packages/platform-objects/src/plugin.ts @@ -3,6 +3,7 @@ import { SetupAppTranslations } from './apps/translations/index.js'; import { MetadataFormsTranslations } from './metadata-translations/index.js'; import { SysMigration } from './system/sys-migration.object.js'; +import { SysMigrationJournal } from './system/sys-migration-journal.object.js'; import { SysSecret } from './system/sys-secret.object.js'; import { attestFreshDatastore } from './system/migration-flag.js'; import type { II18nService, IObjectQLEngine } from '@objectstack/spec/contracts'; @@ -21,6 +22,14 @@ import type { II18nService, IObjectQLEngine } from '@objectstack/spec/contracts' * file collection, the engine's strict value-shape gates (#3438/#4235) * and the migration CLI all read the same ledger, and none of them * should have to drag an unrelated service into the boot to find it. + * - **`sys_migration_journal`** — the per-RUN crash-recovery trace + * (ADR-0119 D2, #4617). Distinct from the flag ledger above in grain + * and purpose: that one records the durable verdict for a named + * migration, this one records what happened inside a single run, so a + * restarted process can tell whether chunk 7 committed. Registered + * unconditionally alongside it because recovery must be discoverable + * with ZERO host wiring — a journal composed by some kernels and not + * others is ADR-0078's silently-inert failure in another costume. * - **`sys_secret`** — the environment's encrypted-secret store * (ADR-0066 D2/④). Same shape as the ledger, sharper stakes (#4270): * its producers span domains — the settings service's encrypted @@ -73,7 +82,7 @@ export class PlatformObjectsPlugin { version: this.version, type: 'plugin', scope: 'system', - objects: [SysMigration, SysSecret], + objects: [SysMigration, SysMigrationJournal, SysSecret], }); } catch { // No manifest service (lean / i18n-only kernels) — the ledger stays diff --git a/packages/platform-objects/src/system/index.ts b/packages/platform-objects/src/system/index.ts index b55bce4ceb..fa69351d89 100644 --- a/packages/platform-objects/src/system/index.ts +++ b/packages/platform-objects/src/system/index.ts @@ -13,6 +13,7 @@ export { SysSetting } from './sys-setting.object.js'; export { SysSecret } from './sys-secret.object.js'; export { SysSettingAudit } from './sys-setting-audit.object.js'; export { SysMigration } from './sys-migration.object.js'; +export { SysMigrationJournal } from './sys-migration-journal.object.js'; export { readDataMigrationFlag, isDataMigrationVerified, diff --git a/packages/platform-objects/src/system/sys-migration-journal.object.ts b/packages/platform-objects/src/system/sys-migration-journal.object.ts new file mode 100644 index 0000000000..6a46f5b97b --- /dev/null +++ b/packages/platform-objects/src/system/sys-migration-journal.object.ts @@ -0,0 +1,188 @@ +// Copyright (c) 2026 ObjectStack. Licensed under the Apache-2.0 license. + +import { ObjectSchema, Field } from '@objectstack/spec/data'; +import { MIGRATION_JOURNAL_KINDS } from '@objectstack/spec/system'; + +/** + * sys_migration_journal — the durable trace of a migration RUN (ADR-0119 D2, #4617) + * + * Rows keyed `(run_id, seq)`. Many rows per run; one run per invocation of + * `runMigrationJournal` (`@objectstack/core`). + * + * ## Why this exists next to `sys_migration` + * + * They answer different questions and must not be conflated. `sys_migration` + * holds ONE row per named migration — the durable verdict ("did it complete + * here, was it shown correct?") that consumers gate on. This object holds MANY + * rows per run — the trace of what actually happened while it ran. + * + * A flag row is written once, at the end, by a run that finished. This journal + * exists precisely because a run may NOT finish. + * + * ## Why a table and not a side-file + * + * ADR-0034 gave the platform real transactions, and ADR-0119 D1 made them + * reachable through the contract — but a transaction cannot be the whole + * answer for migration-class work. A million-row backfill cannot hold one + * write-lock for its duration; `driver-memory`'s `beginTransaction` deep-clones + * the entire database, so "just wrap it" is O(db) per begin; and a process + * KILLED (as distinct from a thrown error) defeats in-process rollback + * entirely. So a big migration commits in pieces, and something durable has to + * record which pieces committed. That is this table. + * + * Keeping it in the database rather than a JSONL side-file (the ADR-0008 + * shape) is a decision, not an implementation detail: the journal is + * data-plane state ABOUT the data plane, so it belongs inside the same + * backup/restore and transaction boundary as the rows it describes. A + * side-file desyncs from the database on restore — and a journal that + * disagrees with the data is worse than none, because it will be believed. + * ADR-0008 solved audit; recovery authority is a different problem. + * + * ## The one invariant a reader must not "clean up" + * + * `chunk_done(i)` is written INSIDE the chunk's own transaction, so + * `done ⇔ committed` holds by construction. `chunk_started(i)` is written + * autonomously BEFORE it. Making both writes autonomous (or both + * transactional) would look tidier and would destroy recovery: the asymmetry + * is what gives `started ∧ ¬done` its single precise meaning — *the outcome is + * unknown* — which is the only state a crash can leave and the only state + * recovery reasons about. + * + * Append-only by contract: the runner never updates or deletes a row, so the + * trace of a failed run survives its compensation. Writes flow through the + * runner in system context; the API surface is read-only diagnostics. + * + * Registered by `PlatformObjectsPlugin` (`./plugin.ts`) alongside + * `sys_migration` — recovery must be discoverable with zero host wiring, so + * the journal cannot be optional infrastructure that some kernels compose and + * others do not (ADR-0078: a journal nobody re-reads is a journal that does + * not exist). The row contract lives in `@objectstack/spec/system` + * (`MigrationJournalEventSchema`) so `@objectstack/core`'s runner can write it + * without depending on this package. + * + * @namespace sys + */ +export const SysMigrationJournal = ObjectSchema.create({ + name: 'sys_migration_journal', + label: 'Migration Journal Entry', + pluralLabel: 'Migration Journal', + icon: 'clipboard-list', + isSystem: true, + managedBy: 'engine-owned', + description: + 'Append-only trace of migration runs: which chunks committed, which were compensated, and where a killed run stopped.', + nameField: 'run_id', // [ADR-0079] canonical primary-title pointer + titleFormat: '{run_id} #{seq} {kind}', + highlightFields: ['run_id', 'seq', 'kind', 'chunk_index', 'created_at'], + listViews: { + recent: { + type: 'grid', + name: 'recent', + label: 'Recent', + columns: ['created_at', 'run_id', 'seq', 'kind', 'chunk_index', 'migration_id'], + sort: [ + { field: 'created_at', order: 'desc' }, + { field: 'seq', order: 'desc' }, + ], + }, + }, + + fields: { + id: Field.text({ + label: 'ID', + readonly: true, + }), + + run_id: Field.text({ + label: 'Run ID', + required: true, + readonly: true, + maxLength: 64, + description: 'Identifies one run of one plan. Rows are keyed (run_id, seq).', + }), + + seq: Field.number({ + label: 'Sequence', + required: true, + readonly: true, + description: + 'Monotonic per-run sequence, from 0. The ORDERING AUTHORITY — created_at can tie at ' + + 'coarse clock resolution and can skew, so recovery never orders by it.', + }), + + kind: Field.select( + MIGRATION_JOURNAL_KINDS.map((value) => ({ label: value, value })), + { + label: 'Kind', + required: true, + readonly: true, + description: 'Event kind. See MIGRATION_JOURNAL_KINDS in @objectstack/spec/system.', + }, + ), + + migration_id: Field.text({ + label: 'Migration ID', + readonly: true, + maxLength: 128, + description: 'The named migration this run belongs to, when it has one — joins to sys_migration.id.', + }), + + plan_hash: Field.text({ + label: 'Plan Hash', + readonly: true, + maxLength: 128, + description: + 'On run_started: hash of the plan shape. A resume whose plan hash differs REFUSES — resuming a ' + + 'changed plan against an old journal would apply chunk boundaries the journal never described.', + }), + + chunk_index: Field.number({ + label: 'Chunk Index', + readonly: true, + description: 'On chunk_started / chunk_done / compensated: the run-global chunk index.', + }), + + attempt: Field.number({ + label: 'Attempt', + readonly: true, + description: + 'Which attempt produced this event. attempt > 1 means a prior outcome was unknown and the ' + + 'callback was asked to recheck by natural key before re-writing.', + }), + + detail: Field.text({ + label: 'Detail (JSON)', + readonly: true, + description: 'JSON payload — the chunk plan on run_started, the error on run_failed or a failed compensation.', + }), + + created_at: Field.datetime({ + label: 'Created At', + readonly: true, + description: 'Wall-clock stamp, for humans reading the trace. Never the ordering authority — that is seq.', + }), + }, + + indexes: [ + // Unique, not merely fast: it makes a duplicate (run_id, seq) a write + // ERROR rather than a silently double-recorded event. A resumed run that + // miscomputed its next sequence must fail loudly — a journal that quietly + // accepts two "chunk 7 done" rows cannot be trusted to say what committed. + { fields: ['run_id', 'seq'], unique: true }, + // The recovery scan's access path: all events for a run, and the + // open-run sweep that looks for run_started without run_done. + { fields: ['run_id', 'kind'], unique: false }, + { fields: ['migration_id'], unique: false }, + ], + + enable: { + trackHistory: false, + searchable: false, + apiEnabled: true, + // Diagnostic read surface only. Writes go through the runner in system + // context — a journal a client could edit could not prove anything about + // what committed, which is the single thing it exists to do. + apiMethods: ['get', 'list'], + clone: false, + }, +}); diff --git a/packages/spec/api-surface.json b/packages/spec/api-surface.json index c92efda61b..0624d5a76f 100644 --- a/packages/spec/api-surface.json +++ b/packages/spec/api-surface.json @@ -1028,6 +1028,8 @@ "LoggingConfig (type)", "LoggingConfigSchema (const)", "METADATA_FORM_REGISTRY (const)", + "MIGRATION_JOURNAL_KINDS (const)", + "MIGRATION_JOURNAL_OBJECT (const)", "MaskingVisibilityRule (type)", "MaskingVisibilityRuleSchema (const)", "MessageQueueConfig (type)", @@ -1094,6 +1096,10 @@ "MiddlewareConfigSchema (const)", "MiddlewareType (type)", "MigrationDependencySchema (const)", + "MigrationJournalEvent (type)", + "MigrationJournalEventSchema (const)", + "MigrationJournalKind (type)", + "MigrationOnCrashPolicy (type)", "MigrationOperation (type)", "MigrationOperationSchema (const)", "MigrationPlan (type)", diff --git a/packages/spec/authorable-surface.json b/packages/spec/authorable-surface.json index c9857043d7..dbfec62d77 100644 --- a/packages/spec/authorable-surface.json +++ b/packages/spec/authorable-surface.json @@ -1,5 +1,5 @@ { - "description": "Ratchet of every AUTHORABLE key in the spec \u2014 what a metadata author may write, which for this platform IS the third-party API. Auto-updated on additions (commit the change). A key that disappears without a tombstone fails gen:schema, because these schemas are not .strict() and Zod would silently strip it. \"[RETIRED]\" marks a tombstoned key that still rejects with an upgrade prescription. See #3855, ADR-0059 \u00a75.", + "description": "Ratchet of every AUTHORABLE key in the spec — what a metadata author may write, which for this platform IS the third-party API. Auto-updated on additions (commit the change). A key that disappears without a tombstone fails gen:schema, because these schemas are not .strict() and Zod would silently strip it. \"[RETIRED]\" marks a tombstoned key that still rejects with an upgrade prescription. See #3855, ADR-0059 §5.", "keys": [ "ai/AIModelConfig:maxTokens", "ai/AIModelConfig:model", @@ -6442,6 +6442,15 @@ "system/MiddlewareConfig:type", "system/MigrationDependency:migrationId", "system/MigrationDependency:package", + "system/MigrationJournalEvent:attempt", + "system/MigrationJournalEvent:chunk_index", + "system/MigrationJournalEvent:created_at", + "system/MigrationJournalEvent:detail", + "system/MigrationJournalEvent:kind", + "system/MigrationJournalEvent:migration_id", + "system/MigrationJournalEvent:plan_hash", + "system/MigrationJournalEvent:run_id", + "system/MigrationJournalEvent:seq", "system/MigrationPlan:dialect", "system/MigrationPlan:estimatedDurationMs", "system/MigrationPlan:reversible", diff --git a/packages/spec/json-schema.manifest.json b/packages/spec/json-schema.manifest.json index 9163533b16..dd0e974144 100644 --- a/packages/spec/json-schema.manifest.json +++ b/packages/spec/json-schema.manifest.json @@ -1366,6 +1366,7 @@ "system/MiddlewareConfig", "system/MiddlewareType", "system/MigrationDependency", + "system/MigrationJournalEvent", "system/MigrationOperation", "system/MigrationPlan", "system/MigrationStatement", diff --git a/packages/spec/src/api/batch.test.ts b/packages/spec/src/api/batch.test.ts index 6de9b107ce..83e6c4fb1f 100644 --- a/packages/spec/src/api/batch.test.ts +++ b/packages/spec/src/api/batch.test.ts @@ -54,7 +54,7 @@ describe('BatchOptionsSchema', () => { it('should use default values', () => { const options = BatchOptionsSchema.parse({}); - // ADR-0118 D4 — `atomic` defaults to FALSE. It declared `true` for as long + // ADR-0119 D4 — `atomic` defaults to FALSE. It declared `true` for as long // as no batch surface honoured it; the declaration was aligned down to the // enforced behaviour so that opting in is explicit and nobody's failure // semantics changed silently. diff --git a/packages/spec/src/api/batch.zod.ts b/packages/spec/src/api/batch.zod.ts index ddc16534a7..27b6e5bf8c 100644 --- a/packages/spec/src/api/batch.zod.ts +++ b/packages/spec/src/api/batch.zod.ts @@ -59,7 +59,7 @@ export type BatchRecord = z.infer; * Configuration options for batch operations */ export const BatchOptionsSchema = lazySchema(() => z.object({ - // ADR-0118 D4. `atomic` declared `.default(true)` while NO enforcement site + // ADR-0119 D4. `atomic` declared `.default(true)` while NO enforcement site // delivered atomicity: `batchData` merely broke its loop, leaving every prior // write committed, and the REST route deliberately forwards the ORIGINAL body // rather than the parsed output, so this default never reached the loop at diff --git a/packages/spec/src/contracts/objectql-engine.ts b/packages/spec/src/contracts/objectql-engine.ts index ec04cbe44e..7b9f5e54ba 100644 --- a/packages/spec/src/contracts/objectql-engine.ts +++ b/packages/spec/src/contracts/objectql-engine.ts @@ -164,7 +164,7 @@ export interface IObjectQLEngine extends IDataEngine { /** Drop the memoized migration-flag reads (the attestation may race a fast boot's first read). */ invalidateDataMigrationFlags(): void; - // ── Transactions (ADR-0118 D1) ─────────────────────────────────────── + // ── Transactions (ADR-0119 D1) ─────────────────────────────────────── /** * Run `callback` inside ONE driver transaction — the ADR-0034 ambient * transaction. The callback receives a context carrying the handle, which @@ -184,13 +184,13 @@ export interface IObjectQLEngine extends IDataEngine { * header; callers that tolerate test doubles keep their runtime * `typeof === 'function'` probes, which types do not replace. * - * TWO CAVEATS ARE PART OF THE DECLARED MEANING (ADR-0118 D1), not + * TWO CAVEATS ARE PART OF THE DECLARED MEANING (ADR-0119 D1), not * behaviour to be discovered: this covers the DEFAULT driver only — objects * routed elsewhere by `setDatasourceMapping` are written outside it — and * when that driver has no `beginTransaction` the callback runs with NO * transaction and NO rollback. A caller that cannot tolerate silently * losing atomicity must fail closed itself rather than assume it held; see - * `batchData`'s atomic gate (ADR-0118 D4). Tightening both is tracked by + * `batchData`'s atomic gate (ADR-0119 D4). Tightening both is tracked by * the ADR's follow-up. * * `trxCtx`/`baseContext` are the engine-local execution-context shape, left diff --git a/packages/spec/src/system/constants/platform-object-names.ts b/packages/spec/src/system/constants/platform-object-names.ts index c08b3d6519..827504738c 100644 --- a/packages/spec/src/system/constants/platform-object-names.ts +++ b/packages/spec/src/system/constants/platform-object-names.ts @@ -61,6 +61,7 @@ export const PLATFORM_OBJECTS_BY_PACKAGE: Readonly z.object({ + run_id: z.string().describe('Identifies one run. Rows are keyed (run_id, seq)'), + seq: z.number().int().min(0) + .describe('Monotonic per-run sequence. Ordering authority — wall-clock timestamps can tie or skew'), + kind: z.enum(MIGRATION_JOURNAL_KINDS).describe('Event kind'), + migration_id: z.string().optional() + .describe('The named migration this run belongs to, when it has one — joins to sys_migration.id'), + plan_hash: z.string().optional() + .describe('On run_started: hash of the plan shape. A resume whose plan hash differs REFUSES rather than resuming a changed plan against an old journal'), + chunk_index: z.number().int().min(0).optional() + .describe('On chunk_started / chunk_done / compensated: the run-global chunk index'), + attempt: z.number().int().min(1).optional() + .describe('Which attempt produced this event. attempt > 1 means a prior outcome was unknown and the callback was asked to recheck by natural key'), + detail: z.string().optional() + .describe('JSON-encoded payload — the chunk plan on run_started, the error on run_failed / a failed compensation'), + created_at: z.string().datetime().optional().describe('Wall-clock stamp, for humans. Never the ordering authority — that is seq'), +}).describe('One event in a migration run journal — the durable trace that lets a killed run be resumed forward or compensated back, with rows proving which')); +export type MigrationJournalEvent = z.infer; + +/** + * What a crashed run should do when it is rediscovered. + * + * `resume` suits an idempotent forward migration (a backfill that can recheck + * by natural key and continue); `compensate` suits work whose partial state is + * worse than no state. Per-plan because only the plan's author knows which of + * those two its steps are — there is no safe global default, and guessing + * would either strand data or undo work that was fine. + */ +export type MigrationOnCrashPolicy = 'resume' | 'compensate'; diff --git a/scripts/adr-anchors.json b/scripts/adr-anchors.json index 4f7a6aaf09..2ce6d6b7d0 100644 --- a/scripts/adr-anchors.json +++ b/scripts/adr-anchors.json @@ -78,12 +78,12 @@ }, { "file": "packages/spec/src/contracts/objectql-engine.ts", - "adrs": ["ADR-0118"], + "adrs": ["ADR-0119"], "invariant": "`transaction` is DECLARED on the `objectql` slot contract — plugin space reaches ADR-0034's ambient transaction by name, not through `as unknown as` casts. Required, not optional, per this file's own rule. Its two caveats (default-driver only; the callback runs with NO transaction when the driver lacks `beginTransaction`) are part of the declared meaning, so a caller that cannot lose atomicity silently must fail closed rather than assume it held." }, { "file": "packages/metadata-protocol/src/protocol.ts", - "adrs": ["ADR-0118"], + "adrs": ["ADR-0119"], "invariant": "`batchData`'s `atomic` is REAL or REFUSED, never silent best-effort. An explicitly atomic batch runs inside ONE `engine.transaction()`, so a failure rolls back every prior write and the response reports zero successes (rows marked ROLLED_BACK / NOT_ATTEMPTED — reporting a rolled-back row as `success: true` was the original bug, not merely the missing transaction). A runtime that cannot roll back gets 501 NOT_IMPLEMENTED; degrading to best-effort is how the flag came to lie." } ]