fix(bullmq): apply configurable job retention to prevent unbounded Redis growth

src/lib/backends/bullmq.ts

@@ -15,7 +15,12 @@
  * @module
  */
 
-import { type Job as BullMQJob, Queue, Worker } from "bullmq";
+import {
+  type DefaultJobOptions,
+  type Job as BullMQJob,
+  Queue,
+  Worker,
+} from "bullmq";
 import type {
   BackendAdapter,
   EnqueueOptions,
@@ -34,15 +39,42 @@ export interface BullMQBackendOptions {
   };
   defaultQueueName?: string;
   concurrency?: number;
+  /**
+   * Default options applied to every enqueued and recurring job, e.g. retention
+   * (`removeOnComplete`/`removeOnFail`), attempts and backoff.
+   *
+   * BullMQ keeps completed and failed jobs in Redis indefinitely by default, so
+   * recurring jobs pile up one record per run forever. To avoid unbounded
+   * growth this backend applies bounded retention by default (see
+   * {@link DEFAULT_JOB_OPTIONS}); whatever you pass here is merged on top and
+   * takes precedence.
+   */
+  defaultJobOptions?: DefaultJobOptions;
 }
 
+/**
+ * Bounded retention applied when the caller does not override it, so Redis does
+ * not grow without limit. Keeps a recent window of completed/failed jobs for
+ * inspection while discarding the rest. Override via
+ * {@link BullMQBackendOptions.defaultJobOptions}.
+ */
+const DEFAULT_JOB_OPTIONS: DefaultJobOptions = {
+  removeOnComplete: { count: 1000 },
+  removeOnFail: { count: 5000 },
+};
+
 class TBullMQBackend implements BackendAdapter {
   private queues: Map<string, Queue> = new Map();
   private workers: Map<string, Worker> = new Map();
   private options: BullMQBackendOptions;
+  private jobOptions: DefaultJobOptions;
 
   constructor(options: BullMQBackendOptions) {
     this.options = options;
+    this.jobOptions = {
+      ...DEFAULT_JOB_OPTIONS,
+      ...options.defaultJobOptions,
+    };
   }
 
   private getOrCreateQueue(queueName: string): Queue {
@@ -51,6 +83,7 @@ class TBullMQBackend implements BackendAdapter {
         queueName,
         new Queue(queueName, {
           connection: this.options.connection,
+          defaultJobOptions: this.jobOptions,
         }),
       );
     }
@@ -107,9 +140,13 @@ class TBullMQBackend implements BackendAdapter {
       jobBody: config.jobBody,
     };
 
+    // Pass retention on the scheduler's job template too: jobs produced by a
+    // scheduler do not inherit the queue's defaultJobOptions, so without this
+    // each recurring run would leave a job record in Redis forever.
     await queue.upsertJobScheduler(schedulerId, repeatOpts, {
       name: config.jobName,
       data: payload,
+      opts: this.jobOptions,
     });
   }
 

src/lib/tests/backends/bullmq_test.ts

@@ -207,7 +207,8 @@ Deno.test({
       twoExecutions,
       new Promise<void>((_, reject) => {
         timeoutId = setTimeout(
-          () => reject(new Error("Timeout waiting for recurring job executions")),
+          () =>
+            reject(new Error("Timeout waiting for recurring job executions")),
           10000,
         ) as unknown as number;
       }),
@@ -300,3 +301,62 @@ Deno.test({
     await backend.close();
   },
 });
+
+Deno.test({
+  name:
+    "BullMQBackend: applies defaultJobOptions retention (no completed pile-up)",
+  ignore: !redisAvailable,
+  sanitizeOps: false,
+  sanitizeResources: false,
+  async fn() {
+    const { BullMQBackend } = await import("../../backends/bullmq.ts");
+    const { Queue } = await import("bullmq");
+    const queueName = `test_bullmq_${testRunId}_retention`;
+
+    // removeOnComplete: true removes each job as soon as it completes, so a
+    // healthy backend leaves zero completed jobs behind. Without the backend
+    // wiring defaultJobOptions through, completed jobs would accumulate.
+    const backend = BullMQBackend({
+      connection: { host: "localhost", port: 6379 },
+      defaultJobOptions: { removeOnComplete: true, removeOnFail: true },
+    });
+
+    let resolveProcessed: () => void;
+    const processed = new Promise<void>((r) => resolveProcessed = r);
+    await backend.listen(async (_payload: JobPayload) => {
+      resolveProcessed();
+      await Promise.resolve();
+    }, { queueNames: [queueName] });
+
+    await backend.enqueue({
+      jobName: "retention_job",
+      queueName,
+      jobBody: { test: true },
+    });
+
+    let timeoutId: number | undefined;
+    await Promise.race([
+      processed,
+      new Promise<void>((_, reject) => {
+        timeoutId = setTimeout(
+          () => reject(new Error("Timeout waiting for retention job")),
+          10000,
+        ) as unknown as number;
+      }),
+    ]);
+    clearTimeout(timeoutId);
+
+    // Let BullMQ move the job to completed and remove it.
+    await new Promise((r) => setTimeout(r, 500));
+
+    const inspector = new Queue(queueName, {
+      connection: { host: "localhost", port: 6379 },
+    });
+    const completedCount = await inspector.getCompletedCount();
+    await inspector.close();
+
+    assertEquals(completedCount, 0);
+
+    await backend.close();
+  },
+});