fix(worker): publish repo-index job pointer in onStarted before superseding

trackLatestJob removes a resource's previous job right after onStarted,
which assumes the parent's latest...JobId pointer has already moved.
repo-index wrote latestIndexingJobId inside process instead, so an abort
or DB error before that write could leave the pointer on a removed job.

- Move the pointer write to an onStarted hook (updateMany keyed by repo
  id), matching the other latestPerResource workloads.
- Reject a latestPerResource workload without onStarted at registration
  so the contract cannot be silently skipped by a future workload.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

Author: brendan-kellam

packages/backend/src/jobManager.test.ts

@@ -157,6 +157,27 @@ describe("BullMQJobManager lifecycle", () => {
         );
     });
 
+    test("rejects a latestPerResource workload that does not publish its job id in onStarted", () => {
+        const manager = new BullMQJobManager({} as Redis);
+        const workload = createWorkload();
+        workload.queueSpec.jobOptions = {
+            ...workload.queueSpec.jobOptions,
+            retention: { mode: "latestPerResource", maxAgeSeconds: 60 },
+        };
+
+        expect(() => manager.register(workload)).toThrow(
+            /must publish its job id to the parent resource in onStarted/,
+        );
+        expect(() =>
+            manager.register(
+                createWorkload({
+                    queueSpec: workload.queueSpec,
+                    onStarted: vi.fn(),
+                }),
+            ),
+        ).not.toThrow();
+    });
+
     test("delegates enqueueing to BullMQClient and returns its job id", async () => {
         const manager = new BullMQJobManager({} as Redis);
         const workload = createWorkload();

packages/backend/src/jobManager.ts

@@ -42,6 +42,14 @@ export class BullMQJobManager implements JobManager {
         if (this.workloads.has(name)) {
             throw new Error(`Workload "${name}" is already registered`);
         }
+        if (
+            workload.queueSpec.jobOptions.retention.mode === "latestPerResource"
+            && !workload.onStarted
+        ) {
+            throw new Error(
+                `Workload "${name}" uses latestPerResource retention and must publish its job id to the parent resource in onStarted`,
+            );
+        }
         this.workloads.set(name, workload);
     }
 
@@ -151,8 +159,10 @@ export class BullMQJobManager implements JobManager {
 
                 const process = async (signal: AbortSignal) => {
                     await workload.onStarted?.(lifecycleContext);
-                    // After onStarted so the parent's `latest...JobId` pointer has
-                    // moved to this job before its predecessor is removed.
+                    // Contract: `latestPerResource` workloads publish this job's id to
+                    // their parent's `latest...JobId` pointer in onStarted (enforced in
+                    // `register`), so the pointer has moved before the superseded job
+                    // is removed.
                     await this.trackLatestJob(spec, job);
                     return workload.process({
                         ...lifecycleContext,

packages/backend/src/repoIndexWorkload.test.ts

@@ -19,18 +19,10 @@ const repoFindUnique = vi.fn();
 const repoUpdate = vi.fn();
 const repoUpdateMany = vi.fn();
 
-const transaction = vi.fn(async (callback: (tx: unknown) => Promise<unknown>) =>
-    callback({
-        repo: {
-            findUnique: repoFindUnique,
-            update: repoUpdate,
-        },
-    }),
-);
-
 const db = {
-    $transaction: transaction,
     repo: {
+        findUnique: repoFindUnique,
+        update: repoUpdate,
         updateMany: repoUpdateMany,
     },
 } as unknown as PrismaClient;
@@ -121,10 +113,20 @@ describe("repoIndexWorkload", () => {
         });
     });
 
+    test("publishes the job as the repository's latest indexing job when it starts", async () => {
+        await workload.onStarted!(lifecycleContext);
+
+        expect(repoUpdateMany).toHaveBeenCalledWith({
+            where: { id: 42 },
+            data: { latestIndexingJobId: "job-1" },
+        });
+    });
+
     test("skips an INDEX job when the repository no longer exists", async () => {
         await workload.process(processContext);
 
         expect(repoUpdate).not.toHaveBeenCalled();
+        expect(repoUpdateMany).not.toHaveBeenCalled();
         expect(lifecycleLogger.debug).toHaveBeenCalledWith(
             "Skipping INDEX job for repo 42: repository no longer exists",
         );

packages/backend/src/repoIndexWorkload.ts

@@ -25,13 +25,22 @@ export const createRepoIndexWorkload = ({
     queueSpec: REPO_INDEX_QUEUE,
     concurrency: settings.maxRepoIndexingJobConcurrency,
     executionLock: REPOSITORY_EXECUTION_LOCK,
-    process: async ({ data, jobId, signal }) => {
+    onStarted: async ({ data: { repoId }, jobId }) => {
+        await db.repo.updateMany({
+            where: {
+                id: repoId,
+            },
+            data: {
+                latestIndexingJobId: jobId,
+            },
+        });
+    },
+    process: async ({ data, signal }) => {
         signal.throwIfAborted();
 
         const start = await prepareRepoIndexJob({
             db,
             repoId: data.repoId,
-            jobId,
         });
 
         if (start.action === "skip") {
@@ -119,45 +128,33 @@ type RepoIndexStartDecision =
 const prepareRepoIndexJob = async ({
     db,
     repoId,
-    jobId,
 }: {
     db: PrismaClient;
     repoId: number;
-    jobId: string;
-}): Promise<RepoIndexStartDecision> =>
-    db.$transaction(async (tx) => {
-        const repo = await tx.repo.findUnique({
-            where: { id: repoId },
-            include: {
-                connections: {
-                    include: {
-                        connection: true,
-                    },
+}): Promise<RepoIndexStartDecision> => {
+    const repo = await db.repo.findUnique({
+        where: { id: repoId },
+        include: {
+            connections: {
+                include: {
+                    connection: true,
                 },
             },
-        });
-
-        if (!repo) {
-            return {
-                action: "skip",
-                reason: "repository no longer exists",
-            };
-        }
-
-        await tx.repo.update({
-            where: {
-                id: repoId,
-            },
-            data: {
-                latestIndexingJobId: jobId,
-            },
-        });
+        },
+    });
 
+    if (!repo) {
         return {
-            action: "run",
-            repo,
+            action: "skip",
+            reason: "repository no longer exists",
         };
-    });
+    }
+
+    return {
+        action: "run",
+        repo,
+    };
+};
 
 const indexRepository = async (
     db: PrismaClient,

packages/shared/src/queue.ts

@@ -48,6 +48,9 @@ export type JobRetention =
     | {
         // Keeps only the most recent job for each resource, keyed by the queue's
         // deduplication id. When a job starts, the job it supersedes is removed.
+        // Workloads on these queues must publish the job id to their parent's
+        // `latest...JobId` pointer in `onStarted`, so the pointer never
+        // references a removed job.
         // `maxAgeSeconds` is an age-only backstop that reclaims jobs whose
         // resource stopped running (e.g. a deleted repo). It must exceed the
         // longest scheduler interval, or a resource's latest job can be