diff --git a/.changeset/21498-cli-compose-migration-recovery.md b/.changeset/21498-cli-compose-migration-recovery.md index e2a75e94493..00b4ba4d3f5 100644 --- a/.changeset/21498-cli-compose-migration-recovery.md +++ b/.changeset/21498-cli-compose-migration-recovery.md @@ -10,4 +10,4 @@ Clause-②: no - **The `os migrate` data commands** (`recorded-by`, `resume`, `value-shapes`, `summary-nulls`, `files-to-references`, `meta --stored`, `audit-metadata-bodies`, `os storage orphans`) now boot with the plugin. A run interrupted before any of its chunks committed now resumes to completion. A command booted over an interrupted run also warns about that run on stderr first. - **Every `os serve` boot** (and so `os start` and `os dev`, which spawn it) composes the plugin beside `PlatformObjectsPlugin`, which registers the journal the scan reads. An interrupted run is reported once at boot, with the `os migrate resume --run ` command that resumes it. Nothing is resumed automatically. A database with no interrupted run prints nothing. A config that composes its own `new MigrationRecoveryPlugin()` keeps that instance. -- **Still refused:** a `recorded-by` run that had committed a chunk before it was interrupted, or that was started with a non-default `--chunk-size`. `resume` now reaches the runner for these runs, and the runner refuses them with `PLAN_CHANGED`. Re-running `os migrate recorded-by --apply` converts whatever rows still hold the sentinel. +- **A run that had committed a chunk, or that was started with a non-default `--chunk-size`,** reaches the runner too. The runner fix that lets it resume is in the `@objectstack/core` entry for #21528. diff --git a/.changeset/21528-core-resume-started-over-plan.md b/.changeset/21528-core-resume-started-over-plan.md new file mode 100644 index 00000000000..0a237e4b5e7 --- /dev/null +++ b/.changeset/21528-core-resume-started-over-plan.md @@ -0,0 +1,15 @@ +--- +'@objectstack/core': patch +--- + +fix(core): a resumed migration run is compared against the chunk plan it started over, so `os migrate resume` completes an interrupted `recorded-by` run that had committed a chunk or was started with a non-default `--chunk-size` (#21528) + +Clause-②: no + +`runMigrationJournal` recomputed a resumed run's chunk plan from the rows `load()` returned at resume time, at the plan's current chunk size, and refused `PLAN_CHANGED` when that plan's hash differed from the one `run_started` recorded. Two kinds of interrupted run could differ. A plan whose `load()` selects only the work still to do, which `recorded-by`'s plan does, returns fewer rows once a chunk has committed. And the plan handed back for a resume carries its own chunk size, not the one the run was started with. So `os migrate resume` listed such a run as `resumable: true`, and `os migrate resume --run --yes` then refused it. + +A resume now reads the chunk plan back from the journal's `run_started` record: + +- **Identity.** The plan's id and step names are hashed with the recorded chunk boundaries and compared with the recorded hash. A plan whose id or steps changed is still refused `PLAN_CHANGED`. The run resumes at the chunk size it started with. +- **Rows.** Each step's rows are bound to that chunk plan. If `load()` returns every row the run started over, each chunk's rows are where the journal put them, as before. If it returns exactly the rows of the chunks not yet committed, those rows go, in order, to those chunks. Any other row count is refused `PLAN_CHANGED`, and the message names the step. +- **Unwind.** If a chunk fails after a resume that bound its rows the second way, the runner compensates the chunks this process committed, newest first. It then stops at the newest chunk an earlier process committed and journals `run_failed`, because `load()` no longer returns that chunk's rows. It does not compensate other rows in their place, and the run ends `failed`. diff --git a/packages/cli/src/commands/migrate/resume.recorded-by.integration.test.ts b/packages/cli/src/commands/migrate/resume.recorded-by.integration.test.ts index 3e384dca3d6..f107eb6f0bd 100644 --- a/packages/cli/src/commands/migrate/resume.recorded-by.integration.test.ts +++ b/packages/cli/src/commands/migrate/resume.recorded-by.integration.test.ts @@ -34,12 +34,22 @@ * composed into every served boot adds no line when there is nothing to * report. * - * ⛔ The interruption is in chunk 0, before any chunk committed. A run that - * had committed a chunk is refused `PLAN_CHANGED` on resume — `recorded-by`'s - * `load()` selects only rows still holding the sentinel, so the chunk plan it - * recomputes no longer hashes to the one the journal recorded. That is a - * defect of the plan's shape, not of this composition, and it is reported on - * its own; this file does not pin around it. + * ## The run's identity survives its own progress (#21528) + * + * `recorded-by`'s `load()` selects only the rows still holding the sentinel, + * so it shrinks as the run commits chunks, and the plan the owner registers + * for resume carries the default chunk size, not the one the run was started + * with. Both used to change the chunk plan a resume recomputed, and the + * runner refused `PLAN_CHANGED` a run the list had just called resumable. + * Three more interrupted runs, made the same way: + * + * 4. killed in chunk 1 after chunk 0 committed (203 rows at the default + * size): it resumes to completion, and chunk 0 is not run again; + * 5. started with a chunk size of 2 (the `--chunk-size 2` an operator + * passes) and killed in chunk 0: it resumes with the journal's size, two + * chunks, not the one chunk the registered plan's default would make; + * 6. the control: a run started by a plan whose step had another name — a + * changed plan — is still refused `PLAN_CHANGED`, and nothing is written. * * Every boot runs in a hook: a case only reads what a boot printed or wrote. */ @@ -97,9 +107,11 @@ function childEnv(overrides: Record): Record { - process.stderr.write('[fixture] run ' + ctx.runId + ' killed in chunk ' + ctx.chunkIndex + '\\n'); - process.kill(process.pid, 'SIGKILL'); - await new Promise(() => {}); -} }] }; +const crashing = { ...owned, steps: [{ + ...step, + ...(process.env.FIXTURE_STEP_NAME ? { name: process.env.FIXTURE_STEP_NAME } : {}), + forward: async (rows, ctx, engine) => { + if (ctx.chunkIndex !== killChunk) return step.forward(rows, ctx, engine); + process.stderr.write('[fixture] run ' + ctx.runId + ' killed in chunk ' + ctx.chunkIndex + '\\n'); + process.kill(process.pid, 'SIGKILL'); + await new Promise(() => {}); + }, +}] }; await core.runMigrationJournal(ql, crashing); process.stderr.write('[fixture] the run finished, so nothing was interrupted\\n'); process.exit(3); @@ -232,58 +251,67 @@ async function readRows(dbFile: string, sql: string, bindings: unknown[] = []): } } -let root: string; -let runId: string; -let resumeList: { payload: any; exitCode: number }; -let resumeAct: { payload: any; exitCode: number }; -let resumed: ProjectCopy; -let scanOutput: string; -let freshOutput: string; - -beforeAll(async () => { - root = mkdtempSync(join(tmpdir(), 'os-21498-')); - const origin = makeProject(root, 'origin'); +interface InterruptedFixture { + /** The interrupted run's id, as the killed child printed it. */ + runId: string; + /** The database the crash left behind. Copy it; never resume over it. */ + origin: ProjectCopy; +} - // ── the crash ────────────────────────────────────────────────────────────── +/** + * Seed `ids` as sentinel rows and kill a recorded-by run in chunk `killChunk`. + * See `CRASH_CHILD` for what the other two options change about the run. + */ +function interruptRun( + name: string, + ids: readonly string[], + opts: { killChunk?: number; chunkSize?: number; stepName?: string } = {}, +): InterruptedFixture { + const origin = makeProject(root, name); + const killChunk = opts.killChunk ?? 0; const crash = spawnSync(process.execPath, ['--input-type=module', '-e', CRASH_CHILD], { cwd: CLI_ROOT, env: childEnv({ OS_ARTIFACT_PATH: join(origin.dir, 'dist', 'objectstack.json'), FIXTURE_PROJECT: origin.dir, FIXTURE_DB: origin.dbFile, - FIXTURE_IDS: JSON.stringify(SENTINEL_IDS), + FIXTURE_IDS: JSON.stringify(ids), FIXTURE_SENTINEL: RECORDED_BY_SENTINEL, + FIXTURE_KILL_CHUNK: String(killChunk), + FIXTURE_CHUNK_SIZE: opts.chunkSize === undefined ? undefined : String(opts.chunkSize), + FIXTURE_STEP_NAME: opts.stepName, }), encoding: 'utf8', timeout: CHILD_BUDGET_MS, }); - const killed = /\[fixture\] run (\S+) killed in chunk 0/.exec(crash.stderr ?? ''); + const killed = new RegExp(`\\[fixture\\] run (\\S+) killed in chunk ${killChunk}\\b`).exec(crash.stderr ?? ''); if (crash.signal !== 'SIGKILL' || !killed) { throw new Error( - `the fixture did not leave an interrupted run (status ${crash.status}, signal ${crash.signal})\n${crash.stderr}`, + `the fixture '${name}' did not leave an interrupted run (status ${crash.status}, signal ${crash.signal})\n${crash.stderr}`, ); } - runId = killed[1]; + return { runId: killed[1], origin }; +} - // Each consumer gets its own copy: a resume concludes the run, a serve boot - // writes rows of its own. - resumed = makeProject(root, 'resumed'); - const scanned = makeProject(root, 'scanned'); - cpSync(join(origin.dir, 'data'), join(resumed.dir, 'data'), { recursive: true }); - cpSync(join(origin.dir, 'data'), join(scanned.dir, 'data'), { recursive: true }); - const fresh = makeProject(root, 'fresh'); +/** A fresh copy of the database a crash left behind. */ +function copyOf(fixture: InterruptedFixture, name: string): ProjectCopy { + const copy = makeProject(root, name); + cpSync(join(fixture.origin.dir, 'data'), join(copy.dir, 'data'), { recursive: true }); + return copy; +} - // ── 1. the real command ──────────────────────────────────────────────────── +/** Each `argv` through the real `os migrate resume --json`, in order, over `project`. */ +async function resumeOver(project: ProjectCopy, ...argvs: string[][]): Promise> { + const out: Array<{ payload: any; exitCode: number }> = []; const savedEnv: Record = {}; const savedCwd = process.cwd(); for (const key of [...OVERRIDING_ENV, 'OS_ARTIFACT_PATH'] as const) savedEnv[key] = process.env[key]; try { for (const key of OVERRIDING_ENV) delete process.env[key]; - process.env.OS_ARTIFACT_PATH = join(resumed.dir, 'dist', 'objectstack.json'); - process.chdir(resumed.dir); - const url = `file:${resumed.dbFile}`; - resumeList = await resumeJson(['--database-url', url]); - resumeAct = await resumeJson(['--run', runId, '--yes', '--database-url', url]); + process.env.OS_ARTIFACT_PATH = join(project.dir, 'dist', 'objectstack.json'); + process.chdir(project.dir); + const url = `file:${project.dbFile}`; + for (const argv of argvs) out.push(await resumeJson([...argv, '--database-url', url])); } finally { process.chdir(savedCwd); for (const [key, value] of Object.entries(savedEnv)) { @@ -291,10 +319,86 @@ beforeAll(async () => { else process.env[key] = value; } } + return out; +} + +/** Journal kinds and chunk indices of one run, in `seq` order. */ +async function journalOf(dbFile: string, id: string): Promise> { + return readRows(dbFile, 'SELECT kind, chunk_index FROM sys_migration_journal WHERE run_id = ? ORDER BY seq', [id]); +} + +/** How many of `ids` still hold the sentinel. */ +async function sentinelCount(dbFile: string, ids: readonly string[]): Promise { + const rows = await readRows( + dbFile, + `SELECT COUNT(*) AS n FROM sys_metadata_history WHERE recorded_by = ? AND id IN (${ids.map(() => '?').join(', ')})`, + [RECORDED_BY_SENTINEL, ...ids], + ); + return Number(rows[0].n); +} + +/** One more row than the default chunk size holds, plus two: chunk 0 is 200 rows, chunk 1 is 3. */ +const COMMITTED_IDS = Array.from({ length: 203 }, (_, i) => `h_21528_c${String(i).padStart(3, '0')}`); +const SIZED_IDS = ['h_21528_s_a', 'h_21528_s_b', 'h_21528_s_c']; +const CHANGED_IDS = ['h_21528_x_a', 'h_21528_x_b', 'h_21528_x_c']; + +let root: string; +let runId: string; +let resumeList: { payload: any; exitCode: number }; +let resumeAct: { payload: any; exitCode: number }; +let resumed: ProjectCopy; +let scanOutput: string; +let freshOutput: string; + +/** The #21528 runs: each fixture, its resumed copy, and what list and act answered. */ +interface ResumedFixture extends InterruptedFixture { + copy: ProjectCopy; + list?: { payload: any; exitCode: number }; + act: { payload: any; exitCode: number }; +} +let afterCommit: ResumedFixture; +let sized: ResumedFixture; +let changed: ResumedFixture; + +beforeAll(async () => { + root = mkdtempSync(join(tmpdir(), 'os-21498-')); + + // ── the crash ────────────────────────────────────────────────────────────── + const interrupted = interruptRun('origin', SENTINEL_IDS); + runId = interrupted.runId; + + // Each consumer gets its own copy: a resume concludes the run, a serve boot + // writes rows of its own. + resumed = copyOf(interrupted, 'resumed'); + const scanned = copyOf(interrupted, 'scanned'); + const fresh = makeProject(root, 'fresh'); + + // ── 1. the real command ──────────────────────────────────────────────────── + [resumeList, resumeAct] = await resumeOver(resumed, [], ['--run', runId, '--yes']); // ── 2 and 3. the real serve boot ─────────────────────────────────────────── scanOutput = await serveBoot(scanned); freshOutput = await serveBoot(fresh); + + // ── 4, 5 and 6. runs whose recomputed chunk plan differs (#21528) ─────────── + const resumeFixture = async ( + fixture: InterruptedFixture, + name: string, + withList: boolean, + ): Promise => { + const copy = copyOf(fixture, name); + const act = ['--run', fixture.runId, '--yes']; + if (!withList) return { ...fixture, copy, act: (await resumeOver(copy, act))[0] }; + const [list, done] = await resumeOver(copy, [], act); + return { ...fixture, copy, list, act: done }; + }; + afterCommit = await resumeFixture(interruptRun('after-commit', COMMITTED_IDS, { killChunk: 1 }), 'after-commit-resumed', true); + sized = await resumeFixture(interruptRun('sized', SIZED_IDS, { chunkSize: 2 }), 'sized-resumed', true); + changed = await resumeFixture( + interruptRun('changed', CHANGED_IDS, { stepName: 'sys_metadata_history.recorded_by: an earlier step' }), + 'changed-resumed', + false, + ); }, HOOK_TIMEOUT_MS); afterAll(() => { @@ -350,3 +454,57 @@ describe('os serve reports an interrupted run through the boot scan (#21498)', ( expect(freshOutput).not.toMatch(/journal scan failed/i); }); }); + +describe("os migrate resume completes a recorded-by run whose remaining rows shrank or whose chunk size was not the default (#21528)", () => { + it('lists a run killed after a committed chunk as resumable, chunk 0 known committed', () => { + const listed = afterCommit.list!.payload.interrupted.find((r: any) => r.runId === afterCommit.runId); + expect(listed, JSON.stringify(afterCommit.list!.payload)).toBeDefined(); + expect(listed.committedChunks).toEqual([0]); + expect(listed.unknownChunks).toEqual([1]); + expect(listed.resumable).toBe(true); + }); + + it('resumes that run to completion without running its committed chunk again', async () => { + expect(afterCommit.act.exitCode, JSON.stringify(afterCommit.act.payload)).toBe(0); + expect(afterCommit.act.payload).toMatchObject({ + runId: afterCommit.runId, status: 'completed', chunksTotal: 2, chunksCommitted: 2, + }); + expect(await sentinelCount(afterCommit.copy.dbFile, COMMITTED_IDS)).toBe(0); + + const journal = await journalOf(afterCommit.copy.dbFile, afterCommit.runId); + // Chunk 0 committed before the kill and is started exactly once; chunk 1 + // is started by the killed run and again by the resume. + expect(journal.filter((e) => e.kind === 'chunk_started').map((e) => e.chunk_index)).toEqual([0, 1, 1]); + expect(journal.filter((e) => e.kind === 'chunk_done').map((e) => e.chunk_index)).toEqual([0, 1]); + expect(journal.at(-1)!.kind).toBe('run_done'); + }); + + it('lists a run started with a chunk size of 2 as resumable', () => { + const listed = sized.list!.payload.interrupted.find((r: any) => r.runId === sized.runId); + expect(listed, JSON.stringify(sized.list!.payload)).toBeDefined(); + expect(listed.unknownChunks).toEqual([0]); + expect(listed.resumable).toBe(true); + }); + + it('resumes that run with the chunk size it started with, from the journal', async () => { + expect(sized.act.exitCode, JSON.stringify(sized.act.payload)).toBe(0); + // Three rows at size 2 are two chunks; the registered plan's default size + // would have made one. + expect(sized.act.payload).toMatchObject({ + runId: sized.runId, status: 'completed', chunksTotal: 2, chunksCommitted: 2, + }); + expect(await sentinelCount(sized.copy.dbFile, SIZED_IDS)).toBe(0); + const journal = await journalOf(sized.copy.dbFile, sized.runId); + expect(journal.filter((e) => e.kind === 'chunk_done').map((e) => e.chunk_index)).toEqual([0, 1]); + expect(journal.at(-1)!.kind).toBe('run_done'); + }); + + it('still refuses PLAN_CHANGED a run another plan started (the control), and writes nothing', async () => { + expect(changed.act.exitCode, JSON.stringify(changed.act.payload)).toBe(1); + expect(changed.act.payload.error).toMatch(/^Refused \(PLAN_CHANGED\)/); + expect(await sentinelCount(changed.copy.dbFile, CHANGED_IDS)).toBe(CHANGED_IDS.length); + expect((await journalOf(changed.copy.dbFile, changed.runId)).map((e) => e.kind)).toEqual([ + 'run_started', 'chunk_started', + ]); + }); +}); diff --git a/packages/core/src/utils/migration-journal.test.ts b/packages/core/src/utils/migration-journal.test.ts index 3e115809106..6d3a8a724c4 100644 --- a/packages/core/src/utils/migration-journal.test.ts +++ b/packages/core/src/utils/migration-journal.test.ts @@ -39,6 +39,7 @@ import { hashMigrationPlan, MigrationJournalRefusal, MigrationPlanRegistry, + type MigrationChunkContext, type MigrationPlan, type MigrationPlanStep, } from './migration-journal'; @@ -413,6 +414,199 @@ describe('plan-hash mismatch on resume (acceptance case 3)', () => { }); }); +// ── a resume compares what the run started over (#21528) ───────────────── + +/** + * Run `plan` until a forward reaches run-global chunk `killAt`, and hand back + * the database as a process killed there leaves it: the runner's own + * `run_started` and every committed chunk, `chunk_started(killAt)` with no + * `chunk_done`. The killed forward writes nothing before it stops, so the + * tables at that moment ARE the post-crash state; a second engine gets a + * copy, because the first one is still inside the dead chunk's transaction. + */ +async function crashAt( + engine: FakeEngine, + plan: MigrationPlan, + killAt: number, + options: { chunkSize?: number } = {}, +): Promise<{ restarted: FakeEngine; runId: string }> { + let runId = ''; + let reached: () => void = () => {}; + const atKill = new Promise((resolve) => { reached = resolve; }); + const crashing: MigrationPlan = { + ...plan, + steps: plan.steps.map((s) => ({ + ...s, + async forward(rows: unknown[], ctx: MigrationChunkContext, e: Parameters[0]) { + if (ctx.chunkIndex !== killAt) return s.forward(rows, ctx, e); + runId = ctx.runId; + reached(); + return new Promise(() => {}); + }, + })), + }; + void runMigrationJournal(asEngine(engine), crashing, options); + await atKill; + const restarted = new FakeEngine(); + restarted.tables = new Map([...engine.tables].map(([k, v]) => [k, v.map((r) => ({ ...r }))])); + return { restarted, runId }; +} + +const ITEMS = 'items'; +const ITEMS_DONE = 'items_done'; + +/** + * A step whose `load()` selects only the work still to do — `items` rows + * without an `items_done` row — so it shrinks as the run commits chunks, the + * shape `recorded-by`'s plan has. Its forward writes through the engine, so a + * rolled-back chunk takes its writes with it. + */ +function makeShrinkingStep( + engine: FakeEngine, + n: number, + overrides: Partial> = {}, +): MigrationPlanStep<{ id: string }> & { forwarded: string[][]; compensated: string[][] } { + engine.tables.set(ITEMS, Array.from({ length: n }, (_, i) => ({ id: `i${i}` }))); + const forwarded: string[][] = []; + const compensated: string[][] = []; + return { + name: 'shrinking', + forwarded, + compensated, + async load(e) { + const done = new Set((await e.find(ITEMS_DONE, {})).map((r: { id: string }) => r.id)); + return (await e.find(ITEMS, {})) + .filter((r: { id: string }) => !done.has(r.id)) + .map((r: { id: string }) => ({ id: r.id })); + }, + async forward(rows, ctx, e) { + forwarded.push(rows.map((r) => r.id)); + for (const r of rows) await e.insert(ITEMS_DONE, { id: r.id }, { context: ctx.context as never }); + }, + async compensate(rows) { compensated.push(rows.map((r) => r.id)); }, + ...overrides, + }; +} + +const doneIds = (e: FakeEngine) => (e.tables.get(ITEMS_DONE) ?? []).map((r) => r.id); +const chunkEvents = async (e: FakeEngine, runId: string, kind: string) => + (await readRunJournal(asEngine(e), runId)).filter((ev) => ev.kind === kind).map((ev) => ev.chunk_index); + +describe('a resume compares what the run started over, not what load() returns now (#21528)', () => { + it('resumes a run killed after a committed chunk when load() selects only the remaining work', async () => { + const engine = new FakeEngine(); + const step = makeShrinkingStep(engine, 5); + const plan: MigrationPlan = { id: 'p', steps: [step], chunkSize: 2, onCrash: 'resume' }; + const { restarted, runId } = await crashAt(engine, plan, 1); + expect(doneIds(restarted)).toEqual(['i0', 'i1']); // chunk 0 committed + + const result = await resumeMigrationJournal(asEngine(restarted), plan, runId); + + expect(result).toMatchObject({ status: 'completed', chunksTotal: 3, chunksCommitted: 3 }); + // The rows load() returns now belong, in order, to chunks 1 and 2; chunk 0 + // is not run again. + expect(step.forwarded).toEqual([['i0', 'i1'], ['i2', 'i3'], ['i4']]); + expect(doneIds(restarted)).toEqual(['i0', 'i1', 'i2', 'i3', 'i4']); + expect(await chunkEvents(restarted, runId, 'chunk_started')).toEqual([0, 1, 1, 2]); + expect(await chunkEvents(restarted, runId, 'chunk_done')).toEqual([0, 1, 2]); + }); + + it('resumes a run whose load() does not shrink with each chunk\'s rows where the journal put them', async () => { + const engine = new FakeEngine(); + const step = makeStep(6); + const plan: MigrationPlan = { id: 'p', steps: [step], chunkSize: 2, onCrash: 'resume' }; + const { restarted, runId } = await crashAt(engine, plan, 1); + + const result = await resumeMigrationJournal(asEngine(restarted), plan, runId); + + expect(result).toMatchObject({ status: 'completed', chunksTotal: 3, chunksCommitted: 3 }); + expect(step.written).toEqual([0, 1, 2, 3, 4, 5]); // each row once; chunk 0 not redone + }); + + it('resumes a run with the chunk size it started with, from the journal — not the plan\'s current one', async () => { + const engine = new FakeEngine(); + const step = makeShrinkingStep(engine, 5); + // Started at 2, the way `--chunk-size 2` starts one; the plan handed back + // for the resume declares no size, so its own would be the default 200. + const { restarted, runId } = await crashAt(engine, { id: 'p', steps: [step] }, 0, { chunkSize: 2 }); + + const result = await resumeMigrationJournal(asEngine(restarted), { id: 'p', steps: [step] }, runId); + + expect(result).toMatchObject({ status: 'completed', chunksTotal: 3, chunksCommitted: 3 }); + expect(step.forwarded).toEqual([['i0', 'i1'], ['i2', 'i3'], ['i4']]); + }); + + it('still refuses PLAN_CHANGED a plan whose declared identity changed (the control), and writes nothing', async () => { + for (const changed of [ + (s: MigrationPlanStep<{ id: string }>): MigrationPlan => ({ id: 'p', steps: [{ ...s, name: 'shrinking, v2' }], chunkSize: 2 }), + (s: MigrationPlanStep<{ id: string }>): MigrationPlan => ({ id: 'p.v2', steps: [s], chunkSize: 2 }), + (s: MigrationPlanStep<{ id: string }>): MigrationPlan => ({ id: 'p', steps: [s, { ...s, name: 'added' }], chunkSize: 2 }), + ]) { + const engine = new FakeEngine(); + const step = makeShrinkingStep(engine, 5); + const { restarted, runId } = await crashAt(engine, { id: 'p', steps: [step], chunkSize: 2 }, 1); + const before = (restarted.tables.get(JOURNAL) ?? []).length; + + await expect(resumeMigrationJournal(asEngine(restarted), changed(step), runId)) + .rejects.toMatchObject({ name: 'MigrationJournalRefusal', code: 'PLAN_CHANGED' }); + expect((restarted.tables.get(JOURNAL) ?? []).length).toBe(before); + expect(doneIds(restarted)).toEqual(['i0', 'i1']); + } + }); + + it('refuses PLAN_CHANGED when load() returns neither every row the run started over nor exactly the rows it had left', async () => { + const engine = new FakeEngine(); + const step = makeShrinkingStep(engine, 5); + const plan: MigrationPlan = { id: 'p', steps: [step], chunkSize: 2 }; + const { restarted, runId } = await crashAt(engine, plan, 1); + // A row appeared under the run: it started over 5 rows and had 3 left, + // and load() now returns 4. + restarted.tables.get(ITEMS)!.push({ id: 'i5' }); + const before = (restarted.tables.get(JOURNAL) ?? []).length; + + const refusal = await resumeMigrationJournal(asEngine(restarted), plan, runId).catch((e: unknown) => e); + + expect(refusal).toBeInstanceOf(MigrationJournalRefusal); + expect(refusal).toMatchObject({ code: 'PLAN_CHANGED' }); + expect((refusal as Error).message).toContain("step 'shrinking'"); // the binding refusal names its step + expect((restarted.tables.get(JOURNAL) ?? []).length).toBe(before); + }); + + it('halts the unwind at a chunk an earlier process committed, rather than compensating other rows', async () => { + const engine = new FakeEngine(); + const step = makeShrinkingStep(engine, 5); + const plan: MigrationPlan = { id: 'p', steps: [step], chunkSize: 2, onCrash: 'resume' }; + const { restarted, runId } = await crashAt(engine, plan, 1); + const failing: MigrationPlan = { + ...plan, + steps: [{ + ...step, + async forward(rows, ctx, e) { + if (ctx.chunkIndex === 2) throw new Error('chunk 2 fails on the resume'); + return step.forward(rows, ctx, e); + }, + }], + }; + + const result = await resumeMigrationJournal(asEngine(restarted), failing, runId); + + // Chunk 1, committed by this process, is compensated with its own rows. + // Chunk 0's rows left load() when it committed, so nothing here knows + // them: the run halts `failed` instead of compensating some other rows + // and journalling a clean unwind. + expect(result.status).toBe('failed'); + expect(step.compensated).toEqual([['i2', 'i3']]); + expect(await chunkEvents(restarted, runId, 'compensated')).toEqual([1]); + const failure = (await readRunJournal(asEngine(restarted), runId)).at(-1)!; + expect(failure).toMatchObject({ kind: 'run_failed', chunk_index: 0 }); + const detail = JSON.parse(failure.detail!); + expect(detail).toMatchObject({ phase: 'compensate', step: 'shrinking' }); + // A halt that says why — not a compensate() that was handed no rows and threw. + expect(detail).toHaveProperty('reason'); + expect(detail).not.toHaveProperty('error'); + }); +}); + // ── LIFO compensation ───────────────────────────────────────────────────── describe('LIFO compensation (ADR-0119 D2 item 4)', () => { diff --git a/packages/core/src/utils/migration-journal.ts b/packages/core/src/utils/migration-journal.ts index df8d8d5009b..b3114af73bb 100644 --- a/packages/core/src/utils/migration-journal.ts +++ b/packages/core/src/utils/migration-journal.ts @@ -163,7 +163,16 @@ export interface MigrationPlanStep { * step 1 (ADR-0117 D8's fail-closed enable gate, generalized). */ preflight?(engine: IObjectQLEngine): Promise; - /** The rows this step processes. Called once, before chunking. */ + /** + * The rows this step processes. Called once, before chunking. + * + * A resume calls it again and binds what it returns to the chunk plan the + * run started with (#21528). Two answers bind, in the order the run first + * loaded them: every row the run started over (a chunk's rows are where the + * journal put them), or exactly the rows of the chunks not yet committed (a + * `load()` that selects only the work still to do shrinks as the run + * commits). Any other count is refused `PLAN_CHANGED`. + */ load(engine: IObjectQLEngine): Promise; /** Forward work for one chunk. Runs INSIDE the chunk's transaction. */ forward(rows: TRow[], ctx: MigrationChunkContext, engine: IObjectQLEngine): Promise; @@ -314,6 +323,12 @@ export function planChunks( * 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. + * + * A new run hashes the boundaries it derives from its rows. A resume hashes + * the boundaries `run_started` recorded (#21528), so it compares the plan's + * declared identity against what the run STARTED over — not against a chunk + * plan recomputed from rows that the run's own progress, or a chunk size + * other than the one it started with, has since changed. */ export function hashMigrationPlan(plan: MigrationPlan, chunks: readonly MigrationChunk[]): string { const shape = JSON.stringify({ @@ -429,6 +444,12 @@ export async function findInterruptedRuns(engine: IObjectQLEngine): Promise string; @@ -437,19 +458,149 @@ export interface RunMigrationJournalOptions { interface LoadedPlan { readonly chunks: MigrationChunk[]; readonly planHash: string; - readonly rowsByStep: unknown[][]; + /** + * A chunk's rows. `undefined` only for a chunk committed before a resume + * whose `load()` no longer returns that chunk's rows — nothing in this + * process can say which rows it covered, so nothing may compensate it. + */ + readonly rowsOf: (c: MigrationChunk) => unknown[] | undefined; } -/** Load every step's rows, derive the chunk plan, hash it. */ -async function loadPlan( - engine: IObjectQLEngine, - plan: MigrationPlan, - chunkSize?: number, -): Promise { +/** Load every step's rows, in declaration order. */ +async function loadRows(engine: IObjectQLEngine, plan: MigrationPlan): Promise { const rowsByStep: unknown[][] = []; for (const step of plan.steps) rowsByStep.push((await step.load(engine)) ?? []); + return rowsByStep; +} + +/** A chunk's rows where the chunk plan put them: `offset`, `length`. */ +const sliceOf = (rowsByStep: unknown[][]) => (c: MigrationChunk): unknown[] => + rowsByStep[c.stepIndex].slice(c.offset, c.offset + c.length); + +/** A new run: derive the chunk plan from the rows, and hash it. */ +function planNewRun(plan: MigrationPlan, rowsByStep: unknown[][], chunkSize?: number): LoadedPlan { const chunks = planChunks(plan, rowsByStep.map((r) => r.length), chunkSize ?? plan.chunkSize); - return { chunks, planHash: hashMigrationPlan(plan, chunks), rowsByStep }; + return { chunks, planHash: hashMigrationPlan(plan, chunks), rowsOf: sliceOf(rowsByStep) }; +} + +/** + * The chunk plan `run_started` recorded, or `undefined` when its `detail` + * carries none — which this runner has written on every run since its first + * commit, so only a journal it did not write lacks one. + * + * The record names each chunk's step, not its index. Steps are recorded in + * declaration order and each one's chunks start at offset 0, so a chunk's step + * is the first at or after the previous chunk's step that carries its name — + * advancing past it on a new offset-0 chunk. A name no step carries maps to + * -1, which no hash the runner wrote can match. + */ +function recordedChunks(plan: MigrationPlan, start: MigrationJournalEvent | undefined): MigrationChunk[] | undefined { + let recorded: unknown; + try { + recorded = start?.detail ? (JSON.parse(start.detail) as { chunks?: unknown }).chunks : undefined; + } catch { + return undefined; + } + if (!Array.isArray(recorded)) return undefined; + const chunks: MigrationChunk[] = []; + let stepIndex = 0; + for (const entry of recorded as Array | null>) { + const { step, offset, length } = entry ?? {}; + if ( + typeof step !== 'string' || !Number.isInteger(offset) || (offset as number) < 0 + || !Number.isInteger(length) || (length as number) < 1 + ) { + return undefined; + } + if (chunks.length > 0 && offset === 0) stepIndex++; + while (stepIndex < plan.steps.length && plan.steps[stepIndex].name !== step) stepIndex++; + chunks.push({ + index: chunks.length, + stepIndex: stepIndex < plan.steps.length ? stepIndex : -1, + stepName: step, + offset: offset as number, + length: length as number, + }); + } + return chunks; +} + +/** + * A resume: compare the plan against what its run STARTED over, then bind the + * rows `load()` returns now to that chunk plan (#21528, ADR-0119 D2 item 5). + * + * The identity check hashes the journal's own chunk boundaries with the plan's + * declared id and step names. So the size the run started with comes back from + * the journal, and a plan whose `load()` selects only the remaining work is + * not mistaken for a changed one; a plan whose id or steps changed still + * refuses `PLAN_CHANGED`. + * + * Binding is per step, against the rows that step's chunks started over (N) + * and the rows in its committed chunks (K). `load()` may return N rows — it + * does not shrink, so a chunk's rows are where the journal put them — or N − K + * — it selects only the work still to do, so those rows belong, in order, to + * the chunks not yet committed. Any other count means the rows moved under the + * run, and the journal's boundaries describe none of them: refused. + */ +function planResumedRun( + plan: MigrationPlan, + runId: string, + start: MigrationJournalEvent | undefined, + rowsByStep: unknown[][], + committed: ReadonlySet, + chunkSize?: number, +): LoadedPlan { + const recorded = recordedChunks(plan, start); + // No recorded chunk plan to read back: reproducing the journal's hash from + // the rows `load()` returns now is the only identity proof left. + const chunks = recorded ?? planNewRun(plan, rowsByStep, chunkSize).chunks; + const planHash = hashMigrationPlan(plan, chunks); + 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.`, + ); + } + + const rowsByChunk = new Map(); + plan.steps.forEach((step, stepIndex) => { + const own = chunks.filter((c) => c.stepIndex === stepIndex); + const startedOver = own.reduce((n, c) => n + c.length, 0); + const inCommitted = own.filter((c) => committed.has(c.index)).reduce((n, c) => n + c.length, 0); + const loaded = rowsByStep[stepIndex]; + if (loaded.length === startedOver) { + for (const c of own) rowsByChunk.set(c.index, loaded.slice(c.offset, c.offset + c.length)); + } else if (loaded.length === startedOver - inCommitted) { + let cursor = 0; + for (const c of own) { + if (committed.has(c.index)) continue; + rowsByChunk.set(c.index, loaded.slice(cursor, cursor + c.length)); + cursor += c.length; + } + } else { + throw new MigrationJournalRefusal( + 'PLAN_CHANGED', + `Refusing to resume run '${runId}': step '${step.name}' now loads ${loaded.length} row(s); the run started ` + + `over ${startedOver}, ${inCommitted} of them in committed chunks. That is neither every row the run started ` + + `over nor exactly the rows it had left, so the chunk boundaries recorded in the journal describe none of them.`, + ); + } + }); + // Only a chunk of a step this plan does not declare is still unbound; with + // a recorded hash that refusal already happened above. + const unbound = chunks.find((c) => !committed.has(c.index) && !rowsByChunk.has(c.index)); + if (unbound) { + throw new MigrationJournalRefusal( + 'PLAN_CHANGED', + `Refusing to resume run '${runId}': the journal's chunk ${unbound.index} belongs to step '${unbound.stepName}', ` + + `which this plan does not declare.`, + ); + } + return { chunks, planHash, rowsOf: (c) => rowsByChunk.get(c.index) }; } /** @@ -457,7 +608,9 @@ async function loadPlan( * * 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. + * step cannot compensate; or a resume's plan disagrees with what its run + * started over — its hash, or rows that bind to that chunk plan under neither + * reading {@link MigrationPlanStep.load} documents. */ export async function runMigrationJournal( engine: IObjectQLEngine, @@ -478,8 +631,6 @@ export async function runMigrationJournal( ); } - const { chunks, planHash, rowsByStep } = await loadPlan(engine, plan, options.chunkSize); - // ── resume bookkeeping ──────────────────────────────────────────────── const resuming = Boolean(options.runId); const runId = options.runId ?? randomUUID(); @@ -494,17 +645,17 @@ export async function runMigrationJournal( 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.`, - ); - } + committed = chunkSetOf(events, 'chunk_done'); + } + + const rowsByStep = await loadRows(engine, plan); + const { chunks, planHash, rowsOf } = resuming + ? planResumedRun( + plan, runId, events.find((e) => e.kind === 'run_started'), rowsByStep, committed, options.chunkSize, + ) + : planNewRun(plan, rowsByStep, options.chunkSize); + + if (resuming) { if (events.some((e) => e.kind === 'run_done')) { return { runId, status: 'completed', chunksTotal: chunks.length, @@ -513,7 +664,6 @@ export async function runMigrationJournal( }; } 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') { @@ -552,7 +702,6 @@ export async function runMigrationJournal( } } - const rowsOf = (c: MigrationChunk): unknown[] => rowsByStep[c.stepIndex].slice(c.offset, c.offset + c.length); const next = (): number => seq++; if (!resuming) { @@ -583,7 +732,8 @@ export async function runMigrationJournal( const attempt = (attemptsByChunk.get(chunk.index) ?? 0) + 1; attemptsByChunk.set(chunk.index, attempt); const step = plan.steps[chunk.stepIndex]; - const rows = rowsOf(chunk); + // Bound for every chunk not committed — `planResumedRun` refuses otherwise. + const rows = rowsOf(chunk)!; // Autonomous, BEFORE the transaction: this is what makes an interrupted // chunk visible as "started, outcome unknown" rather than invisible. @@ -629,7 +779,7 @@ interface UnwindArgs { runId: string; planHash: string; chunks: readonly MigrationChunk[]; - rowsOf: (c: MigrationChunk) => unknown[]; + rowsOf: LoadedPlan['rowsOf']; next: () => number; now: () => string; committed: Set; @@ -660,15 +810,24 @@ async function unwind( 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. + const rows = a.rowsOf(chunk); + + // Nothing to undo this with — no compensate(), or (#21528) a chunk a + // resumed run's earlier process committed, whose rows `load()` no longer + // returns, so nothing here knows which rows it covered. Say so loudly and + // stop: a silent skip, or a compensate() over other rows, would leave the + // rows written and the journal claiming a clean unwind. + const reason = !step.compensate + ? 'step declares no compensate()' + : rows === undefined + ? 'committed before this resume, and load() no longer returns its rows' + : undefined; + if (reason) { 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()', + phase: 'compensate', reason, step: step.name, cause: errText(a.cause), }), }); @@ -682,7 +841,7 @@ async function unwind( 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 step.compensate!(rows!, { runId: a.runId, chunkIndex: index, attempt, context: trxCtx }, engine); await appendEvent( engine, { diff --git a/packages/metadata-protocol/src/migrations/recorded-by-sentinel.test.ts b/packages/metadata-protocol/src/migrations/recorded-by-sentinel.test.ts index f08d972ff0e..ffda179a3bf 100644 --- a/packages/metadata-protocol/src/migrations/recorded-by-sentinel.test.ts +++ b/packages/metadata-protocol/src/migrations/recorded-by-sentinel.test.ts @@ -21,7 +21,7 @@ */ import { describe, it, expect } from 'vitest'; -import { runMigrationJournal, readRunJournal } from '@objectstack/core'; +import { runMigrationJournal, readRunJournal, resumeMigrationJournal, type MigrationPlan } from '@objectstack/core'; // [#5619] The producer's OWN write-verb dispatch decisions. Imported from // `@objectstack/metadata-core` rather than `@objectstack/objectql` because // objectql depends on THIS package — the import that would pin these doubles to @@ -236,6 +236,48 @@ describe('#4556 migration — sentinel → NULL', () => { expect(byId(engine, 'h2').recorded_by).toBe('usr_alice'); }); + it('[#21528] a run killed after a committed chunk, started at chunk size 2, resumes with the plan the owner registers', async () => { + const engine = new FakeEngine(); + seedMixed(engine); + // Started the way `os migrate recorded-by --apply --chunk-size 2` starts + // it: chunk 0 converts h1+h3 and commits, and the process dies inside + // chunk 1 (h5) before writing anything. The forward stops there for good, + // so the tables at that moment are what the kill leaves; the resume gets + // a copy, since this engine is still inside the dead chunk's transaction. + const started = createRecordedBySentinelPlan({ chunkSize: 2 }); + let runId = ''; + let killed: () => void = () => {}; + const atKill = new Promise((resolve) => { killed = resolve; }); + const crashing: MigrationPlan = { + ...started, + steps: [{ + ...started.steps[0]!, + async forward(rows, ctx, e) { + if (ctx.chunkIndex === 0) return started.steps[0]!.forward(rows, ctx, e); + runId = ctx.runId; + killed(); + return new Promise(() => {}); + }, + }], + }; + void runMigrationJournal(asEngine(engine), crashing); + await atKill; + const restarted = new FakeEngine(); + restarted.tables = new Map([...engine.tables].map(([k, v]) => [k, v.map((r) => ({ ...r }))])); + expect(await findSentinelHistoryRows(asEngine(restarted))).toEqual([{ id: 'h5' }]); + + // The registry holds the owner's plan at its default size, and load() now + // returns only h5: the resume still runs the journal's two chunks. + const result = await resumeMigrationJournal(asEngine(restarted), createRecordedBySentinelPlan(), runId); + + expect(result).toMatchObject({ runId, status: 'completed', chunksTotal: 2, chunksCommitted: 2 }); + expect(restarted.history().filter((r) => r.recorded_by === RECORDED_BY_SENTINEL)).toEqual([]); + expect(byId(restarted, 'h2').recorded_by).toBe('usr_alice'); + const events = await readRunJournal(asEngine(restarted), runId); + expect(events.filter((e) => e.kind === 'chunk_done').map((e) => e.chunk_index)).toEqual([0, 1]); + expect(events[events.length - 1]!.kind).toBe('run_done'); + }); + it('findSentinelHistoryRows selects only sentinel rows, by id', async () => { const engine = new FakeEngine(); seedMixed(engine);