diff --git a/.changeset/fix-async-import-cancel-2824.md b/.changeset/fix-async-import-cancel-2824.md new file mode 100644 index 0000000000..a8eaff02fb --- /dev/null +++ b/.changeset/fix-async-import-cancel-2824.md @@ -0,0 +1,27 @@ +--- +'@objectstack/rest': patch +--- + +fix(import): make async import-job cancellation actually stop the worker (#2824) + +Cancelling a running async import used to have no effect on a synchronous +storage driver (better-sqlite3 / wasm fallback): every `await` in the row +loop resolved as a microtask, so a 50k-row import monopolized the Node event +loop for minutes — the cancel route's HTTP handler (and every progress poll) +could never run, so the in-memory flag `shouldCancel` polls was never set. +The job then finished `succeeded` with all rows written despite the user's +cancel. + +Three-part fix: + +- **`runImport` yields one macrotask at every progress boundary** (every + `progressEvery` rows), so pending I/O — the cancel request, progress + polls, any other traffic — gets serviced during a large import. This is + the root-cause fix; it also unblocks progress polling for the wizard. +- **The worker's `shouldCancel` now also reads the durable job row** as a + fallback: a cancel accepted by another process (or after a restart + dropped the in-memory flag) still stops the worker. +- **A late cancel wins the terminal state**: the worker's final patch no + longer overwrites the cancel route's durable `cancelled` with + `succeeded`, and a job cancelled while still `pending` doesn't start at + all. Counts stay truthful — they reflect what was actually written. diff --git a/packages/rest/src/import-job-integration.test.ts b/packages/rest/src/import-job-integration.test.ts index b053d75cff..097f64e7a8 100644 --- a/packages/rest/src/import-job-integration.test.ts +++ b/packages/rest/src/import-job-integration.test.ts @@ -120,8 +120,9 @@ function makeRes() { return res; } -async function boot() { +async function boot(decorateDriver?: (driver: any) => void) { const { driver } = makeMemoryDriver(); + decorateDriver?.(driver); const engine = new ObjectQL(); engine.registerDriver(driver, true); await engine.init(); @@ -234,6 +235,96 @@ describe('async import job — real engine + protocol integration', () => { expect(c._status).toBe(404); }); + // framework#2824 — cancelling a running job must actually stop the worker. + it('cancels a running job mid-flight: the worker stops at the next checkpoint', async () => { + const rows = Array.from({ length: 1000 }, (_, i) => ({ id: `mc${i}`, title: `t${i}` })); + const created = await callCreate(ctx.create, { format: 'json', rows }); + const jobId = created._json.jobId; + + const c = await callJob(ctx.cancel, jobId); + expect(c._json).toMatchObject({ success: true }); + + const done = await waitForTerminal(ctx.progress, jobId); + expect(done.status).toBe('cancelled'); + + // Let the background worker settle (progress stops moving), then prove it + // really stopped early instead of importing all 1000 rows (#2824's bug). + let settled = -1; + for (let i = 0; i < 100; i++) { + const r = await callJob(ctx.progress, jobId); + const p = Number(r._json?.processed ?? 0); + if (p === settled) break; + settled = p; + await new Promise((rr) => setTimeout(rr, 10)); + } + expect(settled).toBeLessThan(1000); + const written = await ctx.engine.find('task', { where: {} }); + expect(written.length).toBeLessThan(1000); + }); + + // framework#2824 — a durable 'cancelled' written by another process (no + // in-memory flag on this node) must stop the worker too. + it('stops the worker when the job row is marked cancelled out-of-band', async () => { + const rows = Array.from({ length: 1000 }, (_, i) => ({ id: `oc${i}`, title: `t${i}` })); + const created = await callCreate(ctx.create, { format: 'json', rows }); + const jobId = created._json.jobId; + + // Simulate a cancel accepted by a different node: write the durable row + // directly, bypassing this server's cancel route and in-memory flag. + await ctx.protocol.updateData({ object: 'sys_import_job', id: jobId, data: { status: 'cancelled' } }); + + const done = await waitForTerminal(ctx.progress, jobId); + expect(done.status).toBe('cancelled'); + let settled = -1; + for (let i = 0; i < 100; i++) { + const r = await callJob(ctx.progress, jobId); + const p = Number(r._json?.processed ?? 0); + if (p === settled) break; + settled = p; + await new Promise((rr) => setTimeout(rr, 10)); + } + expect(settled).toBeLessThan(1000); + }); + + // framework#2824 — a cancel that lands too late to stop the loop must still + // win the terminal state: the worker's final patch may not overwrite the + // durable 'cancelled' with 'succeeded'. + it('keeps the terminal state cancelled when the cancel lands after the last row', async () => { + // Slow the task writes so the cancel deterministically arrives while the + // job is running — but the 3-row job has no mid-loop checkpoint, so the + // loop still completes every row before noticing. + const slowCtx = await boot((driver) => { + const bulkCreate = driver.bulkCreate.bind(driver); + driver.bulkCreate = async (o: string, rows2: any[]) => { + if (o === 'task') await new Promise((r) => setImmediate(r)); + return bulkCreate(o, rows2); + }; + const create = driver.create.bind(driver); + driver.create = async (o: string, data: any) => { + if (o === 'task') await new Promise((r) => setImmediate(r)); + return create(o, data); + }; + }); + const created = await callCreate(slowCtx.create, { + format: 'json', + rows: [{ id: 'lc1', title: 'a' }, { id: 'lc2', title: 'b' }, { id: 'lc3', title: 'c' }], + }); + const jobId = created._json.jobId; + const c = await callJob(slowCtx.cancel, jobId); + expect(c._json).toMatchObject({ success: true }); + + const done = await waitForTerminal(slowCtx.progress, jobId); + // The counts stay truthful (rows were written), but the user's cancel wins + // the status — before the fix this flipped back to 'succeeded'. + expect(done.status).toBe('cancelled'); + // Give the worker's final patch time to land, then re-check it did not + // overwrite the status. + await new Promise((r) => setTimeout(r, 50)); + const after = await callJob(slowCtx.progress, jobId); + expect(after._json.status).toBe('cancelled'); + expect(after._json.processed).toBe(3); + }); + it('cancel on an already-finished job is a no-op success', async () => { const created = await callCreate(ctx.create, { format: 'json', rows: [{ id: 'k', title: 'x' }] }); const jobId = created._json.jobId; diff --git a/packages/rest/src/import-runner-cancel.test.ts b/packages/rest/src/import-runner-cancel.test.ts new file mode 100644 index 0000000000..9c591c372f --- /dev/null +++ b/packages/rest/src/import-runner-cancel.test.ts @@ -0,0 +1,94 @@ +// Copyright (c) 2025 ObjectStack. Licensed under the Apache-2.0 license. + +/** + * runImport() cooperative cancellation (framework#2824). + * + * With a synchronous storage driver every `await` in the row loop resolves as + * a microtask, so the loop used to monopolize the event loop for the whole + * import — the HTTP cancel handler could never run, so the flag `shouldCancel` + * polls was never set. The runner now yields one macrotask at every progress + * boundary; these tests pin that behaviour by scheduling the cancel signal on + * the macrotask queue (exactly where an HTTP handler lives) and asserting the + * loop actually stops. + */ + +import { describe, it, expect, vi } from 'vitest'; +import { runImport, type ImportProtocolLike } from './import-runner'; +import type { ExportFieldMeta } from './export-format.js'; + +const metaMap = new Map([ + ['name', { name: 'name', type: 'text' }], +]); + +const baseOpts = { + objectName: 'task', + metaMap, + writeMode: 'insert' as const, + matchFields: [] as string[], + dryRun: false, + runAutomations: false, + trimWhitespace: true, + createMissingOptions: false, + skipBlankMatchKey: false, +}; + +function rowsOf(n: number): Array> { + return Array.from({ length: n }, (_, i) => ({ name: `r${i}` })); +} + +/** Protocol whose writes resolve synchronously (microtask-only), like better-sqlite3. */ +function syncProtocol(): ImportProtocolLike { + return { + findData: vi.fn(async () => []), + createData: vi.fn(async (args: { data: { name: string } }) => ({ id: `id_${args.data.name}` })), + updateData: vi.fn(async () => ({})), + createManyData: vi.fn(async (args: { records: any[] }) => ({ + records: args.records.map((r) => ({ id: `id_${r.name}`, ...r })), + })), + }; +} + +describe('runImport — cooperative cancellation (framework#2824)', () => { + it('yields the event loop at progress boundaries so a macrotask can set the cancel flag', async () => { + let cancelRequested = false; + // Simulates the cancel route: it can only run when the loop yields a + // macrotask. Before the fix this callback fired after the import finished. + setImmediate(() => { cancelRequested = true; }); + + const summary = await runImport({ + ...baseOpts, p: syncProtocol(), rows: rowsOf(1000), progressEvery: 200, + shouldCancel: () => cancelRequested, + }); + + expect(summary.cancelled).toBe(true); + expect(summary.processed).toBe(200); // stopped at the first checkpoint + expect(summary.created).toBe(200); // rows written so far are reported truthfully + }); + + it('runs to completion when nobody cancels', async () => { + const summary = await runImport({ + ...baseOpts, p: syncProtocol(), rows: rowsOf(450), progressEvery: 200, + shouldCancel: () => false, + }); + expect(summary.cancelled).toBe(false); + expect(summary.processed).toBe(450); + expect(summary.created).toBe(450); + }); + + it('yields at progress boundaries even without a shouldCancel callback', async () => { + // The synchronous import route passes no shouldCancel; the yield must still + // happen so concurrent HTTP requests are serviced during a large import. + let macrotaskRan = false; + setImmediate(() => { macrotaskRan = true; }); + let observedMidLoop = false; + + await runImport({ + ...baseOpts, p: syncProtocol(), rows: rowsOf(400), progressEvery: 200, + onProgress: (pr) => { if (pr.processed === 400) observedMidLoop = macrotaskRan; }, + }); + + // By the final progress report the pre-scheduled macrotask has run, + // proving the loop reached the event loop's timer/check phases mid-import. + expect(observedMidLoop).toBe(true); + }); +}); diff --git a/packages/rest/src/import-runner.ts b/packages/rest/src/import-runner.ts index 59bdd8a821..22f32366a6 100644 --- a/packages/rest/src/import-runner.ts +++ b/packages/rest/src/import-runner.ts @@ -137,6 +137,20 @@ function toFailedResult(rowNo: number, err: unknown): ImportRowResult { /** Upper bound on rows in one createManyData batch (framework#2678 suggests 100-500). */ const MAX_CREATE_BATCH_SIZE = 200; +/** + * Yield one macrotask so the host's event loop can service pending I/O. + * With a synchronous storage driver (better-sqlite3 and the wasm fallback) + * every `await` in the row loop resolves as a microtask, so a large import + * otherwise monopolizes the event loop for its whole duration: HTTP cancel + * and progress requests sit unserviced, and the cooperative `shouldCancel` + * flag has nobody able to set it (framework#2824). + */ +const yieldToEventLoop = (): Promise => + new Promise((resolve) => { + if (typeof setImmediate === 'function') setImmediate(resolve); + else setTimeout(resolve, 0); + }); + export function runImport(opts: RunImportOptions): Promise { const { p, objectName, environmentId, context, rows, metaMap, @@ -367,8 +381,11 @@ export function runImport(opts: RunImportOptions): Promise { await flushPendingCreates(); if (onProgress) await onProgress(snapshot(processed)); } - if (shouldCancel && processed < rows.length && (processed % progressEvery === 0)) { - if (await shouldCancel()) { cancelled = true; break; } + if (processed < rows.length && processed % progressEvery === 0) { + // Yield BEFORE polling the flag: a cancel request can only set it + // once its HTTP handler gets event-loop time (framework#2824). + await yieldToEventLoop(); + if (shouldCancel && (await shouldCancel())) { cancelled = true; break; } } } diff --git a/packages/rest/src/rest-server.ts b/packages/rest/src/rest-server.ts index f4c08d442c..ff89cf1e18 100644 --- a/packages/rest/src/rest-server.ts +++ b/packages/rest/src/rest-server.ts @@ -3658,6 +3658,22 @@ export class RestServer { // are registered inside registerDataActionEndpoints (before the greedy // CRUD `:object/:id`), so the literal `import/jobs` segments win. + // Shared loader: fetch one job row by id. Used by the read routes, the + // cancel route, and the background worker's durable cancellation checks. + const loadImportJob = async (p: any, jobId: string, environmentId?: string, context?: any): Promise => { + const r = await p.findData({ + object: IMPORT_JOB_OBJECT, + query: { $filter: { id: jobId }, $top: 1 }, + ...(environmentId ? { environmentId } : {}), + ...(context ? { context } : {}), + }); + const rows = Array.isArray(r?.records) ? r.records + : Array.isArray(r?.data) ? r.data + : Array.isArray(r?.rows) ? r.rows + : Array.isArray(r) ? r : []; + return rows[0]; + }; + // POST /data/:object/import/jobs — create an async import job. this.routeManager.register({ method: 'POST', @@ -3731,6 +3747,13 @@ export class RestServer { // import can be logically rolled back later. const captureUndo = !prepared.dryRun && prepared.rows.length <= IMPORT_JOB_UNDO_MAX_ROWS; void (async () => { + // Cancelled while still pending? Don't start (and don't let + // the 'running' patch below overwrite the durable 'cancelled'). + if (this.cancelledImportJobs.has(jobId)) { + this.cancelledImportJobs.delete(jobId); + await patch({ status: 'cancelled', completed_at: new Date().toISOString() }); + return; + } await patch({ status: 'running', started_at: new Date().toISOString() }); try { const summary = await runImport({ @@ -3744,10 +3767,33 @@ export class RestServer { skipped_count: pr.skipped, error_count: pr.errors, }), - shouldCancel: () => this.cancelledImportJobs.has(jobId), + shouldCancel: async () => { + if (this.cancelledImportJobs.has(jobId)) return true; + // Durable fallback: the cancel route also writes + // status='cancelled' to the job row, so a cancel + // accepted by another process (or after a restart + // dropped the in-memory flag) still stops the worker. + try { + const row = await loadImportJob(p, jobId, environmentId, context); + return String(row?.status ?? '') === 'cancelled'; + } catch { return false; } + }, }); + // A cancel that lands after the last checkpoint must still + // win the terminal state: the cancel route already marked + // the durable row 'cancelled', and a late 'succeeded' here + // would silently overwrite it (framework#2824). Counts stay + // truthful either way — they reflect what was written. + let finalStatus = summary.cancelled ? 'cancelled' : 'succeeded'; + if (finalStatus === 'succeeded' && this.cancelledImportJobs.has(jobId)) finalStatus = 'cancelled'; + if (finalStatus === 'succeeded') { + try { + const row = await loadImportJob(p, jobId, environmentId, context); + if (String(row?.status ?? '') === 'cancelled') finalStatus = 'cancelled'; + } catch { /* keep succeeded */ } + } await patch({ - status: summary.cancelled ? 'cancelled' : 'succeeded', + status: finalStatus, processed_rows: summary.processed, created_count: summary.created, updated_count: summary.updated, @@ -3778,21 +3824,6 @@ export class RestServer { }, }); - // Shared loader for the read routes: fetch one job row by id. - const loadImportJob = async (p: any, jobId: string, environmentId?: string, context?: any): Promise => { - const r = await p.findData({ - object: IMPORT_JOB_OBJECT, - query: { $filter: { id: jobId }, $top: 1 }, - ...(environmentId ? { environmentId } : {}), - ...(context ? { context } : {}), - }); - const rows = Array.isArray(r?.records) ? r.records - : Array.isArray(r?.data) ? r.data - : Array.isArray(r?.rows) ? r.rows - : Array.isArray(r) ? r : []; - return rows[0]; - }; - // POST /data/import/jobs/:jobId/cancel — request cancellation. this.routeManager.register({ method: 'POST',