fix(db): jobs.update writes only patched columns; late progress cannot resurrect a finished job

Co-Authored-By: Claude Opus 5.5 <[email protected]>

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<void> {
-      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<string, unknown> = { 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<JobRecord[]> {

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 } });
+});