JavaScript

The JavaScript client is the reference implementation: every other SDK is a transliteration of it. It is ESM-only, requires Node 22 or newer (engines.node: ">=22.0.0"), and ships the broker client and the streaming SDK in one package.

npm i queen-mq

yarn add queen-mq

pnpm add queen-mq

bun add queen-mq

import { Queen } from 'queen-mq'

The package also exports Admin, the five defaults objects (CLIENT_DEFAULTS, QUEUE_DEFAULTS, CONSUME_DEFAULTS, POP_DEFAULTS, BUFFER_DEFAULTS), and the streaming entry points Stream, tokenBucketGate and slidingWindowGate. Named ESM exports mean a bundler can drop the streaming code if you only import Queen.

Construct a client

The constructor accepts three shapes: a URL string, an array of URLs, or a config object.

const client = new Queen('http://localhost:6632')

const client = new Queen(['http://broker-a:6632', 'http://broker-b:6632'])

const client = new Queen({
  urls: ['http://broker-a:6632', 'http://broker-b:6632'],
  bearerToken: process.env.QUEEN_TOKEN,
  timeoutMillis: 30000,
  loadBalancingStrategy: 'affinity',
  handleSignals: false,
})

With one URL the client talks to it directly. With more than one it builds a load balancer and enables failover, so a 5xx or a network error marks that backend unhealthy and the request is retried against the next one. A 4xx is never retried across backends, and a 429 is retried in place against the same backend (see 429 and 403).

Every URL is validated at construction. Passing neither url nor urls throws Must provide urls or url in configuration.

Client options

Defaults come from CLIENT_DEFAULTS in utils/defaults.js.

Push

const res = await client
.queue('test-queue-v2')
.push([{ data: { message: 'Hello, world!' } }])

push() takes one item or an array. Each item’s body is data, or payload, or, if neither key is present, the item itself. A transactionId you do not set is minted as a UUIDv7 client-side. A traceId is forwarded only if it is a valid UUID.

The call returns a thenable builder, so await executes it. The resolved value is the broker’s array, one entry per item in request order:

const res = await client.queue('orders').push([{ data: { id: 1 } }])
// res[0] === { index: 0, message_id: '...', transaction_id: '...', queueName: 'orders', status: 'queued' }

status is queued, duplicate, buffered or failed. buffered means the broker’s database write failed and the message went to the broker’s disk spool for replay. The HTTP status is still 201, so checking the code alone hides it. The push callbacks below only bucket queued, duplicate and failed, so a buffered item fires none of them: read status off the returned array if you need to see it.

Supply your own transactionId to make a push idempotent inside the queue’s dedup window:

const res1 = await client
.queue('test-queue-v2')
.push([{ transactionId: 'test-transaction-id', data: { message: 'Hello, world!' } }])

const res2 = await client
.queue('test-queue-v2')
.push([{ transactionId: 'test-transaction-id', data: { message: 'Hello, world!' } }])
// res1[0].status === 'queued', res2[0].status === 'duplicate'

Before awaiting, you can attach callbacks that split the response by status:

await client
  .queue('orders')
  .push(items)
  .onSuccess(queued => metrics.add(queued.length))
  .onDuplicate(dups => log.warn({ n: dups.length }, 'already accepted'))
  .onError((failed, err) => log.error(err))

Without an onError callback, a failed item makes the await throw. With one, it does not.

Pop

const res = await client
.queue('test-queue-v2-pop-non-empty')
.batch(1)
.wait(true)
.pop()

pop() resolves to an array of messages, or []. Each message carries id, transactionId, traceId, data, producerSub, createdAt, partitionId, partition, leaseId and consumerGroup.

wait defaults to true on the builder, so a bare pop() is a 30-second long poll. Set .wait(false) for an immediate return and .timeoutMillis(n) to change the poll window.

Two knobs matter for throughput. .batch(n) caps how many messages come back; .partitions(n) claims up to n partitions in one call, with batch acting as a global cap across all of them and one leaseId covering the lot, so renewing once extends them all. .group('name') selects a consumer group; without one the pop uses queue mode.

.autoAck(true) on a pop commits the cursor inside the pop transaction. That is at-most-once delivery: a crash after the response is a lost message.

Consume

await client
.queue('test-queue-v2-consume')
.batch(1)
.limit(1)
.each()
.consume(async msg => {
    msgToReturn = msg
})

consume(handler) starts concurrency workers, each in its own poll loop, and resolves when every worker stops. It never returns on its own unless you set limit, idleMillis or an AbortSignal.

Without .each() the handler receives the array of popped messages, even when batch is 1. That differs from Python, Go and PHP, which pass a single message in that case.

Each delivered message also gets a trace() method that posts a trace event and never throws:

await msg.trace({ traceName: ['tenant-acme'], eventType: 'step', data: { stage: 'charged' } })

What autoAck actually does

autoAck in consume() is client-side. The consumer never sends autoAck to the server; it calls POST /api/v1/ack itself after your handler returns, with status completed on return and failed on throw. Because an ack is an offset commit, a nack clamps the group’s cursor at the failed message, so everything after it in the popped batch will be redelivered. With .each() the consumer detects this and abandons the rest of the batch rather than producing duplicates and rejected acks. Without .each() your handler owns the whole batch, so it should either succeed for all of it or throw.

Attaching .onSuccess() or .onError() to a consume forces autoAck off, on the assumption that the callback acks. If it does not, nothing is acked and the lease expires.

Acknowledge manually

const messages = await client.queue('orders').group('billing').batch(10).pop()
for (const msg of messages) { await handle(msg) }
await client.ack(messages, true, { group: 'billing' })

ack(message, status, context) takes one message or an array. status is true (completed), false (failed), or one of the strings completed, failed, retry, dlq. context.group names the consumer group; context.error records a failure reason that lands on the DLQ row if this nack exhausts the retry budget.

partitionId is mandatory on every ack: the client throws (batch) or returns { success: false } (single) without it, because a transactionId alone is not unique across partitions. Pass the message objects you got from the pop and it is already there.

A rejected ack still arrives as HTTP 200 with success: false on the item, so the per-item flag is the only signal that the broker accepted it. The client checks the response length against the number of acks sent and throws if they disagree, rather than misattributing a result to the wrong message.

Per-message statuses in one batch call:

messages[0]._status = 'dlq'
messages[0]._error = 'unparseable payload'
await client.ack(messages, true, { group: 'billing' })

Transactions

await client
    .transaction()
    .queue('test-queue-v2-txn-basic-b')
    .push([{ data: { value: messages[0].data.value + 1 } }])
    .ack(messages[0])
    .commit()

transaction() bundles pushes and acks into one PostgreSQL transaction, all or nothing. The builder collects .queue(name).partition(p).push(items) and .ack(messages, status) operations and sends them on .commit(), together with the leaseIds of every acked message as requiredLeases, so the commit fails if a lease expired in the meantime. Committing with no operations throws.

This is the primitive for a pipeline stage: ack the input and push the output atomically, so a crash cannot produce a stage that consumed without emitting.

Lease renewal

await client.renew(messages)

renew() accepts a lease ID, a message, or an array, and de-duplicates the lease IDs. A multi-partition pop shares one lease across every claimed partition, so passing the whole array issues one HTTP call, not N.

Client-side buffering

await client
  .queue('events')
  .buffer({ messageCount: 500, timeMillis: 200 })
  .push([{ data: { seen: Date.now() } }])

With buffer(), push() returns { buffered: true, count: N } immediately and the messages are flushed when the buffer reaches messageCount (default 100) or timeMillis (default 1000) elapses since the first message, whichever comes first. Buffers are keyed per queue/partition.

Flush explicitly with await client.queue('events').flushBuffer() for one queue/partition or await client.flushAllBuffers() for all of them. client.getBufferStats() returns { activeBuffers, totalBufferedMessages, oldestBufferAge, flushesPerformed }.

Admin

const overview = await client.admin.getOverview()
const queues = await client.admin.listQueues({ limit: 50 })
const lagging = await client.admin.getLaggingConsumers(120)

client.admin is a lazily created singleton over the observability and management routes: resources (overview, namespaces, tasks, queues, partitions), messages, traces, status and analytics, consumer groups (list, get, lagging, delete-for-queue, seek), health, metrics and the two maintenance-mode switches. The full route list, with the access level each one requires, is in Reference.

Dead-letter queue

const { messages } = await client.queue('orders').dlq('billing').limit(50).offset(0).get()

dlq() reads GET /api/v1/dlq and swallows errors to { messages: [], total: 0 }. Each row carries queue, partition, consumerGroup, errorMessage, retryCount, data, createdAt and failedAt.

To reprocess one dead letter, admin.retryMessage(partitionId, transactionId) calls the broker’s replay route, which re-pushes the stored payload and then drops the DLQ row. The replayed message is a new message: it gets a fresh identity and lands at the tail of its partition, so ordering relative to the original is not preserved. There is no bulk redrive: to replay many rows, read them and push them yourself.

429 and 403

const client = new Queen({
  url: 'https://cell.example',
  bearerToken: token,
  retry429: { maxAttempts: 20, baseMs: 500, capMs: 30000 },
})

A 429 is retried in place against the same backend. The delay honours Retry-After (seconds) when present, otherwise baseMs * 2^attempt capped at capMs, and both get ±20 % jitter to avoid a synchronised herd. maxAttempts defaults to 10 for push and admin calls; a long-poll pop (wait: true) retries unboundedly unless you set maxAttempts, in which case it applies to both. A 429 never marks the backend unhealthy.

A 403 is terminal. The consume worker logs it with its code and throws out of the worker rather than retrying, because cluster_suspended, storage_quota_exceeded, feature_gated and forbidden never resolve on their own. Errors carry .status and, when the body has one, .code, so you can branch without matching on message text.

Logging

Console logging is off unless QUEEN_CLIENT_LOG=true (Node) or window.QUEEN_CLIENT_LOG === true (browser). Passing a logger overrides the env gate entirely, so the logger is then always active. It must implement info, warn and error; debug falls back to info.

const client = new Queen({ url, logger: pino(), structuredLogs: true })

By default the logger receives one formatted string, [operation] {json}. With structuredLogs: true the detail keys become top-level fields and the operation becomes the message, pino/bunyan style. Use it only with a logger that accepts a leading merge object.

Graceful shutdown

await client.close()

close() flushes every buffer, tears down the buffer manager, removes the signal handlers and destroys the HTTP agent. That last step matters: the client owns its own undici Agent so close() can release the keep-alive sockets. Without it, Node’s event loop stays pinned by open sockets and the process never exits on its own.

Unless you pass handleSignals: false, the constructor registers SIGINT and SIGTERM handlers that call close() and process.exit(0). A second signal exits immediately with code 1. When Queen is embedded in an application with its own lifecycle, either construct with handleSignals: false or call client.disableGracefulShutdown(); client.enableGracefulShutdown() puts them back.

Streaming

import { Stream } from 'queen-mq'

const stream = Stream.from(client.queue('events').group('agg'))

The windowed streaming SDK ships in the same package and builds on the same client. It has its own concepts (windows, watermarks, gates, state) and is documented separately in Use Queen.