refactor!: extract magic numbers into named constants

Author: RomainLanz

benchmark/boringnode/harness.ts

@@ -54,7 +54,7 @@ export async function run(options: BenchmarkOptions): Promise<BenchmarkResult> {
     locations: [''],
     worker: {
       concurrency: options.concurrency,
-      pollingInterval: 1, // Very short polling for benchmarks
+      idleDelay: 1, // Very short delay for benchmarks
     },
   }
 

examples/config.ts

@@ -28,7 +28,6 @@ export const config: QueueManagerConfig = {
 
   worker: {
     concurrency: 5,
-    pollingInterval: '10ms',
   },
 
   locations: ['./examples/jobs/**/*.ts'],

src/constants.ts

@@ -0,0 +1,33 @@
+/**
+ * Default job priority (1-10 scale, lower = higher priority)
+ */
+export const DEFAULT_PRIORITY = 5
+
+/**
+ * Multiplier used in score calculation: priority * multiplier + timestamp
+ *
+ * This ensures higher priority jobs are processed first,
+ * while preserving FIFO order within the same priority.
+ * The value (1e13) leaves room for ~300 years of millisecond timestamps.
+ */
+export const PRIORITY_SCORE_MULTIPLIER = 1e13
+
+/**
+ * Default delay when the worker is idle (no jobs in queue)
+ */
+export const DEFAULT_IDLE_DELAY = '2s'
+
+/**
+ * Default interval between stalled job checks
+ */
+export const DEFAULT_STALLED_INTERVAL = '30s'
+
+/**
+ * Default threshold after which a job is considered stalled
+ */
+export const DEFAULT_STALLED_THRESHOLD = '30s'
+
+/**
+ * Default delay before retrying after an error
+ */
+export const DEFAULT_ERROR_RETRY_DELAY = '5s'

src/drivers/knex_adapter.ts

@@ -2,6 +2,8 @@ import KnexPkg from 'knex'
 import type { Knex } from 'knex'
 import type { Adapter, AcquiredJob } from '../contracts/adapter.js'
 import type { JobData } from '../types/main.js'
+import { DEFAULT_PRIORITY } from '../constants.js'
+import { calculateScore } from '../utils.js'
 
 export interface KnexAdapterOptions {
   connection: Knex
@@ -168,8 +170,8 @@ export class KnexAdapter implements Adapter {
       // Move them to pending
       for (const job of delayedJobs) {
         const jobData: JobData = JSON.parse(job.data)
-        const priority = jobData.priority ?? 5
-        const score = priority * 1e13 + now
+        const priority = jobData.priority ?? DEFAULT_PRIORITY
+        const score = calculateScore(priority, now)
 
         await trx(this.#tableName).where('id', job.id).where('queue', queue).update({
           status: 'pending',
@@ -223,8 +225,8 @@ export class KnexAdapter implements Adapter {
       })
     } else {
       // Move back to pending
-      const priority = jobData.priority ?? 5
-      const score = priority * 1e13 + now
+      const priority = jobData.priority ?? DEFAULT_PRIORITY
+      const score = calculateScore(priority, now)
 
       await this.#connection(this.#tableName).where('id', jobId).where('queue', queue).update({
         status: 'pending',
@@ -244,9 +246,9 @@ export class KnexAdapter implements Adapter {
   async pushOn(queue: string, jobData: JobData): Promise<void> {
     await this.#ensureTableExists()
 
-    const priority = jobData.priority ?? 5
+    const priority = jobData.priority ?? DEFAULT_PRIORITY
     const timestamp = Date.now()
-    const score = priority * 1e13 + timestamp
+    const score = calculateScore(priority, timestamp)
 
     await this.#connection(this.#tableName).insert({
       id: jobData.id,
@@ -327,8 +329,8 @@ export class KnexAdapter implements Adapter {
         } else {
           // Recover: increment stalledCount and put back in pending
           jobData.stalledCount = currentStalledCount + 1
-          const priority = jobData.priority ?? 5
-          const score = priority * 1e13 + now
+          const priority = jobData.priority ?? DEFAULT_PRIORITY
+          const score = calculateScore(priority, now)
 
           await trx(this.#tableName)
             .where('id', row.id)

src/drivers/redis_adapter.ts

@@ -1,6 +1,8 @@
 import { Redis, type RedisOptions } from 'ioredis'
 import type { Adapter, AcquiredJob } from '../contracts/adapter.js'
 import type { JobData } from '../types/main.js'
+import { DEFAULT_PRIORITY } from '../constants.js'
+import { calculateScore } from '../utils.js'
 
 const redisKey = 'jobs'
 type RedisConfig = Redis | RedisOptions
@@ -25,6 +27,8 @@ const ACQUIRE_JOB_SCRIPT = `
     for i = 1, #ready_jobs do
       local job_data = ready_jobs[i]
       local job = cjson.decode(job_data)
+      -- Score = priority * 1e13 + timestamp
+      -- Lower score = higher priority, FIFO within same priority
       local priority = job.priority or 5
       local timestamp = tonumber(now)
       local score = priority * 10000000000000 + timestamp
@@ -115,6 +119,8 @@ const RETRY_JOB_SCRIPT = `
   if retry_at and retry_at > now then
     redis.call('ZADD', delayed_key, retry_at, job_data)
   else
+    -- Score = priority * 1e13 + timestamp
+    -- Lower score = higher priority, FIFO within same priority
     local priority = job.priority or 5
     local score = priority * 10000000000000 + now
     redis.call('ZADD', pending_key, score, job_data)
@@ -164,6 +170,8 @@ const RECOVER_STALLED_JOBS_SCRIPT = `
         -- Recover: increment stalledCount and put back in pending
         job.stalledCount = current_stalled_count + 1
         local job_data = cjson.encode(job)
+        -- Score = priority * 1e13 + timestamp
+        -- Lower score = higher priority, FIFO within same priority
         local priority = job.priority or 5
         local score = priority * 10000000000000 + now
         redis.call('ZADD', pending_key, score, job_data)
@@ -297,12 +305,9 @@ export class RedisAdapter implements Adapter {
   }
 
   async pushOn(queue: string, jobData: JobData): Promise<void> {
-    const priority = jobData.priority ?? 5
-
-    // Use priority as primary score, add timestamp for FIFO order within same priority
-    // Date.now() precision is sufficient but perfect FIFO within the same millisecond is not guaranteed
+    const priority = jobData.priority ?? DEFAULT_PRIORITY
     const timestamp = Date.now()
-    const score = priority * 1e13 + timestamp
+    const score = calculateScore(priority, timestamp)
 
     await this.#connection.zadd(`${redisKey}::${queue}`, score, JSON.stringify(jobData))
   }

src/types/main.ts

@@ -57,10 +57,10 @@ export interface WorkerConfig {
   concurrency?: number
 
   /**
-   * How often to poll for new jobs when the queue is empty.
+   * Delay between queue polls when idle (no jobs available).
    * @default '2s'
    */
-  pollingInterval?: Duration
+  idleDelay?: Duration
 
   /**
    * Maximum duration a job can run before being timed out.

src/utils.ts

@@ -1,6 +1,7 @@
 import { parse as parseDuration } from '@lukeed/ms'
 import type { Duration } from './types/main.js'
 import * as errors from './exceptions.js'
+import { PRIORITY_SCORE_MULTIPLIER } from './constants.js'
 
 export function parse(duration: Duration): number {
   if (typeof duration === 'number') {
@@ -15,3 +16,15 @@ export function parse(duration: Duration): number {
 
   return milliseconds
 }
+
+/**
+ * Calculate the score for job ordering in the queue.
+ * Lower scores are processed first.
+ *
+ * @param priority - Job priority (1-10, lower = higher priority)
+ * @param timestamp - Timestamp in milliseconds
+ * @returns Score for queue ordering
+ */
+export function calculateScore(priority: number, timestamp: number): number {
+  return priority * PRIORITY_SCORE_MULTIPLIER + timestamp
+}

src/worker.ts

@@ -10,11 +10,17 @@ import type { QueueManagerConfig, WorkerCycle } from './types/main.js'
 import { Locator } from './locator.js'
 import type { JobOptions } from './types/main.js'
 import type { Job } from './job.js'
+import {
+  DEFAULT_IDLE_DELAY,
+  DEFAULT_STALLED_INTERVAL,
+  DEFAULT_STALLED_THRESHOLD,
+  DEFAULT_ERROR_RETRY_DELAY,
+} from './constants.js'
 
 export class Worker {
   readonly #id: string
   readonly #config: QueueManagerConfig
-  readonly #pollingInterval: number
+  readonly #idleDelay: number
   readonly #stalledInterval: number
   readonly #stalledThreshold: number
   readonly #maxStalledCount: number
@@ -39,9 +45,9 @@ export class Worker {
     this.#id = randomUUID()
 
     // Parse worker config once at construction
-    this.#pollingInterval = parse(config.worker?.pollingInterval ?? '2s')
-    this.#stalledInterval = parse(config.worker?.stalledInterval ?? '30s')
-    this.#stalledThreshold = parse(config.worker?.stalledThreshold ?? '30s')
+    this.#idleDelay = parse(config.worker?.idleDelay ?? DEFAULT_IDLE_DELAY)
+    this.#stalledInterval = parse(config.worker?.stalledInterval ?? DEFAULT_STALLED_INTERVAL)
+    this.#stalledThreshold = parse(config.worker?.stalledThreshold ?? DEFAULT_STALLED_THRESHOLD)
     this.#maxStalledCount = config.worker?.maxStalledCount ?? 1
     this.#concurrency = config.worker?.concurrency ?? 1
     this.#gracefulShutdown = config.worker?.gracefulShutdown ?? true
@@ -148,14 +154,18 @@ export class Worker {
         yield* this.#fillPool(queues)
 
         if (this.#pool.isEmpty()) {
-          yield { type: 'idle', suggestedDelay: this.#pollingInterval }
+          yield { type: 'idle', suggestedDelay: this.#idleDelay }
           continue
         }
 
         const completed = await this.#pool.waitForNextCompletion()
         yield { type: 'completed', queue: completed.queue, job: completed.job }
       } catch (error) {
-        yield { type: 'error', error: error as Error, suggestedDelay: parse('5s') }
+        yield {
+          type: 'error',
+          error: error as Error,
+          suggestedDelay: parse(DEFAULT_ERROR_RETRY_DELAY),
+        }
       }
     }
   }