Skip to content
bunqueue bunqueue
Get started
Get started

SDKs: TypeScript, Python, PHP, Go, Rust, Elixir

guide · sdks

Client SDKs for every runtime.

Official client SDKs for Node.js, Deno, Bun, Python, PHP, Go, Rust, Elixir and Cloudflare Workers. Each speaks the native TCP protocol and MessagePack through an idiomatic Queue and Worker API.

6 official SDKs, one queue1 formal, versioned wire protocolproduce in one language, consume in anotherretries, priorities, cron, DLQ: all server-side

The design is simple: the server owns every queue semantic, retries with backoff, priorities, scheduling, stall detection, the dead letter queue. Your applications only add jobs and process them. A Next.js API written in TypeScript can enqueue work that a Python service consumes, both against the same queue, with no shared runtime and no translation layer.

Supported platforms

Section titled “Supported platforms”
PlatformPackageDistribution
Node.js ≥ 20, Bun, Deno ≥ 2, Cloudflare Workersbunqueue-clientnpm · source
Python ≥ 3.9bunqueue-clientPyPI · source
PHP ≥ 8.1bunqueue/clientPackagist · source
Go ≥ 1.26.5github.com/egeominotti/bunqueue/sdk/gogo get, source
Rust ≥ 1.85bunqueue-clientcrates.io · source
Elixir ≥ 1.15bunqueue_clientHex upcoming, source
Bun, embedded in process, no serverbunqueueQuick Start

Every SDK is versioned independently from the server and follows semantic versioning; each changelog is linked in Resources. The wire protocol is a formal, public, versioned contract (protocolVersion: 2, negotiated via Hello), so a client written today keeps working across server upgrades, and services in different languages interoperate on the same queues out of the box.

Getting started

Section titled “Getting started”

1. Run the server

Section titled “1. Run the server”

The server is the only component that requires Bun, and only if you run it with bunx. Docker and the prebuilt binary need nothing at all.

Terminal window
bunx bunqueue start

One command, no install. Without --data-path (or BUNQUEUE_DATA_PATH) the queue is in-memory: pass it, e.g. --data-path ./data/bunq.db, to persist jobs to SQLite.

Port 6789 serves the TCP protocol used by the SDKs, port 6790 serves the HTTP API with /health and /metrics. Additional deployment options are covered in Running the Server.

2. Install the client

Section titled “2. Install the client”
Terminal window
npm install bunqueue-client

3. Produce jobs

Section titled “3. Produce jobs”
import { Queue } from 'bunqueue-client';


const queue = new Queue('emails');


await queue.add('welcome', { to: 'user@example.com' });

4. Process jobs

Section titled “4. Process jobs”
import { Worker } from 'bunqueue-client';


const worker = new Worker('emails', async (job) => {
  await sendEmail(job.data.to);
  return { sent: true };
});


worker.on('completed', (job) => console.log('done:', job.id));
worker.on('error', (err) => console.error(err)); // always attach (see Worker semantics)

Run either file with the runtime you already use:

Terminal window
node --experimental-strip-types app.ts    # Node 22 or later
bun app.ts                                # Bun
deno run -A app.ts                        # Deno 2 or later
python app.py                             # Python
php worker.php                            # PHP
go run .                                  # Go
cargo run                                 # Rust
mix run app.exs                           # Elixir

Producer and worker are usually separate services, often in different languages: a Next.js API adds jobs, a Python service processes them, against the same queue and the same protocol. Constructors default to host: 'localhost' and port: 6789, so no options are needed for a local setup.

Protocol and architecture

Section titled “Protocol and architecture”

Understanding four facts about the transport explains most SDK behavior:

  1. Framing: every message is a 4-byte big-endian length prefix followed by a standard msgpack map. Maximum frame size is 64 MB; maximum job payload is 10 MB.
  2. Pipelining: every request carries a reqId the server echoes back, so many commands are in flight on one socket concurrently. A single connection is usually all a service needs.
  3. Authentication-first: when a token is configured, Auth is guaranteed to be the first frame on every (re)connection, in every SDK: TypeScript through synchronous write ordering, Python through a connection lock (safe under free-threaded concurrency), and PHP, Go, Rust, and Elixir inside the connect sequence itself.
  4. The job name travels inside data: add('welcome', {to}) is encoded as data: { name: 'welcome', to }. Non-object payloads are wrapped as { name, payload }.

Connection options

Section titled “Connection options”
const queue = new Queue('emails', {
  host: 'queue.example.com',
  port: 6789,
  token: process.env.BUNQUEUE_TOKEN,
  tls: true,               // or { caFile } or { rejectUnauthorized: false }
  commandTimeoutMs: 10000, // default
  maxInFlight: 10_000,     // backpressure bound; 0 = unbounded (default)
  poolSize: 4,             // producer-side connection pool; default 1
});

The command timeout governs how long each in-flight command waits for a response. TypeScript and Python keep the TCP connect timeout internal; PHP, Go, and Rust expose it separately. Elixir applies its timeout to connect and command exchange.

Connection resilience

Section titled “Connection resilience”

Every SDK preserves the same recovery invariants; implementation details follow the runtime:

MechanismAvailability and behavior
Lazy reconnectAll six: a lost connection reconnects on the next call; workers re-register after every connection generation
Worker retry backoffAll six retry a failed pull loop without spinning; TypeScript, Python, PHP, Go, and Rust use bounded backoff, while Elixir uses a short fixed delay
Producer fast-fail windowTypeScript and Python throttle repeated failed connection attempts (500 ms → 5 s)
TCP keepaliveTypeScript and Python enable an approximately 15 s idle probe where the OS exposes the controls
Timeout-driven teardownAll six discard a stream whose framing state is ambiguous; TypeScript/Python tolerate a configurable consecutive-timeout threshold, PHP/Go/Rust/Elixir tear down immediately
BackpressureTypeScript optionally parks callers at maxInFlight; the synchronous clients naturally serialize/bound calls
Auth orderingAll six prevent any command racing ahead of Auth after a reconnect

Producing jobs

Section titled “Producing jobs”

Job options

Section titled “Job options”

Every option is validated server-side; identical semantics in every SDK. Naming follows each language’s idiom: TypeScript and PHP use camelCase keys (attempts, jobId, removeOnComplete), Python uses snake_case (attempts, job_id, remove_on_complete), Go takes a bunqueue.JobOptions map with the camelCase keys, Rust uses typed JobOptions fields, and Elixir accepts keyword or map options. Unknown options raise an error in every SDK instead of being silently dropped.

OptionTypeDefaultNotes
prioritynumber0Higher runs sooner; −1 000 000 … 1 000 000
delayms0Up to 1 year
attemptsnumber3Max attempts including the first; up to 1000
backoffnumber | { type, delay }1000type: 'fixed' or 'exponential'; delay up to 1 day
ttlms-Expires the job if not processed in time
timeoutms-Per-job processing timeout, up to 1 day
jobIdstring-Custom id; idempotent, re-adding an unfinished id is a no-op
deduplication{ id, ttl?, extend?, replace? }-Dedup window keyed on id
debounce{ id, ttl? }-Collapses bursts to the last job
dependsOnstring[]-Job ids that must complete first
parentId / childrenIdsstring / string[]-Flow relationships (usually set via FlowProducer)
tags / groupIdstring[] / string-Metadata; groupId also scopes group rate limits
lifobooleanfalseOrders only among LIFO jobs; mixed queues stay FIFO
removeOnComplete / removeOnFailbooleanfalseDrop the job record at the terminal state
durablebooleanfalseBypass the server write buffer, fsync before the ACK returns
repeat{ every } or { pattern, tz? } + limit?-Repeatable jobs (see Cron)
stallTimeoutms-Per-job stall detection override
stackTraceLimit / keepLogs / sizeLimitnumber-Failure stack cap, retained log lines, payload cap
await queue.add('report', data, { priority: 10, delay: 5000, attempts: 5 });
await queue.add('charge', payment, { jobId: `order-${orderId}`, durable: true });

Idempotency and bulk

Section titled “Idempotency and bulk”

jobId makes add idempotent: re-adding an id whose job is still unfinished (waiting, active, waiting-children) returns the existing job instead of creating a duplicate. This holds for addBulk too, each bulk entry’s jobId is preserved on the wire, so an idempotent batch ingest can be re-run safely after a crash:

await queue.addBulk(
  orders.map((o) => ({ name: 'ingest', data: o, opts: { jobId: `order-${o.id}` } }))
);

Producer throughput

Section titled “Producer throughput”

For high-volume producers the TypeScript SDK can fan commands across a round-robin connection pool, one line, no API change (TypeScript only):

const queue = new Queue('ingest', { poolSize: 4 });

The pool is producer-side only: workers intentionally keep a single connection, because job leases (lock tokens) and worker registration are per-connection server state. In every SDK, addBulk is the first tool for throughput: one round-trip for the whole batch.

Processing jobs

Section titled “Processing jobs”

Worker options

Section titled “Worker options”
OptionDefaultNotes
concurrency4 in TypeScript/Python/Go/Rust; 1 in Elixir/PHPJobs processed in parallel except PHP, which is sequential by design
batchSizeusually 10; Elixir defaults to concurrencyJobs fetched per PULLB, capped by free slots and the server max (1000)
pollTimeoutMsusually 5000; Elixir 1000Server-side long-poll; max 30 000
lockTtlMs30 000Job lease TTL
heartbeatIntervalS10; Go defaults to disabledWorker + per-job lock heartbeats; 0 disables (Go: negative or DisableHeartbeat: true)
ackBatchoffOpt-in ACK batching (below; TypeScript and Python)
autoruntrue where exposedTypeScript/Python can start at construction; PHP, Go, Rust, and Elixir start explicitly

Names follow each language: Python, Rust, and Elixir use snake_case; PHP uses camelCase array keys; Go uses a WorkerOptions struct (PollTimeoutMs, LockTtlMs, …).

Lease model

Section titled “Lease model”

A pulled job carries a lock token. The worker heartbeats every active job’s lock on the heartbeat interval, so a job that legitimately runs longer than the lock TTL survives. If the worker dies, the lease expires and the server requeues the job (or moves it to the DLQ once maxStalls is exceeded), at-least-once delivery, so make handlers idempotent.

PHP is the deliberate sequential exception: it can heartbeat between jobs, but it cannot interrupt a running user callback. A PHP handler that can exceed lockTtlMs must call $job->extendLock(...) from the callback or split the work into shorter jobs.

Failures

Section titled “Failures”
import { UnrecoverableError } from 'bunqueue-client';


const worker = new Worker('emails', async (job) => {
  if (!isValid(job.data)) {
    throw new UnrecoverableError('malformed payload'); // skip retries → DLQ
  }
  return await send(job.data);
});

A thrown error fails the job with its message and the leading stack lines (persisted server-side, capped by stackTraceLimit); the server applies the retry/backoff policy and eventually the dead letter queue. UnrecoverableError bypasses retries entirely.

Worker events

Section titled “Worker events”

TypeScript, Python, PHP, and Go expose worker lifecycle listeners. Rust and Elixir use their structured telemetry callback for transport/command lifecycle and normal language control flow for per-job handler outcomes.

worker.on('completed', (job, result) => log.info('done', job.id));
worker.on('failed', (job, err) => log.warn('failed', job.id, err.message));
worker.on('error', (err) => log.error(err)); // always attach

In TypeScript, always attach an error listener: per Node EventEmitter semantics an unhandled error event throws. The other SDKs swallow listener exceptions. Every worker frees each job’s concurrency slot before emitting, so a throwing listener cannot leak a slot or degrade throughput in any SDK. The error itself is yours to observe.

ACK batching (high volume)

Section titled “ACK batching (high volume)”

Opt-in: coalesce completed-job acknowledgements into ACKB round-trips. Available in TypeScript and Python; the PHP and Go workers acknowledge each job individually.

const worker = new Worker('ingest', process, {
  concurrency: 32,
  ackBatch: { enabled: true, maxSize: 50, maxDelayMs: 5 },
});

Semantics are strict: a job stays active, its lock still heartbeated, until the server confirms the batch; the batch is flushed on close(); every job settles exactly once even if an event listener throws. Defaults are off, so nothing changes unless you enable it.

Graceful shutdown

Section titled “Graceful shutdown”
await worker.close();        // stop pulling, flush batched ACKs, drain in-flight jobs
await worker.close(true);    // force: skip the in-flight drain

Query, control and operations

Section titled “Query, control and operations”

All SDKs implement the core producer, lookup, wait, queue-control, DLQ, scheduler, rate-limit, and global-concurrency operations with identical server semantics. TypeScript and PHP use camelCase, Python/Rust/Elixir use snake_case, and Go uses exported Go style (GetJobCounts, RetryDlq, …). The broader webhook/stats/metrics convenience surface remains richest in TypeScript, Python, PHP, and Go; Rust and Elixir can always use the documented protocol or HTTP operations until typed helpers are added.

AreaMethods
QuerygetJob, getJobByCustomId, getJobs + per-state helpers (getWaiting, getActive, getCompleted, getFailed, getDelayed, getPrioritized, getWaitingChildren), getJobState (Python name: get_state), getResult, getProgress, counts + counts-per-priority, getChildrenValues, job logs
WaitingwaitForJob(id, ttlMs), resolves with the result; throws on timeout (CommandTimeoutError) and on a failed job (CommandError), never silently returns undefined
Controlpause / resume / isPaused, drain, obliterate, clean, remove, discard, promoteJob(s), retryJob (failed → waiting), updateJobData, updateJobProgress, changeJobPriority, changeJobDelay, moveJobToWait/Delayed/Completed/Failed, extendJobLock
DLQgetDlq, retryDlq(id?), purgeDlq, retryJobs (BullMQ semantics: retries the whole DLQ by default, or pass count to retry only the first N entries, honored by server 2.8.29 and later; Python retry_dlq accepts count= too), DLQ configuration
SchedulersupsertJobScheduler (cron pattern or fixed interval, timezone, skipIfNoWorker, preventOverlap, per-scheduler job options), removeJobScheduler, getJobScheduler(s), addCron, every
Adminrate limiting, global concurrency, stall configuration, webhooks, getStats, getMetrics, listQueues, getWorkers

A representative slice, in every language:

const job = await queue.getJob(id);            // null when missing
const state = await queue.getJobState(id);     // 'waiting' | 'active' | ...
const result = await queue.waitForJob(id, 30_000);
const counts = await queue.getJobCounts();     // { waiting, active, completed, ... }
await queue.pause();
const dropped = await queue.drain();           // number of removed jobs
await queue.retryDlq();                        // re-queue dead-lettered jobs

Two behaviors worth knowing:

  • Not-found is null/None/nil, never an exception, getJob, getJobByCustomId and getJobScheduler map the server’s not-found response for you.
  • Progress updates require an active job; the server rejects progress updates on waiting jobs by design.

Flows

Section titled “Flows”

Pipelines and parent/child trees with automatic ordering, children complete before their parent, results are readable from the parent:

import { FlowProducer } from 'bunqueue-client';


const flow = new FlowProducer();


// Sequential pipeline
await flow.addChain([
  { name: 'extract', queueName: 'etl' },
  { name: 'transform', queueName: 'etl' },
  { name: 'load', queueName: 'etl' },
]);


// Tree: parent waits for its children
const node = await flow.add({
  name: 'assemble', queueName: 'orders',
  children: [
    { name: 'reserve-stock', queueName: 'orders' },
    { name: 'charge-card', queueName: 'orders' },
  ],
});


// Fan-in: N parallel jobs converge into one
await flow.addBulkThen(parts, { name: 'merge', queueName: 'orders' });

Fan-in (addBulkThen, N parallel jobs converging into one final job) is available in TypeScript and Python.

Creation is transactional per call: if any push fails, every job already created by that call is cancelled (rollback). getFlow(id) reconstructs a tree from a root id, it returns null for a missing root, skips children that were since removed (removeOnComplete), and is cycle-safe, so it always returns the surviving partial tree instead of throwing.

Observability

Section titled “Observability”

All six SDKs expose opt-in, dependency-free structured telemetry. Bring OpenTelemetry, Prometheus, Logger, tracing, or your own collector; SDKs stay silent by default. Consumer callback exceptions and Rust callback panics are isolated so an observer can never break transport or worker correctness.

import { Queue, consoleLogger, type TelemetryEvent } from 'bunqueue-client';


const queue = new Queue('emails', {
  logger: consoleLogger('info'),           // or any { debug, info, warn, error }
  onTelemetry: (e: TelemetryEvent) => {
    if (e.type === 'command') histogram.observe({ cmd: e.cmd }, e.durationMs);
    if (e.type === 'reconnect_scheduled') reconnects.inc();
  },
});


// Connection is an EventEmitter for imperative lifecycle hooks:
queue.connection.on('connect', (i) => log.info('link up', i));
queue.connection.on('disconnect', (i) => log.warn('link down', i));
queue.connection.on('reconnect_scheduled', (i) => log.warn('retrying', i));

The idiomatic event sets cover connection/reconnection, authentication, command latency and outcome, timeout, transport error, and close as applicable to each runtime; worker retry events are added where the client has a retry loop. Tokens, job payloads, results, private keys, and CA contents are never recorded. Scrape the server’s HTTP /metrics endpoint for authoritative queue and broker metrics.

Simple Mode

Section titled “Simple Mode”

Bunqueue bundles a Queue and a Worker into a single object, a 1:1 port of the official client’s Simple Mode. It brings routes, onion middleware, in-process retry strategies, a circuit breaker, batch accumulation, event triggers, job TTL, priority aging, cooperative cancellation, and deduplication or debounce defaults. Available in TypeScript and Python; in PHP and Go, compose Queue and Worker directly.

import { Bunqueue } from 'bunqueue-client';


const app = new Bunqueue('notifications', {
  routes: {
    'send-email': async (job) => ({ sent: true }),
    'send-sms': async (job) => ({ sent: true }),
  },
  concurrency: 10,
  retry: { maxAttempts: 5, strategy: 'jitter' },
  circuitBreaker: { threshold: 5, resetTimeout: 30_000 },
});


app.use(async (job, next) => {
  console.time(job.name);
  const result = await next();
  console.timeEnd(job.name);
  return result;
});


await app.add('send-email', { to: 'alice@example.com' });
await app.cron('daily-digest', '0 9 * * *', { to: 'all' });

Scheduler job options are honored end to end: the template’s attempts, backoff, timeout, delay, remove_on_complete / remove_on_fail, priority and deduplication are applied to every cron-spawned job, in every SDK (scheduler support included: upsertJobScheduler exists in PHP and Go too).

One difference from the Bun original: embedded: true is not available, it requires the in-process Bun runtime. Connect to a server instead; the option raises a clear error if you try.

Cloudflare Workers

Section titled “Cloudflare Workers”

The same client runs inside Workers. Enable Node.js compatibility and add jobs directly from your fetch handlers:

wrangler.toml
compatibility_flags = ["nodejs_compat"]
compatibility_date = "2025-01-01"
import { Queue } from 'bunqueue-client';


export default {
  async fetch(req: Request, env: Env): Promise<Response> {
    const queue = new Queue('signups', { host: env.BQ_HOST, port: 6789, token: env.BQ_TOKEN, tls: true });
    const job = await queue.add('welcome', await req.json());
    queue.close();
    return Response.json({ queued: job.id });
  },
};

Workers are request-scoped, so there is no long-lived worker loop. Instead: produce from fetch handlers, and consume in batches from a Cron Trigger, pull, process, acknowledge, return. Both patterns work with Simple Mode too. Two requirements apply: the server must be reachable from the internet, and TLS needs a publicly trusted certificate.

Security

Section titled “Security”

Authentication uses server-side tokens; transport security uses native TLS.

const queue = new Queue('emails', {
  host: 'queue.example.com',
  port: 6789,
  token: process.env.BUNQUEUE_TOKEN,   // server started with AUTH_TOKENS=...
  tls: true,                            // or { caFile: './ca.pem' } for a custom CA
});

Certificate verification is on by default for every TLS connection in every SDK, a wrong or missing CA rejects the connection instead of silently connecting; only an explicit opt-out (rejectUnauthorized: false in TypeScript, {"verify": False} in Python, ['verifyPeer' => false] in PHP, InsecureSkipVerify: true in Go, verify: false in Elixir) switches to encryption-only mode for development. Rust deliberately exposes no insecure TLS mode. Start the server with AUTH_TOKENS to require authentication, and with TLS_CERT_FILE plus TLS_KEY_FILE for encrypted transport. Full hardening guidance lives in the deployment guide.

Errors and guarantees

Section titled “Errors and guarantees”

Every SDK exposes typed equivalents of the same error categories, so retry logic can branch precisely (PHP uses exception classes under Bunqueue\Exception; Go uses errors.As; Rust uses the Error enum; Elixir returns {:error, exception} or raises through bang APIs):

ErrorMeaning
ConnectionLink lost or server unreachable (in-flight commands reject; the next call reconnects)
TimeoutNo response within the command timeout, including waitForJob timeout
CommandThe server answered ok: false (validation, failed wait, or a write-side not-found)
AuthenticationToken rejected during the connection handshake
Protocol / serializationInvalid map, frame, MessagePack extension, or oversized outgoing body
Unrecoverable processingRaised/returned by your processor to skip retries and fail terminally

Delivery is at-least-once: a worker crash after processing but before the ACK means the job runs again, design handlers to be idempotent (the jobId option helps on the producing side). Two numeric-precision rules:

  • JavaScript numbers are IEEE 754 doubles, exact up to 2⁵³. Pass larger 64-bit identifiers (snowflake ids) as strings; never put BigInt in job data.
  • Python, PHP, Go, Rust, and Elixir integers outside the int32 range are recursively encoded as float64 on the wire (exact up to 2⁵³, safe for millisecond timestamps); the same string rule applies to larger identifiers.

Write a client in any language

Section titled “Write a client in any language”

The wire protocol is small and fully documented: length-prefixed msgpack frames, one command per message, plain request/response. If your language is not covered yet, you can build a client against the wire protocol specification and certify it with the conformance suite: point its runner at your client and it tells you exactly what to fix. The official SDKs are built and certified the same way.

Two environments are out of scope today. Browsers have no raw-TCP API, so route through your own backend instead of connecting directly. WebAssembly is feasible but not one portable runtime: a WASM target becomes official only when it can pass the same conformance suite without weakening authentication, TLS, or framing.

Resources

Section titled “Resources”