diff --git a/README.md b/README.md index 6bc6462..d9f6976 100644 --- a/README.md +++ b/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 }) ``` @@ -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 diff --git a/src/contracts/adapter.ts b/src/contracts/adapter.ts index 64dc591..4f12e75 100644 --- a/src/contracts/adapter.ts +++ b/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 + + /** Releases what the watcher opened. */ + close(): Promise +} + /** * Adapter interface for queue storage backends. * @@ -100,6 +129,21 @@ export interface Adapter { */ popFrom(queue: string): Promise + /** + * 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 + /** * Recover stalled jobs that have been active for too long. * A stalled job is one where the worker stopped responding (e.g., crash). diff --git a/src/drivers/knex_adapter.ts b/src/drivers/knex_adapter.ts index 8d1d87d..3bef291 100644 --- a/src/drivers/knex_adapter.ts +++ b/src/drivers/knex_adapter.ts @@ -1,4 +1,5 @@ -import { randomUUID } from 'node:crypto' +import { createHash, randomUUID } from 'node:crypto' +import { setTimeout as sleep } from 'node:timers/promises' import KnexPkg from 'knex' import type { Knex } from 'knex' import type { @@ -7,6 +8,7 @@ import type { JobLease, PushResult, StalledJobsRecovery, + WorkWatcher, } from '../contracts/adapter.js' import type { AdapterFactoryContext, @@ -35,6 +37,15 @@ import { import { KnexQueueSchemaService } from '../services/knex_queue_schema.js' import { scheduleDatesMigrationRequiredMessage } from '../services/schedule_dates.js' import { SCHEDULE_TIMING_COLUMNS } from '../services/schedule_timing.js' +import { + dispatchWakeMessage, + encodeWakeMessage, + startHeartbeat, + subscriptionHeartbeat, + type WakeListener, + type WakeMessage, +} from './wake_messages.js' +import { WorkerWakeup } from './worker_wakeup.js' export { KnexQueueSchemaService } from '../services/knex_queue_schema.js' export type { ScheduleDatesMigrationOptions } from '../services/schedule_dates.js' @@ -47,6 +58,87 @@ export interface KnexAdapterOptions { ownsConnection?: boolean /** Receives the retention prunings that fail. Defaults to the console. */ logger?: Logger + /** + * PostgreSQL only. Wake workers through LISTEN/NOTIFY instead of polling. + * Defaults to false. Enable it on every process that dispatches jobs, not + * only on workers: a job dispatched without it wakes no worker. + * + * Each worker holds one more connection, and every NOTIFY reaches every + * worker: suited to a few workers with a high concurrency. Not supported + * behind a pooler in transaction mode, such as PgBouncer. + */ + notifications?: boolean +} + +/** + * The part of a node-postgres client the LISTEN connection uses. + */ +interface PgListenClient { + connect(): Promise + end(): Promise + query(sql: string, values?: unknown[]): Promise<{ rows: DbRow[] }> + on( + event: 'notification', + listener: (message: { channel: string; payload?: string }) => void + ): void + on(event: 'error' | 'end', listener: (error?: Error) => void): void + removeAllListeners(): void + /** The socket of node-postgres, to destroy a connection that does not end. */ + connection?: { stream?: { destroy(): void } } +} + +/** + * The channel of a jobs table. PostgreSQL rejects a channel name over 63 + * bytes in pg_notify: a longer table name is hashed. + */ +function wakeChannelName(table: string): string { + if (Buffer.byteLength(table) <= 63) return table + return `queue_jobs_${createHash('sha256').update(table).digest('hex').slice(0, 40)}` +} + +/** + * End a LISTEN or probe client. A graceful end waits for the server, which a + * stalled connection never answers: its socket is destroyed after a while. + */ +async function endListenClient(client: PgListenClient): Promise { + client.removeAllListeners() + client.on('error', () => {}) + const timer = setTimeout( + () => client.connection?.stream?.destroy(), + subscriptionHeartbeat.timeout + ) + await client.end().catch(() => {}) + clearTimeout(timer) +} + +/** The message of a job of `queue` ready now, or due at `at`. */ +function jobMessage(queue: string, at?: number): WakeMessage { + return at === undefined ? { ready: [queue] } : { due: { [queue]: at } } +} + +const LISTEN_RECONNECT_DELAY = 1000 +const LISTEN_PROBE_TIMEOUT = 2000 +const LISTEN_MAX_RECONNECT_DELAY = 30_000 +/** + * Every NOTIFY reaches every LISTEN connection of the database: outside of + * transactions, a queue gets at most one signal per interval per adapter. + */ +const SIGNAL_INTERVAL = 50 + +/** + * A watcher and the next schedule run it was told about. + */ +interface ScheduleWatcher { + onWake: WakeListener + at: number | null +} + +/** Signals that wait for the next NOTIFY of an adapter. */ +interface PendingSignals { + ready: Set + /** The earliest due time by queue. */ + due: Map + sent: Promise } /** @@ -73,6 +165,14 @@ interface ScheduleRow extends DbRow { created_at: number | string | null } +/** + * Options of the `knex()` factory. + */ +export type KnexFactoryOptions = Pick< + KnexAdapterOptions, + 'tableName' | 'schedulesTableName' | 'notifications' +> + /** * Create a new Knex adapter factory. * Accepts either a Knex instance or a Knex configuration object. @@ -82,14 +182,22 @@ interface ScheduleRow extends DbRow { * * When passing a Knex instance, the caller is responsible for * managing the connection lifecycle. + * + * ```ts + * knex(config) + * knex(config, 'custom_jobs_table') + * knex(config, { tableName: 'custom_jobs_table', notifications: true }) + * ``` */ -export function knex(config: KnexConfig, tableName?: string) { +export function knex(config: KnexConfig, options?: string | KnexFactoryOptions) { + const resolved = typeof options === 'string' ? { tableName: options } : options + return (context?: AdapterFactoryContext) => { const isKnexInstance = typeof config === 'function' const connection = isKnexInstance ? config : KnexPkg(config) return new KnexAdapter({ + ...resolved, connection, - tableName, ownsConnection: !isKnexInstance, logger: context?.logger, }) @@ -106,15 +214,390 @@ export class KnexAdapter implements Adapter { readonly #schedulesTable: string readonly #ownsConnection: boolean readonly #logger: Logger + /** Set on PostgreSQL with notifications enabled: the jobs table names the channel. */ + readonly #wakeChannel?: string + /** Signals asked for since the last NOTIFY: see #signal. */ + #pendingSignals?: PendingSignals + /** Watchers of this adapter: see claimDueSchedule. */ + readonly #scheduleWatchers = new Set() #workerId: string = '' #scheduleDatesCheck: Promise | undefined + /** Defined on PostgreSQL with notifications enabled. See `Adapter.watch`. */ + readonly watch?: (queues: readonly string[], signal?: AbortSignal) => Promise + constructor(config: KnexAdapterOptions) { this.#connection = config.connection this.#jobsTable = config.tableName ?? 'queue_jobs' this.#schedulesTable = config.schedulesTableName ?? 'queue_schedules' this.#ownsConnection = config.ownsConnection ?? false this.#logger = config.logger ?? consoleLogger + + if (this.#connection.client.dialect === 'postgresql' && config.notifications === true) { + this.#wakeChannel = wakeChannelName(this.#jobsTable) + this.watch = async (queues, signal) => { + const wakeup = new WorkerWakeup() + const close = await this.#listen(queues, (at) => wakeup.notify(at), signal) + return { + reset: () => wakeup.reset(), + wait: (maxDelay, abort, cooldown) => wakeup.wait(maxDelay, abort, cooldown), + close, + } + } + } + } + + /** + * NOTIFY the workers. On a transaction, PostgreSQL delivers the + * notification on commit, and drops it on rollback. + */ + async #notify(connection: Knex, message: WakeMessage): Promise { + if (!this.#wakeChannel) return + + const query = connection.raw('select pg_notify(?, ?)', [ + this.#wakeChannel, + encodeWakeMessage(message), + ]) + + // A failed statement aborts the transaction: let the application see it. + if (connection.isTransaction) { + await query + return + } + + // The job is stored: a lost signal only delays it until the next poll. + try { + await query + } catch (error) { + debug('failed to notify workers: %O', error) + } + } + + /** + * Signal work of `queue`: ready now, or due at `at`. + * + * In a transaction, the NOTIFY joins it. Otherwise, the work is stored and + * the signal is sent in the background: PostgreSQL serializes the commits + * that NOTIFY, and delivers each NOTIFY to every LISTEN connection. An + * adapter sends at most one NOTIFY per SIGNAL_INTERVAL, for all its queues; + * the signals asked for meanwhile join the next one, with the earliest due + * time of each queue. That is enough: promoting a delayed job announces the + * next one. `destroy()` waits for them. + */ + async #signal(queue: string, at?: number): Promise { + if (!this.#wakeChannel) return + const ready = at === undefined || at <= Date.now() + if (this.#connection.isTransaction) { + return this.#notify(this.#connection, jobMessage(queue, ready ? undefined : at)) + } + + const pending = this.#pendingSignals ?? { + ready: new Set(), + due: new Map(), + sent: Promise.resolve(), + } + if (ready) { + pending.ready.add(queue) + } else { + pending.due.set(queue, Math.min(pending.due.get(queue) ?? Infinity, at!)) + } + if (this.#pendingSignals) return + + this.#pendingSignals = pending + pending.sent = (async () => { + while (pending.ready.size > 0 || pending.due.size > 0) { + const message: WakeMessage = {} + if (pending.ready.size > 0) message.ready = [...pending.ready] + if (pending.due.size > 0) message.due = Object.fromEntries(pending.due) + pending.ready.clear() + pending.due.clear() + await this.#notify(this.#connection, message) + await sleep(SIGNAL_INTERVAL) + } + this.#pendingSignals = undefined + })() + } + + /** + * Insert a job row and signal it. In a transaction, the insert and the + * NOTIFY run in one statement, so the signal costs no round trip. + */ + async #insertJob(row: Record, queue: string, at?: number): Promise { + const insert = this.#connection(this.#jobsTable).insert(row) + if (!this.#wakeChannel || !this.#connection.isTransaction) { + await insert + await this.#signal(queue, at) + return + } + + await this.#connection.raw('with inserted as (?) select pg_notify(?, ?)', [ + insert, + this.#wakeChannel, + encodeWakeMessage(jobMessage(queue, at)), + ]) + } + + /** Tell watchers the next schedule run, read in the same transaction. */ + async #notifySchedules(connection: Knex): Promise { + if (!this.#wakeChannel) return + const at = await this.#nextScheduleAt(connection) + if (at !== null) await this.#notify(connection, { schedule: at }) + } + + /** + * LISTEN on a connection outside of the pool, held for the whole + * subscription. A lost connection is replaced with a growing delay. + * + * The client is created from the driver of Knex rather than with + * acquireRawConnection, which only returns it once connected: ending the + * subscription must be able to abort a connection still in its handshake. + */ + async #listen( + queues: readonly string[], + onWake: WakeListener, + signal?: AbortSignal + ): Promise<() => Promise> { + const channel = this.#wakeChannel! + const knexClient = this.#connection.client + const watcher: ScheduleWatcher = { onWake, at: null } + let client: PgListenClient | undefined + let closed = false + let reconnectTimer: NodeJS.Timeout | undefined + let reconnectDelay = LISTEN_RECONNECT_DELAY + let listening = false + // A client ended in its handshake never settles connect(): closing does. + const closing = Promise.withResolvers() + closing.promise.catch(() => {}) + + const release = async () => { + const current = client + client = undefined + if (!current) return + await endListenClient(current) + } + + // A connection config provider is called again once its settings expire, + // as the pool of Knex does, for rotated credentials for instance. + let providedSettings: object | undefined + let settingsExpired: (() => boolean) | null = null + // Only the latest lookup is cached: an abandoned one may answer later. + let lookups = 0 + const createClient = async (): Promise => { + const provider = knexClient.connectionConfigProvider + if (!provider) return new knexClient.driver.Client(knexClient.connectionSettings) + if (providedSettings && !settingsExpired?.()) { + return new knexClient.driver.Client(providedSettings) + } + + const lookup = ++lookups + const { expirationChecker, ...settings } = await provider() + if (lookup === lookups) { + providedSettings = settings + settingsExpired = expirationChecker ?? null + } + return new knexClient.driver.Client(settings) + } + + // Signals sent before listening, or while disconnected, are lost. The + // LISTEN connection reads them: the pool keeps a connection it adds. + const announce = async (connection: PgListenClient) => { + const read = async (query: Knex.QueryBuilder) => { + const { sql, bindings } = query.toSQL().toNative() + return (await connection.query(sql, [...bindings])).rows + } + onWake(Date.now()) + // One per queue: promoting a job only announces the next one of its queue. + const rows = await read( + this.#connection(this.#jobsTable) + .whereIn('queue', [...queues]) + .where('status', 'delayed') + .groupBy('queue') + .min('execute_at as at') + ) + for (const row of rows) { + if (row.at !== null) onWake(Number(row.at)) + } + const [next] = await read(this.#nextScheduleQuery(this.#connection)) + watcher.at = next?.at === null || next?.at === undefined ? null : Number(next.at) + if (watcher.at !== null) onWake(watcher.at) + } + + const onLost = (error?: Error) => { + listening = false + if (closed || reconnectTimer) return + this.#logger.error({ err: error }, 'Lost the LISTEN connection, reconnecting') + void release() + reconnectTimer = setTimeout(() => { + reconnectTimer = undefined + if (closed) return + connect().then( + () => (reconnectDelay = LISTEN_RECONNECT_DELAY), + (connectError) => { + reconnectDelay = Math.min(reconnectDelay * 2, LISTEN_MAX_RECONNECT_DELAY) + onLost(connectError) + } + ) + }, reconnectDelay) + } + + // A pooler in transaction mode, such as PgBouncer, accepts LISTEN but + // never delivers: check that a NOTIFY sent by another connection, outside + // of the pool, arrives. + let probe: { payload: string; received: PromiseWithResolvers } | undefined + const verify = async () => { + probe = { + payload: encodeWakeMessage({ probe: randomUUID() }), + received: Promise.withResolvers(), + } + const { received, payload } = probe + const sender = await createClient() + // An error event without a listener would crash the process. + const failed = Promise.withResolvers() + failed.promise.catch(() => {}) + sender.on('error', (error) => failed.reject(error)) + let timer: NodeJS.Timeout | undefined + const timeout = new Promise((_resolve, reject) => { + timer = setTimeout( + () => + reject( + new Error( + 'LISTEN received no notification within 2s. Behind a pooler in transaction ' + + 'mode, such as PgBouncer, do not enable notifications' + ) + ), + LISTEN_PROBE_TIMEOUT + ) + }) + // The timeout covers the connection of the sender too: it may stall. + const exchange = (async () => { + await sender.connect() + await sender.query('select pg_notify($1, $2)', [channel, payload]) + await received.promise + })() + exchange.catch(() => {}) + try { + await Promise.race([exchange, failed.promise, timeout, closing.promise]) + } finally { + clearTimeout(timer) + probe = undefined + void endListenClient(sender) + } + } + + const connect = async () => { + // Set when the caller gave up on this attempt: a late client stays unused. + let abandoned = false + const start = async () => { + // A connection config provider may look settings up: it may stall too. + const connection = await createClient() + if (abandoned) return + client = connection + connection.on('error', onLost) + connection.on('end', () => onLost()) + connection.on('notification', (message) => { + if (message.channel !== channel || !message.payload) return + if (message.payload === probe?.payload) return probe.received.resolve() + dispatchWakeMessage(message.payload, queues, onWake, (at) => { + watcher.at = at + onWake(at) + }) + }) + await connection.connect() + // Announcements read unqualified tables, as the pool does. + await knexClient.setSchemaSearchPath(connection) + await connection.query(`LISTEN "${channel.replaceAll('"', '""')}"`) + if (closed) return + await verify() + listening = true + await announce(connection) + } + // A connection may stall before the heartbeat watches it: give up on + // it, so that the caller falls back to polling or reconnects. + let timer: NodeJS.Timeout | undefined + const stalled = new Promise((_resolve, reject) => { + timer = setTimeout( + () => reject(new Error('The LISTEN connection did not start in time')), + LISTEN_PROBE_TIMEOUT + subscriptionHeartbeat.timeout + ) + }) + const starting = start() + starting.catch(() => {}) + try { + await Promise.race([starting, stalled, closing.promise]) + } catch (error) { + abandoned = true + throw error + } finally { + clearTimeout(timer) + } + if (closed) return release() + } + const stopHeartbeat = startHeartbeat( + () => (listening ? client?.query('select 1') : undefined), + () => onLost(new Error('The LISTEN connection did not answer a ping')) + ) + + signal?.throwIfAborted() + this.#scheduleWatchers.add(watcher) + const close = async () => { + closed = true + closing.reject(new Error('The LISTEN subscription is closed')) + stopHeartbeat() + clearTimeout(reconnectTimer) + signal?.removeEventListener('abort', abort) + this.#scheduleWatchers.delete(watcher) + await release() + } + // Ends the first connection, even in its handshake. + const abort = () => void close() + signal?.addEventListener('abort', abort, { once: true }) + + try { + await connect() + signal?.throwIfAborted() + } catch (error) { + // The connection may have failed with an error event: no reconnection. + await close() + throw error + } + + return close + } + + async #announceNextSchedule(watcher: ScheduleWatcher, onlyFuture = false): Promise { + watcher.at = await this.#nextScheduleAt() + if (watcher.at === null) return + if (onlyFuture && watcher.at <= Date.now()) return + watcher.onWake(watcher.at) + } + + /** + * Next run of a schedule that claimDueSchedule can claim: a schedule past + * its run limit or end date stays active but never runs again. + */ + async #nextScheduleAt(connection: Knex = this.#connection): Promise { + const row = await this.#nextScheduleQuery(connection) + return row?.at === null || row?.at === undefined ? null : Number(row.at) + } + + #nextScheduleQuery(connection: Knex) { + return connection(this.#schedulesTable) + .where('status', 'active') + .whereNotNull('next_run_at') + .where((builder) => { + builder.whereNull('run_limit').orWhereRaw('run_count < run_limit') + }) + .where((builder) => { + builder + .whereNull('to_date') + .orWhere((range) => + range + .where('to_date', '>=', Date.now()) + .whereRaw('?? >= ??', ['to_date', 'next_run_at']) + ) + }) + .min('next_run_at as at') + .first() } setWorkerId(workerId: string): void { @@ -122,6 +605,7 @@ export class KnexAdapter implements Adapter { } async destroy(): Promise { + await this.#pendingSignals?.sent if (this.#ownsConnection) { await this.#connection.destroy() } @@ -321,6 +805,20 @@ export class KnexAdapter implements Adapter { execute_at: null, }) } + + // Workers only know the next delayed job: announce the one after it. + // Due rows left are being promoted by concurrent transactions. + if (this.#wakeChannel) { + const next = await trx(this.#jobsTable) + .where('queue', queue) + .where('status', 'delayed') + .where('execute_at', '>', now) + .min('execute_at as at') + .first() + if (next?.at !== null && next?.at !== undefined) { + await this.#notify(trx, jobMessage(queue, Number(next.at))) + } + } }) } @@ -518,6 +1016,7 @@ export class KnexAdapter implements Adapter { score: null, execute_at: retryAt.getTime(), }) + if (updated > 0) await this.#signal(queue, retryAt.getTime()) return updated > 0 } @@ -533,6 +1032,7 @@ export class KnexAdapter implements Adapter { score, execute_at: null, }) + if (updated > 0) await this.#signal(queue) return updated > 0 } @@ -546,22 +1046,22 @@ export class KnexAdapter implements Adapter { const score = calculateScore(priority, timestamp) if (jobData.dedup) { - return this.#pushWithDedup(queue, jobData, { + const result = await this.#pushWithDedup(queue, jobData, { id: jobData.id, queue, status: 'pending', data: JSON.stringify(jobData), score, }) + // Other outcomes leave the existing job as it was. + if (result.outcome === 'added') await this.#signal(queue) + return result } - await this.#connection(this.#jobsTable).insert({ - id: jobData.id, - queue, - status: 'pending', - data: JSON.stringify(jobData), - score, - }) + await this.#insertJob( + { id: jobData.id, queue, status: 'pending', data: JSON.stringify(jobData), score }, + queue + ) } async pushLater(jobData: JobData, delay: number): Promise { @@ -572,22 +1072,30 @@ export class KnexAdapter implements Adapter { const executeAt = Date.now() + delay if (jobData.dedup) { - return this.#pushWithDedup(queue, jobData, { + const result = await this.#pushWithDedup(queue, jobData, { id: jobData.id, queue, status: 'delayed', data: JSON.stringify(jobData), execute_at: executeAt, }) + // A due time no job has would wake watchers in vain, and could hide + // the one of another job signalled with it. + if (result.outcome === 'added') await this.#signal(queue, executeAt) + return result } - await this.#connection(this.#jobsTable).insert({ - id: jobData.id, + await this.#insertJob( + { + id: jobData.id, + queue, + status: 'delayed', + data: JSON.stringify(jobData), + execute_at: executeAt, + }, queue, - status: 'delayed', - data: JSON.stringify(jobData), - execute_at: executeAt, - }) + executeAt + ) } async #pushWithDedup( @@ -773,6 +1281,7 @@ export class KnexAdapter implements Adapter { for (let start = 0; start < rows.length; start += SQL_INSERT_CHUNK_SIZE) { await connection(this.#jobsTable).insert(rows.slice(start, start + SQL_INSERT_CHUNK_SIZE)) } + await this.#notify(connection, jobMessage(queue)) } if (this.#connection.isTransaction) { @@ -860,6 +1369,7 @@ export class KnexAdapter implements Adapter { } } + if (recovered > 0) await this.#notify(trx, jobMessage(queue)) return { recovered, exceeded } }) } @@ -941,6 +1451,7 @@ export class KnexAdapter implements Adapter { ]), ...definition, }) + await this.#notifySchedules(this.#connection) return id } @@ -991,6 +1502,7 @@ export class KnexAdapter implements Adapter { if (Object.keys(data).length > 0) { await this.#connection(this.#schedulesTable).where('id', id).update(data) + await this.#notifySchedules(this.#connection) } } @@ -998,6 +1510,7 @@ export class KnexAdapter implements Adapter { await this.#assertScheduleDatesMigrated() await this.#connection(this.#schedulesTable).where('id', id).delete() + await this.#notifySchedules(this.#connection) } async claimDueSchedule(): Promise { @@ -1005,7 +1518,7 @@ export class KnexAdapter implements Adapter { const now = Date.now() - return this.#claimTransaction(this.#schedulesTable, async (trx) => { + const claimed = await this.#claimTransaction(this.#schedulesTable, async (trx) => { // Find one due schedule with row locking let query = trx(this.#schedulesTable) .where('status', 'active') @@ -1059,10 +1572,25 @@ export class KnexAdapter implements Adapter { last_run_at: now, run_count: newRunCount, }) + await this.#notifySchedules(trx) // Return schedule data (before update state for payload) return this.#rowToScheduleData(row) }) + + // A watcher woken for a schedule that expired meanwhile, past its end + // date or run limit, waits for nothing: tell it the next claimable one. + // A due one is locked by a concurrent claim, which notifies on commit: + // waking for it again would only spin. + if (!claimed) { + for (const watcher of this.#scheduleWatchers) { + if (watcher.at !== null && watcher.at <= now) { + await this.#announceNextSchedule(watcher, true) + } + } + } + + return claimed } #rowToScheduleData(row: ScheduleRow): ScheduleData { diff --git a/src/drivers/redis_adapter.ts b/src/drivers/redis_adapter.ts index a8eb730..4103a8f 100644 --- a/src/drivers/redis_adapter.ts +++ b/src/drivers/redis_adapter.ts @@ -8,8 +8,11 @@ import type { JobLease, PushResult, StalledJobsRecovery, + WorkWatcher, } from '../contracts/adapter.js' -import { consoleLogger } from '../logger.js' +import { consoleLogger, type Logger } from '../logger.js' +import { RedisWorkWatcher } from './redis_work_watcher.js' +import { subscriptionHeartbeat } from './wake_messages.js' import type { AdapterFactoryContext, DedupOutcome } from '../types/main.js' import type { JobData, @@ -35,6 +38,7 @@ import { REMOVE_JOB_SCRIPT, RENEW_JOBS_SCRIPT, RETRY_JOB_SCRIPT, + SEED_MARKERS_SCRIPT, UPDATE_SCHEDULE_SCRIPT, UPSERT_SCHEDULE_SCRIPT, } from './redis_scripts.js' @@ -45,6 +49,8 @@ const schedulesDueKey = 'schedules::due' // Schedule hashes live one level below the index keys so no schedule id can collide with them. const scheduleDataPrefix = 'schedules::data::' const legacyScheduleDataPrefix = 'schedules::' +// Below the due index: a 0.7 schedule hash lives at `schedules::`. +const schedulesMarkerKey = 'schedules::due::marker' /** * An ioredis connection, or the options to create one. @@ -69,7 +75,7 @@ function isRedisConnection(config?: RedisConfig): config is Redis { export function redis(config?: RedisConfig) { return ({ logger }: AdapterFactoryContext = { logger: consoleLogger }) => { if (isRedisConnection(config)) { - return new RedisAdapter(config, false) + return new RedisAdapter(config, false, logger) } const options: RedisOptions = { @@ -86,17 +92,77 @@ export function redis(config?: RedisConfig) { logger.error({ err: error }, 'Redis connection error') }) - return new RedisAdapter(connection, true) + return new RedisAdapter(connection, true, logger) } } export class RedisAdapter implements Adapter { readonly #connection: Redis readonly #ownsConnection: boolean + readonly #logger: Logger #workerId: string = '' - constructor(connection: Redis, ownsConnection: boolean = false) { + constructor(connection: Redis, ownsConnection: boolean = false, logger: Logger = consoleLogger) { this.#connection = connection this.#ownsConnection = ownsConnection + this.#logger = logger + } + + /** + * Wait for work with BZPOPMIN on the markers of the queues and schedules, + * through a dedicated connection: a blocking command holds its connection. + */ + async watch(queues: readonly string[], signal?: AbortSignal): Promise { + signal?.throwIfAborted() + const markers = queues.map((queue) => ({ + key: this.#getKeys(queue).marker, + source: this.#getKeys(queue).delayed, + })) + markers.push({ key: schedulesMarkerKey, source: schedulesDueKey }) + + // A command interrupted by a reconnection fails instead of blocking again. + const blocking = this.#connection.duplicate({ autoResendUnfulfilledCommands: false }) + blocking.on('error', (error) => { + this.#logger.error({ err: error }, 'Redis blocking connection error') + }) + // A command queued during a silent handshake, or dropped on reconnection, + // never settles: abort and a deadline reject, so that the worker polls. + const aborted = Promise.withResolvers() + aborted.promise.catch(() => {}) + const abort = () => aborted.reject(signal!.reason) + signal?.addEventListener('abort', abort, { once: true }) + const deadline = setTimeout( + () => aborted.reject(new Error('The Redis blocking connection did not start in time')), + subscriptionHeartbeat.interval + ) + + try { + const seedKeys = queues.flatMap((queue) => { + const keys = this.#getKeys(queue) + return [keys.pending, keys.delayed, keys.marker] + }) + seedKeys.push(schedulesDueKey, schedulesMarkerKey) + // A user without the @blocking ACL category cannot wait: probe BZPOPMIN, + // so that such a worker polls instead. In MULTI, it returns at once. + const probe = blocking + .multi() + .bzpopmin(`${markers[0].key}::probe`, 0) + .exec() + .catch((error: { previousErrors?: Error[] }) => { + // EXECABORT tells nothing: the NOPERM error is the one to log. + throw error.previousErrors?.[0] ?? error + }) + const seed = this.#connection.eval(SEED_MARKERS_SCRIPT, seedKeys.length, ...seedKeys) + await Promise.race([Promise.all([probe, seed]), aborted.promise]) + signal?.throwIfAborted() + } catch (error) { + blocking.disconnect() + throw error + } finally { + clearTimeout(deadline) + signal?.removeEventListener('abort', abort) + } + + return new RedisWorkWatcher(this.#connection, blocking, markers, this.#logger) } #getKeys(queue: string) { @@ -106,6 +172,7 @@ export class RedisAdapter implements Adapter { delayed: `${redisKey}::${queue}::delayed`, active: `${redisKey}::${queue}::active`, overlay: `${redisKey}::${queue}::metadata`, + marker: `${redisKey}::${queue}::marker`, completed: `${redisKey}::${queue}::completed`, completedIndex: `${redisKey}::${queue}::completed::index`, failed: `${redisKey}::${queue}::failed`, @@ -146,12 +213,13 @@ export class RedisAdapter implements Adapter { const result = await this.#connection.eval( ACQUIRE_JOB_SCRIPT, - 5, + 6, keys.data, keys.pending, keys.active, keys.delayed, keys.overlay, + keys.marker, this.#workerId, now.toString(), leaseToken @@ -261,12 +329,13 @@ export class RedisAdapter implements Adapter { const retried = await this.#connection.eval( RETRY_JOB_SCRIPT, - 5, + 6, keys.data, keys.active, keys.pending, keys.delayed, keys.overlay, + keys.marker, job.id, retryAt ? retryAt.getTime().toString() : '0', now.toString(), @@ -320,12 +389,13 @@ export class RedisAdapter implements Adapter { const [payloadData, payloadIsUndefined] = encodeRedisJobPayloadOverlay(jobData.payload) const result = (await this.#connection.eval( PUSH_DEDUP_JOB_SCRIPT, - 5, + 6, keys.data, keys.delayed, dedupKey, keys.pending, keys.overlay, + keys.marker, jobData.id, JSON.stringify(jobData), executeAt.toString(), @@ -333,17 +403,19 @@ export class RedisAdapter implements Adapter { jobData.dedup.extend ? '1' : '0', jobData.dedup.replace ? '1' : '0', payloadData, - payloadIsUndefined + payloadIsUndefined, + '1' )) as [string, string] return { outcome: result[0] as DedupOutcome, jobId: result[1] } } await this.#connection.eval( PUSH_DELAYED_JOB_SCRIPT, - 3, + 4, keys.data, keys.delayed, keys.overlay, + keys.marker, jobData.id, JSON.stringify(jobData), executeAt.toString() @@ -361,12 +433,13 @@ export class RedisAdapter implements Adapter { const [payloadData, payloadIsUndefined] = encodeRedisJobPayloadOverlay(jobData.payload) const result = (await this.#connection.eval( PUSH_DEDUP_JOB_SCRIPT, - 5, + 6, keys.data, keys.pending, dedupKey, keys.delayed, keys.overlay, + keys.marker, jobData.id, JSON.stringify(jobData), score.toString(), @@ -374,17 +447,19 @@ export class RedisAdapter implements Adapter { jobData.dedup.extend ? '1' : '0', jobData.dedup.replace ? '1' : '0', payloadData, - payloadIsUndefined + payloadIsUndefined, + '0' )) as [string, string] return { outcome: result[0] as DedupOutcome, jobId: result[1] } } await this.#connection.eval( PUSH_JOB_SCRIPT, - 3, + 4, keys.data, keys.pending, keys.overlay, + keys.marker, jobData.id, JSON.stringify(jobData), score.toString() @@ -414,6 +489,7 @@ export class RedisAdapter implements Adapter { multi.zadd(keys.pending, score, job.id) } + multi.zadd(keys.marker, 0, '0') await multi.exec() } @@ -439,11 +515,12 @@ export class RedisAdapter implements Adapter { const [recovered, ...exceeded] = (await this.#connection.eval( RECOVER_STALLED_JOBS_SCRIPT, - 4, + 5, keys.data, keys.active, keys.pending, keys.overlay, + keys.marker, now.toString(), stalledThreshold.toString(), maxStalledCount.toString(), @@ -506,11 +583,12 @@ export class RedisAdapter implements Adapter { const unfinalizedClaim = (await this.#connection.eval( UPSERT_SCHEDULE_SCRIPT, - 4, + 5, scheduleKey, schedulesIndexKey, schedulesDueKey, `${legacyScheduleDataPrefix}${id}`, + schedulesMarkerKey, id, now.toString(), JSON.stringify(scheduleData), @@ -611,9 +689,10 @@ export class RedisAdapter implements Adapter { await this.#connection.eval( UPDATE_SCHEDULE_SCRIPT, - 2, + 3, scheduleKey, schedulesDueKey, + schedulesMarkerKey, id, JSON.stringify(data) ) @@ -647,9 +726,10 @@ export class RedisAdapter implements Adapter { while (true) { const result = (await this.#connection.eval( CLAIM_SCHEDULE_SCRIPT, - 2, + 3, schedulesDueKey, scheduleDataPrefix, + schedulesMarkerKey, now.toString(), claimToken )) as ['claimed' | 'unfinalized', string] | null @@ -716,9 +796,10 @@ export class RedisAdapter implements Adapter { await this.#connection.eval( FINALIZE_CRON_SCHEDULE_SCRIPT, - 2, + 3, `${scheduleDataPrefix}${claim.id}`, schedulesDueKey, + schedulesMarkerKey, claim.id, claim.cronExpression, claim.configRevision, diff --git a/src/drivers/redis_scripts.ts b/src/drivers/redis_scripts.ts index 9efe584..103218d 100644 --- a/src/drivers/redis_scripts.ts +++ b/src/drivers/redis_scripts.ts @@ -1,5 +1,100 @@ import { REDIS_DEDUP_LUA, REDIS_JOB_STORAGE_LUA } from './redis_job_storage.js' +/** + * Markers wake idle workers, as in BullMQ. A marker key is a sorted set that + * workers wait on with BZPOPMIN: each marker wakes one worker only, and a + * marker written while every worker is busy waits for the first idle one. + * + * - Member '0', score 0: a job is ready. Many pushes leave one marker. The + * worker that pops a job writes it again while jobs remain, to wake the + * next worker. + * - Member '1': the earliest known time at which a delayed job or a schedule + * becomes due. A worker holds it until then. + * + * The marker key of a script is its last key. + */ +const MARKER_LUA = ` + local marker_key = KEYS[#KEYS] + + local function mark_ready() + redis.call('ZADD', marker_key, 0, '0') + end + + -- A later time would hide the earlier one. + local function mark_due(at) + local current = redis.call('ZSCORE', marker_key, '1') + if not current or tonumber(current) > at then + redis.call('ZADD', marker_key, at, '1') + end + end + + -- Exactly the first score of a sorted set, once the previous one is handled. + local function mark_first_of(key) + local first = redis.call('ZRANGE', key, 0, 0, 'WITHSCORES') + if #first > 0 then + redis.call('ZADD', marker_key, first[2], '1') + else + redis.call('ZREM', marker_key, '1') + end + end +` + +/** + * Writes the markers of work stored before a worker waits for it, or by an + * older version of this package. KEYS = per queue its pending, delayed and + * marker keys, then the schedules due index and its marker. + */ +export const SEED_MARKERS_SCRIPT = ` + local function seed(source_key, marker_key, ready) + local first = redis.call('ZRANGE', source_key, 0, 0, 'WITHSCORES') + if #first == 0 then + return + end + if ready then + redis.call('ZADD', marker_key, 0, '0') + return + end + local current = redis.call('ZSCORE', marker_key, '1') + if not current or tonumber(current) > tonumber(first[2]) then + redis.call('ZADD', marker_key, first[2], '1') + end + end + + for i = 1, #KEYS - 2, 3 do + seed(KEYS[i], KEYS[i + 2], true) + seed(KEYS[i + 1], KEYS[i + 2], false) + end + seed(KEYS[#KEYS - 1], KEYS[#KEYS], false) +` + +/** + * Marks the first entry of a sorted set due after now, ARGV[1]. KEYS = the + * sorted set (delayed jobs or schedules), its marker. A worker woken by a + * due marker runs it: what is due now is its own work, and the marker of + * the next entry is not lost if its cycle writes none. + */ +export const ADVANCE_MARKER_SCRIPT = ` + local next = redis.call('ZRANGEBYSCORE', KEYS[1], '(' .. ARGV[1], '+inf', 'WITHSCORES', 'LIMIT', 0, 1) + if #next == 0 then + return + end + local current = redis.call('ZSCORE', KEYS[2], '1') + if not current or tonumber(current) > tonumber(next[2]) then + redis.call('ZADD', KEYS[2], next[2], '1') + end +` + +/** + * Puts back a marker a worker popped but does not act on: ARGV = member, + * score. A due time only replaces a later one. + */ +export const RELEASE_MARKER_SCRIPT = ` + local current = redis.call('ZSCORE', KEYS[1], ARGV[1]) + if not current or tonumber(current) > tonumber(ARGV[2]) then + redis.call('ZADD', KEYS[1], ARGV[2], ARGV[1]) + end +` + /** * Lua script for pushing a job to the queue. * Stores job data in the central hash and adds jobId to pending ZSET. @@ -13,9 +108,11 @@ export const PUSH_JOB_SCRIPT = ` local score = tonumber(ARGV[3]) ${REDIS_JOB_STORAGE_LUA} +${MARKER_LUA} store_job_data(data_key, overlay_key, job_id, job_data) redis.call('ZADD', pending_key, score, job_id) + mark_ready() return 1 ` @@ -54,9 +151,18 @@ export const PUSH_DEDUP_JOB_SCRIPT = ` local replace = tonumber(ARGV[6]) local payload_data = ARGV[7] local payload_is_undefined = tonumber(ARGV[8]) + local delayed = ARGV[9] == '1' ${REDIS_JOB_STORAGE_LUA} ${REDIS_DEDUP_LUA} +${MARKER_LUA} + + -- The job may be skipped: a spurious marker only costs one worker a look. + if delayed then + mark_due(score) + else + mark_ready() + end local existing_result = resolve_dedup_existing_job( data_key, @@ -95,9 +201,11 @@ export const PUSH_DELAYED_JOB_SCRIPT = ` local execute_at = tonumber(ARGV[3]) ${REDIS_JOB_STORAGE_LUA} +${MARKER_LUA} store_job_data(data_key, overlay_key, job_id, job_data) redis.call('ZADD', delayed_key, execute_at, job_id) + mark_due(execute_at) return 1 ` @@ -120,6 +228,7 @@ export const ACQUIRE_JOB_SCRIPT = ` local lease_token = ARGV[3] ${REDIS_JOB_STORAGE_LUA} +${MARKER_LUA} -- Process delayed jobs: move ready jobs to pending local ready_job_ids = redis.call('ZRANGEBYSCORE', delayed_key, 0, now) @@ -134,6 +243,8 @@ ${REDIS_JOB_STORAGE_LUA} redis.call('ZREM', delayed_key, job_id) end end + + mark_first_of(delayed_key) end -- Pop highest priority job (lowest score) @@ -142,6 +253,11 @@ ${REDIS_JOB_STORAGE_LUA} return nil end + -- Wake the next idle worker while jobs remain. + if redis.call('ZCARD', pending_key) > 0 then + mark_ready() + end + local job_id = result[1] local job_data = redis.call('HGET', data_key, job_id) if not job_data then @@ -318,6 +434,7 @@ export const RETRY_JOB_SCRIPT = ` ${REDIS_JOB_STORAGE_LUA} ${REDIS_LEASE_LUA} +${MARKER_LUA} -- Verify the caller still holds the job if not holds_lease(active_key, job_id, lease_token) then @@ -345,12 +462,14 @@ ${REDIS_LEASE_LUA} -- Add back to pending or delayed if retry_at and retry_at > now then redis.call('ZADD', delayed_key, retry_at, job_id) + mark_due(retry_at) 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_id) + mark_ready() end return 1 @@ -378,6 +497,7 @@ export const RECOVER_STALLED_JOBS_SCRIPT = ` local lease_token_prefix = ARGV[6] ${REDIS_JOB_STORAGE_LUA} +${MARKER_LUA} local recovered = 0 -- The first element is replaced by the recovered count before returning. @@ -433,6 +553,9 @@ ${REDIS_JOB_STORAGE_LUA} end result[1] = recovered + if recovered > 0 then + mark_ready() + end return result ` @@ -590,6 +713,7 @@ export const UPSERT_SCHEDULE_SCRIPT = ` local next_run_at = ARGV[4] ${SCHEDULE_DUE_INDEX_LUA} +${MARKER_LUA} -- The same test as migrate(): an indexed id with a 0.7 hash at its old key. if redis.call('SISMEMBER', schedules_index_key, id) == 1 @@ -642,6 +766,7 @@ ${SCHEDULE_DUE_INDEX_LUA} ) redis.call('SADD', schedules_index_key, id) sync_schedule_due_index(schedule_key, due_key, id) + mark_first_of(due_key) -- A finished schedule never keeps a claim token, so an empty next run with -- one is a cron claim not finalized yet. @@ -668,9 +793,11 @@ export const UPDATE_SCHEDULE_SCRIPT = ` local updates = cjson.decode(ARGV[2]) ${SCHEDULE_DUE_INDEX_LUA} +${MARKER_LUA} if redis.call('EXISTS', schedule_key) == 0 then redis.call('ZREM', due_key, id) + mark_first_of(due_key) return 0 end @@ -685,6 +812,7 @@ ${SCHEDULE_DUE_INDEX_LUA} end sync_schedule_due_index(schedule_key, due_key, id) + mark_first_of(due_key) return 1 ` @@ -703,9 +831,11 @@ export const FINALIZE_CRON_SCHEDULE_SCRIPT = ` local next_run_at = ARGV[5] ${SCHEDULE_DUE_INDEX_LUA} +${MARKER_LUA} if redis.call('EXISTS', schedule_key) == 0 then redis.call('ZREM', due_key, id) + mark_first_of(due_key) return 0 end @@ -723,6 +853,7 @@ ${SCHEDULE_DUE_INDEX_LUA} redis.call('HDEL', schedule_key, 'claim_token', 'claimed_at') end sync_schedule_due_index(schedule_key, due_key, id) + mark_first_of(due_key) return 0 end @@ -739,6 +870,7 @@ ${SCHEDULE_DUE_INDEX_LUA} redis.call('HSET', schedule_key, 'next_run_at', next_run_at) redis.call('HDEL', schedule_key, 'claim_token', 'claimed_at') sync_schedule_due_index(schedule_key, due_key, id) + mark_first_of(due_key) return 1 ` @@ -852,6 +984,7 @@ ${SCHEDULE_DUE_INDEX_LUA} * KEYS[2] = schedule key prefix (e.g. "schedules::data::") * ARGV[1] = now (epoch milliseconds) * ARGV[2] = claim token, kept by the schedule when it is a cron + * KEYS[3] = schedules marker, written when the due index changes * * Returns nil, or `{ 'claimed' | 'unfinalized', }`. */ @@ -862,14 +995,22 @@ export const CLAIM_SCHEDULE_SCRIPT = ` local claim_token = ARGV[2] ${SCHEDULE_DUE_INDEX_LUA} +${MARKER_LUA} + + local skipped = false while true do local candidates = redis.call('ZRANGEBYSCORE', due_key, '-inf', tostring(now), 'LIMIT', 0, 1) if #candidates == 0 then + if skipped then + mark_first_of(due_key) + end return nil end + skipped = true + local id = candidates[1] local schedule_key = prefix .. id @@ -952,6 +1093,7 @@ ${SCHEDULE_DUE_INDEX_LUA} 'claimed_at', new_claimed_at) sync_schedule_due_index(schedule_key, due_key, id) + mark_first_of(due_key) -- Return the schedule data (before update) as JSON return { 'claimed', cjson.encode(schedule) } diff --git a/src/drivers/redis_work_watcher.ts b/src/drivers/redis_work_watcher.ts new file mode 100644 index 0000000..eef59d3 --- /dev/null +++ b/src/drivers/redis_work_watcher.ts @@ -0,0 +1,213 @@ +import { setTimeout as sleep } from 'node:timers/promises' +import type { Redis } from 'ioredis' +import type { WorkWatcher } from '../contracts/adapter.js' +import type { Logger } from '../logger.js' +import { ADVANCE_MARKER_SCRIPT, RELEASE_MARKER_SCRIPT } from './redis_scripts.js' +import { subscriptionHeartbeat } from './wake_messages.js' + +/** + * A marker key, and the sorted set whose due times it announces: the + * delayed jobs of a queue, or the schedules. + */ +export interface RedisMarker { + key: string + source: string +} + +interface PoppedMarker { + key: string + member: string + score: number +} + +const ERROR_RETRY_DELAY_MS = 1_000 + +/** + * Waits for the markers that writers add for the queues and schedules (see + * MARKER_LUA). BZPOPMIN pops each marker for one waiting worker only, so a + * job wakes one worker, not all of them: `cooldown` does not apply. + * + * A popped due time is held while the worker waits for it, and put back + * when the worker stops waiting first, so that another idle worker gets it. + * A marker popped once the worker stopped waiting is put back as well. + */ +export class RedisWorkWatcher implements WorkWatcher { + readonly #connection: Redis + readonly #blocking: Redis + readonly #markers: readonly RedisMarker[] + readonly #logger: Logger + /** The key prefix of the connection: replies of BZPOPMIN include it. */ + readonly #prefix: string + /** Due times popped by the current wait, by marker key. */ + readonly #held = new Map() + /** The BZPOPMIN in flight. It may outlive the wait that sent it. */ + #reading?: Promise + /** Receives the result of the BZPOPMIN in flight while a wait reads it. */ + #deliver?: (marker: PoppedMarker | null) => void + #closed = false + /** Ends the BZPOPMIN in flight when its connection closes. */ + #onClose?: () => void + + constructor(connection: Redis, blocking: Redis, markers: readonly RedisMarker[], logger: Logger) { + this.#connection = connection + this.#blocking = blocking + this.#markers = markers + this.#logger = logger + this.#prefix = blocking.options?.keyPrefix ?? '' + blocking.on('close', () => this.#onClose?.()) + } + + reset(): void {} + + async wait(maxDelay: number, signal: AbortSignal): Promise { + const deadline = Date.now() + maxDelay + + try { + while (!signal.aborted && !this.#closed) { + const now = Date.now() + const [heldKey, heldAt] = this.#nextHeld() + if (heldKey !== undefined && heldAt <= now) { + this.#held.delete(heldKey) + await this.#advance(heldKey, now) + return true + } + if (deadline <= now) return false + + const marker = await this.#read(Math.min(deadline, heldAt), signal) + if (!marker) continue + if (marker.member === '0') return true + if (marker.score <= Date.now()) { + await this.#advance(marker.key, Date.now()) + return true + } + + const held = this.#held.get(marker.key) + if (held === undefined || marker.score < held) this.#held.set(marker.key, marker.score) + } + return false + } finally { + for (const [key, at] of this.#held) void this.#release({ key, member: '1', score: at }) + this.#held.clear() + } + } + + async close(): Promise { + this.#closed = true + this.#deliver?.(null) + this.#blocking.disconnect() + } + + #nextHeld(): [string | undefined, number] { + let next: [string | undefined, number] = [undefined, Infinity] + for (const [key, at] of this.#held) { + if (at < next[1]) next = [key, at] + } + return next + } + + /** + * Resolve with the next popped marker, or null at `until`, on abort, or + * when the BZPOPMIN in flight times out. + */ + #read(until: number, signal: AbortSignal): Promise { + if (!this.#reading) this.#reading = this.#pop(until) + + return new Promise((resolve) => { + const done = (marker: PoppedMarker | null) => { + clearTimeout(timer) + signal.removeEventListener('abort', abort) + if (this.#deliver === done) this.#deliver = undefined + resolve(marker) + } + const abort = () => done(null) + const timer = setTimeout(abort, until - Date.now()) + signal.addEventListener('abort', abort, { once: true }) + this.#deliver = done + }) + } + + async #pop(until: number): Promise { + // A command blocked longer may not survive a reconnection, as in BullMQ. + // It also bounds how long a silent connection goes unnoticed. + const timeout = Math.min(Math.max(until - Date.now(), 10), subscriptionHeartbeat.interval) + let watchdog: NodeJS.Timeout | undefined + let marker: PoppedMarker | null = null + + try { + const command = this.#blocking.bzpopmin( + ...this.#markers.map((marker) => marker.key), + timeout / 1000 + ) + // ioredis may never settle a command on a connection that went silent, + // nor one it dropped on reconnection: past the timeout, give up on it. + command.catch(() => {}) + const stuck = new Promise<'stuck'>((resolve) => { + watchdog = setTimeout(() => resolve('stuck'), timeout + subscriptionHeartbeat.timeout) + }) + // ioredis reconnects a closed connection, but drops its command. + const closed = new Promise<'closed'>((resolve) => { + this.#onClose = () => resolve('closed') + }) + const reply = await Promise.race([command, stuck, closed]) + if (this.#closed) { + marker = null + } else if (reply === 'closed') { + marker = this.#mayBeLost() + } else if (reply === 'stuck') { + this.#logger.error( + { err: new Error('BZPOPMIN did not return in time') }, + 'Redis blocking connection is silent, reconnecting' + ) + this.#blocking.disconnect(true) + marker = this.#mayBeLost() + } else if (reply) { + marker = this.#parse(reply) + } + } catch (error) { + if (!this.#closed) { + this.#logger.error({ err: error as Error }, 'Failed to wait for work on Redis') + await sleep(ERROR_RETRY_DELAY_MS) + marker = this.#mayBeLost() + } + } finally { + clearTimeout(watchdog) + this.#onClose = undefined + this.#reading = undefined + } + + if (this.#deliver) { + this.#deliver(marker) + } else if (marker) { + await this.#release(marker) + } + } + + /** + * Redis may have popped a marker whose reply was lost with the connection: + * look for work rather than wait for the next poll. + */ + #mayBeLost(): PoppedMarker { + return { key: this.#markers[0].key, member: '0', score: 0 } + } + + #parse([prefixedKey, member, score]: [string, string, string]): PoppedMarker | null { + const key = prefixedKey.slice(this.#prefix.length) + if (!this.#markers.some((marker) => marker.key === key)) return null + return { key, member, score: Number(score) } + } + + /** Put back a marker this worker does not act on, for another worker. */ + async #release(marker: PoppedMarker): Promise { + await this.#connection + .eval(RELEASE_MARKER_SCRIPT, 1, marker.key, marker.member, marker.score.toString()) + .catch((error) => this.#logger.error({ err: error }, 'Failed to put back a Redis marker')) + } + + /** Mark the next due time of a marker whose due time this worker runs. */ + async #advance(key: string, now: number): Promise { + const source = this.#markers.find((marker) => marker.key === key)!.source + await this.#connection + .eval(ADVANCE_MARKER_SCRIPT, 2, source, key, now.toString()) + .catch((error) => this.#logger.error({ err: error }, 'Failed to advance a Redis marker')) + } +} diff --git a/src/drivers/wake_messages.ts b/src/drivers/wake_messages.ts new file mode 100644 index 0000000..d2e2ec7 --- /dev/null +++ b/src/drivers/wake_messages.ts @@ -0,0 +1,105 @@ +/** + * Receives the time (epoch ms) at which work may become due: now for a job + * ready to run, later for a delayed job, a retry, or a schedule. + */ +export type WakeListener = (at: number) => void + +/** + * Message published on the LISTEN channel of the Knex adapter. + * + * - `{ ready, due }`: jobs of the `ready` queues are ready now, and a job of + * each `due` queue becomes due at its time (epoch ms). Without a time, a + * writer whose clock runs ahead cannot delay the watcher. + * - `{ schedule }`: the next schedule run after a schedule changed, read by + * the writer so watchers do not each read it. + * - `{ probe }`: sent by a watcher to check that its connection receives + * notifications. Other watchers ignore it. + */ +export type WakeMessage = + | { ready?: string[]; due?: Record } + | { schedule: number } + | { probe: string } + +export function encodeWakeMessage(message: WakeMessage): string { + return JSON.stringify(message) +} + +/** + * Route a raw message of the wake channel: jobs of the watched queues to + * `onWake`, the next schedule run to `onSchedule`. Others are ignored. + */ +export function dispatchWakeMessage( + raw: string, + queues: readonly string[], + onWake: WakeListener, + onSchedule: WakeListener +): void { + let message: Partial<{ ready: unknown; due: unknown; schedule: unknown }> | null + + try { + message = JSON.parse(raw) + } catch { + return + } + // Any role allowed to connect can notify the channel, with any payload. + if (!message || typeof message !== 'object') return + + if (typeof message.schedule === 'number') { + onSchedule(message.schedule) + return + } + + if (Array.isArray(message.ready) && message.ready.some((queue) => queues.includes(queue))) { + onWake(Date.now()) + } + if (message.due && typeof message.due === 'object') { + for (const [queue, at] of Object.entries(message.due)) { + if (typeof at === 'number' && queues.includes(queue)) onWake(at) + } + } +} + +/** + * How often a watcher checks its connection, and how long the connection has + * to answer: a PING on the LISTEN connection, or the timeout of BZPOPMIN on + * Redis. Tests shorten them. + */ +export const subscriptionHeartbeat = { interval: 10_000, timeout: 5_000 } + +/** + * Ping a subscription connection on an interval. A connection can stop + * delivering without closing, behind a stuck proxy or a half-open TCP + * connection: `onSilent` is called when a ping gets no answer in time, to + * reconnect. `ping` returns undefined to skip a beat, while reconnecting. + * + * Returns a function that stops the heartbeat. + */ +export function startHeartbeat( + ping: () => Promise | undefined, + onSilent: () => void +): () => void { + let timeout: NodeJS.Timeout | undefined + + const interval = setInterval(() => { + if (timeout) return + const answer = ping() + if (!answer) return + + const timer = setTimeout(() => { + timeout = undefined + onSilent() + }, subscriptionHeartbeat.timeout) + timeout = timer + const answered = () => { + clearTimeout(timer) + if (timeout === timer) timeout = undefined + } + answer.then(answered, answered) + }, subscriptionHeartbeat.interval) + interval.unref() + + return () => { + clearInterval(interval) + clearTimeout(timeout) + } +} diff --git a/src/drivers/worker_wakeup.ts b/src/drivers/worker_wakeup.ts new file mode 100644 index 0000000..e891dcc --- /dev/null +++ b/src/drivers/worker_wakeup.ts @@ -0,0 +1,89 @@ +/** + * Lets an idle worker sleep until an adapter signals that work may be due. + * + * Signals for now wake the current wait, or the next one when they arrive + * while the worker is looking for work. Signals for later are kept as timers, + * so a delayed job, a retry, or a schedule wakes the worker when it is due. + * + * ``` + * cycle: reset() ─► stalled check, schedules, pop ─► wait(maxDelay) + * ▲ notify(now) here sets `#requested`: the wait returns at once + * ``` + */ +export class WorkerWakeup { + /** Bounds the timers kept: past it, the latest are dropped and found by polling. */ + static readonly MAX_TIMERS = 1024 + + #requested = false + /** Future due times, sorted ascending, without duplicates. */ + #timers: number[] = [] + /** Re-evaluates the current wait. */ + #interrupt?: () => void + + notify(at: number): void { + if (at <= Date.now()) { + this.#requested = true + } else if (!this.#insert(at)) { + return + } + + this.#interrupt?.() + } + + /** + * Call before looking for work: the signals received until then are + * covered by this look, those received from now on wake the next wait. + */ + reset(): void { + this.#requested = false + const now = Date.now() + const due = this.#timers.findIndex((at) => at > now) + this.#timers.splice(0, due === -1 ? this.#timers.length : due) + } + + /** + * Resolve true on a signal for now or when the next timer is due, false + * after `maxDelay` or when `signal` aborts. Signals and timers wake no + * earlier than `cooldown`. + */ + async wait(maxDelay: number, signal: AbortSignal, cooldown = 0): Promise { + const start = Date.now() + const deadline = start + maxDelay + const quietUntil = start + cooldown + + while (!signal.aborted) { + const now = Date.now() + if (deadline <= now) return false + + const due = this.#requested ? now : (this.#timers[0] ?? Infinity) + const wakeAt = Math.max(due, quietUntil) + if (wakeAt <= now) return true + + await new Promise((resolve) => { + const done = () => { + clearTimeout(timer) + signal.removeEventListener('abort', done) + this.#interrupt = undefined + resolve() + } + const timer = setTimeout(done, Math.min(deadline, wakeAt) - now) + signal.addEventListener('abort', done, { once: true }) + this.#interrupt = done + }) + } + + return false + } + + /** Returns false when `at` was already known or past the bound. */ + #insert(at: number): boolean { + let index = this.#timers.findIndex((timer) => timer >= at) + if (index === -1) index = this.#timers.length + if (this.#timers[index] === at) return false + if (index >= WorkerWakeup.MAX_TIMERS) return false + + this.#timers.splice(index, 0, at) + if (this.#timers.length > WorkerWakeup.MAX_TIMERS) this.#timers.pop() + return true + } +} diff --git a/src/worker_session.ts b/src/worker_session.ts index ba08baf..e422d95 100644 --- a/src/worker_session.ts +++ b/src/worker_session.ts @@ -6,7 +6,7 @@ import { JobPool, type PoolEntry } from './job_pool.js' import { WorkerHeartbeat, type HeartbeatOperationWrapper } from './worker_heartbeat.js' import { Locator } from './locator.js' import type { SingleJobDispatchRequest } from './job_dispatch_runtime.js' -import type { Adapter, AcquiredJob } from './contracts/adapter.js' +import type { Adapter, AcquiredJob, WorkWatcher } from './contracts/adapter.js' import type { JobExecutionOutcome, JobExecutionRuntime } from './job_runtime.js' import type { Logger, WorkerCycle } from './types/main.js' import { DEFAULT_ERROR_RETRY_DELAY, UNKNOWN_JOB_RETRY_DELAY } from './constants.js' @@ -80,6 +80,10 @@ export class WorkerSession { #completionDelayController?: AbortController #lastStalledCheck = 0 #delayController?: AbortController + /** Set while `start()` holds a watcher: idle waits sleep until work may be due. */ + #watcher?: WorkWatcher + /** The last wait ended on a wake: if the next look finds no work, it cools down. */ + #woken = false constructor(options: WorkerSessionOptions) { this.#workerId = options.workerId @@ -122,11 +126,20 @@ export class WorkerSession { const completion = Promise.withResolvers() this.#startCompletion = completion.promise + // Polls until the watcher is ready: a slow handshake delays no job. + const watchAbort = new AbortController() + const watching = this.#watch(watchAbort.signal) + try { debug('starting worker %s on queues: %O', this.#workerId, this.#queues) for await (const cycle of generator) { - if (cycle.type === 'started' || cycle.type === 'completed') { + if (cycle.type === 'started') { + this.#woken = false + continue + } + + if (cycle.type === 'completed') { continue } @@ -141,9 +154,15 @@ export class WorkerSession { debug('worker %s is idle, waiting for %dms', this.#workerId, delay) } - await this.#waitBeforeNextCycle(delay) + await this.#waitBeforeNextCycle(cycle.type === 'idle' && this.#watcher ? null : delay) } } finally { + this.#watcher = undefined + watchAbort.abort() + const watcher = await watching + await watcher?.close().catch((error) => { + this.#logger.error({ err: error, workerId: this.#workerId }, 'Failed to stop watching') + }) completion.resolve() if (this.#startCompletion === completion.promise) { this.#startCompletion = undefined @@ -270,14 +289,66 @@ export class WorkerSession { } } - async #waitBeforeNextCycle(delay: number): Promise { + /** + * Watch for work through the adapter, when it supports it. A failure is + * logged and the worker keeps polling. `signal` aborts a watcher that is + * not ready when the session stops. + */ + async #watch(signal: AbortSignal): Promise { + if (!this.#adapter.watch) return + + try { + const watcher = await this.#adapter.watch(this.#queues, signal) + if (!signal.aborted) { + this.#watcher = watcher + debug('worker %s watches for work', this.#workerId) + // The current wait polls: end it, so the next one sleeps until work. + this.#delayController?.abort() + this.#completionDelayController?.abort() + } + return watcher + } catch (error) { + if (signal.aborted) return + this.#logger.error( + { err: error as Error, workerId: this.#workerId }, + 'Failed to watch for work, polling instead' + ) + } + } + + /** + * Sleep until work may be due. After a wake that found no work, the + * watcher cools down for `idleDelay`: idle workers woken together look for + * work at most as often as when polling. + */ + async #waitForWork(watcher: WorkWatcher, signal: AbortSignal): Promise { + const cooldown = this.#woken ? this.#settings.idleDelay : 0 + this.#woken = await watcher.wait(this.#untilStalledCheck(), signal, cooldown) + } + + /** + * Time left before the next stalled check. A watching worker sleeps at + * most this long: the check doubles as the poll that catches lost signals. + */ + #untilStalledCheck(): number { + return Math.max(this.#lastStalledCheck + this.#settings.stalledInterval - Date.now(), 0) + } + + /** + * Sleep for `delay`, or until work may be due when `delay` is null. + */ + async #waitBeforeNextCycle(delay: number | null): Promise { if (!this.#running) return const controller = new AbortController() this.#delayController = controller try { - await setTimeout(delay, undefined, { signal: controller.signal }) + if (delay === null) { + if (this.#watcher) await this.#waitForWork(this.#watcher, controller.signal) + } else { + await setTimeout(delay, undefined, { signal: controller.signal }) + } } catch (error) { if (!controller.signal.aborted) { throw error @@ -295,6 +366,7 @@ export class WorkerSession { try { while (this.#running) { try { + this.#watcher?.reset() await this.#checkStalledJobs() if (!this.#running) break @@ -365,7 +437,11 @@ export class WorkerSession { this.#completionDelayController = controller try { - await setTimeout(this.#settings.idleDelay, undefined, { signal: controller.signal }) + if (this.#watcher) { + await this.#waitForWork(this.#watcher, controller.signal) + } else { + await setTimeout(this.#settings.idleDelay, undefined, { signal: controller.signal }) + } } catch (error) { if (!controller.signal.aborted) { throw error @@ -404,40 +480,28 @@ export class WorkerSession { return completed } + /** + * Acquire jobs one at a time until the pool is full or the queues run dry. + * An empty queue then costs one acquisition, not one per free slot: a + * watching worker looks for work on every wake. + */ async #fillPool(): Promise { - const slotsAvailable = this.#settings.concurrency - this.#pool.size + const cycles: StartedCycle[] = [] - if (slotsAvailable <= 0) return { cycles: [], errors: [] } + while (this.#running && this.#pool.size < this.#settings.concurrency) { + let cycle: StartedCycle | null - let probe: StartedCycle | null - - try { - probe = await this.#acquireAndStartNextJob() - } catch (error) { - return { cycles: [], errors: [error] } - } - - if (!probe) return { cycles: [], errors: [] } - if (!this.#running) return { cycles: [probe], errors: [] } - - const acquisitions: Array> = [] - for (let index = 1; index < slotsAvailable; index++) { - acquisitions.push(this.#acquireAndStartNextJob()) - } - - const results = await Promise.allSettled(acquisitions) - const cycles = [probe] - const errors: unknown[] = [] - - for (const result of results) { - if (result.status === 'rejected') { - errors.push(result.reason) - } else if (result.value) { - cycles.push(result.value) + try { + cycle = await this.#acquireAndStartNextJob() + } catch (error) { + return { cycles, errors: [error] } } + + if (!cycle) break + cycles.push(cycle) } - return { cycles, errors } + return { cycles, errors: [] } } async #acquireAndStartNextJob(): Promise { diff --git a/tests/_utils/register_work_watcher_suite.ts b/tests/_utils/register_work_watcher_suite.ts new file mode 100644 index 0000000..4f27aa3 --- /dev/null +++ b/tests/_utils/register_work_watcher_suite.ts @@ -0,0 +1,261 @@ +import { setTimeout } from 'node:timers/promises' +import { test as JapaTest } from '@japa/runner' +import type { Adapter, WorkWatcher } from '../../src/contracts/adapter.js' + +interface WorkWatcherSuiteOptions { + test: typeof JapaTest + /** Creates an adapter that implements `watch`. */ + createAdapter: () => Adapter + /** See `inTransaction` of the driver test suite. */ + inTransaction?: (operation: (adapter: Adapter) => Promise) => Promise +} + +const job = (id: string) => ({ id, name: 'TestJob', payload: {}, attempts: 0 }) + +/** + * Watch, as a worker does: `next()` looks for work (reset), then waits and + * resolves with the time the wait took, or null when nothing woke it. + */ +async function watch(adapter: Adapter, queues = ['default']) { + const watcher: WorkWatcher = await adapter.watch!(queues) + // The worker looks for work once ready: what is stored is covered. + watcher.reset() + + return { + watcher, + close: () => watcher.close(), + async next(maxDelay = 2000): Promise { + watcher.reset() + const start = Date.now() + const woken = await watcher.wait(maxDelay, new AbortController().signal, 0) + return woken ? Date.now() - start : null + }, + /** Wait for the first wake at least `atLeast` ms after now. */ + async wakeAfter(atLeast: number, maxDelay = 2000): Promise { + const start = Date.now() + while (Date.now() - start < maxDelay) { + const woken = await this.next(maxDelay - (Date.now() - start)) + if (woken === null) return null + if (Date.now() - start >= atLeast) return Date.now() - start + } + return null + }, + } +} + +/** + * Wakes of adapters that implement `watch`. The watcher and the writer use + * separate adapters, as a worker and a dispatcher would. Due times are + * short, so the suite checks when a wait ends, within a margin. + */ +export function registerWorkWatcherSuite({ + test, + createAdapter, + inTransaction, +}: WorkWatcherSuiteOptions) { + test('wakes for a push to its queue only', async ({ assert, cleanup }) => { + const watching = await watch(createAdapter()) + cleanup(watching.close) + const writer = createAdapter() + + const waiting = watching.next() + await writer.pushOn('other', job('other-queue')) + await setTimeout(100) + await writer.pushOn('default', job('default-queue')) + + const elapsed = await waiting + assert.isNotNull(elapsed) + assert.isAtLeast(elapsed!, 100, 'not woken by the other queue') + assert.isBelow(elapsed!, 400) + }) + + test('wakes when a delayed job and a retry become due', async ({ assert, cleanup }) => { + const watching = await watch(createAdapter()) + cleanup(watching.close) + const writer = createAdapter() + + await writer.pushLaterOn('default', job('delayed'), 300) + assert.closeTo((await watching.wakeAfter(250))!, 300, 100) + await writer.popFrom('default') + + await writer.pushOn('default', job('retried')) + await watching.next() + const acquired = await writer.popFrom('default') + await writer.retryJob(acquired!, 'default', new Date(Date.now() + 300)) + assert.closeTo((await watching.wakeAfter(250))!, 300, 100) + }) + + test('wakes when a schedule becomes due', async ({ assert, cleanup }) => { + const watching = await watch(createAdapter()) + cleanup(watching.close) + const writer = createAdapter() + + await writer.upsertSchedule({ + id: 'watched-schedule', + name: 'TestJob', + payload: {}, + everyMs: 300, + timezone: 'UTC', + nextRunAt: new Date(Date.now() + 300), + }) + assert.closeTo((await watching.wakeAfter(250))!, 300, 100) + + // Its next run, written by the claim. + assert.isNotNull(await writer.claimDueSchedule()) + assert.closeTo((await watching.wakeAfter(250))!, 300, 100) + }) + + test('wakes for work stored before watching', async ({ assert, cleanup }) => { + const writer = createAdapter() + await writer.pushLaterOn('default', job('stored-delayed'), 300) + await writer.upsertSchedule({ + id: 'stored-schedule', + name: 'TestJob', + payload: {}, + everyMs: 60_000, + timezone: 'UTC', + nextRunAt: new Date(Date.now() + 600), + }) + + const watching = await watch(createAdapter()) + cleanup(watching.close) + + assert.closeTo((await watching.wakeAfter(250))!, 300, 100) + await writer.popFrom('default') + assert.closeTo((await watching.wakeAfter(200))!, 300, 100) + }) + + test('wakes for the next delayed job once one is promoted', async ({ assert, cleanup }) => { + const writer = createAdapter() + await writer.pushLaterOn('default', job('first-delayed'), 100) + await writer.pushLaterOn('default', job('second-delayed'), 400) + + const watching = await watch(createAdapter()) + cleanup(watching.close) + + assert.closeTo((await watching.wakeAfter(50))!, 100, 80) + assert.equal((await writer.popFrom('default'))?.id, 'first-delayed') + assert.closeTo((await watching.wakeAfter(200))!, 300, 100) + }) + + test('wakes for the next delayed job of each queue', async ({ assert, cleanup }) => { + const writer = createAdapter() + await writer.pushLaterOn('first', job('first-queue'), 150) + await writer.pushLaterOn('second', job('second-queue'), 450) + + const watching = await watch(createAdapter(), ['first', 'second']) + cleanup(watching.close) + + assert.closeTo((await watching.wakeAfter(100))!, 150, 80) + await writer.popFrom('first') + assert.closeTo((await watching.wakeAfter(200))!, 300, 100) + }) + + test('wakes for the next schedule past an expired one', async ({ assert, cleanup }) => { + const writer = createAdapter() + const now = Date.now() + await writer.upsertSchedule({ + id: 'expired', + name: 'TestJob', + payload: {}, + everyMs: 60_000, + timezone: 'UTC', + to: new Date(now - 500), + nextRunAt: new Date(now - 1000), + }) + await writer.upsertSchedule({ + id: 'next', + name: 'TestJob', + payload: {}, + everyMs: 60_000, + timezone: 'UTC', + nextRunAt: new Date(now + 400), + }) + + const worker = createAdapter() + const watching = await watch(worker) + cleanup(watching.close) + + // A worker woken for the expired schedule claims nothing. + assert.isNull(await worker.claimDueSchedule()) + const expected = 400 - (Date.now() - now) + assert.closeTo((await watching.wakeAfter(expected - 100))!, expected, 100) + }) + + test('wakes for the next schedule when the announced one expires before running', async ({ + assert, + cleanup, + }) => { + const writer = createAdapter() + const now = Date.now() + await writer.upsertSchedule({ + id: 'expiring', + name: 'TestJob', + payload: {}, + everyMs: 60_000, + timezone: 'UTC', + to: new Date(now + 150), + nextRunAt: new Date(now + 100), + }) + await writer.upsertSchedule({ + id: 'next', + name: 'TestJob', + payload: {}, + everyMs: 60_000, + timezone: 'UTC', + nextRunAt: new Date(now + 600), + }) + + const worker = createAdapter() + const watching = await watch(worker) + cleanup(watching.close) + + // Woken for the expiring schedule once it has also passed its end date. + await setTimeout(200) + assert.isNull(await worker.claimDueSchedule()) + const expected = 600 - (Date.now() - now) + assert.closeTo((await watching.wakeAfter(expected - 100))!, expected, 100) + }) + + test('wakes for the next schedule once one is deleted', async ({ assert, cleanup }) => { + const writer = createAdapter() + const schedule = (id: string, inMs: number) => + writer.upsertSchedule({ + id, + name: 'TestJob', + payload: {}, + everyMs: 60_000, + timezone: 'UTC', + nextRunAt: new Date(Date.now() + inMs), + }) + await schedule('first', 200) + await schedule('second', 500) + + const watching = await watch(createAdapter()) + cleanup(watching.close) + await writer.deleteSchedule('first') + + assert.closeTo((await watching.wakeAfter(400))!, 500, 100) + }) + + if (inTransaction) { + test('wakes for a push in a transaction once it commits', async ({ assert, cleanup }) => { + const watching = await watch(createAdapter()) + cleanup(watching.close) + + let waiting = watching.next(300) + await inTransaction(async (adapter) => { + await adapter.pushOn('default', job('committed')) + assert.isNull(await waiting, 'no wake before the commit') + waiting = watching.next() + }) + assert.isNotNull(await waiting) + + await inTransaction(async (adapter) => { + await adapter.pushOn('default', job('rolled-back')) + throw new Error('rollback') + }).catch(() => {}) + assert.isNull(await watching.next(300), 'no wake after a rollback') + }) + } +} diff --git a/tests/_utils/silent_proxy.ts b/tests/_utils/silent_proxy.ts new file mode 100644 index 0000000..2d5de1e --- /dev/null +++ b/tests/_utils/silent_proxy.ts @@ -0,0 +1,48 @@ +import { createServer, connect, type Socket } from 'node:net' + +/** + * TCP proxy to a server. `silence()` stops forwarding on open connections + * without closing them, as a stuck network path would: all of them, or the + * one from the client port `clientPort`. Later connections are forwarded. + */ +export async function startSilentProxy(target: { host: string; port: number }) { + const connections = new Map() + + const server = createServer((client) => { + const upstream = connect(target) + const entry = { client, upstream, silent: false } + const clientPort = client.remotePort! + connections.set(clientPort, entry) + + for (const [from, to] of [ + [client, upstream], + [upstream, client], + ]) { + from.on('data', (chunk) => { + if (!entry.silent) to.write(chunk) + }) + from.on('error', () => to.destroy()) + from.on('close', () => { + connections.delete(clientPort) + to.destroy() + }) + } + }) + await new Promise((resolve) => server.listen(0, '127.0.0.1', resolve)) + + return { + port: (server.address() as { port: number }).port, + silence(clientPort?: number) { + for (const [port, entry] of connections) { + if (clientPort === undefined || port === clientPort) entry.silent = true + } + }, + async close() { + for (const { client, upstream } of connections.values()) { + client.destroy() + upstream.destroy() + } + await new Promise((resolve) => server.close(resolve)) + }, + } +} diff --git a/tests/adapter.spec.ts b/tests/adapter.spec.ts index 7903838..28cba20 100644 --- a/tests/adapter.spec.ts +++ b/tests/adapter.spec.ts @@ -1,22 +1,26 @@ import { mkdtemp, rm } from 'node:fs/promises' import { tmpdir } from 'node:os' import { join } from 'node:path' +import { createServer, type AddressInfo } from 'node:net' import Knex from 'knex' import type { Knex as KnexTypes } from 'knex' import { test } from '@japa/runner' import type { Assert } from '@japa/assert' -import { Redis } from 'ioredis' +import { Redis, type RedisOptions } from 'ioredis' import { CronExpressionParser } from 'cron-parser' import { MemoryAdapter } from './_mocks/memory_adapter.js' import { redis, RedisAdapter } from '../src/drivers/redis_adapter.js' -import { KnexAdapter } from '../src/drivers/knex_adapter.js' +import { knex, KnexAdapter } from '../src/drivers/knex_adapter.js' import { KnexQueueSchemaService } from '../src/services/knex_queue_schema.js' import { registerDriverTestSuite } from './_utils/register_driver_test_suite.js' +import { registerWorkWatcherSuite } from './_utils/register_work_watcher_suite.js' import { withRedisWriteSpy } from './_utils/with_redis_write_spy.js' import { withKnexQuerySpy } from './_utils/with_knex_query_spy.js' import { pauseKnexDeletes } from './_utils/pause_deletes.js' import { registerRetentionRaceSuite } from './_utils/register_retention_race_suite.js' import { MemoryLogger } from './_mocks/memory_logger.js' +import { startSilentProxy } from './_utils/silent_proxy.js' +import { subscriptionHeartbeat } from '../src/drivers/wake_messages.js' import { pgDeleteFailureSql, registerPruneFailureSuite, @@ -115,6 +119,236 @@ test.group('Adapter | Redis', (group) => { createAdapter: () => new RedisAdapter(connection), }) + registerWorkWatcherSuite({ + test, + createAdapter: () => new RedisAdapter(connection), + }) + + test('dispatches and watches for an ACL user without channel permissions', async ({ + assert, + cleanup, + }) => { + // New ACL users get no channel permission by default since Redis 7. + await connection.call( + 'ACL', + 'SETUSER', + 'queue-no-channels', + 'on', + '>pw', + '~*', + '+@all', + 'resetchannels' + ) + const restricted = new Redis({ + host: process.env.REDIS_HOST || 'localhost', + port: Number.parseInt(process.env.REDIS_PORT || '6379', 10), + keyPrefix: KEY_PREFIX, + db: 15, + username: 'queue-no-channels', + password: 'pw', + }) + cleanup(async () => { + restricted.disconnect() + await connection.call('ACL', 'DELUSER', 'queue-no-channels') + }) + const adapter = new RedisAdapter(restricted) + const watcher = await adapter.watch(['default']) + cleanup(() => watcher.close()) + + const waiting = watcher.wait(2000, new AbortController().signal, 0) + await adapter.pushOn('default', { id: 'acl-job', name: 'TestJob', payload: {}, attempts: 0 }) + assert.isTrue(await waiting) + }) + + test('refuses to watch for an ACL user that cannot block', async ({ assert, cleanup }) => { + await connection.call( + 'ACL', + 'SETUSER', + 'queue-no-blocking', + 'on', + '>pw', + '~*', + '&*', + '+@all', + '-@blocking' + ) + const restricted = new Redis({ + host: process.env.REDIS_HOST || 'localhost', + port: Number.parseInt(process.env.REDIS_PORT || '6379', 10), + keyPrefix: KEY_PREFIX, + db: 15, + username: 'queue-no-blocking', + password: 'pw', + }) + cleanup(async () => { + restricted.disconnect() + await connection.call('ACL', 'DELUSER', 'queue-no-blocking') + }) + + // The worker polls instead of logging a failed wait every second. + await assert.rejects(() => new RedisAdapter(restricted).watch(['default']), /NOPERM/) + }) + + test('wakes one waiting watcher per marker', async ({ assert, cleanup }) => { + const writer = new RedisAdapter(connection) + const job = (id: string) => ({ id, name: 'TestJob', payload: {}, attempts: 0 }) + // Pushed while every worker is busy: two jobs leave one marker. + await writer.pushOn('default', job('first')) + await writer.pushOn('default', job('second')) + + const watchers = await Promise.all([ + new RedisAdapter(connection).watch(['default']), + new RedisAdapter(connection).watch(['default']), + ]) + const controller = new AbortController() + cleanup(async () => { + controller.abort() + await Promise.all(watchers.map((watcher) => watcher.close())) + }) + const woken: number[] = [] + for (const [index, watcher] of watchers.entries()) { + void watcher.wait(5000, controller.signal, 0).then((wake) => wake && woken.push(index)) + } + await new Promise((resolve) => setTimeout(resolve, 200)) + assert.lengthOf(woken, 1) + + // The woken worker takes a job while another remains: the next one wakes. + await writer.popFrom('default') + await new Promise((resolve) => setTimeout(resolve, 200)) + assert.sameMembers(woken, [0, 1]) + }) + + test('hands a due time to another watcher when its watcher stops waiting', async ({ + assert, + cleanup, + }) => { + const first = await new RedisAdapter(connection).watch(['default']) + const second = await new RedisAdapter(connection).watch(['default']) + cleanup(() => Promise.all([first.close(), second.close()]).then(() => {})) + + const busy = new AbortController() + const firstWait = first.wait(5000, busy.signal, 0) + await new Promise((resolve) => setTimeout(resolve, 50)) + const writer = new RedisAdapter(connection) + await writer.pushLaterOn( + 'default', + { id: 'later', name: 'TestJob', payload: {}, attempts: 0 }, + 300 + ) + await new Promise((resolve) => setTimeout(resolve, 100)) + + // The first watcher holds the due time, then its worker gets busy. + busy.abort() + assert.isFalse(await firstWait) + const start = Date.now() + assert.isTrue(await second.wait(2000, new AbortController().signal, 0)) + assert.closeTo(Date.now() - start, 200, 100) + }) + + test('aborts a blocking connection stuck in its handshake', async ({ assert, cleanup }) => { + // Accepts connections and never answers. + const blackhole = createServer(() => {}) + await new Promise((resolve) => blackhole.listen(0, '127.0.0.1', resolve)) + cleanup(() => void blackhole.close()) + const port = (blackhole.address() as AddressInfo).port + const direct = new Redis(connection.options) + direct.duplicate = (override?: Partial) => + new Redis({ ...connection.options, host: '127.0.0.1', port, ...override }) + cleanup(() => direct.disconnect()) + + const controller = new AbortController() + const adapter = new RedisAdapter(direct, false, new MemoryLogger()) + const watching = adapter.watch(['default'], controller.signal) + setTimeout(() => controller.abort(), 100) + + await assert.rejects(() => watching) + }).timeout(2000) + + test('gives up on a blocking connection that does not start', async ({ assert, cleanup }) => { + shortenHeartbeat(cleanup) + // Accepts connections and never answers, as a connection whose command was lost. + const blackhole = createServer(() => {}) + await new Promise((resolve) => blackhole.listen(0, '127.0.0.1', resolve)) + cleanup(() => void blackhole.close()) + const port = (blackhole.address() as AddressInfo).port + const direct = new Redis(connection.options) + direct.duplicate = (override?: Partial) => + new Redis({ ...connection.options, host: '127.0.0.1', port, ...override }) + cleanup(() => direct.disconnect()) + + // Without an abort, the worker falls back to polling. + const adapter = new RedisAdapter(direct, false, new MemoryLogger()) + await assert.rejects(() => adapter.watch(['default']), /did not start in time/) + }).timeout(2000) + + test('reconnects a blocking connection that stops answering', async ({ assert, cleanup }) => { + shortenHeartbeat(cleanup) + const target = { + host: process.env.REDIS_HOST || 'localhost', + port: Number.parseInt(process.env.REDIS_PORT || '6379', 10), + } + const proxy = await startSilentProxy(target) + cleanup(() => proxy.close()) + // Only the blocking connection goes through the proxy. + const direct = new Redis({ ...target, keyPrefix: KEY_PREFIX, db: 15 }) + direct.duplicate = (override?: Partial) => + new Redis({ host: '127.0.0.1', port: proxy.port, keyPrefix: KEY_PREFIX, db: 15, ...override }) + cleanup(() => direct.disconnect()) + + const logger = new MemoryLogger() + const watcher = await new RedisAdapter(direct, false, logger).watch(['default']) + cleanup(() => watcher.close()) + const silences = () => + logger.logs.filter((entry) => entry.message.includes('connection is silent')).length + const waiting = watcher.wait(2000, new AbortController().signal, 0) + await new Promise((resolve) => setTimeout(resolve, 50)) + proxy.silence() + + const writer = new RedisAdapter(connection) + const job = (id: string) => ({ id, name: 'TestJob', payload: {}, attempts: 0 }) + // The marker of this push may be lost with the silent connection: the + // watcher wakes to look once it gives up on it. + await writer.pushOn('default', job('during-silence')) + assert.isTrue(await waiting) + await writer.popFrom('default') + + assert.equal(silences(), 1) + + // The new connection receives the next marker. + const next = watcher.wait(2000, new AbortController().signal, 0) + await writer.pushOn('default', job('after-reconnection')) + assert.isTrue(await next) + assert.equal(silences(), 1) + }).timeout(5000) + + test('reconnects a blocking connection the server closes', async ({ assert, cleanup }) => { + const logger = new MemoryLogger() + const watcher = await new RedisAdapter(connection, false, logger).watch(['default']) + cleanup(() => watcher.close()) + const waiting = watcher.wait(3000, new AbortController().signal, 0) + await new Promise((resolve) => setTimeout(resolve, 50)) + + // Its reply may be lost: the watcher looks for work at once, not after + // the timeout of BZPOPMIN. + const start = Date.now() + const clients = (await connection.client('LIST')) as string + const blocked = clients.split('\n').find((line) => line.includes('cmd=bzpopmin'))! + await connection.client('KILL', 'ID', blocked.match(/id=(\d+)/)![1]) + assert.isTrue(await waiting) + assert.isBelow(Date.now() - start, 500) + + // The new connection receives the next marker. + const next = watcher.wait(3000, new AbortController().signal, 0) + await new RedisAdapter(connection).pushOn('default', { + id: 'after-kill', + name: 'TestJob', + payload: {}, + attempts: 0, + }) + assert.isTrue(await next) + assert.isBelow(Date.now() - start, 1000) + }).timeout(5000) + test('listSchedules should use bounded network round-trips as schedule count grows', async ({ assert, }) => { @@ -2123,6 +2357,13 @@ test.group('Adapter | Redis', (group) => { * A jobs table as the previous versions left it: no unique index, and expired * owners that still hold their dedup id. */ +/** Check watcher connections every 50 ms, with 300 ms to answer. */ +function shortenHeartbeat(cleanup: (callback: () => void) => void) { + const defaults = { ...subscriptionHeartbeat } + Object.assign(subscriptionHeartbeat, { interval: 50, timeout: 300 }) + cleanup(() => Object.assign(subscriptionHeartbeat, defaults)) +} + async function assertDedupMigration( assert: Assert, connection: ReturnType, @@ -2354,6 +2595,747 @@ test.group('Adapter | Knex (PostgreSQL)', (group) => { }, }) + registerWorkWatcherSuite({ + test, + inTransaction: (operation) => + connection.transaction((trx) => + operation( + new KnexAdapter({ connection: trx, tableName, schedulesTableName, notifications: true }) + ) + ), + createAdapter: () => + new KnexAdapter({ connection, tableName, schedulesTableName, notifications: true }), + }) + + test('announces a future delayed job past rows another transaction promotes', async ({ + assert, + cleanup, + }) => { + const row = (id: string, executeAt: number) => ({ + id, + queue: 'default', + status: 'delayed', + data: JSON.stringify({ id, name: 'TestJob', payload: {}, attempts: 0 }), + execute_at: executeAt, + }) + const now = Date.now() + const nextAt = now + 400 + await connection(tableName).insert([ + row('locked', now - 2000), + row('own', now - 1000), + row('next', nextAt), + ]) + + const watcher = await new KnexAdapter({ + connection, + tableName, + schedulesTableName, + notifications: true, + }).watch!(['default']) + cleanup(() => watcher.close()) + + // Another worker holds the oldest due row while it promotes it. + const trx = await connection.transaction() + cleanup(async () => { + await trx.rollback() + }) + await trx(tableName).where('id', 'locked').forUpdate().first() + + await new KnexAdapter({ + connection, + tableName, + schedulesTableName, + notifications: true, + }).popFrom('default') + watcher.reset() + assert.isTrue(await watcher.wait(2000, new AbortController().signal, 0)) + assert.isAtLeast(Date.now(), nextAt - 20, 'not woken for the locked row') + }) + + /** + * Replace the node-postgres client the adapter creates for LISTEN. Each + * client gets the behavior `behave` returns for its 1-based creation + * number: 'real' for a working connection, or a fake one. + */ + type FakeBehavior = { + connect?: () => Promise + query?: () => Promise + end?: () => Promise + } + + function fakeListenClients( + cleanup: (callback: () => void) => void, + behave: (index: number) => 'real' | FakeBehavior + ) { + type PgClient = new (settings: unknown) => { + end(): Promise + on(event: string, listener: (error?: Error) => void): unknown + emit(event: string, error?: Error): unknown + } + const client = connection.client as KnexTypes.Client & { driver: { Client: PgClient } } + const driver = client.driver + const RealClient = driver.Client + client.driver = { ...driver } + cleanup(() => { + client.driver = driver + }) + + const created: Array<{ + ended: boolean + destroyed?: boolean + emit: (event: string, error?: Error) => void + }> = [] + const FakeClient = class { + #listeners = new Map void>() + #behavior: FakeBehavior + state = { + ended: false, + destroyed: false, + emit: (event: string, error?: Error) => this.#listeners.get(event)?.(error), + } + + // As node-postgres: the socket, destroyed when end() does not settle. + connection = { stream: { destroy: () => (this.state.destroyed = true) } } + + constructor(behavior: FakeBehavior) { + this.#behavior = behavior + created.push(this.state) + } + on(event: string, listener: (error?: Error) => void) { + this.#listeners.set(event, listener) + } + removeAllListeners() { + this.#listeners.clear() + } + // As node-postgres: connect() never settles once ended in its handshake. + async connect() { + await this.#behavior.connect?.() + } + async query() { + await this.#behavior.query?.() + } + async end() { + this.state.ended = true + await this.#behavior.end?.() + } + } + client.driver.Client = function (settings: unknown) { + const behavior = behave(created.length + 1) + if (behavior !== 'real') return new FakeClient(behavior) + + const real = new RealClient(settings) + const state = { + ended: false, + emit: (event: string, error?: Error) => real.emit(event, error), + } + created.push(state) + const end = real.end.bind(real) + real.end = () => { + state.ended = true + return end() + } + return real + } as unknown as PgClient + return created + } + + test('rejects a LISTEN connection that receives no notification', async ({ assert, cleanup }) => { + // As behind PgBouncer in transaction mode: LISTEN succeeds, nothing arrives. + fakeListenClients(cleanup, () => ({})) + const watching = new KnexAdapter({ + connection, + tableName, + schedulesTableName, + notifications: true, + }) + + await assert.rejects(() => watching.watch!(['default']), /do not enable notifications/) + }).timeout(5000) + + test('rejects a LISTEN probe whose connection fails', async ({ assert, cleanup }) => { + // The probe sends its notification, then its connection fails. + const created = fakeListenClients(cleanup, (index) => + index === 1 + ? 'real' + : { + async query() { + setTimeout(() => created[1].emit('error', new Error('Connection terminated')), 10) + }, + } + ) + const watching = new KnexAdapter({ + connection, + tableName, + schedulesTableName, + notifications: true, + }) + + await assert.rejects(() => watching.watch!(['default']), 'Connection terminated') + }) + + test('destroys a LISTEN probe connection that does not end', async ({ assert, cleanup }) => { + shortenHeartbeat(cleanup) + const never = () => new Promise(() => {}) + const created = fakeListenClients(cleanup, (index) => + index === 1 ? 'real' : { connect: never, end: never } + ) + const watching = new KnexAdapter({ + connection, + tableName, + schedulesTableName, + notifications: true, + }) + + await assert.rejects(() => watching.watch!(['default']), /do not enable notifications/) + await new Promise((resolve) => setTimeout(resolve, 400)) + assert.isTrue(created[1].destroyed) + }).timeout(5000) + + test('aborts a watch whose connection settings lookup stalls', async ({ assert, cleanup }) => { + const stalled = Knex({ client: 'pg', connection: () => new Promise(() => {}) }) + cleanup(() => stalled.destroy()) + const watching = new KnexAdapter({ + connection: stalled, + tableName, + schedulesTableName, + notifications: true, + }) + const controller = new AbortController() + + const watcher = watching.watch!(['default'], controller.signal) + setTimeout(() => controller.abort(), 50) + + await assert.rejects(() => watcher) + }).timeout(2000) + + test('aborts a first LISTEN connection still in its handshake', async ({ assert, cleanup }) => { + const created = fakeListenClients(cleanup, () => ({ connect: () => new Promise(() => {}) })) + const watching = new KnexAdapter({ + connection, + tableName, + schedulesTableName, + notifications: true, + }) + const controller = new AbortController() + + const watcher = watching.watch!(['default'], controller.signal) + while (created.length < 1) await new Promise((resolve) => setTimeout(resolve, 5)) + controller.abort() + + await assert.rejects(() => watcher) + assert.isTrue(created[0].ended) + }) + + test('rejects a LISTEN probe whose connection stalls', async ({ assert, cleanup }) => { + // The LISTEN connection works, the probe never completes its handshake. + fakeListenClients(cleanup, (index) => + index === 1 ? 'real' : { connect: () => new Promise(() => {}) } + ) + const watching = new KnexAdapter({ + connection, + tableName, + schedulesTableName, + notifications: true, + }) + + await assert.rejects(() => watching.watch!(['default']), /do not enable notifications/) + }).timeout(5000) + + test('ignores a notification that is not a wake message', async ({ assert, cleanup }) => { + const options = { connection, tableName, schedulesTableName, notifications: true } + const watcher = await new KnexAdapter(options).watch!(['default']) + cleanup(() => watcher.close()) + watcher.reset() + const waiting = watcher.wait(2000, new AbortController().signal, 0) + + await connection.raw(`select pg_notify(?, 'null')`, [tableName]) + const job = { id: 'after-null', name: 'TestJob', payload: {}, attempts: 0 } + await new KnexAdapter(options).pushOn('default', job) + assert.isTrue(await waiting) + }) + + test('watches tables found through the search path', async ({ assert, cleanup }) => { + const schema = 'queue_search_path' + await connection.raw('create schema if not exists ??', [schema]) + cleanup(() => connection.raw('drop schema ?? cascade', [schema])) + // Tables only the search path finds, unlike those of the other tests. + const tables = { tableName: 'tenant_jobs', schedulesTableName: 'tenant_schedules' } + for (const [table, like] of [ + [tables.tableName, tableName], + [tables.schedulesTableName, schedulesTableName], + ]) { + await connection.raw('create table ?? (like ?? including all)', [`${schema}.${table}`, like]) + } + const tenant = Knex({ ...connection.client.config, searchPath: [schema] }) + cleanup(() => tenant.destroy()) + const options = { connection: tenant, ...tables, notifications: true } + + const watcher = await new KnexAdapter(options).watch!(['default']) + cleanup(() => watcher.close()) + watcher.reset() + const waiting = watcher.wait(2000, new AbortController().signal, 0) + + const job = { id: 'tenant-job', name: 'TestJob', payload: {}, attempts: 0 } + await new KnexAdapter(options).pushOn('default', job) + assert.isTrue(await waiting) + }) + + test('does not wake again for a due schedule a concurrent claim holds', async ({ + assert, + cleanup, + }) => { + await connection(schedulesTableName).insert({ + id: 'locked', + name: 'TestJob', + payload: '{}', + every_ms: 60_000, + timezone: 'UTC', + next_run_at: Date.now() + 100, + run_count: 0, + status: 'active', + created_at: Date.now(), + }) + const worker = new KnexAdapter({ + connection, + tableName, + schedulesTableName, + notifications: true, + }) + const watcher = await worker.watch!(['default']) + cleanup(() => watcher.close()) + + // Another worker is claiming the schedule when it becomes due. + const trx = await connection.transaction() + cleanup(async () => { + await trx.rollback() + }) + await trx(schedulesTableName).where('id', 'locked').forUpdate().first() + await new Promise((resolve) => setTimeout(resolve, 150)) + + assert.isNull(await worker.claimDueSchedule()) + watcher.reset() + assert.isFalse( + await watcher.wait(100, new AbortController().signal, 0), + 'the commit of the other claim notifies' + ) + }) + + test('calls the connection provider again when its settings expire', async ({ + assert, + cleanup, + }) => { + let provided = 0 + let expired = false + const provider = Knex({ + client: 'pg', + connection: async () => { + provided++ + expired = false + return { + host: process.env.PG_HOST || 'localhost', + port: Number.parseInt(process.env.PG_PORT || '5432', 10), + user: process.env.PG_USER || 'postgres', + password: process.env.PG_PASSWORD || 'postgres', + database: process.env.PG_DATABASE || 'queue_test', + application_name: `listen-${provided}`, + expirationChecker: () => expired, + } + }, + }) + cleanup(() => provider.destroy()) + const listenerName = async () => { + const result = await connection.raw( + // The LISTEN connection, not the probe that notifies through another one. + `select application_name from pg_stat_activity where application_name like 'listen-%' and query not like '%pg_notify%'` + ) + return result.rows.map((row: { application_name: string }) => row.application_name) + } + + const subscriber = new KnexAdapter({ + connection: provider, + notifications: true, + tableName, + schedulesTableName, + logger: new MemoryLogger(), + }) + const watcher = await subscriber.watch!(['default']) + cleanup(() => watcher.close()) + const [first] = await listenerName() + + // Rotated credentials: the provider marks its settings expired. + expired = true + await connection.raw( + `select pg_terminate_backend(pid) from pg_stat_activity where application_name = ?`, + [first] + ) + await new Promise((resolve) => setTimeout(resolve, 1500)) + + const [second] = await listenerName() + assert.isDefined(second) + assert.notEqual(second, first) + }) + + test('keeps the settings of the latest connection provider lookup', async ({ + assert, + cleanup, + }) => { + shortenHeartbeat(cleanup) + // A lookup gated here answers once a later one restored LISTEN. + let gate: PromiseWithResolvers | undefined + let expired = false + let provided = 0 + const provider = Knex({ + client: 'pg', + connection: async () => { + const name = gate ? 'listen-stale' : `listen-fresh-${++provided}` + const answered = gate?.promise + gate = undefined + await answered + return { + host: process.env.PG_HOST || 'localhost', + port: Number.parseInt(process.env.PG_PORT || '5432', 10), + user: process.env.PG_USER || 'postgres', + password: process.env.PG_PASSWORD || 'postgres', + database: process.env.PG_DATABASE || 'queue_test', + application_name: name, + expirationChecker: () => expired, + } + }, + }) + cleanup(() => provider.destroy()) + const listeners = async () => { + const result = await connection.raw( + `select application_name from pg_stat_activity where application_name like 'listen-%' and query not like '%pg_notify%'` + ) + return result.rows.map((row: { application_name: string }) => row.application_name) + } + const terminate = (name: string) => + connection.raw( + `select pg_terminate_backend(pid) from pg_stat_activity where application_name = ?`, + [name] + ) + const waitFor = async (name: string, timeout: number) => { + const deadline = Date.now() + timeout + while (Date.now() < deadline) { + const names = await listeners() + if (names.includes(name)) return names + await new Promise((resolve) => setTimeout(resolve, 50)) + } + return listeners() + } + + const watcher = await new KnexAdapter({ + connection: provider, + notifications: true, + tableName, + schedulesTableName, + logger: new MemoryLogger(), + }).watch!(['default']) + cleanup(() => watcher.close()) + + // The next lookup stalls past the startup deadline, the one after it works. + const stale = Promise.withResolvers() + gate = stale + expired = true + await terminate('listen-fresh-1') + assert.include(await waitFor('listen-fresh-2', 8000), 'listen-fresh-2') + expired = false + stale.resolve() + await new Promise((resolve) => setTimeout(resolve, 100)) + + // Reconnecting reuses the cached settings: those of a lookup after the stale one. + await terminate('listen-fresh-2') + await new Promise((resolve) => setTimeout(resolve, 1500)) + const names = await listeners() + assert.lengthOf(names, 1) + assert.match(names[0], /^listen-fresh-/) + }).timeout(15_000) + + test('does not reconnect a LISTEN connection that failed to start', async ({ + assert, + cleanup, + }) => { + const created = fakeListenClients(cleanup, () => ({ + async query() { + const error = new Error('Connection terminated') + created[0].emit('error', error) + throw error + }, + })) + + const subscriber = new KnexAdapter({ + connection, + tableName, + schedulesTableName, + logger: new MemoryLogger(), + notifications: true, + }) + await assert.rejects(() => subscriber.watch!(['default']), 'Connection terminated') + + // Past the reconnection delay + await new Promise((resolve) => setTimeout(resolve, 1200)) + assert.lengthOf(created, 1) + }) + + test('aborts a LISTEN reconnection still in its handshake when closed', async ({ + assert, + cleanup, + }) => { + // The first connection and its probe work, the next never completes its handshake. + const created = fakeListenClients(cleanup, (index) => + index <= 2 ? 'real' : { connect: () => new Promise(() => {}) } + ) + + const subscriber = new KnexAdapter({ + connection, + tableName, + schedulesTableName, + logger: new MemoryLogger(), + notifications: true, + }) + const watcher = await subscriber.watch!(['default']) + created[0].emit('error', new Error('Connection terminated')) + + while (created.length < 3) await new Promise((resolve) => setTimeout(resolve, 20)) + await watcher.close() + assert.isTrue(created[2].ended) + }) + + test('reconnects a LISTEN connection whose replacement stalls', async ({ assert, cleanup }) => { + shortenHeartbeat(cleanup) + // The first connection and its probe work, the next never completes its handshake. + const created = fakeListenClients(cleanup, (index) => + index === 3 ? { connect: () => new Promise(() => {}) } : 'real' + ) + const options = { connection, tableName, schedulesTableName, notifications: true } + const watcher = await new KnexAdapter({ ...options, logger: new MemoryLogger() }).watch!([ + 'default', + ]) + cleanup(() => watcher.close()) + created[0].emit('error', new Error('Connection terminated')) + + // The stalled connection is given up, and the next one listens. + const deadline = Date.now() + 8000 + while (created.length < 5 && Date.now() < deadline) { + await new Promise((resolve) => setTimeout(resolve, 50)) + } + assert.lengthOf(created, 5) + assert.isTrue(created[2].ended) + await new Promise((resolve) => setTimeout(resolve, 200)) + watcher.reset() + const waiting = watcher.wait(2000, new AbortController().signal, 0) + const job = { id: 'after-stall', name: 'TestJob', payload: {}, attempts: 0 } + await new KnexAdapter(options).pushOn('default', job) + assert.isTrue(await waiting) + }).timeout(10_000) + + test('signals no due time for a delayed job deduplicated away', async ({ assert, cleanup }) => { + const options = { connection, tableName, schedulesTableName, notifications: true } + const watcher = await new KnexAdapter(options).watch!(['default']) + cleanup(() => watcher.close()) + watcher.reset() + const start = Date.now() + const waiting = watcher.wait(2000, new AbortController().signal, 0) + + const writer = new KnexAdapter(options) + const job = (id: string, dedup?: { id: string }) => ({ + id, + name: 'TestJob', + payload: {}, + attempts: 0, + dedup, + }) + await writer.pushLaterOn('default', job('owner', { id: 'slot' }), 3000) + // Skipped: no job is due at 200ms. Its time would hide the next one. + await writer.pushLaterOn('default', job('duplicate', { id: 'slot' }), 200) + await writer.pushLaterOn('default', job('next'), 500) + + assert.isTrue(await waiting) + assert.isAtLeast(Date.now() - start, 450) + }) + + test('signals jobs of a table whose name is too long for a channel', async ({ + assert, + cleanup, + }) => { + // A schema and a table name of 63 bytes each, the PostgreSQL maximum. + const schema = `tenant_${'x'.repeat(56)}` + const longTableName = `${schema}.queue_jobs_test` + await connection.raw('create schema if not exists ??', [schema]) + cleanup(() => connection.raw('drop schema ?? cascade', [schema])) + await connection.raw('create table ?? (like ?? including all)', [longTableName, tableName]) + const options = { tableName: longTableName, schedulesTableName, notifications: true } + + const subscriber = new KnexAdapter({ connection, ...options }) + const watcher = await subscriber.watch!(['default']) + cleanup(() => watcher.close()) + watcher.reset() + const waiting = watcher.wait(2000, new AbortController().signal, 0) + + const job = { id: 'long-table-job', name: 'TestJob', payload: {}, attempts: 0 } + await connection.transaction((trx) => + new KnexAdapter({ connection: trx, ...options }).pushOn('default', job) + ) + assert.isTrue(await waiting) + assert.isNotNull(await subscriber.getJob(job.id, 'default')) + }) + + test('reconnects a LISTEN connection that stops answering', async ({ assert, cleanup }) => { + shortenHeartbeat(cleanup) + const proxy = await startSilentProxy({ + host: process.env.PG_HOST || 'localhost', + port: Number.parseInt(process.env.PG_PORT || '5432', 10), + }) + cleanup(() => proxy.close()) + // The LISTEN connection goes through the proxy. + const knexClient = connection.client as KnexTypes.Client & { + connectionConfigProvider?: () => Promise + } + knexClient.connectionConfigProvider = async () => ({ + host: '127.0.0.1', + port: proxy.port, + user: process.env.PG_USER || 'postgres', + password: process.env.PG_PASSWORD || 'postgres', + database: process.env.PG_DATABASE || 'queue_test', + }) + cleanup(() => { + knexClient.connectionConfigProvider = undefined + }) + + const logger = new MemoryLogger() + const watcher = await new KnexAdapter({ + connection, + tableName, + schedulesTableName, + logger, + notifications: true, + }).watch!(['default']) + cleanup(() => watcher.close()) + watcher.reset() + const waiting = watcher.wait(4000, new AbortController().signal, 0) + proxy.silence() + + // Its notification is lost: the new connection announces it. + await new KnexAdapter({ + connection, + tableName, + schedulesTableName, + notifications: true, + }).pushOn('default', { + id: 'after-silence', + name: 'TestJob', + payload: {}, + attempts: 0, + }) + assert.isTrue(await waiting) + assert.include( + logger.logs.map((entry) => String(entry.obj?.err)), + 'Error: The LISTEN connection did not answer a ping' + ) + }).timeout(6000) + + test('a watcher woken for nothing sleeps through its cooldown', async ({ assert, cleanup }) => { + const watcher = await new KnexAdapter({ + connection, + tableName, + schedulesTableName, + notifications: true, + }).watch!(['default']) + cleanup(() => watcher.close()) + watcher.reset() + + const start = Date.now() + const waiting = watcher.wait(2000, new AbortController().signal, 300) + await new KnexAdapter({ + connection, + tableName, + schedulesTableName, + notifications: true, + }).pushOn('default', { + id: 'during-cooldown', + name: 'TestJob', + payload: {}, + attempts: 0, + }) + assert.isTrue(await waiting) + assert.closeTo(Date.now() - start, 300, 80) + }) + + test('sends one notification per interval for a burst of pushes to several queues', async ({ + assert, + cleanup, + }) => { + const listener = (await connection.client.acquireRawConnection()) as { + query(sql: string): Promise + on(event: 'notification', listener: (message: { payload?: string }) => void): void + } + cleanup(() => connection.client.destroyRawConnection(listener)) + const payloads: string[] = [] + listener.on('notification', (message) => payloads.push(message.payload ?? '')) + await listener.query(`LISTEN "${tableName}"`) + + const writer = new KnexAdapter({ + connection, + tableName, + schedulesTableName, + notifications: true, + }) + const start = Date.now() + for (let index = 0; index < 50; index++) { + // Spread over queues: the interval holds for the adapter. + await writer.pushOn(`queue-${index % 5}`, { + id: `burst-${index}`, + name: 'TestJob', + payload: {}, + attempts: 0, + }) + } + const elapsed = Date.now() - start + await writer.destroy() + await new Promise((resolve) => setTimeout(resolve, 50)) + + assert.isAtLeast(payloads.length, 1) + assert.isAtMost(payloads.length, Math.ceil(elapsed / 50) + 1) + }) + + test('watches without holding a connection of the pool', async ({ assert, cleanup }) => { + // The worker holds a pool connection while its watcher starts. + const trx = await connection.transaction() + cleanup(async () => { + await trx.rollback() + }) + await trx.raw('select 1') + const pool = ( + connection.client as KnexTypes.Client & { + pool: { numUsed(): number; numFree(): number } + } + ).pool + const before = pool.numUsed() + pool.numFree() + + const watcher = await new KnexAdapter({ + connection, + tableName, + schedulesTableName, + notifications: true, + }).watch!(['default']) + cleanup(() => watcher.close()) + + assert.equal(pool.numUsed() + pool.numFree(), before) + }) + + test('knex() factory accepts a table name or options', async ({ assert }) => { + const job = { id: 'factory-job', name: 'TestJob', payload: {}, attempts: 0 } + + await knex(connection, tableName)().pushOn('default', job) + + const polling = knex(connection, { tableName, schedulesTableName })() + assert.isUndefined(polling.watch, 'polls by default') + assert.isNotNull(await polling.getJob(job.id, 'default'), 'the string sets the jobs table') + + const listening = knex(connection, { tableName, schedulesTableName, notifications: true })() + assert.isFunction(listening.watch) + }) + registerPruneFailureSuite({ test, createAdapter: (logger) => { diff --git a/tests/worker_session.spec.ts b/tests/worker_session.spec.ts index 256172d..7731110 100644 --- a/tests/worker_session.spec.ts +++ b/tests/worker_session.spec.ts @@ -11,6 +11,8 @@ import { import { ControllableAdapter } from './_mocks/controllable_adapter.js' import { trackPromise } from './_utils/track_promise.js' import { MemoryLogger } from './_mocks/memory_logger.js' +import type { WorkWatcher } from '../src/contracts/adapter.js' +import { WorkerWakeup } from '../src/drivers/worker_wakeup.js' type SessionOverrides = Partial> & { jobExecutionRuntime?: Partial @@ -81,20 +83,13 @@ test.group('WorkerSession', () => { assert.deepEqual(adapter.polledQueues, ['critical', 'default', 'low']) }) - test('fans out remaining acquisitions after the initial probe finds work', async ({ - assert, - cleanup, - }) => { + test('acquires one job at a time until the pool is full', async ({ assert, cleanup }) => { const adapter = new ControllableAdapter() const executionGate = Promise.withResolvers() - let executionsStarted = 0 - adapter.acquisitions.block(2, 3) - const session = createSession({ adapter, jobExecutionRuntime: { async execute() { - executionsStarted++ await executionGate.promise return { type: 'completed' } }, @@ -103,12 +98,11 @@ test.group('WorkerSession', () => { }) cleanup(async () => { - adapter.releaseAll() executionGate.resolve() await session.stop() }) - for (let index = 1; index <= 3; index++) { + for (let index = 1; index <= 4; index++) { await adapter.pushOn('default', { id: `backlog-job-${index}`, name: 'TestJob', @@ -118,18 +112,11 @@ test.group('WorkerSession', () => { }) } - const cycle = session.processCycle() - await adapter.acquisitions.waitForStarted(3) - - assert.equal(adapter.acquisitions.calls, 3) - - adapter.acquisitions.release(2, 3) - - assert.equal((await cycle)?.type, 'started') - assert.equal(executionsStarted, 3) + assert.equal((await session.processCycle())?.type, 'started') + assert.equal(adapter.acquisitions.calls, 3, 'stops once the pool is full') }) - test('uses one probe per subsequent empty tick after processing one job', async ({ + test('uses one acquisition per empty tick after processing one job', async ({ assert, cleanup, }) => { @@ -147,14 +134,14 @@ test.group('WorkerSession', () => { }) assert.equal((await session.processCycle())?.type, 'started') - assert.equal(adapter.acquisitions.calls, 10) + assert.equal(adapter.acquisitions.calls, 2, 'stops at the first empty acquisition') assert.equal((await session.processCycle())?.type, 'completed') assert.equal((await session.processCycle())?.type, 'idle') - assert.equal(adapter.acquisitions.calls, 11) + assert.equal(adapter.acquisitions.calls, 3) assert.equal((await session.processCycle())?.type, 'idle') - assert.equal(adapter.acquisitions.calls, 12) + assert.equal(adapter.acquisitions.calls, 4) }) test('reuses one completion wait across idle ticks and observes a later job', async ({ @@ -796,76 +783,200 @@ test.group('WorkerSession', () => { await Promise.all([stop, start]) }) - test('waits for sibling acquisitions when one acquisition fails', async ({ assert, cleanup }) => { + test('emits a started job before a later acquisition error', async ({ assert, cleanup }) => { const adapter = new ControllableAdapter() - let jobCompleted = false - adapter.acquisitions.block(2, 3).fail(2, new Error('Failed to acquire job')) - adapter.finalizations.block(1) + adapter.acquisitions.fail(2, new Error('Failed to acquire the next job')) + const session = createSession({ adapter, settings: { concurrency: 2 } }) + cleanup(() => session.stop()) + await adapter.pushOn('default', { + id: 'success-before-error', + name: 'TestJob', + payload: {}, + attempts: 0, + priority: 0, + }) + + const startedCycle = await session.processCycle() + const errorCycle = await session.processCycle() + + assert.equal(startedCycle?.type, 'started') + assert.equal(errorCycle?.type, 'error') + }) +}) + +/** Hands the watcher of `watch` to the test: `notify` wakes it. */ +class WatchableAdapter extends ControllableAdapter { + wakeup?: WorkerWakeup + watchError?: Error + /** The cooldown of each wait. */ + cooldowns: number[] = [] + + async watch(_queues: readonly string[], _signal?: AbortSignal): Promise { + if (this.watchError) throw this.watchError + const wakeup = new WorkerWakeup() + this.wakeup = wakeup + return { + reset: () => wakeup.reset(), + wait: (maxDelay, signal, cooldown) => { + this.cooldowns.push(cooldown) + return wakeup.wait(maxDelay, signal, cooldown) + }, + close: async () => { + this.wakeup = undefined + }, + } + } +} + +const job = (id: string) => ({ id, name: 'TestJob', payload: {}, attempts: 0 }) + +test.group('WorkerSession | work watcher', () => { + test('sleeps until a wake signal instead of polling every idleDelay', async ({ + assert, + cleanup, + }) => { + const adapter = new WatchableAdapter() + const executed = Promise.withResolvers() const session = createSession({ adapter, + settings: { idleDelay: 10 }, jobExecutionRuntime: { async execute() { - jobCompleted = true + executed.resolve() return { type: 'completed' } }, }, - settings: { concurrency: 3 }, }) + cleanup(() => session.stop()) + const start = session.start() + await adapter.acquisitions.waitForSettled(1) + await setTimeout(100) + assert.equal(adapter.acquisitions.calls, 1) + + await adapter.pushOn('default', job('woken')) + adapter.wakeup!.notify(Date.now()) + await executed.promise + + await session.stop() + await start + assert.isUndefined(adapter.wakeup, 'stopping closes the watcher') + }) + + test('wakes when a due time is reached', async ({ assert, cleanup }) => { + const adapter = new WatchableAdapter() + const session = createSession({ adapter }) + cleanup(() => session.stop()) + + void session.start() + await adapter.acquisitions.waitForSettled(1) + adapter.wakeup!.notify(Date.now() + 150) + + await setTimeout(75) + assert.equal(adapter.acquisitions.calls, 1) + await setTimeout(150) + assert.equal(adapter.acquisitions.calls, 2) + }) + + test('keeps a wake received while looking for work', async ({ cleanup }) => { + const adapter = new WatchableAdapter() + adapter.acquisitions.block(1) + const session = createSession({ adapter }) cleanup(async () => { - adapter.releaseAll() + adapter.acquisitions.releaseAll() await session.stop() }) - await adapter.pushOn('default', { - id: 'late-job-after-sibling-error', - name: 'TestJob', - payload: {}, - attempts: 0, - priority: 0, - }) + void session.start() + await adapter.acquisitions.waitForStarted(1) + adapter.wakeup!.notify(Date.now()) + adapter.acquisitions.release(1) - const start = session.start() - await adapter.acquisitions.waitForStarted(3) - adapter.acquisitions.release(2) - await adapter.acquisitions.waitForSettled(2) + // Lost, the signal would leave the session asleep until the stalled check. + await adapter.acquisitions.waitForStarted(2) + }) - const stop = trackPromise(session.stop()) - assert.isFalse(stop.settled) + test('polls and stops while the watcher is not ready', async ({ assert }) => { + const adapter = new WatchableAdapter() + let aborted = false + adapter.watch = (_queues, signal) => + new Promise((_resolve, reject) => { + signal?.addEventListener('abort', () => { + aborted = true + reject(signal.reason) + }) + }) + const logger = new MemoryLogger() + const session = createSession({ adapter, logger, settings: { idleDelay: 10 } }) - adapter.acquisitions.release(3) - await adapter.finalizations.waitForStarted() - assert.isTrue(jobCompleted) - assert.isFalse(stop.settled) + const start = session.start() + await adapter.acquisitions.waitForStarted(3) + await session.stop() + await start - adapter.finalizations.release(1) - await Promise.all([stop.promise, start]) - assert.equal(adapter.acquisitions.calls, 3) - assert.isNull(await adapter.getJob('late-job-after-sibling-error', 'default')) + assert.isTrue(aborted, 'stopping aborts the pending watch') + assert.deepEqual(logger.logs, [], 'an aborted watch is not an error') }) - test('emits successful acquisitions before a sibling acquisition error', async ({ + test('follows the wakes of a watcher that gets ready while polling', async ({ assert, cleanup, }) => { - const adapter = new ControllableAdapter() - adapter.acquisitions.fail(2, new Error('Failed to acquire sibling job')) - const session = createSession({ adapter, settings: { concurrency: 2 } }) + const adapter = new WatchableAdapter() + const ready = Promise.withResolvers() + const watch = adapter.watch.bind(adapter) + adapter.watch = async (queues, signal) => { + await ready.promise + return watch(queues, signal) + } + const session = createSession({ adapter, settings: { idleDelay: 60_000 } }) cleanup(() => session.stop()) - await adapter.pushOn('default', { - id: 'success-before-error', - name: 'TestJob', - payload: {}, - attempts: 0, - priority: 0, - }) + void session.start() + await adapter.acquisitions.waitForSettled(1) + await setTimeout(20) // the session now sleeps for idleDelay + ready.resolve() + while (!adapter.wakeup) await setTimeout(1) + adapter.wakeup.notify(Date.now() + 100) + + // Without it, the first poll would wait for idleDelay. + await setTimeout(200) + assert.isAtLeast(adapter.acquisitions.calls, 2) + }) - const startedCycle = await session.processCycle() - const errorCycle = await session.processCycle() + test('polls every idleDelay when watching fails', async ({ assert, cleanup }) => { + const adapter = new WatchableAdapter() + adapter.watchError = new Error('LISTEN is not supported') + const logger = new MemoryLogger() + const session = createSession({ adapter, logger, settings: { idleDelay: 10 } }) + cleanup(() => session.stop()) - assert.equal(startedCycle?.type, 'started') - assert.equal(errorCycle?.type, 'error') + void session.start() + await adapter.acquisitions.waitForStarted(3) + + const errors = logger.logs.filter((entry) => entry.level === 'error') + assert.equal(errors[0].message, 'Failed to watch for work, polling instead') + assert.strictEqual(errors[0].obj?.err, adapter.watchError) + }) + + test('cools down for idleDelay after a wake that found no work', async ({ assert, cleanup }) => { + const adapter = new WatchableAdapter() + const session = createSession({ adapter, settings: { idleDelay: 50 } }) + cleanup(() => session.stop()) + + void session.start() + while (adapter.cooldowns.length === 0) await setTimeout(1) + + // Another worker took the job: this wake finds nothing. + adapter.wakeup!.notify(Date.now()) + while (adapter.cooldowns.length < 2) await setTimeout(1) + assert.equal(adapter.cooldowns[1], 50) + + // A wake that finds a job: no cooldown once it is done. + await adapter.pushOn('default', job('found')) + adapter.wakeup!.notify(Date.now()) + while (adapter.cooldowns.length < 3) await setTimeout(1) + assert.deepEqual(adapter.cooldowns, [0, 50, 0]) }) })