JavaScript client

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

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

retry429: { maxAttempts, baseMs, capMs } overrides the HTTP 429 backoff policy, which is separate from retryAttempts; the defaults, the Retry-After contract and the jitter are in What the clients do with a 429.

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, carries nine keys: leaseTime, retryLimit, priority, delayedProcessing, windowBuffer, maxSize, retentionSeconds, completedRetentionSeconds and encryptionEnabled, with the values listed under Queue option defaults.

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 merges since 1.6.0: an option the body does not carry keeps the value the queue already has, an explicit null restores that option’s default, and top-level "mode": "replace" puts back the pre-1.6.0 behaviour of resetting everything the body omits. This client has no mode setter, so create() always merges. The mechanics and the full option list are in Queue options.
• Because config() merges over QUEUE_DEFAULTS, those nine keys are on the wire whether you named them or not, so they land on the client’s default rather than on the queue’s stored value. The twelve options outside that object, dedupWindowSeconds and retentionEnabled among them, are the ones a partial call now leaves alone.
• 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 sends traceId only when it is a valid UUID. The wire field is always payload; data is a client-side alias.

The broker’s push path stores no trace id, so it is dropped and the message pops back with traceId: null. Push inside a transaction when a trace id has to survive. See Pick your SDK.

const res = await client
  .queue('orders')
  .partition('customer-42')
  .push([{ data: { orderId: 9137, amount: 99.5 } }])

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

const first = await client
  .queue('payments')
  .partition('customer-42')
  .push([{ transactionId: 'order-9137-paid', data: { orderId: 9137, amount: 99.5 } }])

const retry = await client
  .queue('payments')
  .partition('customer-42')
  .push([{ transactionId: 'order-9137-paid', data: { orderId: 9137, amount: 99.5 } }])
// retry[0].status is 'duplicate': the second push wrote nothing
// and answers with the first message's id.

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 on the push route page. 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 } once the buffer has accepted every item. While the buffer sits at maxSize the await parks until the flusher drains below the bound, so under overload a buffered push is no longer instantaneous: the producer degrades to the flush rate instead of growing the heap. Accepted is not sent; the messages still leave with the flush.

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 messages = await client
  .queue('orders')
  .batch(10)
  .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.

popResult()

async popResult() → { messages, autopilot }. It is the same call as pop() and returns the messages plus the broker’s account of how it sized the claim: partitions, batch, and an optional waitMillis pacing hint the consume loop honours in place of its own delay between empty polls. autopilot is null when the pop did not engage autopilot, when the broker is older than 1.2, or when the answer was a bodiless 204, which has no body to carry an echo. See pop autopilot.

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('orders')
  .group('billing')
  .subscriptionMode('all')
  .limit(1)
  .each()
  .consume(async (message) => {
    console.log(message.data)
  })

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.

The buffer is bounded and lossless under errors. At maxSize waiting messages (default 4 * messageCount), further pushes await the flusher instead of growing the heap. A batch whose POST fails goes back to the front of the buffer, in order, and is retried every retryDelayMillis until it lands or the client closes; it is never dropped. A broker outage therefore shows up as parked producers and a full buffer, not as silent loss.

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).

Since 1.6.0 that route runs on a move: one transaction claims the newest dead-letter row at the address, pushes it back into the log and deletes the row, and the replayed message carries the deterministic transaction id dlq:<dead-letter row id>. Two consequences for a caller. A second call for the same address answers 404, because the row is gone, so a retry after a lost response is safe rather than a second copy. And when the destination’s dedup window already holds that transaction id the verdict is result: "duplicate", which means nothing was written and the dead-letter row was kept (dlqRowRemoved: false). The shape of the 200 body is a superset of what this method has always returned, so nothing that parses replayedAs needs changing. See Messages and dead letters.

TransactionBuilder

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

await client
  .queue('orders')
  .group('invoicing')
  .subscriptionMode('all')
  .each()
  .autoAck(false) // the acknowledgement belongs to the transaction, not to the loop
  .limit(1)
  .idleMillis(5000)
  .consume(async (message) => {
    // commit() throws when the broker rejects the bundle, so reaching the
    // line after it means the ack and the push are both durable.
    await client
      .transaction()
      .queue('invoices')
      .push([{ data: { orderId: message.data.orderId, invoiced: true } }])
      .ack(message, 'completed', { consumerGroup: 'invoicing' })
      .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.

A transaction is atomic, not exactly-once end to end; see the limit.

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

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'. Every window option object also accepts these five, which control when a window closes and what happens to whatever arrives after it did:

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. The code values are the proxy’s stable contract, listed under Behind the proxy. 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: a 204 carries no body at all.

Same surface, other languages

Parity here is a claim about the surface, not about shared code: each client is written natively in its own idiom. These sixteen rows are that surface, read out of clients/ on this tree. Six of them are full parity across all six clients; the other ten are where a client stops.

The ten divergent rows, with their conditions:

• Key/value. The seven operations are get, getMany, getPrefix, put, putIfAbsent, delete and incr. C++ wraps five of them: getMany and getPrefix have no method, and both remain reachable through POST /api/v1/kv on the shared HttpClient. getPrefix inside a transaction is refused by the broker in every client, which is its rule and not a client’s gap.
• Timers. C++ has schedule and cancel only; peek and list are reads and stay on GET /api/v1/timers/{queue}[/{timerKey}]. There is no reschedule operation anywhere, because schedule is the upsert and status reports which happened; Python, Go, Rust and PHP still spell a reschedule alias over it, JavaScript and C++ do not.
• once. JavaScript (kv.once, transaction().once) and Python (kv.once, transaction().once) fold putIfAbsent plus required into the question people actually ask, “did I win?”. The other four write the putIfAbsent themselves, which is the same wire and one line longer.
• kv on a streams operator context. No client has it, so it is not a row. Inside a stream the state primitive is state_ops, which commits with the sink and the ack in the cycle’s own transaction; a key/value write from an operator would not, and that atomicity is the thing the stream already gives you for free.
• Admin facade. C++ has no Admin class. Every management and observability route is still reachable, through the shared HttpClient returned by get_http_client().
• affinity. The C++ LoadBalancer implements round-robin and "session" only, and load_balancing_strategy = "affinity" falls through to round-robin silently. The strategy matters because it keeps one consumer’s pops on one backend, which is what keeps two clients from contending on the same partition claim.
• Streams. The Stream builder, the four window kinds, gates and the /streams/v1/* runtime are in the JavaScript, Python, Go and Rust clients. PHP and C++ have none of it.
• Lease renewal. In C++, renew_lease() sets a flag the consumer loop does not act on: the worker carries an unimplemented placeholder where the timer would go. Call client.renew(message) yourself before the lease expires.
• Per-message trace. Where it exists it is attached by the consume loop, so a message returned by a manual pop() never carries it. Rust records a trace through Admin::record_trace instead of hanging a method off the message. C++ cannot hang one off a JSON object at all, and its trace hook is an explicit no-op.
• Host override. JavaScript (hostHeader) and Rust (host_header) advertise a request authority, and a TLS SNI, independent of the address the socket dials. That is what addresses a named tenant cluster behind a shared proxy endpoint. The other four clients send the dialled address.
• pop failures. Go, Rust and PHP surface a 4xx, an exhausted 429 budget, a terminal 403 or a network fault to the caller, so an empty result means an empty queue. JavaScript, Python and C++ log the failure and return an empty result, which does not distinguish an empty queue from revoked credentials.

The js suite in the test matrix runs clients/client-js/test-v2, the broker-free unit tests including the proxy 429 contract and then the live-broker integration run, on the single, ha and tenanted topologies. The unit tests are green at 91 of 91 on this tree.