From a0fd2735a1f60a8b5cd508795ae28812d97054c3 Mon Sep 17 00:00:00 2001 From: Claude Date: Sun, 2 Aug 2026 15:36:38 +0000 Subject: [PATCH] =?UTF-8?q?feat(core,platform-objects,spec):=20ADR-0119=20?= =?UTF-8?q?D2=20=E2=80=94=20the=20migration-journal=20runner=20(#4617)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A migration killed mid-run is now either resumable to completion or compensable to clean, with journal rows proving which. ADR-0119 D1 made `engine.transaction()` contract-reachable, which answers multi-write atomicity that fits in one transaction. Migration-class work does not fit: a million-row backfill cannot hold one write-lock, driver-memory's `beginTransaction` deep-clones the whole database, `transaction()` binds the default driver only, and a KILLED process defeats in-process rollback entirely. So the unit of atomicity is the chunk, and durability across chunks is a journal. - `runMigrationJournal` (@objectstack/core): preflight dry-run across every step before any step writes; chunked writes each inside `engine.transaction()`; LIFO compensation newest-first on failure; re-entrant forward recovery under a per-plan `onCrash` policy; at-least-once with an `attempt` counter, reusing bulk-write.ts's delivery contract rather than re-deriving it. - `sys_migration_journal` (@objectstack/platform-objects): rows keyed (run_id, seq) under a unique index, registered unconditionally beside sys_migration so recovery is discoverable with zero host wiring (ADR-0078). Distinct in grain from sys_migration, which holds one verdict per named migration; this holds many rows per run. - Row contract + object-name constant in @objectstack/spec/system, so core's runner writes the journal without depending on platform-objects. The invariant carrying the design: `chunk_done(i)` is written INSIDE the chunk's transaction so `done ⇔ committed` holds by construction, while `chunk_started(i)` is written autonomously before it. That asymmetry gives `started ∧ ¬done` exactly one meaning — outcome unknown — which is the only state a crash leaves and the only state recovery reasons about. The runner refuses rather than degrades: no rollback capability, a failed preflight, an uncompensable plan declaring onCrash:'compensate', or a resume whose plan hash disagrees with the journal. A compensation failure halts and is journalled, and the run ends `failed` rather than `compensated` — 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 was the same condition in this runner and in batchData's atomic gate, and two copies drift by one clause and leave one caller believing it has atomicity it does not have. It moves to @objectstack/core as a type predicate; metadata-protocol imports it. Boot reconciliation and `os migrate resume` land separately; the discovery primitive they consume, `findInterruptedRuns`, is exported here. Docs: ADR-0118 (plugin-reachable transactions) is renumbered ADR-0119. It merged a day after an unrelated ADR-0118 (非用户 actor 的平台契约), and the earlier merge holds the number. Its Status line now cites the implementing PR and its tests instead of a dangling "this PR", and records that D2/D3 remain unimplemented. Refs: ADR-0119 D2, #4617, #4612, ADR-0034, ADR-0060, ADR-0078, ADR-0117 D8 Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01NKcGqCYCCpMkB5UW8jNPXx --- .changeset/adr-0119-d2-migration-journal.md | 78 ++ ...adr-0119-plugin-reachable-transactions.md} | 4 +- content/docs/references/system/migration.mdx | 25 +- ...e-transactions-and-honest-atomic-batch.md} | 5 +- packages/core/src/index.ts | 5 + .../core/src/utils/migration-journal.test.ts | 472 ++++++++++++ packages/core/src/utils/migration-journal.ts | 670 ++++++++++++++++++ packages/metadata-protocol/src/host-engine.ts | 4 +- .../src/protocol.batch-atomic.test.ts | 12 +- packages/metadata-protocol/src/protocol.ts | 45 +- .../src/sys-metadata-repository.ts | 2 +- .../src/protocol-batch-atomic.test.ts | 8 +- packages/platform-objects/src/plugin.test.ts | 16 +- packages/platform-objects/src/plugin.ts | 11 +- packages/platform-objects/src/system/index.ts | 1 + .../system/sys-migration-journal.object.ts | 188 +++++ packages/spec/api-surface.json | 6 + packages/spec/authorable-surface.json | 11 +- packages/spec/json-schema.manifest.json | 1 + packages/spec/src/api/batch.test.ts | 2 +- packages/spec/src/api/batch.zod.ts | 2 +- .../spec/src/contracts/objectql-engine.ts | 6 +- .../system/constants/platform-object-names.ts | 1 + packages/spec/src/system/migration.zod.ts | 81 +++ scripts/adr-anchors.json | 4 +- 25 files changed, 1607 insertions(+), 53 deletions(-) create mode 100644 .changeset/adr-0119-d2-migration-journal.md rename .changeset/{adr-0118-plugin-reachable-transactions.md => adr-0119-plugin-reachable-transactions.md} (96%) rename docs/adr/{0118-plugin-reachable-transactions-and-honest-atomic-batch.md => 0119-plugin-reachable-transactions-and-honest-atomic-batch.md} (95%) create mode 100644 packages/core/src/utils/migration-journal.test.ts create mode 100644 packages/core/src/utils/migration-journal.ts create mode 100644 packages/platform-objects/src/system/sys-migration-journal.object.ts 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." } ]