From f061e360928c453b9a3664340bafc1e9930d2da2 Mon Sep 17 00:00:00 2001 From: Lucas Winther Date: Fri, 30 Oct 2026 03:01:05 +0100 Subject: [PATCH] fix(db): jobs.update writes only patched columns; late progress cannot resurrect a finished job Co-Authored-By: Claude Opus 5.5 --- packages/db/src/repos/jobs.ts | 20 ++++++++++++++------ packages/db/test/repos.test.ts | 25 +++++++++++++++++++++++++ 2 files changed, 39 insertions(+), 6 deletions(-) diff --git a/packages/db/src/repos/jobs.ts b/packages/db/src/repos/jobs.ts index 0713e72..3850ef6 100644 --- a/packages/db/src/repos/jobs.ts +++ b/packages/db/src/repos/jobs.ts @@ -44,13 +44,21 @@ export function jobsRepo(db: Db) { } }, + /** + * Single UPDATE writing only the patched columns (plus updated_at). A progress-only patch + * (progress and/or message) is guarded by status = 'running', so a late report on a + * finished job is a no-op. + */ async update(id: string, patch: JobPatch): Promise { - const cur = await get(id); - if (!cur) return; - const next = { ...cur, ...patch }; - await db.sql`UPDATE jobs SET status = ${next.status}, progress = ${next.progress}, message = ${next.message ?? null}, - result_json = ${next.result === null || next.result === undefined ? null : JSON.stringify(next.result)}, - error = ${next.error ?? null}, updated_at = ${new Date().toISOString()} WHERE id = ${id}`; + const set: Record = { updated_at: new Date().toISOString() }; + if (patch.status !== undefined) set.status = patch.status; + if (patch.progress !== undefined) set.progress = patch.progress; + if (patch.message !== undefined) set.message = patch.message; + if (patch.result !== undefined) set.result_json = patch.result === null ? null : JSON.stringify(patch.result); + if (patch.error !== undefined) set.error = patch.error; + const progressOnly = patch.status === undefined && patch.result === undefined && patch.error === undefined; + if (progressOnly) await db.sql`UPDATE jobs SET ${db.sql(set)} WHERE id = ${id} AND status = 'running'`; + else await db.sql`UPDATE jobs SET ${db.sql(set)} WHERE id = ${id}`; }, async list(kind?: string): Promise { diff --git a/packages/db/test/repos.test.ts b/packages/db/test/repos.test.ts index 8169fb9..684b455 100644 --- a/packages/db/test/repos.test.ts +++ b/packages/db/test/repos.test.ts @@ -80,6 +80,7 @@ forEachDb("jobs lifecycle", async (db) => { const id = await j.insert("import", { url: "x" }); const got = await j.get(id); expect(got).toMatchObject({ kind: "import", status: "queued", progress: 0, payload: { url: "x" }, result: null, error: null, message: null }); + await j.claimNext(["import"]); await j.update(id, { progress: 0.5, message: "half" }); await j.update(id, { status: "done", result: { ok: 1 } }); expect(await j.get(id)).toMatchObject({ status: "done", progress: 0.5, message: "half", result: { ok: 1 } }); @@ -123,3 +124,27 @@ forEachDb("two concurrent claimNext calls never claim the same job", async (db) const res = await Promise.all([solo.claimNext(["k"]), solo.claimNext(["k"])]); expect(res.filter((x) => x !== null).map((x) => x!.id)).toEqual([only]); }); + +forEachDb("late progress update does not resurrect a done job", async (db) => { + const j = jobsRepo(db); + const id = await j.insert("k", {}); + await j.claimNext(["k"]); + await j.update(id, { status: "done", progress: 1, result: { ok: true } }); + await j.update(id, { progress: 0.4, message: "stale" }); + expect(await j.get(id)).toMatchObject({ status: "done", progress: 1, message: null, result: { ok: true } }); + const f = await j.insert("k", {}); + await j.update(f, { status: "failed", error: "boom" }); + await j.update(f, { progress: 0.9 }); + expect(await j.get(f)).toMatchObject({ status: "failed", error: "boom", progress: 0 }); +}); + +forEachDb("update only touches patched columns", async (db) => { + const j = jobsRepo(db); + const id = await j.insert("k", { a: 1 }); + await j.claimNext(["k"]); + await j.update(id, { progress: 0.5, message: "working" }); + await j.update(id, { message: "still" }); + expect(await j.get(id)).toMatchObject({ status: "running", progress: 0.5, message: "still", result: null, error: null, payload: { a: 1 } }); + await j.update(id, { result: { r: 1 } }); + expect(await j.get(id)).toMatchObject({ status: "running", progress: 0.5, message: "still", result: { r: 1 } }); +});