fix!: reject a second Worker started in the same process

Each Worker is meant to run in its own process. Starting a second one
while another ran in the process called QueueManager.init() again, which
destroyed the Adapters of the first Worker: it then produced error
cycles forever, without saying why.

worker.start() now throws the new E_WORKER_ALREADY_RUNNING error when
another Worker holds the process, before init() can touch the
QueueManager. A Worker holds the process from start() until its stop()
completes, after its last job is finalized; start() releases it only if
initialization fails before any session exists, and a restart requested
during stop() reserves it again once that stop completes.
processCycle() and process(), low-level tools used by tests, are not
guarded.

BREAKING CHANGE: a second worker.start() while another Worker runs in
the same process rejects with E_WORKER_ALREADY_RUNNING.

Author: RomainLanz

.changelog/one-worker-per-process.md

@@ -0,0 +1,20 @@
+# One Worker per Process
+
+## Bug Fix
+
+Starting a second Worker in a process where another one was running re-initialized the
+`QueueManager`, which destroyed the Adapters of the first Worker. The first Worker then produced
+`error` cycles forever, without saying why.
+
+`worker.start()` now throws the new `E_WORKER_ALREADY_RUNNING` error when another Worker is running
+in the process, before touching the `QueueManager`. The first Worker keeps working. A Worker holds
+the process from `start()` until its `stop()` completes, after its last job is finalized; then
+another Worker can start.
+
+Run one Worker per process. A Worker listens on one Adapter: to process the queues of several
+Adapters, start one process per Adapter.
+
+## Breaking Changes
+
+A second `worker.start()` while another Worker runs in the same process now rejects with
+`E_WORKER_ALREADY_RUNNING` instead of starting.

README.md

@@ -761,8 +761,8 @@ Schedules at every boot:
 - it keeps its next run, unless its timing changes (cron expression, interval, timezone, `from`,
   `to`, or `limit`). A payload change alone does not move the next run.
 
-A Schedule and every Job it dispatches stay on the same Adapter. Start a Worker for each Adapter
-that owns Schedules.
+A Schedule and every Job it dispatches stay on the same Adapter. Start a Worker, in its own
+process, for each Adapter that owns Schedules.
 
 </details>
 
@@ -818,6 +818,11 @@ const config = {
 }
 ```
 
+Run one Worker per process. A Worker listens on one Adapter, so to process the queues of several
+Adapters, start one process per Adapter. Starting a second Worker while another one runs in the
+same process throws `E_WORKER_ALREADY_RUNNING`: initializing it would destroy the Adapters of the
+first one. A Worker holds the process until its `stop()` completes.
+
 A job is stalled when its worker stops renewing it for longer than `stalledThreshold`, usually
 because the process crashed. A stalled job goes back to the queue up to `maxStalledCount` times.
 After that, the worker that detects it fails it like any other permanent failure: `failed()`

src/exceptions.ts

@@ -56,6 +56,12 @@ export const E_INVALID_TIMEOUT = createError<[timeout: string]>(
   500
 )
 
+export const E_WORKER_ALREADY_RUNNING = createError(
+  'Another Worker is already running in this process. Run one Worker per process.',
+  'E_WORKER_ALREADY_RUNNING',
+  500
+)
+
 export const E_QUEUE_NOT_INITIALIZED = createError(
   'QueueManager is not initialized. Call QueueManager.init() before using it.',
   'E_QUEUE_NOT_INITIALIZED',

src/worker.ts

@@ -60,6 +60,9 @@ export class Worker {
   #session?: WorkerSession
   #stopOperation?: Promise<void>
   #shutdownGeneration = 0
+
+  /** The Worker whose `start()` is running in this process. */
+  static #running?: Worker
   #shutdownHandler?: () => Promise<void>
 
   /** Unique identifier for this worker instance. */
@@ -142,23 +145,55 @@ export class Worker {
    * This method blocks until the Worker is stopped. Jobs are processed
    * concurrently up to the configured concurrency limit.
    *
+   * Run one Worker per process: starting a second one while another runs
+   * throws, since initializing it would destroy the Adapters of the first.
+   * A Worker holds the process from `start()` until its `stop()` completes.
+   *
    * @param queues - Queue names to process in priority order
    */
   async start(queues: string[] = ['default']): Promise<void> {
-    while (this.#stopOperation) {
-      await this.#stopOperation
-    }
+    this.#reserveProcess()
 
-    const shutdownGeneration = this.#shutdownGeneration
-    await this.init()
+    let session: WorkerSession
 
-    if (shutdownGeneration !== this.#shutdownGeneration) return
+    try {
+      while (this.#stopOperation) {
+        await this.#stopOperation
+      }
+
+      // A stop that completed meanwhile released the process.
+      this.#reserveProcess()
+
+      const shutdownGeneration = this.#shutdownGeneration
+      await this.init()
+
+      if (shutdownGeneration !== this.#shutdownGeneration) return
+
+      session = this.#useSession(queues)
+      this.#setupGracefulShutdown()
+    } catch (error) {
+      // Nothing runs for this Worker yet: another one can start.
+      if (!this.#session) this.#releaseProcess()
+      throw error
+    }
 
-    const session = this.#useSession(queues)
-    this.#setupGracefulShutdown()
     await session.start()
   }
 
+  #reserveProcess(): void {
+    if (Worker.#running && Worker.#running !== this) {
+      throw new errors.E_WORKER_ALREADY_RUNNING()
+    }
+
+    Worker.#running = this
+  }
+
+  #releaseProcess(): void {
+    if (Worker.#running === this) {
+      Worker.#running = undefined
+    }
+  }
+
   /**
    * Stop the active session and wait until it reaches quiescence.
    *
@@ -204,6 +239,9 @@ export class Worker {
     }
 
     this.#removeShutdownHandlers()
+
+    // Every job has been finalized: the Adapters are no longer needed.
+    this.#releaseProcess()
   }
 
   /**

tests/worker.spec.ts

@@ -2850,3 +2850,172 @@ test.group('Worker | Scheduler Integration', () => {
     assert.includeMembers(executedJobs, ['scheduled', 'regular'])
   })
 })
+
+test.group('Worker | one per process', (group) => {
+  group.each.teardown(() => Locator.clear())
+
+  const config = (adapter: MemoryAdapter) => ({
+    default: 'memory',
+    adapters: { memory: () => adapter },
+    logger: new MemoryLogger(),
+    worker: { idleDelay: 5, gracefulShutdown: false },
+  })
+
+  async function waitFor(condition: () => boolean) {
+    for (let i = 0; i < 200 && !condition(); i++) {
+      await setTimeout(5)
+    }
+  }
+
+  test('start() rejects a second Worker and keeps the first one working', async ({
+    assert,
+    cleanup,
+  }) => {
+    let executed = 0
+
+    class CountedJob extends Job {
+      async execute() {
+        executed++
+      }
+    }
+
+    Locator.register('CountedJob', CountedJob)
+    const adapter = new MemoryAdapter()
+    const first = new Worker(config(adapter))
+    const second = new Worker(config(adapter))
+    cleanup(() => first.stop())
+
+    const running = first.start(['default'])
+    await setTimeout(20)
+
+    await assert.rejects(() => second.start(['default']), errors.E_WORKER_ALREADY_RUNNING)
+
+    // Starting the second Worker did not re-initialize the QueueManager and
+    // destroy the Adapter of the first one.
+    await adapter.push({ id: 'after-second', name: 'CountedJob', payload: {}, attempts: 0 })
+    await waitFor(() => executed === 1)
+    assert.equal(executed, 1)
+
+    await first.stop()
+    await running
+  })
+
+  test('start() accepts another Worker once the first one stopped', async ({ cleanup }) => {
+    const adapter = new MemoryAdapter()
+    const first = new Worker(config(adapter))
+    const second = new Worker(config(adapter))
+    cleanup(async () => {
+      await first.stop()
+      await second.stop()
+    })
+
+    const firstRun = first.start(['default'])
+    await setTimeout(20)
+    await first.stop()
+    await firstRun
+
+    const secondRun = second.start(['default'])
+    await setTimeout(20)
+    await second.stop()
+    await secondRun
+  })
+
+  test('a second start() of the same Worker keeps the process reserved', async ({
+    assert,
+    cleanup,
+  }) => {
+    const first = new Worker(config(new MemoryAdapter()))
+    const other = new Worker(config(new MemoryAdapter()))
+    cleanup(() => first.stop())
+
+    const running = first.start(['default'])
+    await setTimeout(20)
+
+    // The loop already runs: this call returns at once and must not release the process.
+    await first.start(['default'])
+    await assert.rejects(() => other.start(['default']), errors.E_WORKER_ALREADY_RUNNING)
+
+    await first.stop()
+    await running
+  })
+
+  test('the process stays reserved until stop() has finalized every job', async ({
+    assert,
+    cleanup,
+  }) => {
+    class BlockedJob extends Job {
+      async execute() {}
+    }
+
+    const fixture = createWorkerFixture()
+    fixture.adapter.finalizations.block(1)
+    const other = new Worker(config(new MemoryAdapter()))
+    cleanup(() => fixture.cleanup())
+
+    await fixture.push(BlockedJob, { id: 'blocked-finalization' })
+    const running = fixture.start()
+    await fixture.adapter.finalizations.waitForStarted(1)
+
+    // The stop has begun, but the finalization still needs the Adapter.
+    const stop = trackPromise(fixture.worker.stop())
+    await setTimeout(10)
+    await assert.rejects(() => other.start(['default']), errors.E_WORKER_ALREADY_RUNNING)
+    assert.isFalse(stop.settled)
+
+    fixture.adapter.finalizations.release(1)
+    await stop.promise
+    await running
+
+    const otherRun = other.start(['default'])
+    await setTimeout(20)
+    await other.stop()
+    await otherRun
+  })
+
+  test('a restart requested during stop() reserves the process again', async ({
+    assert,
+    cleanup,
+  }) => {
+    class BlockedJob extends Job {
+      async execute() {}
+    }
+
+    const fixture = createWorkerFixture()
+    fixture.adapter.finalizations.block(1)
+    const other = new Worker(config(new MemoryAdapter()))
+    cleanup(() => fixture.cleanup())
+
+    await fixture.push(BlockedJob, { id: 'blocked-before-restart' })
+    const firstRun = fixture.start()
+    await fixture.adapter.finalizations.waitForStarted(1)
+
+    const stop = trackPromise(fixture.worker.stop())
+    const restart = fixture.start()
+    fixture.adapter.finalizations.release(1)
+    await stop.promise
+    await firstRun
+    await fixture.adapter.acquisitions.waitForStarted(2)
+
+    // The restarted Worker holds the process again.
+    await assert.rejects(() => other.start(['default']), errors.E_WORKER_ALREADY_RUNNING)
+
+    await fixture.worker.stop()
+    await restart
+  })
+
+  test('a start that fails to initialize does not block the next Worker', async ({
+    assert,
+    cleanup,
+  }) => {
+    const broken = new Worker({ default: 'memory', adapters: {} })
+    await assert.rejects(() => broken.start(['default']), errors.E_CONFIGURATION_ERROR)
+
+    const worker = new Worker(config(new MemoryAdapter()))
+    cleanup(() => worker.stop())
+
+    const run = worker.start(['default'])
+    await setTimeout(20)
+    await worker.stop()
+    await run
+  })
+})