From 8cf4a05ca81e9821d18d2c41633a133fa4cd840c Mon Sep 17 00:00:00 2001 From: Marco Beretta <81851188+berry-13@users.noreply.github.com> Date: Wed, 29 Jul 2026 02:27:58 +0200 Subject: [PATCH] fix(import): serialize job mutations and freeze terminal jobs The background run reads the cancel flag and writes progress in separate round trips, so a DELETE landing between one of those reads and its write was overwritten by the stale snapshot the read returned: the job went back to active, the next isCancelled said so, and the import kept going. Every mutation now goes through the per-key lock that only confirmStart held, and a job that has reached a terminal status no longer accepts a status or phase change - only the partial report describing what it wrote before it stopped. --- packages/api/src/import/job.spec.ts | 70 +++++++++++++++++++++++++++++ packages/api/src/import/job.ts | 51 ++++++++++++++++++--- 2 files changed, 114 insertions(+), 7 deletions(-) diff --git a/packages/api/src/import/job.spec.ts b/packages/api/src/import/job.spec.ts index 502de5f8b6..365dd51543 100644 --- a/packages/api/src/import/job.spec.ts +++ b/packages/api/src/import/job.spec.ts @@ -98,6 +98,76 @@ describe('ImportJobStore', () => { expect(owned?.updatedAt).toBe(job.updatedAt); }); + describe('terminal jobs', () => { + it('ignores a phase update that lands after cancellation', async () => { + const job = await store.create({ userId: 'u1', filepath: '/tmp/a.zip', filename: 'a.zip' }); + await store.cancel('u1', job.jobId); + + const patched = await store.patch('u1', job.jobId, { phase: 'conversations' }); + + expect(patched?.phase).toBe('cancelled'); + expect(patched?.status).toBe('cancelled'); + expect(await store.isCancelled('u1', job.jobId)).toBe(true); + }); + + it('still records the partial report a cancelled run produced', async () => { + const job = await store.create({ userId: 'u1', filepath: '/tmp/a.zip', filename: 'a.zip' }); + await store.cancel('u1', job.jobId); + + const report = { + imported: 12, + skipped: 0, + assetsImported: 3, + assetsUnavailable: 0, + errors: [], + }; + const patched = await store.patch('u1', job.jobId, { report }); + + expect(patched?.report).toEqual(report); + expect(patched?.phase).toBe('cancelled'); + }); + + /** The background run reads the cancel flag and writes progress in two + * separate round trips. Without serialization the progress write lands on a + * job it read before the cancellation, resurrecting `status: 'active'` — + * and the very next `isCancelled` then tells the run to keep importing. */ + it('does not let a progress update in flight during a cancellation resurrect the job', async () => { + const backing = new Keyv(); + const read = backing.get.bind(backing); + /** Only the first read stalls, which is the progress update's: the + * cancellation then reads and writes entirely inside that window, so the + * progress write is left holding a job snapshot taken before it. */ + let stall = true; + jest.spyOn(backing, 'get').mockImplementation(async (key: string) => { + const value = await read(key); + if (stall) { + stall = false; + await new Promise((resolve) => setTimeout(resolve, 20)); + } + return value; + }); + const slowStore = new ImportJobStore(backing, 60000); + const job = await slowStore.create({ + userId: 'u1', + filepath: '/tmp/a.zip', + filename: 'a.zip', + }); + + const progress = slowStore.patch('u1', job.jobId, { + progress: { + conversations: { done: 5, total: 10 }, + messages: { done: 40, total: 80 }, + assets: { done: 0, total: 0 }, + }, + }); + await new Promise((resolve) => setTimeout(resolve, 1)); + await slowStore.cancel('u1', job.jobId); + await progress; + + expect(await slowStore.isCancelled('u1', job.jobId)).toBe(true); + }); + }); + describe('confirmStart', () => { it('returns not_found for a job that does not exist', async () => { expect(await store.confirmStart('u1', 'missing')).toEqual({ status: 'not_found' }); diff --git a/packages/api/src/import/job.ts b/packages/api/src/import/job.ts index 5a2bdde6af..eb18a29ae1 100644 --- a/packages/api/src/import/job.ts +++ b/packages/api/src/import/job.ts @@ -18,14 +18,38 @@ export type StartTransitionResult = | { status: 'not_found' } | { status: 'conflict'; job: ImportJob }; +/** + * Statuses a job never moves out of. Reaching one freezes `status` and + * `phase`: the background run's `isCancelled`/`onPhase`/`onProgress` + * callbacks each read and write separately, so a `DELETE` landing between + * one of those reads and its write would otherwise walk a cancelled job back + * into `phase: 'conversations'` — leaving the client polling forever a job + * whose run has already stopped. Every other field (`report`, `progress`) + * still applies, so a cancelled job still gains the partial report + * describing what was written before it stopped. + */ +const TERMINAL_STATUSES = new Set(['cancelled', 'completed', 'failed']); + +type ImportJobPatch = Omit, 'userId' | 'jobId'>; + +function applyTerminalGuard(existing: ImportJob, patch: ImportJobPatch): ImportJobPatch { + if (!TERMINAL_STATUSES.has(existing.status)) { + return patch; + } + const { status: _status, phase: _phase, ...rest } = patch; + return rest; +} + export class ImportJobStore { private readonly store: Keyv; private readonly ttl: number; - /** Serializes `confirmStart` calls sharing the same job key. `Keyv`'s + /** Serializes every mutation sharing the same job key. `Keyv`'s * `get`/`set` pair is not itself atomic — without this, two racing * `/start` requests can both read `awaiting_confirmation` before either - * write lands, launching the background run twice over the same - * archive. Scoped to this process; see `confirmStart`. */ + * write lands (launching the background run twice over the same archive), + * and a progress update that read an active job can land after a + * cancellation and resurrect it. Scoped to this process; see + * `confirmStart`. */ private readonly transitionLocks = new Map>(); constructor(store: Keyv, ttl: number = DEFAULT_TTL) { @@ -63,21 +87,34 @@ export class ImportJobStore { return job ?? null; } - async patch( + /** The read-modify-write itself, without the lock. Only ever called from + * inside `withTransitionLock`; re-entering the lock here would deadlock + * `confirmStart`, which already holds it. */ + private async applyPatch( userId: string, jobId: string, - patch: Omit, 'userId' | 'jobId'>, + patch: ImportJobPatch, ): Promise { const existing = await this.get(userId, jobId); if (!existing) { return null; } - const updated: ImportJob = { ...existing, ...patch, updatedAt: Date.now() }; + const updated: ImportJob = { + ...existing, + ...applyTerminalGuard(existing, patch), + updatedAt: Date.now(), + }; await this.store.set(this.key(userId, jobId), updated, this.ttl); return updated; } + async patch(userId: string, jobId: string, patch: ImportJobPatch): Promise { + return this.withTransitionLock(this.key(userId, jobId), () => + this.applyPatch(userId, jobId, patch), + ); + } + async cancel(userId: string, jobId: string): Promise { const updated = await this.patch(userId, jobId, { status: 'cancelled', @@ -129,7 +166,7 @@ export class ImportJobStore { if (existing.phase !== 'awaiting_confirmation') { return { status: 'conflict', job: existing }; } - const updated = await this.patch(userId, jobId, { phase: 'queued' }); + const updated = await this.applyPatch(userId, jobId, { phase: 'queued' }); if (!updated) { return { status: 'not_found' }; }