fix(sync): honor retries and failed hook with shared job runtime

Author: RomainLanz

README.md

@@ -292,6 +292,17 @@ import { sync } from '@boringnode/queue/drivers/sync_adapter'
 const adapter = sync() // Jobs execute immediately
 ```
 
+Use the `sync` adapter for tests and lightweight local development only.
+
+- `await MyJob.dispatch(payload).run()` waits for the job to fully finish.
+- Retries are executed inline, not by a background worker.
+- If you configure backoff, the adapter will `sleep` between attempts.
+- This means the caller can stay blocked for the full retry duration.
+
+Example: with `maxRetries: 3` and an exponential backoff of `1s`, `2s`, `4s`,
+the request or command that dispatched the job can stay busy for about 7 seconds
+before the job exhausts its retries and runs `failed()`.
+
 ## Job Options
 
 ```typescript
@@ -338,6 +349,14 @@ export default class ReliableJob extends Job<Payload> {
 }
 ```
 
+`maxRetries` can be defined directly on the job options, and `retry.backoff`
+controls the delay between attempts.
+
+> With the `sync` adapter, these delays happen inline in the caller via
+> `sleep`. If a job fails repeatedly, `dispatch().run()` will take as long as
+> the total backoff duration. Use a worker-backed adapter when you do not want
+> retries to slow down the request/command that dispatched the job.
+
 <details>
 <summary><strong>Available strategies</strong></summary>
 

src/drivers/sync_adapter.ts

@@ -1,5 +1,7 @@
+import { setTimeout as sleep } from 'node:timers/promises'
 import { Locator } from '../locator.js'
 import { QueueManager } from '../queue_manager.js'
+import { JobExecutionRuntime } from '../job_runtime.js'
 import type { Adapter, AcquiredJob } from '../contracts/adapter.js'
 import type {
   JobContext,
@@ -30,7 +32,7 @@ export class SyncAdapter implements Adapter {
   }
 
   pushOn(queue: string, jobData: JobData): Promise<void> {
-    return this.#execute(jobData.name, jobData.payload, queue)
+    return this.#execute(jobData, queue)
   }
 
   pushLater(jobData: JobData, delay: number): Promise<void> {
@@ -39,7 +41,7 @@ export class SyncAdapter implements Adapter {
 
   pushLaterOn(queue: string, jobData: JobData, delay: number): Promise<void> {
     setTimeout(() => {
-      void this.#execute(jobData.name, jobData.payload, queue)
+      void this.#execute(jobData, queue)
     }, delay)
 
     return Promise.resolve()
@@ -142,27 +144,58 @@ export class SyncAdapter implements Adapter {
     return Promise.resolve(null)
   }
 
-  async #execute(jobName: string, payload: unknown, queue: string = 'default'): Promise<void> {
-    const JobClass = Locator.get(jobName)
+  async #execute(jobData: JobData, queue: string = 'default'): Promise<void> {
+    const JobClass = Locator.get(jobData.name)
 
     if (!JobClass) {
-      throw new Error(`Job class ${jobName} not found.`)
-    }
-
-    const context: JobContext = {
-      jobId: `sync-${Date.now()}`,
-      name: jobName,
-      attempt: 1,
-      queue,
-      priority: DEFAULT_PRIORITY,
-      acquiredAt: new Date(),
-      stalledCount: 0,
+      throw new Error(`Job class ${jobData.name} not found.`)
     }
 
+    const options = JobClass.options || {}
+    const configResolver = QueueManager.getConfigResolver()
+    const runtime = JobExecutionRuntime.from({
+      jobName: jobData.name,
+      options,
+      retryConfig: configResolver.resolveRetryConfig(queue, options),
+      defaultTimeout: configResolver.getWorkerTimeout(),
+    })
     const jobFactory = QueueManager.getJobFactory()
-    const jobInstance = jobFactory ? await jobFactory(JobClass) : new JobClass()
-
-    jobInstance.$hydrate(payload, context)
-    await jobInstance.execute()
+    let attempts = jobData.attempts
+
+    while (true) {
+      const context: JobContext = {
+        jobId: jobData.id,
+        name: jobData.name,
+        attempt: attempts + 1,
+        queue,
+        priority: jobData.priority ?? DEFAULT_PRIORITY,
+        acquiredAt: new Date(),
+        stalledCount: jobData.stalledCount ?? 0,
+      }
+
+      const jobInstance = jobFactory ? await jobFactory(JobClass) : new JobClass()
+
+      try {
+        await runtime.execute(jobInstance, jobData.payload, context)
+        return
+      } catch (error) {
+        const outcome = runtime.resolveFailure(error as Error, attempts)
+
+        if (outcome.type === 'failed') {
+          await jobInstance.failed?.(outcome.hookError)
+          return
+        }
+
+        attempts++
+
+        if (outcome.type === 'retry' && outcome.retryAt) {
+          const delay = outcome.retryAt.getTime() - Date.now()
+
+          if (delay > 0) {
+            await sleep(delay)
+          }
+        }
+      }
+    }
   }
 }

src/job_runtime.ts

@@ -0,0 +1,169 @@
+import * as errors from './exceptions.js'
+import type { Job } from './job.js'
+import type { Duration, JobContext, JobOptions, RetryConfig } from './types/main.js'
+import { parse } from './utils.js'
+
+export type JobExecutionOutcome =
+  | { type: 'completed' }
+  | { type: 'retry'; retryAt?: Date }
+  | {
+      type: 'failed'
+      reason: 'timeout' | 'no-retries' | 'max-attempts'
+      storageError: Error
+      hookError: Error
+    }
+
+type JobExecutionRuntimeConfig = {
+  jobName: string
+  options?: JobOptions
+  retryConfig: RetryConfig
+  defaultTimeout?: Duration
+}
+
+type JobExecutionRuntimeFactoryOptions = {
+  jobName: string
+  options?: JobOptions
+  retryConfig: RetryConfig
+  defaultTimeout?: Duration
+}
+
+/**
+ * Shared execution policy for a single job runtime.
+ *
+ * It encapsulates timeout resolution and retry/failure decisions so the
+ * worker and the sync adapter follow the same execution rules.
+ */
+export class JobExecutionRuntime {
+  readonly #jobName: string
+  readonly #options: JobOptions
+  readonly #retryConfig: RetryConfig
+  readonly #timeout?: number
+
+  /**
+   * Build a runtime from already-resolved queue/job execution config.
+   */
+  static from({
+    jobName,
+    options,
+    retryConfig,
+    defaultTimeout,
+  }: JobExecutionRuntimeFactoryOptions): JobExecutionRuntime {
+    return new JobExecutionRuntime({
+      jobName,
+      options,
+      retryConfig,
+      defaultTimeout,
+    })
+  }
+
+  get maxRetries(): number | undefined {
+    return this.#retryConfig.maxRetries
+  }
+
+  /**
+   * Create a runtime with fully resolved retry and timeout settings.
+   */
+  constructor({ jobName, options, retryConfig, defaultTimeout }: JobExecutionRuntimeConfig) {
+    this.#jobName = jobName
+    this.#options = options || {}
+    this.#retryConfig = retryConfig
+
+    const timeout = this.#options.timeout ?? defaultTimeout
+    this.#timeout = timeout === undefined ? undefined : parse(timeout)
+  }
+
+  /**
+   * Execute a hydrated job instance and enforce the configured timeout.
+   */
+  async execute(instance: Job, payload: unknown, context: JobContext): Promise<void> {
+    if (this.#timeout === undefined) {
+      instance.$hydrate(payload, context)
+      return instance.execute()
+    }
+
+    const signal = AbortSignal.timeout(this.#timeout)
+    instance.$hydrate(payload, context, signal)
+
+    const { abortPromise, cleanupAbortListener } = this.#createTimeoutAbortRace(
+      signal,
+      instance.constructor.name
+    )
+
+    try {
+      await Promise.race([instance.execute(), abortPromise])
+    } finally {
+      cleanupAbortListener()
+    }
+  }
+
+  /**
+   * Convert an execution error into a retry or permanent-failure outcome.
+   */
+  resolveFailure(error: Error, attempts: number): JobExecutionOutcome {
+    if (error instanceof errors.E_JOB_TIMEOUT && this.#options.failOnTimeout) {
+      return {
+        type: 'failed',
+        reason: 'timeout',
+        storageError: error,
+        hookError: error,
+      }
+    }
+
+    if (typeof this.#retryConfig.maxRetries === 'undefined' || this.#retryConfig.maxRetries <= 0) {
+      return {
+        type: 'failed',
+        reason: 'no-retries',
+        storageError: error,
+        hookError: error,
+      }
+    }
+
+    if (attempts >= this.#retryConfig.maxRetries) {
+      return {
+        type: 'failed',
+        reason: 'max-attempts',
+        storageError: error,
+        hookError: new errors.E_JOB_MAX_ATTEMPTS_REACHED([this.#jobName], { cause: error }),
+      }
+    }
+
+    if (this.#retryConfig.backoff) {
+      return {
+        type: 'retry',
+        retryAt: this.#retryConfig.backoff().getNextRetryAt(attempts + 1),
+      }
+    }
+
+    return { type: 'retry' }
+  }
+
+  /**
+   * Create the timeout race used to abort a job execution cleanly.
+   */
+  #createTimeoutAbortRace(signal: AbortSignal, runtimeJobName: string) {
+    const timeout = this.#timeout!
+    let abortHandler: (() => void) | undefined
+
+    const abortPromise = new Promise<never>((_, reject) => {
+      abortHandler = () => {
+        reject(new errors.E_JOB_TIMEOUT([runtimeJobName, timeout]))
+      }
+
+      if (signal.aborted) {
+        abortHandler()
+        return
+      }
+
+      signal.addEventListener('abort', abortHandler, { once: true })
+    })
+
+    return {
+      abortPromise,
+      cleanupAbortListener: () => {
+        if (abortHandler) {
+          signal.removeEventListener('abort', abortHandler)
+        }
+      },
+    }
+  }
+}

src/queue_config_resolver.ts

@@ -54,9 +54,10 @@ export class QueueConfigResolver {
   /**
    * Resolve the retry policy for a job using priority: job > queue > global.
    */
-  resolveRetryConfig(queue: string, jobRetryConfig?: RetryConfig): RetryConfig {
+  resolveRetryConfig(queue: string, jobOptions?: JobOptions): RetryConfig {
     const queueConfig = this.#queueConfigs.get(queue)
     const queueRetryConfig = queueConfig?.retry || {}
+    const jobRetryConfig = this.#normalizeJobRetryConfig(jobOptions)
 
     const maxRetries =
       jobRetryConfig?.maxRetries ??
@@ -95,4 +96,22 @@ export class QueueConfigResolver {
   getWorkerTimeout(): Duration | undefined {
     return this.#workerTimeout
   }
+
+  /**
+   * Normalize job retry settings so top-level `maxRetries` participates in the
+   * merge like `retry.maxRetries`.
+   */
+  #normalizeJobRetryConfig(jobOptions?: JobOptions): RetryConfig | undefined {
+    if (
+      !jobOptions ||
+      (jobOptions.retry === undefined && jobOptions.maxRetries === undefined)
+    ) {
+      return undefined
+    }
+
+    return {
+      ...jobOptions.retry,
+      maxRetries: jobOptions.retry?.maxRetries ?? jobOptions.maxRetries,
+    }
+  }
 }