fix(queue): handle DLQ jobs in cancellation and add monitor tests

src/queue/managedQueue.ts

@@ -350,6 +350,31 @@ export class ManagedQueue<T = unknown, R = unknown> implements ManagedQueuePort<
         return false;
       }
 
+      // If the job is in the DLQ (failed state from stall/max-attempts), removeAsync
+      // cannot handle it. Pause the queue to prevent the worker from picking up the
+      // job, retry it (moves from DLQ to waiting), then remove and resume.
+      if (stateBefore === "failed") {
+        this.queue.pause();
+        try {
+          await this.queue.retryJob(jobId);
+          await this.queue.removeAsync(jobId);
+        } finally {
+          this.queue.resume();
+        }
+        const stateAfter = await this.queue.getJobState(jobId);
+        if (stateAfter === "unknown") {
+          logger.info(`Removed DLQ job ${jobId} from "${this.queue.name}"`);
+          try {
+            getLogStore().deleteMany([jobId]);
+          } catch {
+            logger.warn(`Failed to clean persisted logs for DLQ job ${jobId}`);
+          }
+          return true;
+        }
+        logger.warn(`Job ${jobId} still present after retry+remove in "${this.queue.name}" (state: ${stateAfter})`);
+        return false;
+      }
+
       // removeAsync handles queue, waitingDeps, and waitingChildren states.
       await this.queue.removeAsync(jobId);
 

src/web/jobCanceller.ts

@@ -136,7 +136,10 @@ export class JobCanceller {
     if (chainResult) {
       jobsToDelete = chainResult.siblings.map((job) => ({ id: job.id, queueName: job.queue }));
     } else {
-      jobsToDelete = [{ id: jobId, queueName: "" }];
+      // Not part of a chain — use the cached entry's queue name for targeted removal.
+      const cachedEntry = this.deps.getCachedJob(jobId);
+      const queueName = cachedEntry?.queue ?? "";
+      jobsToDelete = [{ id: jobId, queueName }];
     }
 
     let overallStatus = true;
@@ -151,14 +154,20 @@ export class JobCanceller {
           if (removed) {
             this.deps.evictJob(jobToDelete.id);
             this.markCancelled(jobToDelete.id);
+          } else if (job.state === "failed" || job.state === "completed") {
+            // Job is in a terminal state (e.g. DLQ) that cancelJob can't remove.
+            // Evict from cache so it doesn't linger as a stale entry.
+            this.deps.evictJob(jobToDelete.id);
           } else {
             overallStatus = false;
           }
         } else {
-          overallStatus = false;
+          // Job no longer exists in the queue — evict the stale cache entry.
+          this.deps.evictJob(jobToDelete.id);
         }
       } else {
-        overallStatus = false;
+        // Queue not found — evict the orphaned cache entry.
+        this.deps.evictJob(jobToDelete.id);
       }
     }