Repository navigation
feat(worker): stop polling idle workers on Redis and PostgreSQL - #34
Open
RomainLanz wants to merge 11 commits into
Open
RomainLanz wants to merge 11 commits into
RomainLanz wants to merge 11 commits into
Conversation
An idle worker polled its queues every idleDelay (2s by default): on PostgreSQL, several transactions per queue on every poll, and up to idleDelay of latency before a new job ran. Adapters may now implement subscribe(queues, onWake, signal). onWake receives the time at which work becomes due: now for a new job, later for a delayed job, a retry, or a schedule run. A subscribed worker sleeps until a signal for now, the next due time it was told about, or the next stalled check, which is also the poll that finds lost signals. The worker polls until its subscription is ready, and aborts it when it stops first. Adapters announce now, the next delayed job of each queue, and the next schedule that can run when they subscribe and after each reconnection. A pop that promotes delayed jobs announces the next delayed job of its queue. - Redis publishes from the Lua scripts that write the job or schedule, so the signal goes with the write in one round trip, and listens on a duplicated connection. - Knex on PostgreSQL sends pg_notify on the connection of the write: a job dispatched in an application transaction is signaled on commit, and not at all on rollback. A connection outside the pool runs LISTEN; the adapter owns its client before it connects, so it can end it in its handshake, and calls a connection provider again once its settings expire. MySQL, SQLite, and Kysely keep polling. The knex() factory now accepts options as its second argument, including notifications: false for poolers without LISTEN support such as PgBouncer in transaction mode. A table name string still works.
Tests against PgBouncer, Redis ACLs and multiple hosts found cases where wake signals broke writes, slowed workers or went unnoticed: - Redis: a user without channel permissions failed every push after its writes, because PUBLISH raised inside the scripts. Scripts now use redis.pcall, so the write succeeds and the worker polls. The channel also includes the database, since pub/sub ignores it. - PostgreSQL behind PgBouncer in transaction mode: LISTEN succeeded but never received anything, so jobs waited for the stalled check. The subscriber now sends a probe notification and falls back to polling with a clear message when it does not arrive. - A table name over 63 bytes failed pg_notify. The channel name is hashed in that case. - "Now" signals carried the producer clock: a producer ahead of the worker delayed jobs until the stalled check. They no longer carry a timestamp, so the worker uses its own clock. - Sequential pushes lost about a third of their throughput, as PostgreSQL serializes NOTIFY commits. Signals for ready jobs are now coalesced per queue, and a delayed job is inserted and signalled in one statement.
A subscription connection can stop delivering without closing, behind a stuck proxy or a half-open TCP connection. The worker then ran jobs only at its stalled check, 30 s by default, until it restarted. The subscriber now sends PING on Redis, or SELECT 1 on the LISTEN connection, every 10 s, and reconnects when it gets no answer within 5 s. The reconnection announces due work, as after any reconnection.
A load test showed the wake signals did not scale. Every push woke every idle worker, and each ran a full cycle: on PostgreSQL with 100 workers at 1000 jobs/s, throughput fell from 635 to 234 jobs/s. Every NOTIFY also reached every LISTEN connection, an idle PostgreSQL worker held 3 connections instead of 1, and every schedule claim made every worker read the next schedule. Adapters now implement watch(), which returns a watcher the worker waits on, instead of subscribe(), which pushed signals to it. Redis waits on markers with BZPOPMIN, as BullMQ does. A push adds a marker to its queue; BZPOPMIN hands it to one waiting worker, and a worker that pops a job adds it again while jobs remain. Delayed jobs and schedules keep their next due time in a marker that one worker holds, and puts back when it stops waiting first. Markers outlive a reconnection, so the pub/sub channel and its announcements are gone. PostgreSQL keeps LISTEN/NOTIFY, which wakes every worker, and bounds what it costs: - A worker woken for nothing waits idleDelay before it wakes again, so idle workers query no more often than when polling. - Outside of a transaction, a queue gets at most one NOTIFY per 50 ms per adapter. - Schedule changes carry the next run, read once by the writer. - The LISTEN connection reads what it announces, and the probe uses a short-lived connection, so the pool does not grow.
ioredis reconnects a closed connection but never settles the command that was in flight on it. After a server-side kill, the watcher waited for its watchdog, up to the BZPOPMIN timeout plus 5 s, and work due meanwhile started that late. The watcher now ends the read when the connection closes, and looks for work at once, since Redis may have popped a marker whose reply was lost.
After a probe found a job, the worker asked for one job per free slot at once. A worker woken for a single job with a concurrency of 10 sent 10 acquisitions, 9 of them empty: on Redis, about 6 times the commands per job of polling. The worker now acquires one job at a time until its pool is full or a queue runs dry, as BullMQ does. An empty queue costs one acquisition.
On PostgreSQL, LISTEN/NOTIFY costs grow with the number of workers: each worker holds one more connection, and every NOTIFY reaches every LISTEN connection. A load test with 100 workers doubled the database CPU, and with max_connections at 100, 40 workers fit instead of 90. Enabled by default, an upgrade alone could exhaust the connections of a deployment. Workers on PostgreSQL now poll unless the adapter is created with `notifications: true`, on every process that dispatches jobs. An adapter now sends at most one NOTIFY per 50 ms for all its queues, instead of one per queue: with 50 queues, the limit did not hold.
Four failure paths escaped the fallback to polling: - The PostgreSQL probe armed its 2s timeout before its connection completed. A probe connection that stalled left the rejection unobserved, and Node exited instead of falling back to polling. - Aborting a Redis watch only disconnected the blocking client. A PING queued during a silent handshake never settled, so stopping the worker hung. Abort now rejects the PING. - A notification whose payload parsed to null threw in the pg event handler. Any role that can connect can send one, and it killed the worker. Payloads that are not objects are now ignored. - The LISTEN connection did not apply the Knex searchPath, so its announcement queries failed for unqualified tables found only through it. It now sets the search path as the pool does.
A replacement LISTEN connection that stalled in its handshake or setup never failed: the heartbeat skips a connection until it listens, so the watcher stayed without notifications until the worker restarted, and closing it waited on a server that never answered. The whole startup of a LISTEN connection now has a deadline, after which the connection is dropped and the reconnection goes on. A connection that does not end gracefully is destroyed. A deduplicated dispatch signalled its attempted due time even when it added no job. That time could replace the real due time of another job in the same notification, and the wake found nothing to promote, so nothing announced the real one. Only added jobs signal now.
The probe client had no error listener: a connection failure while it waited for its notification emitted an unhandled error event and killed the process, instead of rejecting watch(). Its errors now reject the probe. The probe client also ended gracefully without a bound, so a connection stalled in its handshake kept its socket open and could hold the process at shutdown. It now ends like the LISTEN client: its socket is destroyed when end() does not settle in time. Both keep an error listener while they end.
A LISTEN connection looked its settings up, through the async provider of Knex, before its startup deadline and abort handling applied. A lookup that stalled made aborting the watch wait for it, so stopping a worker hung, and a reconnection could stall forever. The lookup now runs within the bounded startup, and a client it returns once the attempt was given up stays unused.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Problem
An idle worker polls its queues every
idleDelay(2 s by default). Each poll claims the due schedules and pops every queue, so many workers send many requests for no work, and a new job waits up toidleDelaybefore it runs.Solution
Adapters can implement an optional
watch(queues, signal). It returns a watcher the worker waits on when it is idle, instead of polling. The worker still checks its queues everystalledInterval, in case a wake was lost.Redis, on by default. Workers wait on markers with
BZPOPMIN, as BullMQ does:BZPOPMINhands it to one waiting worker;PostgreSQL (Knex), opt-in with
knex(config, { notifications: true }). WorkersLISTEN, and a push sendspg_notifyon its own connection, so a job dispatched in a transaction wakes workers when it commits. It stays off by default because its cost grows with the number of workers: one more connection per worker, and everyNOTIFYreaches every worker. Its cost is bounded:idleDelaybefore it wakes again;NOTIFYper 50 ms outside transactions;LISTENconnection pings every 10 s, and reconnects when it gets no answer.MySQL, SQLite, Kysely: no change.
The worker now acquires jobs one at a time until its pool is full or the queues are empty. Before, a worker with a concurrency of 10 that found one job sent 9 more acquisitions at once, which a wake per job made costly.
Upgrading
notifications: true. Set it on every process that dispatches jobs, not only on workers. Leave it off behind PgBouncer in transaction mode.Proof
Three orbs ran
main(ea13357), the first version of this PR (f8bd75c, pub/sub), and the marker version, against real Redis and PostgreSQL. In the load tests, each database ran in Docker capped at 2 CPUs.mainmain's level. On Redis, quiet pickup p50 / p99 went from 444 / 1786 ms to 3 / 215 ms.LISTENconnection silenced by a proxy recovers. The first version of this PR stayed at about 6 s of latency until the worker restarted.Limits:
mainper job under load, and pushes 7 to 9% slower at 1000 jobs/s with 32 workers or fewer, with Redis capped at 2 CPUs.Tests
tests/worker_session.spec.ts:tests/_utils/register_work_watcher_suite.ts, for Redis and Knex (PostgreSQL): wakes for pushes, delayed jobs, retries and schedules; work stored before watching; the next delayed job of each queue; expired and deleted schedules; transaction commit and rollback.NOTIFYper interval across queues;LISTENconnection, aLISTENbehind a pooler that receives nothing, handshakes aborted on close;knex()options.The full suite passes (1730 tests) on Redis, PostgreSQL, MySQL 8.4 and SQLite.