feat: add support for job Signal

close #88

Author: Julien-R44

docs/docs/pages/guides/creating-jobs.mdx

@@ -98,11 +98,34 @@ export default class SendEmailJob extends Job<SendEmailJobData, SendEmailJobRetu
     // Logger instance
     this.logger.info('Processing email job', { to: emailData.to })
 
+    // AbortSignal - aborted when the worker is shutting down or the job is cancelled
+    const signal = this.signal
+
     return { messageId: 'abc123', success: true }
   }
 }
 ```
 
+### Abort Signal
+
+When the worker shuts down gracefully or a job is explicitly cancelled, BullMQ aborts the `AbortSignal` available via `this.signal`. You can use it to stop long-running work early:
+
+```typescript
+export default class CrawlPagesJob extends Job<CrawlData, void> {
+  async process(): Promise<void> {
+    // Pass it directly to fetch or any AbortSignal-aware API
+    const response = await fetch(this.data.url, { signal: this.signal })
+
+    // Or check it manually in a loop
+    for (const page of this.data.pages) {
+      if (this.signal?.aborted) break
+
+      await this.crawlPage(page)
+    }
+  }
+}
+```
+
 ## Default Queue
 
 Set a default queue for a job class:

packages/core/src/job/base_job.ts

@@ -58,6 +58,7 @@ export abstract class BaseJob<DataType, ReturnType> {
   worker!: BullWorker<DataType, ReturnType>
   logger!: Logger
   token?: string
+  signal?: AbortSignal
   error?: Error
 
   abstract process(...args: any[]): Promise<ReturnType>
@@ -131,6 +132,7 @@ export abstract class BaseJob<DataType, ReturnType> {
     job: BullJob<DataType, ReturnType>,
     token: string | undefined,
     logger: Logger,
+    signal: AbortSignal | undefined,
   ) {
     if (jobClass.encrypted) {
       job.data = jobClass.decrypt(job.data as string)
@@ -140,6 +142,7 @@ export abstract class BaseJob<DataType, ReturnType> {
     this.job = job
     this.data = job.data
     this.token = token
+    this.signal = signal
     this.logger = logger.child({ jobName: job.name, jobId: job.id })
   }
 

packages/core/src/worker/worker.ts

@@ -46,7 +46,7 @@ export class Worker<KnownQueues extends Record<string, QueueConfig> = Queues> {
 
     return BullMqFactory.createWorker(
       String(this.#queueName),
-      async (job, token) => this.#processJob(job, token),
+      async (job, token, signal) => this.#processJob(job, token, signal),
       {
         autorun: false,
         connection,
@@ -105,8 +105,13 @@ export class Worker<KnownQueues extends Record<string, QueueConfig> = Queues> {
   /**
    * Create a job instance with all dependencies configured
    */
-  async #createJobInstance(options: { job: BullJob; token?: string; error?: Error }) {
-    const { job, token, error } = options
+  async #createJobInstance(options: {
+    job: BullJob
+    token?: string
+    signal?: AbortSignal
+    error?: Error
+  }) {
+    const { job, token, signal, error } = options
 
     const JobClass = this.#getJobClass(job.name)
     const adonisLogger = this.#createTracedLogger()
@@ -116,7 +121,7 @@ export class Worker<KnownQueues extends Record<string, QueueConfig> = Queues> {
     resolver.bindValue(Logger, jobLogger)
 
     const instance = await resolver.make(JobClass)
-    instance.$init(this.#bullWorker!, JobClass, job, token, jobLogger)
+    instance.$init(this.#bullWorker!, JobClass, job, token, jobLogger, signal)
 
     if (error) instance.$setError(error)
 
@@ -138,8 +143,12 @@ export class Worker<KnownQueues extends Record<string, QueueConfig> = Queues> {
   /**
    * Start processing a job
    */
-  async #processJob(job: BullJob, token?: string) {
-    const { instance: jobInstance, resolver } = await this.#createJobInstance({ job, token })
+  async #processJob(job: BullJob, token?: string, signal?: AbortSignal) {
+    const { instance: jobInstance, resolver } = await this.#createJobInstance({
+      job,
+      token,
+      signal,
+    })
 
     this.#setDefaultSpanAttributes(jobInstance)