Fix BullMQ removal processor arguments (#783)

src/queue/index.ts

@@ -149,6 +149,23 @@ export async function addRemovalJob(
 export async function recoverStuckRemoving() {
   if (!removeQueue) return;
   try {
+    // A failed job is durable proof that removal was requested. Retry it even
+    // if the final failure handler already changed the repository to ERROR.
+    const failedJobs = await removeQueue.getJobs(["failed"]);
+    for (const job of failedJobs) {
+      const repoId = job.data?.repoId;
+      if (!repoId) continue;
+      try {
+        await addRemovalJob(repoId);
+        logger.info("requeued failed removal", { repoId });
+      } catch (e) {
+        logger.warn("failed removal recovery failed", {
+          ...serializeError(e),
+          repoId,
+        });
+      }
+    }
+
     const stuck = await AnonymizedRepositoryModel.find(
       { status: RepositoryStatus.REMOVING },
       { repoId: 1 }

src/queue/processes/removeCache.ts

@@ -31,4 +31,15 @@ export async function processRemoveCache(
   }
 }
 
-export default processRemoveCache;
+/**
+ * BullMQ passes its lock token as the second processor argument. Do not let
+ * that token occupy the database-injection slot used by processRemoveCache.
+ */
+export function createRemoveCacheProcessor(database?: Database) {
+  return async (
+    job: SandboxedJob<RepoJobData, void>,
+    _lockToken?: string
+  ) => processRemoveCache(job, database);
+}
+
+export default createRemoveCacheProcessor();

src/queue/processes/removeRepository.ts

@@ -50,4 +50,15 @@ export async function processRemoveRepository(
   }
 }
 
-export default processRemoveRepository;
+/**
+ * BullMQ calls sandbox processors with (job, lockToken). Keep that transport
+ * signature separate from the injectable worker function used by tests.
+ */
+export function createRemoveRepositoryProcessor(database?: Database) {
+  return async (
+    job: SandboxedJob<RepoJobData, void>,
+    _lockToken?: string
+  ) => processRemoveRepository(job, database);
+}
+
+export default createRemoveRepositoryProcessor();

test/backend-reliability.test.js

@@ -12,9 +12,11 @@ const {
 } = require("../src/server/routes/repository-private");
 const {
   processRemoveRepository,
+  createRemoveRepositoryProcessor,
 } = require("../src/queue/processes/removeRepository");
 const {
   processRemoveCache,
+  createRemoveCacheProcessor,
 } = require("../src/queue/processes/removeCache");
 const { addRemovalJob } = require("../src/queue");
 const {
@@ -181,6 +183,22 @@ describe("removal workers", function () {
     expect(statuses[statuses.length - 1][1]).to.equal(failure.message);
   });
 
+  it("does not treat the BullMQ lock token as a database dependency", async function () {
+    const statuses = [];
+    const database = {
+      connect: async () => undefined,
+      getRepository: async () => ({
+        updateStatus: async (status) => statuses.push(status),
+        remove: async () => undefined,
+      }),
+    };
+    const processor = createRemoveRepositoryProcessor(database);
+
+    await processor(job, "bullmq-lock-token");
+
+    expect(statuses).to.include("removing");
+  });
+
   it("does not add a duplicate when a removal job is live", async function () {
     let additions = 0;
     const queue = {
@@ -238,6 +256,22 @@ describe("removal workers", function () {
     }
     expect(caught).to.equal(failure);
   });
+
+  it("keeps the BullMQ lock token out of cache worker injection", async function () {
+    let removed = false;
+    const processor = createRemoveCacheProcessor({
+      connect: async () => undefined,
+      getRepository: async () => ({
+        removeCache: async () => {
+          removed = true;
+        },
+      }),
+    });
+
+    await processor(job, "bullmq-lock-token");
+
+    expect(removed).to.equal(true);
+  });
 });
 
 describe("repository expiration maintenance", function () {