@zudojs/queue

OVERVIEW

Some work should not happen while a user waits. Sending an email, resizing an image or calling a slow third-party API can take seconds, and if it fails you want to try again later. A job queue is a list of such tasks that your program works through in the background, one at a time or a few at once.

@zudojs/queue gives you that list. You add a job (a name plus some data), you register a processor (the function that does the work for that name), and the queue runs the processor for each job. If the processor throws, the queue can retry with a delay and, when every attempt fails, park the job in a dead-letter store so it is not silently lost.

Everything in this package runs in memory inside one Node.js process. Jobs disappear when the process exits. That is ideal for development, tests and small apps; a durable backend would need a custom Queue implementation.

INSTALLATION

Install the package. The three packages it depends on are pulled in automatically.

QUICK START

This example creates a queue, registers a processor for jobs named "send-email", adds one job, and waits for it to finish.

What you should see: waiting, then the "Sending ..." line, then completed. The queue starts polling for work the moment you call process(), so you never have to start it by hand.

JOBS AND PROCESSORS

A job is one unit of work. It has a name (which processor should run it), data (the input), and a state that the queue moves through as work happens. You never build a job by hand; queue.add() builds it and returns it.

A processor is an async function registered with queue.process(name, fn). The queue calls it with two arguments: the job, and a JobContext. The context carries an AbortSignal that fires if the job times out or the queue shuts down, the 1-based attemptNumber, and an updateProgress() method.

Inside a processor, job.attempt counts the attempts that have already failed, so it starts at 0: a job with attempts: 3 runs with job.attempt equal to 0, then 1, then 2. For a human-readable count use the context's attemptNumber (new in v1.4.0), which is 1-based and always job.attempt + 1, the same convention as ctx.attempt in @zudojs/scheduler. job.maxAttempts holds the total.

Job states

The JobState enum lists every state. Its values are lowercase strings, so job.state === "completed" works.

Reporting progress and honouring cancellation

This processor reports progress with createJobProgress() and stops early if the signal is aborted.

A processor returns either nothing or a JobResult, built with createJobResult(data, durationMs) or createJobErrorResult(message, durationMs). Only a result with success: false, or a thrown error, counts as a failure. Each job gets a 30-second timeout unless you pass timeout.

JOB OPTIONS

The third argument to queue.add() is a JobOptions object. Every field is optional. You can also set defaults for the whole queue with QueueOptions.defaultJobOptions.

Dates and other non-JSON values in data

add() copies the payload through the queue's serializer. Since v1.4.0 the default JsonSerializer preserves types, so a Date in data reaches the processor as a Date, matching what Queue<{ dueAt: Date }> promises. BigInt, Map, Set, Uint8Array and Error round-trip too. Output for plain JSON data is unchanged. No options are needed:

The default uses the tagged format of @zudojs/serialization. createJsonSerializer() also defaults to preserveTypes: true; pass serializer: createJsonSerializer({ preserveTypes: false }) for plain JSON, where a Date becomes its ISO string (type such fields as string). Class instances still lose their prototype through JSON; to keep them, pass serializer: PassthroughSerializer (or serializePayloads: false) and the in-memory queue stores the payload by reference. Before v1.4.0 the default was plain JSON: a Date arrived as a string, a BigInt made add() throw, and a Map or Set arrived as {}.

This example uses priority, delay and deduplication together. JobPriorityLevels is a set of named numbers you can use instead of guessing.

getNextJob() only peeks; it does not change the job. The duplicate add() throws a JobDuplicateError from @zudojs/errors.

RETRIES AND BACKOFF

A retry means running the same job again after it fails. Backoff is the pause before each retry. Pausing matters: if a service is down, hitting it again instantly just fails again, and hundreds of jobs doing that at once make things worse.

Set attempts to more than 1 and the queue retries. Add a backoff to control the pause. Two shapes exist: fixed (same pause every time) and exponential (the pause doubles each time, up to maxDelay). A jitter setting adds randomness so failed jobs do not all retry in the same instant.

This processor fails twice and succeeds on the third attempt, with a 50 ms fixed pause between tries.

You can also compute delays yourself. calculateRetryDelay(attempt, backoff) is the function the queue uses internally.

DEAD LETTER

When a job has used all its attempts, the queue does not delete it. It moves the job to a dead-letter store: a holding area for jobs that could not be done. You can inspect them, log them or re-add them by hand. The term comes from postal services, where undeliverable mail goes to a "dead letter office".

Every queue has an in-memory dead-letter store by default. Read it with getDeadLetterJobs().

Each entry is a DeadLetterJob: a copy of the job as it was when it ran out of attempts (so its state still reads "failed"), the error, the number of attempts, and a reason string. To keep dead jobs somewhere else, pass your own DeadLetterStore as QueueOptions.deadLetterStore; createInMemoryDeadLetterStore() shows the shape to copy.

Retention is bounded

Since v1.3.0 the default in-memory store keeps the most recent 1000 dead-lettered jobs (DEFAULT_DEAD_LETTER_JOBS) and evicts the oldest past that. Before v1.3.0 it was unbounded, so a queue that dead-lettered steadily grew until the process ran out of memory. Build the store yourself to choose a different cap.

WORKERS

A worker is a loop that repeatedly asks a queue "anything to do?", claims one job, and runs it. The in-memory queue already contains such a loop, which is why the Quick Start needed no worker. Creating a Worker switches that loop off as a consumer (queue.setAutoProcess(false)), so the worker is the only thing running jobs and worker.stop() really stops consumption. A separate Worker is useful when you want its own concurrency limit, its own middleware, start/stop control, or per-worker statistics.

A worker never runs a processor itself. It calls queue.claimNextJob() (which marks the job active so nobody else takes it) and then queue.runJob(), so retries, dead-lettering and middleware all still apply.

Start a worker, let it process a job, then stop it cleanly.

Worker options

MIDDLEWARE

Middleware is a function that wraps every processor call. It receives a context with the job and a next() function; calling next() runs the rest of the chain and, finally, the processor. Use it for things every job needs, such as logging or timing, without repeating code in each processor.

Pass middleware in QueueOptions.middleware. This example adds the built-in logging middleware and one custom timer.

What you should see: "Job processing started" with the job details, then "send-email took 0ms", then "Job processing completed". Middleware runs in array order, outermost first.

EVENTS

The queue can tell you when things happen: a job was created, started, completed, failed, cancelled, or will retry, and a worker started, stopped or hit an error. Since v1.4.0 every queue has an event emitter (an object you can subscribe to) out of the box: a queue created without one gets an in-memory emitter, reachable as queue.events. Before v1.4.0 the default was a silent no-op. To share one emitter between queues, or to plug in your own, create it and hand it to the queue as QueueOptions.eventEmitter.

Subscribe with on(event, handler). It returns a function that unsubscribes.

Worker events and Queue.events

Before v1.3.0 the last four rows above were declared in QueueEventMap but nothing emitted them, so subscribing to worker:started for readiness never fired. They fire now. A worker reports its own lifecycle on the emitter belonging to the queue it consumes, reached through the new Queue.events property — you do not pass the emitter to createWorker.

Queue.events is optional on the Queue interface, so reach it with queue.events?.. The in-memory queue always has one: a queue created without an eventEmitter exposes its own in-memory emitter, so you can subscribe with no set-up at all:

PAUSING, STATS AND SHUTDOWN

pause() stops the queue from starting new jobs; resume() starts them again. By default a paused queue also rejects add(); set pauseRejectsAdd: false in QueueOptions to keep accepting jobs while paused.

close() waits for running jobs (up to closeTimeout, 30 s), cancels pending timers and forgets every job. Always call it when your program shuts down. After closing, add() throws QueueDisposedError.

Use createQueueManager() when you have several queues and want to close them all at once.

API REFERENCE

Everything below is exported from @zudojs/queue unless a note says otherwise.

Functions

Queue methods

Queue options

Types and constants

Errors

These are thrown by the queue but live in @zudojs/errors. Import them from there.

COMMON MISTAKES

• → Passing a plain string as the queue name. TypeScript reports a type error on createInMemoryQueue("emails"). Wrap it: createQueueName("emails").
• → Importing errors from @zudojs/queue. The import is undefined at runtime and instanceof checks silently fail. Import JobTimeoutError and friends from @zudojs/errors.
• → Reading job.state from the object add() returned. Jobs are immutable snapshots, so it stays "waiting" forever. Call queue.getJob(job.id) to see the current state.
• → Expecting attempts: 3 to retry instantly. Without a backoff the default is exponential starting at 1 second, so a test that waits 200 ms sees only one attempt. Pass createFixedBackoff(10) in tests.
• → Adding class instances or cyclic objects. The default serializer keeps Date, BigInt, Map, Set, Uint8Array and Error, but class instances lose their methods and cyclic objects throw JobSerializationError. Put IDs and plain data in the job, or pass serializer: PassthroughSerializer to store payloads by reference.
• → Treating job.attempt as 1-based. It is 0 on the first run. Use the processor context's ctx.attemptNumber (1-based, equal to job.attempt + 1) for a human-readable count.
• → Forgetting close(). Jobs you never awaited are abandoned and pending retries are lost when the process exits. Close every queue (or manager.closeAll()) during shutdown.

RELATED PACKAGES

• → @zudojs/scheduler — runs a job on a timetable (cron, every N minutes). A queue runs a job because something happened. A scheduler handler often just adds a job to a queue.
• → @zudojs/errors — the QueueError family you catch and inspect.
• → @zudojs/events — application-wide event bus; implement QueueEventEmitter on top of it to forward queue events.
• → @zudojs/serialization — the JSON serializer behind JsonSerializer.
• → @zudojs/messaging — message brokers and pub/sub between services, rather than background work inside one.

COMPLETE EXPORT INDEX

Every name @zudojs/queue exports from its package root at v1.3.0 — 87 in total, generated from the package’s own entry point rather than written by hand. The sections above explain the ones you reach for most; this is the exhaustive list, so nothing shipped is undocumented. Names not covered above are typically internal helpers and supporting types.