|
21 | 21 | */ |
22 | 22 |
|
23 | 23 | import { describe, it, expect } from 'vitest'; |
24 | | -import { runMigrationJournal, readRunJournal } from '@objectstack/core'; |
| 24 | +import { runMigrationJournal, readRunJournal, resumeMigrationJournal, type MigrationPlan } from '@objectstack/core'; |
25 | 25 | // [#5619] The producer's OWN write-verb dispatch decisions. Imported from |
26 | 26 | // `@objectstack/metadata-core` rather than `@objectstack/objectql` because |
27 | 27 | // objectql depends on THIS package — the import that would pin these doubles to |
@@ -236,6 +236,48 @@ describe('#4556 migration — sentinel → NULL', () => { |
236 | 236 | expect(byId(engine, 'h2').recorded_by).toBe('usr_alice'); |
237 | 237 | }); |
238 | 238 |
|
| 239 | + it('[#21528] a run killed after a committed chunk, started at chunk size 2, resumes with the plan the owner registers', async () => { |
| 240 | + const engine = new FakeEngine(); |
| 241 | + seedMixed(engine); |
| 242 | + // Started the way `os migrate recorded-by --apply --chunk-size 2` starts |
| 243 | + // it: chunk 0 converts h1+h3 and commits, and the process dies inside |
| 244 | + // chunk 1 (h5) before writing anything. The forward stops there for good, |
| 245 | + // so the tables at that moment are what the kill leaves; the resume gets |
| 246 | + // a copy, since this engine is still inside the dead chunk's transaction. |
| 247 | + const started = createRecordedBySentinelPlan({ chunkSize: 2 }); |
| 248 | + let runId = ''; |
| 249 | + let killed: () => void = () => {}; |
| 250 | + const atKill = new Promise<void>((resolve) => { killed = resolve; }); |
| 251 | + const crashing: MigrationPlan = { |
| 252 | + ...started, |
| 253 | + steps: [{ |
| 254 | + ...started.steps[0]!, |
| 255 | + async forward(rows, ctx, e) { |
| 256 | + if (ctx.chunkIndex === 0) return started.steps[0]!.forward(rows, ctx, e); |
| 257 | + runId = ctx.runId; |
| 258 | + killed(); |
| 259 | + return new Promise<void>(() => {}); |
| 260 | + }, |
| 261 | + }], |
| 262 | + }; |
| 263 | + void runMigrationJournal(asEngine(engine), crashing); |
| 264 | + await atKill; |
| 265 | + const restarted = new FakeEngine(); |
| 266 | + restarted.tables = new Map([...engine.tables].map(([k, v]) => [k, v.map((r) => ({ ...r }))])); |
| 267 | + expect(await findSentinelHistoryRows(asEngine(restarted))).toEqual([{ id: 'h5' }]); |
| 268 | + |
| 269 | + // The registry holds the owner's plan at its default size, and load() now |
| 270 | + // returns only h5: the resume still runs the journal's two chunks. |
| 271 | + const result = await resumeMigrationJournal(asEngine(restarted), createRecordedBySentinelPlan(), runId); |
| 272 | + |
| 273 | + expect(result).toMatchObject({ runId, status: 'completed', chunksTotal: 2, chunksCommitted: 2 }); |
| 274 | + expect(restarted.history().filter((r) => r.recorded_by === RECORDED_BY_SENTINEL)).toEqual([]); |
| 275 | + expect(byId(restarted, 'h2').recorded_by).toBe('usr_alice'); |
| 276 | + const events = await readRunJournal(asEngine(restarted), runId); |
| 277 | + expect(events.filter((e) => e.kind === 'chunk_done').map((e) => e.chunk_index)).toEqual([0, 1]); |
| 278 | + expect(events[events.length - 1]!.kind).toBe('run_done'); |
| 279 | + }); |
| 280 | + |
239 | 281 | it('findSentinelHistoryRows selects only sentinel rows, by id', async () => { |
240 | 282 | const engine = new FakeEngine(); |
241 | 283 | seedMixed(engine); |
|
0 commit comments