test(adapter): move queue-isolation tests from fake adapter to shared driver suite

Author: RomainLanz

tests/_mocks/memory_adapter.ts

@@ -13,6 +13,7 @@ import { parse } from '../../src/utils.js'
 interface ActiveJob {
   job: JobData
   acquiredAt: number
+  queue: string
 }
 
 interface DelayedJob {
@@ -119,7 +120,7 @@ export class MemoryAdapter implements Adapter {
     }
 
     const acquiredAt = Date.now()
-    this.#activeJobs.set(job.id, { job, acquiredAt })
+    this.#activeJobs.set(job.id, { job, acquiredAt, queue })
 
     return { ...job, acquiredAt }
   }
@@ -187,6 +188,10 @@ export class MemoryAdapter implements Adapter {
     let recovered = 0
 
     for (const [jobId, active] of this.#activeJobs.entries()) {
+      if (active.queue !== queue) {
+        continue
+      }
+
       const isStalled = now - active.acquiredAt > stalledThreshold
 
       if (!isStalled) {
@@ -219,7 +224,7 @@ export class MemoryAdapter implements Adapter {
 
   async getJob(jobId: string, queue: string): Promise<JobRecord | null> {
     const active = this.#activeJobs.get(jobId)
-    if (active) {
+    if (active && active.queue === queue) {
       return { status: 'active', data: active.job }
     }
 

tests/_utils/register_driver_test_suite.ts

@@ -146,6 +146,27 @@ export function registerDriverTestSuite(options: DriverTestSuiteOptions) {
     assert.equal(record!.data.id, 'job-active')
   })
 
+  test('getJob should not return active job from another queue', async ({ assert }) => {
+    const adapter = await options.createAdapter()
+    adapter.setWorkerId('worker-1')
+
+    await adapter.pushOn('queue-a', {
+      id: 'job-active-other-queue',
+      name: 'TestJob',
+      payload: {},
+      attempts: 0,
+    })
+
+    await adapter.popFrom('queue-a')
+
+    const wrongQueueRecord = await adapter.getJob('job-active-other-queue', 'queue-b')
+    assert.isNull(wrongQueueRecord)
+
+    const rightQueueRecord = await adapter.getJob('job-active-other-queue', 'queue-a')
+    assert.isNotNull(rightQueueRecord)
+    assert.equal(rightQueueRecord!.status, 'active')
+  })
+
   test('getJob should return null for non-existent job', async ({ assert }) => {
     const adapter = await options.createAdapter()
     adapter.setWorkerId('worker-1')
@@ -592,6 +613,50 @@ export function registerDriverTestSuite(options: DriverTestSuiteOptions) {
     assert.isNull(job3)
   })
 
+  test('recoverStalledJobs should only recover jobs from the targeted queue', async ({ assert }) => {
+    const adapter = await options.createAdapter()
+    adapter.setWorkerId('worker-1')
+
+    await adapter.pushOn('queue-a', {
+      id: 'job-stalled-a',
+      name: 'TestJob',
+      payload: null,
+      attempts: 0,
+    })
+    await adapter.pushOn('queue-b', {
+      id: 'job-stalled-b',
+      name: 'TestJob',
+      payload: null,
+      attempts: 0,
+    })
+
+    const jobA = await adapter.popFrom('queue-a')
+    const jobB = await adapter.popFrom('queue-b')
+    assert.isNotNull(jobA)
+    assert.isNotNull(jobB)
+
+    await new Promise((resolve) => setTimeout(resolve, 50))
+
+    const recoveredA = await adapter.recoverStalledJobs('queue-a', 10, 1)
+    assert.equal(recoveredA, 1)
+
+    const recoveredJobA = await adapter.popFrom('queue-a')
+    assert.isNotNull(recoveredJobA)
+    assert.equal(recoveredJobA!.id, 'job-stalled-a')
+
+    const queueBPending = await adapter.popFrom('queue-b')
+    assert.isNull(queueBPending)
+
+    await new Promise((resolve) => setTimeout(resolve, 50))
+
+    const recoveredB = await adapter.recoverStalledJobs('queue-b', 10, 1)
+    assert.equal(recoveredB, 1)
+
+    const recoveredJobB = await adapter.popFrom('queue-b')
+    assert.isNotNull(recoveredJobB)
+    assert.equal(recoveredJobB!.id, 'job-stalled-b')
+  })
+
   test('completeJob with undefined retention should remove job (default behavior)', async ({
     assert,
   }) => {

tests/fake_adapter.spec.ts

@@ -1,4 +1,3 @@
-import { setTimeout } from 'node:timers/promises'
 import { test } from '@japa/runner'
 import { Job } from '../src/job.js'
 import { fake } from '../src/drivers/fake_adapter.js'
@@ -126,71 +125,4 @@ test.group('FakeAdapter', () => {
     await adapter.destroy()
   })
 
-  test('should recover stalled jobs only for targeted queue', async ({ assert }) => {
-    const adapter = fake()()
-
-    await adapter.pushOn('queue-a', {
-      id: 'job-a',
-      name: 'JobA',
-      payload: null,
-      attempts: 0,
-    })
-
-    await adapter.pushOn('queue-b', {
-      id: 'job-b',
-      name: 'JobB',
-      payload: null,
-      attempts: 0,
-    })
-
-    const jobA = await adapter.popFrom('queue-a')
-    const jobB = await adapter.popFrom('queue-b')
-
-    assert.isNotNull(jobA)
-    assert.isNotNull(jobB)
-
-    await setTimeout(2)
-
-    const recoveredA = await adapter.recoverStalledJobs('queue-a', 0, 1)
-    assert.equal(recoveredA, 1)
-
-    const recoveredJobA = await adapter.popFrom('queue-a')
-    assert.equal(recoveredJobA?.id, 'job-a')
-
-    const noneInB = await adapter.popFrom('queue-b')
-    assert.isNull(noneInB)
-
-    await setTimeout(2)
-
-    const recoveredB = await adapter.recoverStalledJobs('queue-b', 0, 1)
-    assert.equal(recoveredB, 1)
-
-    const recoveredJobB = await adapter.popFrom('queue-b')
-    assert.equal(recoveredJobB?.id, 'job-b')
-
-    await adapter.destroy()
-  })
-
-  test('should ignore active jobs from other queues in getJob', async ({ assert }) => {
-    const adapter = fake()()
-
-    await adapter.pushOn('queue-a', {
-      id: 'job-a',
-      name: 'JobA',
-      payload: null,
-      attempts: 0,
-    })
-
-    const jobA = await adapter.popFrom('queue-a')
-    assert.isNotNull(jobA)
-
-    const wrongQueue = await adapter.getJob('job-a', 'queue-b')
-    assert.isNull(wrongQueue)
-
-    const correctQueue = await adapter.getJob('job-a', 'queue-a')
-    assert.isNotNull(correctQueue)
-    assert.equal(correctQueue?.status, 'active')
-
-    await adapter.destroy()
-  })
 })