diff --git a/src/lib/backends/bullmq.ts b/src/lib/backends/bullmq.ts index 93206eb..17b96a2 100644 --- a/src/lib/backends/bullmq.ts +++ b/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 = new Map(); private workers: Map = 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, }); } diff --git a/src/lib/tests/backends/bullmq_test.ts b/src/lib/tests/backends/bullmq_test.ts index 0d5aa9f..588ba40 100644 --- a/src/lib/tests/backends/bullmq_test.ts +++ b/src/lib/tests/backends/bullmq_test.ts @@ -207,7 +207,8 @@ Deno.test({ twoExecutions, new Promise((_, 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((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((_, 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(); + }, +});