JavaScript client

One npm package, queen-mq, carries both the broker client and the streaming SDK. It is ESM-only ("type": "module"), requires Node 22 or newer, and depends on undici for its own HTTP dispatcher so close() can release keep-alive sockets deterministically. Version 1.0.0, aligned with the broker.

import { Queen, Admin, Stream, tokenBucketGate, slidingWindowGate } from 'queen-mq'

The package also exports the default objects themselves (CLIENT_DEFAULTS, QUEUE_DEFAULTS, CONSUME_DEFAULTS, POP_DEFAULTS, BUFFER_DEFAULTS), so an application can read a default rather than restate it.

Constructor

new Queen(config)

config is a URL string, an array of URL strings, or an options object. A string or array is equivalent to passing { url } / { urls }; everything else takes the default below. URLs are validated at construction, and an object with neither url nor urls throws Must provide urls or url in configuration.

Three consequences worth knowing before you tune anything:

• loadBalancingStrategy, affinityHashRing, enableFailover and healthRetryAfterMillis are inert with a single URL. The load balancer is only constructed when urls has more than one entry; a single-URL client goes straight to one HttpClient.
• handleSignals: true installs process.exit(). The handler flushes buffers and then exits the process. Set it to false when the client is embedded in an application that owns its own shutdown, and call close() yourself.
• headers: { Host } cannot work. fetch() treats host as a forbidden header name and drops it silently. The client detects it, maps it onto hostHeader, and warns on console.warn regardless of the log setting, because a wrong Host at a multi-tenant proxy does not fail loudly: it lands your traffic on the wrong cluster.
• retryAttempts and failover are mutually exclusive paths. With one URL, or with enableFailover: false, a call makes up to retryAttempts attempts against the same backend (3 total by default). With several URLs and failover on, retryAttempts is not consulted at all: the call walks the backend list, trying each at most once.

retry429

Separate from retryAttempts, which covers 5xx and network faults. This policy covers only HTTP 429 from a proxy.

A Retry-After header (seconds) always wins over the computed delay. Every delay gets ±20% jitter. A 429 is retried against the same backend and never triggers failover, because rate limiting is not a health signal.

hostHeader

A queen_proxy deployment picks the tenant cluster from the first DNS label of the Host header. When the base URL points at a shared address (an IP, a cell endpoint, a local rig), hostHeader rewrites the request authority (and TLS SNI) while the socket still dials the configured address, the same contract as curl --resolve. Accepts a bare authority only ('acme.eu1.example', 'acme.local:6711'); anything URL-shaped throws at construction.

Queen

ack(message, status, context)

message is one message object, one transaction-id string, or an array of either. status is true → completed, false → failed, or a status string passed through verbatim ('retry', 'dlq'). context accepts { group, error }.

partitionId is mandatory on every message object. Without it the call fails (Message must have partitionId property to ensure message uniqueness): for a single message as a returned { success: false, error }, for a batch as a thrown Error. A leaseId on the message is forwarded when present.

Per-message statuses inside a batch work by tagging items with _status and _error. If any item in the array carries either key, the client switches to per-item mode for the whole batch and the status argument becomes only the fallback.

renew(messageOrLeaseId)

Accepts a lease-id string, a message object, or an array of either. Lease ids are deduplicated first: with multi-partition pop every message in one batch shares a single leaseId, and one extend call renews every claimed partition, so passing the whole message array issues one HTTP call, not N.

QueueBuilder

queen.queue(name) returns a fresh builder. Every configuration method returns this; the terminal methods are push(), pop(), consume(), create(), delete(), dlq() and flushBuffer().

Addressing

Queue lifecycle

QUEUE_DEFAULTS, the object config() merges into:

Keys you add to config() are sent verbatim, so any option /configure accepts can be set even though it is not in the defaults object. Two things follow:

• /configure is a full replace. Keys absent from the body reset to the broker’s own defaults, not to whatever the queue had before. Send the complete intended configuration every time. The full option list lives in Reference.
• Calling create() without config() sends QUEUE_DEFAULTS as-is, which means leaseTime: 300, whereas a queue created implicitly by a first push gets a 60-second lease.

Producing

Each item may be { data }, { payload }, or a bare object used as the payload itself. The client mints a UUIDv7 transactionId when the item does not carry one, and forwards traceId only when it is a valid UUID. The wire field is always payload; data is a client-side alias.

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

Deduplication is keyed on transactionId and enforced in SQL before an offset is allocated, so a duplicate writes nothing:

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'

PushBuilder

Thenable: awaiting it executes the push. then() is idempotent: a second await on the same builder resolves without re-sending.

Awaiting a non-buffered push resolves to the broker’s per-item array. The four possible item statuses are described in Reference. With no onError callback registered, at least one failed item makes the await throw; with one registered, it does not throw. Awaiting a buffered push resolves to { buffered: true, count } immediately. The messages have not been sent yet.

Consuming

Both pop() and consume() read these; the defaults differ between the two.

pop()

async pop() → array of message objects, [] when there is nothing to return.

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

Route selection: a queue name with a non-Default partition uses /api/v1/pop/queue/:queue/partition/:partition; a queue name alone uses /api/v1/pop/queue/:queue; namespace or task without a queue uses /api/v1/pop. No queue, namespace or task throws.

consume(handler, options)

Returns a ConsumeBuilder, thenable like the others. options.signal accepts an AbortSignal that stops every worker at the top of its next iteration.

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

Registering either callback forces autoAck to false for the run, to prevent a double-ack when the callback acks by hand.

Worker loop behaviour, all of it observable:

• A handler throw with autoAck on nacks the message and does not rethrow, so the worker keeps going.
• In each() mode, a nack aborts the rest of the popped batch. The nack releases the lease and clamps the broker cursor at the failed message, so everything after it will be redelivered; continuing would only produce duplicates and rejected acks.
• A 429 that escapes the HTTP layer’s own retry sleeps for Retry-After (or 1 s) and continues.
• A network error sleeps 1 s and continues.
• A 403 stops the worker and rethrows. cluster_suspended cannot resolve itself, and none of the other proxy 403 codes are worth hot-looping.
• Any other error is rethrown.

message.trace(config)

Messages delivered through consume() are given a trace() method that posts to /api/v1/traces:

await msg.trace({
  traceName: ['tenant-acme', 'room-123'],
  eventType: 'info',
  data: { text: 'started processing' }
})

data is required; traceName accepts a string or an array of strings; eventType defaults to 'info'. It never throws: failures return { success: false, error }.

Buffering

A buffer flushes when it reaches messageCount messages or timeMillis elapses since its first message, whichever comes first. flushBuffer() cancels the timer and drains in messageCount-sized batches. Buffers are keyed on "<queue>/<partition>", so two builders on the same queue and partition share one buffer.

Dead-letter queue

dlq(consumerGroup = null) returns a DLQBuilder. Throws with no queue name. The builder’s partition is inherited from the queue builder unless it is Default.

The DLQ is read-only through this builder. Replay lives on the admin façade: admin.retryMessage(partitionId, transactionId).

TransactionBuilder

queen.transaction(). Pushes and acks bundled into one PostgreSQL transaction, all-or-nothing.

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

Lease ids are deduplicated before requiredLeases is sent. Item payloads follow the same data / payload / bare-object rule as push(), and so does the id: a transactionId you set on the item is sent, and one you omit is minted client side. That id is the deduplication key of the pushed message, so setting it deterministically is what makes a retried transaction safe.

Admin

queen.admin (getter) or new Admin(httpClient). Every method is async and returns the broker’s JSON body verbatim. Methods that take a params object turn it into a query string, dropping null and undefined.

Resources

Messages and traces

Status and analytics

Consumer groups

seekConsumerGroup posts options as the body verbatim: { toEnd: true } or { timestamp: '<RFC3339>' }.

System

Streaming SDK

Stream builds an immutable operator chain over a queue and runs it against the broker’s /streams/v1/* endpoints. Every combinator returns a new Stream.

Stream.from(queen.queue('orders'), options?) // → Stream

Window option objects share gracePeriod, idleFlushMs, eventTime, allowedLateness and onLate ('drop' | 'include'); the size argument differs per window: windowTumbling({ seconds }), windowSliding({ size, slide }), windowSession({ gap }), windowCron({ every }) where every is one of 'second' | 'minute' | 'hour' | 'day' | 'week'.

gate(fn) receives (value, ctx) with ctx.state (per-key, persisted only on ALLOW), ctx.streamTimeMs and ctx.partitionId. Returning false halts the batch: the runner acks the allowed prefix and does not release the source lease, so the denied message and its successors are redelivered in their original order when the lease expires.

Two ready-made gate functions ship with the package: tokenBucketGate({ capacity, refillPerSec, costFn?, allowZeroCost? }) and slidingWindowGate({ limit, windowSec, costFn? }).

run(runOptions)

async, resolves to a Runner.

The chain’s shape is fingerprinted into a config hash. Re-deploying a different chain under the same queryId is rejected at registration unless reset: true is passed. Only the operator kinds and their structural config are hashed: user closures cannot be serialised stably, so changing the body of a map does not change the hash.

Runner exposes stop() (graceful drain) and metrics(), which returns queryId, serverQueryId, and the counters cyclesTotal, flushCyclesTotal, messagesTotal, pushItemsTotal, stateOpsTotal, lateEventsTotal, errorsTotal, lastCycleAt, lastFlushAt, lastError.

Logging

With no logger configured, logging is off unless QUEEN_CLIENT_LOG=true is set in the environment (Node) or window.QUEEN_CLIENT_LOG === true (browser). Output goes to console as [timestamp] [LEVEL] [operation] {json}.

A custom logger is always active, env var or not. It must implement info, warn and error; debug falls back to info. By default it is called with one formatted string. With structuredLogs: true it is called pino-style (info(fields, message)), where fields is the detail object plus an operation key. Use that only with a logger that accepts a leading merge object.

Errors

Failures throw an Error with .status (HTTP status) and, when the body carried one, .code. The message is the body’s error string when present. A 429 also carries .retryAfterSeconds. Proxy codes: rate_limited, quota_exceeded on 429; cluster_suspended, storage_quota_exceeded, feature_gated, forbidden on 403. A timeout throws with name === 'AbortError' and a .timeout field. Network faults from undici arrive as fetch failed (ECONNREFUSED). The underlying cause is unwrapped into the message rather than left in error.cause.

A 204 response resolves to null; the broker sends no body at all on 204.

Same surface, other languages