refactor(redis): remove verrou in favor of atomic lua script

Author: RomainLanz

benchmark/bullmq/harness.ts

@@ -1,4 +1,4 @@
-import { Queue, Worker } from 'bullmq'
+import { Queue, Worker, type ConnectionOptions } from 'bullmq'
 import { Redis } from 'ioredis'
 import { barrier, type BenchmarkOptions, type BenchmarkResult } from '../helpers.ts'
 
@@ -19,7 +19,7 @@ export async function run(options: BenchmarkOptions): Promise<BenchmarkResult> {
   await clearQueue(connection)
 
   const queue = new Queue('benchmark', {
-    connection,
+    connection: connection as ConnectionOptions,
     defaultJobOptions: {
       removeOnComplete: true,
       removeOnFail: true,
@@ -43,7 +43,7 @@ export async function run(options: BenchmarkOptions): Promise<BenchmarkResult> {
       next()
     },
     {
-      connection,
+      connection: connection as ConnectionOptions,
       concurrency: options.concurrency,
     }
   )

benchmark/run.ts

@@ -69,9 +69,7 @@ async function runBenchmarks(config: BenchmarkConfig = defaultConfig) {
     grouped.get(key)!.push(result)
   }
 
-  console.log(
-    'Jobs\tConc.\t@boringnode/queue\t\tBullMQ\t\t\tDiff'
-  )
+  console.log('Jobs\tConc.\t@boringnode/queue\t\tBullMQ\t\t\tDiff')
   console.log('-'.repeat(90))
 
   for (const [key, group] of grouped) {
@@ -82,9 +80,7 @@ async function runBenchmarks(config: BenchmarkConfig = defaultConfig) {
     const boringnodeStr = boringnode
       ? `${boringnode.elapsed}ms (${boringnode.jobsPerSecond.toFixed(0)} j/s)`
       : '-'
-    const bullmqStr = bullmq
-      ? `${bullmq.elapsed}ms (${bullmq.jobsPerSecond.toFixed(0)} j/s)`
-      : '-'
+    const bullmqStr = bullmq ? `${bullmq.elapsed}ms (${bullmq.jobsPerSecond.toFixed(0)} j/s)` : '-'
 
     let diff = ''
     if (boringnode && bullmq) {

package.json

@@ -36,8 +36,7 @@
   },
   "dependencies": {
     "@lukeed/ms": "^2.0.2",
-    "@poppinss/utils": "^6.10.1",
-    "@verrou/core": "^0.5.1"
+    "@poppinss/utils": "^6.10.1"
   },
   "devDependencies": {
     "@adonisjs/eslint-config": "^2.1.2",

src/contracts/adapter.ts

@@ -1,20 +1,55 @@
-import type { LeaseManager } from '#contracts/lease_manager'
-import type { JobData, LeaseConfig } from '#types/main'
+import type { JobData } from '#types/main'
+
+export interface AcquiredJob extends JobData {
+  acquiredAt: number
+}
 
 export interface Adapter {
-  createLeaseManager(config: LeaseConfig): LeaseManager
+  /**
+   * Set the worker ID for this adapter instance.
+   * Required before calling pop methods when consuming jobs.
+   */
+  setWorkerId(workerId: string): void
 
-  size(): Promise<number>
-  sizeOf(queue: string): Promise<number>
+  /**
+   * Pop the next available job from the default queue.
+   * The driver handles locking internally.
+   */
+  pop(): Promise<AcquiredJob | null>
+
+  /**
+   * Pop the next available job from a specific queue.
+   * The driver handles locking internally.
+   */
+  popFrom(queue: string): Promise<AcquiredJob | null>
+
+  /**
+   * Blocking pop that waits for a job to be available.
+   * Supported by Redis adapter.
+   */
+  popAndWait?(queue: string, timeout: number): Promise<AcquiredJob | null>
+
+  /**
+   * Mark a job as completed and remove it from active set.
+   */
+  completeJob(jobId: string, queue: string): Promise<void>
+
+  /**
+   * Mark a job as failed permanently.
+   */
+  failJob(jobId: string, queue: string, error?: Error): Promise<void>
+
+  /**
+   * Retry a job - move back to pending queue with incremented attempts.
+   */
+  retryJob(jobId: string, queue: string, retryAt?: Date): Promise<void>
 
   push(jobData: JobData): Promise<void>
   pushOn(queue: string, jobData: JobData): Promise<void>
-
   pushLater(jobData: JobData, delay: number): Promise<void>
   pushLaterOn(queue: string, jobData: JobData, delay: number): Promise<void>
 
-  pop(): Promise<JobData | null>
-  popFrom(queue: string): Promise<JobData | null>
-
+  size(): Promise<number>
+  sizeOf(queue: string): Promise<number>
   destroy(): Promise<void>
 }