feat(worker): stop polling idle workers on Redis and PostgreSQL

README.md

@@ -411,8 +411,15 @@ import Knex from 'knex'
 const connection = Knex({ client: 'pg', connection: '...' })
 const adapter = knex(connection)
 
-// Custom table name
+// Custom table names
 const adapter = knex(config, 'custom_jobs_table')
+const adapter = knex(config, {
+  tableName: 'custom_jobs_table',
+  schedulesTableName: 'custom_schedules_table',
+})
+
+// PostgreSQL: wake Workers with LISTEN/NOTIFY instead of polling (see Worker Configuration)
+const adapter = knex(config, { notifications: true })
 ```
 
 </details>
@@ -932,7 +939,7 @@ const config = {
   worker: {
     adapter: 'redis', // Registered Adapter listened to by this Worker
     concurrency: 5, // Parallel jobs
-    idleDelay: '2s', // Poll interval when idle
+    idleDelay: '2s', // Poll interval when idle. PostgreSQL: wait after a wake that found nothing
     timeout: '1m', // Default job timeout
     stalledThreshold: '30s', // When to consider job stalled
     stalledInterval: '30s', // How often to check
@@ -943,6 +950,30 @@ const config = {
 }
 ```
 
+On Redis, an idle Worker does not poll: it waits on a marker with `BZPOPMIN`, as in BullMQ, until a
+job is dispatched, or a delayed job, a retry, or a schedule becomes due. A dispatched job wakes one
+idle Worker, which wakes the next one while jobs remain. The Worker still checks its queues every
+`stalledInterval`, in case a wake was lost, and holds one more connection to wait for work.
+A connection that goes silent is detected within 15 s; until then, the Worker finds work at its
+stalled check.
+
+The process that dispatches a job wakes the Workers. A job dispatched by an older version of the
+package, or written to Redis without the Adapter, wakes none: the Worker finds it at its next
+stalled check. Upgrade the processes that dispatch jobs with, or before, the Workers.
+
+On PostgreSQL, Workers poll every `idleDelay` by default. With `knex(config, { notifications: true })`,
+they `LISTEN` for a `NOTIFY` instead, and a job dispatched in a transaction wakes them when it
+commits. Enable it on every process that dispatches jobs, not only on Workers. It suits a few
+Workers with a high `concurrency`:
+
+- Each Worker holds one more connection: size `max_connections` for 2 connections per Worker,
+  or dispatchers may get none.
+- Every `NOTIFY` reaches every Worker. A Worker woken for a job another Worker took waits
+  `idleDelay` before it looks again, so idle Workers never query more often than when polling.
+- It does not work behind a pooler in transaction mode, such as PgBouncer.
+
+MySQL, SQLite, and Kysely poll every `idleDelay`.
+
 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

src/contracts/adapter.ts

@@ -55,6 +55,35 @@ export interface StalledJobsRecovery {
   exceeded: AcquiredJob[]
 }
 
+/**
+ * Lets an idle worker wait for work instead of polling. See `Adapter.watch`.
+ *
+ * ```
+ * cycle:  reset() ─► stalled check, schedules, pop ─► wait() ─► cycle
+ * ```
+ */
+export interface WorkWatcher {
+  /**
+   * Called before the worker looks for work: work signaled until then is
+   * covered by that look.
+   */
+  reset(): void
+
+  /**
+   * Resolve true when work may be due on the watched queues or schedules,
+   * false after `maxDelay` or when `signal` aborts.
+   *
+   * `cooldown` is non-zero when the previous wake found no work, as when
+   * another worker took the job first. An adapter that wakes every worker for
+   * the same work must not wake this one again before it elapses: idle workers
+   * then look for work at most once per `cooldown`, as when polling.
+   */
+  wait(maxDelay: number, signal: AbortSignal, cooldown: number): Promise<boolean>
+
+  /** Releases what the watcher opened. */
+  close(): Promise<void>
+}
+
 /**
  * Adapter interface for queue storage backends.
  *
@@ -100,6 +129,21 @@ export interface Adapter {
    */
   popFrom(queue: string): Promise<AcquiredJob | null>
 
+  /**
+   * Optional. Lets an idle worker wait until work may be due on `queues` or
+   * the schedules, instead of polling. Without it, the worker polls every
+   * `idleDelay`.
+   *
+   * A wake is a hint and may be lost, so the worker still polls once per
+   * stalled interval. The worker polls until `watch` resolves, and aborts
+   * `signal` when it stops first: the adapter then releases what it opened
+   * and rejects.
+   *
+   * @param queues - The queues the worker processes
+   * @param signal - Aborted when the worker stops before the watcher is ready
+   */
+  watch?(queues: readonly string[], signal?: AbortSignal): Promise<WorkWatcher>
+
   /**
    * Recover stalled jobs that have been active for too long.
    * A stalled job is one where the worker stopped responding (e.g., crash).