fix(rest): make async import-job cancellation actually stop the worker (#2824)

.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.

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;

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<string, ExportFieldMeta>([
+  ['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<Record<string, any>> {
+  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);
+  });
+});

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<void> =>
+  new Promise<void>((resolve) => {
+    if (typeof setImmediate === 'function') setImmediate(resolve);
+    else setTimeout(resolve, 0);
+  });
+
 export function runImport(opts: RunImportOptions): Promise<ImportRunSummary> {
   const {
     p, objectName, environmentId, context, rows, metaMap,
@@ -367,8 +381,11 @@ export function runImport(opts: RunImportOptions): Promise<ImportRunSummary> {
         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; }
       }
     }
 

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<any | undefined> => {
+            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<any | undefined> => {
-            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',