feat: add repeating jobs with external cancellation support

Author: RomainLanz

README.md

@@ -29,6 +29,7 @@ npm install @boringnode/queue
 - **Priority Queues**: Process high-priority jobs first
 - **Retry with Backoff**: Automatic retries with exponential, linear, or fixed backoff strategies
 - **Job Timeout**: Automatically fail or retry jobs that exceed a time limit
+- **Repeating Jobs**: Schedule jobs to repeat at fixed intervals
 
 ## Quick Start
 
@@ -51,10 +52,6 @@ export default class SendEmailJob extends Job<SendEmailPayload> {
     queue: 'email',
   }
 
-  constructor(payload: SendEmailPayload, context: JobContext) {
-    super(payload, context)
-  }
-
   async execute(): Promise<void> {
     console.log(`[Attempt ${this.context.attempt}] Sending email to: ${this.payload.to}`)
   }
@@ -135,7 +132,7 @@ interface QueueManagerConfig {
   // Worker configuration
   worker: {
     concurrency: number
-    pollingInterval: string
+    idleDelay: Duration
   }
 
   // Job discovery locations
@@ -242,9 +239,9 @@ Schedule jobs to run in the future:
 ```typescript
 // Various time formats
 await SendEmailJob.dispatch(payload).in('30s') // 30 seconds
-await SendEmailJob.dispatch(payload).in('5m') // 5 minutes
-await SendEmailJob.dispatch(payload).in('2h') // 2 hours
-await SendEmailJob.dispatch(payload).in('1d') // 1 day
+await SendEmailJob.dispatch(payload).in('5m')  // 5 minutes
+await SendEmailJob.dispatch(payload).in('2h')  // 2 hours
+await SendEmailJob.dispatch(payload).in('1d')  // 1 day
 ```
 
 ## Priority
@@ -358,15 +355,18 @@ export default class MyJob extends Job<Payload> {
 
 ### Context Properties
 
-| Property       | Type   | Description                                     |
-| -------------- | ------ | ----------------------------------------------- |
-| `jobId`        | string | Unique identifier for this job                  |
-| `name`         | string | Job class name                                  |
-| `attempt`      | number | Current attempt number (1-based)                |
-| `queue`        | string | Queue name this job is being processed from     |
-| `priority`     | number | Job priority (lower = higher priority)          |
-| `acquiredAt`   | Date   | When this job was acquired by the worker        |
-| `stalledCount` | number | Times this job was recovered from stalled state |
+| Property          | Type                | Description                                       |
+|-------------------|---------------------|---------------------------------------------------|
+| `jobId`           | string              | Unique identifier for this job                    |
+| `name`            | string              | Job class name                                    |
+| `attempt`         | number              | Current attempt number (1-based)                  |
+| `queue`           | string              | Queue name this job is being processed from       |
+| `priority`        | number              | Job priority (lower = higher priority)            |
+| `acquiredAt`      | Date                | When this job was acquired by the worker          |
+| `stalledCount`    | number              | Times this job was recovered from stalled state   |
+| `isRepeating`     | boolean             | Whether this job is configured to repeat          |
+| `repeatRemaining` | number \| undefined | Remaining repetitions (undefined = infinite)      |
+| `repeatId`        | string \| undefined | Unique ID for the repeat chain (for cancellation) |
 
 ## Dependency Injection
 
@@ -417,6 +417,88 @@ export default class SendEmailJob extends Job<SendEmailPayload> {
 
 Without a `jobFactory`, jobs are instantiated with `new JobClass(payload, context)`.
 
+## Repeating Jobs
+
+Schedule jobs to repeat automatically at fixed intervals:
+
+```typescript
+// Repeat every 5 seconds indefinitely
+await SyncJob.dispatch({ source: 'api' }).every('5s')
+
+// Repeat every hour, 10 times total
+await CleanupJob.dispatch({ days: 30 }).every('1h').times(10)
+
+// Combine with delay (start after 30 seconds, then repeat every minute)
+await ReportJob.dispatch({ type: 'daily' }).in('30s').every('1m')
+```
+
+### Cancelling a Repeating Job
+
+When dispatching a repeating job, you receive a `repeatId` that can be used to cancel the entire repeat chain from anywhere:
+
+```typescript
+import { QueueManager } from '@boringnode/queue'
+
+// Dispatch returns jobId and repeatId
+const { jobId, repeatId } = await SyncJob.dispatch({ source: 'api' }).every('5s')
+
+console.log(`Started repeating job ${jobId} with repeat chain ${repeatId}`)
+
+// Later, cancel the repeat chain from anywhere
+if (repeatId) {
+  await QueueManager.cancelRepeat(repeatId)
+}
+```
+
+The `repeatId` is also available inside the job via `this.context.repeatId`.
+
+### Stopping from Within the Job
+
+A job can stop its own repetition by calling `this.stopRepeating()`:
+
+```typescript
+import { Job } from '@boringnode/queue'
+import type { JobContext } from '@boringnode/queue/types'
+
+export default class SyncJob extends Job<SyncPayload> {
+  static readonly jobName = 'SyncJob'
+
+  async execute(): Promise<void> {
+    const result = await this.syncData()
+
+    // Stop repeating when sync is complete
+    if (result.isComplete) {
+      this.stopRepeating()
+    }
+  }
+}
+```
+
+### Repeat Context
+
+Jobs have access to repeat information via `this.context`:
+
+```typescript
+async execute(): Promise<void> {
+  if (this.context.isRepeating) {
+    console.log(`Repeating job, ${this.context.repeatRemaining ?? 'infinite'} runs remaining`)
+  }
+}
+```
+
+| Property          | Type                | Description                                       |
+|-------------------|---------------------|---------------------------------------------------|
+| `isRepeating`     | boolean             | Whether this job is configured to repeat          |
+| `repeatRemaining` | number \| undefined | Remaining repetitions (undefined = infinite)      |
+| `repeatId`        | string \| undefined | Unique ID for the repeat chain (for cancellation) |
+
+### How Repeating Works
+
+- Each repeat creates a **new job** with a new ID
+- The payload is **preserved** across repeats
+- Failed jobs do **not** repeat (only successful completions trigger the next run)
+- The repeat interval is the delay **between** job completions
+
 ## Job Discovery
 
 The queue manager automatically discovers and registers jobs from the specified locations:
@@ -459,7 +541,7 @@ By default, a simple console logger is used that only outputs warnings and error
 Performance comparison with BullMQ using realistic jobs (5ms simulated work per job):
 
 | Jobs | Concurrency | @boringnode/queue | BullMQ | Diff        |
-| ---- | ----------- | ----------------- | ------ | ----------- |
+|------|-------------|-------------------|--------|-------------|
 | 100  | 1           | 562ms             | 596ms  | 5.7% faster |
 | 100  | 5           | 116ms             | 117ms  | ~same       |
 | 100  | 10          | 62ms              | 62ms   | ~same       |
@@ -495,4 +577,4 @@ npm run benchmark -- --duration=10
 [typescript-image]: https://img.shields.io/badge/Typescript-294E80.svg?style=for-the-badge&logo=typescript
 [typescript-url]: https://www.typescriptlang.org
 [license-image]: https://img.shields.io/npm/l/@boringnode/queue?color=blueviolet&style=for-the-badge
-[license-url]: LICENSE.md
+[license-url]: LICENSE.md

examples/app.ts

@@ -2,6 +2,7 @@ import { config } from './config.js'
 import { QueueManager } from '../src/queue_manager.js'
 import SendEmailJob from './jobs/send_email_job.js'
 import SyncJob from './jobs/sync_job.js'
+import MetricsJob from './jobs/metrics_job.js'
 
 await QueueManager.init(config)
 
@@ -13,4 +14,13 @@ for (let i = 0; i < 10; i++) {
   await SendEmailJob.dispatch({ to: 'romain.lanz@pm.me' + i })
 }
 
+// Example: Dispatch a repeating job and get the repeatId for later cancellation
+const { jobId, repeatId } = await MetricsJob.dispatch({ endpoint: '/api/health' }).every('10s')
+
+console.log(`Started metrics collection job ${jobId}`)
+console.log(`To cancel this repeating job, use: await QueueManager.cancelRepeat('${repeatId}')`)
+
+// Example: Cancel a repeating job after some condition
+// await QueueManager.cancelRepeat(repeatId)
+
 await QueueManager.destroy()

examples/jobs/metrics_job.ts

@@ -0,0 +1,37 @@
+import { Job } from '../../src/job.js'
+import type { JobOptions } from '../../src/types/index.js'
+
+interface MetricsJobPayload {
+  endpoint: string
+}
+
+/**
+ * Example job that collects metrics at regular intervals.
+ * Demonstrates repeating jobs with external cancellation.
+ */
+export default class MetricsJob extends Job<MetricsJobPayload> {
+  static readonly jobName = 'MetricsJob'
+
+  static options: JobOptions = {
+    queue: 'metrics',
+  }
+
+  async execute(): Promise<void> {
+    const repeatInfo = this.context.isRepeating
+      ? ` (repeat ${this.context.repeatRemaining ?? '∞'} remaining, repeatId: ${this.context.repeatId})`
+      : ''
+
+    console.log(
+      `[Job ${this.context.jobId}] Collecting metrics from ${this.payload.endpoint}${repeatInfo}`
+    )
+
+    // Simulate metrics collection
+    const metrics = {
+      timestamp: new Date().toISOString(),
+      cpu: Math.random() * 100,
+      memory: Math.random() * 100,
+    }
+
+    console.log(`[Job ${this.context.jobId}] Metrics:`, metrics)
+  }
+}

examples/jobs/send_email_job.ts

@@ -1,4 +1,3 @@
-import { setTimeout } from 'node:timers/promises'
 import { Job } from '../../src/job.js'
 import type { JobOptions } from '../../src/types/index.js'
 
@@ -14,7 +13,6 @@ export default class SendEmailJob extends Job<SendEmailPayload> {
   }
 
   async execute(): Promise<void> {
-    await setTimeout(1000)
     console.log(`[Attempt ${this.context.attempt}] Sending email to: ${this.payload.to}`)
   }
 }

examples/worker.ts

@@ -2,4 +2,4 @@ import { Worker } from '../src/worker.js'
 import { config } from './config.js'
 
 const worker = new Worker(config)
-await worker.start(['default', 'email', 'reports'])
+await worker.start(['default', 'email', 'reports', 'metrics'])

src/contracts/adapter.ts

@@ -149,4 +149,22 @@ export interface Adapter {
    * Called when the worker stops or the adapter is no longer needed.
    */
   destroy(): Promise<void>
+
+  /**
+   * Cancel a repeating job chain.
+   *
+   * After calling this, `isRepeatCancelled` will return true for this groupId,
+   * and the worker will not re-dispatch jobs with this groupId.
+   *
+   * @param groupId - The repeat chain identifier (from RepeatConfig.groupId)
+   */
+  cancelRepeat(groupId: string): Promise<void>
+
+  /**
+   * Check if a repeat chain has been cancelled.
+   *
+   * @param groupId - The repeat chain identifier to check
+   * @returns True if the repeat chain has been cancelled
+   */
+  isRepeatCancelled(groupId: string): Promise<boolean>
 }