# SDKs: TypeScript, Python, PHP, Go, Rust, Elixir Six production-grade official SDKs for Node.js, Deno, Bun, Python, PHP, Go, Rust, Elixir and Cloudflare Workers. Canonical: https://bunqueue.dev/guide/sdks/ --- import { Tabs, TabItem } from '@astrojs/starlight/components';
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 queue 1 formal, versioned wire protocol produce in one language, consume in another retries, 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 | Platform | Package | Distribution | | ----------------------------------------------- | ---------------------------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------ | | Node.js ≥ 20, Bun, Deno ≥ 2, Cloudflare Workers | `bunqueue-client` | [npm](https://www.npmjs.com/package/bunqueue-client) · [source](https://github.com/egeominotti/bunqueue/tree/main/sdk/typescript) | | Python ≥ 3.9 | `bunqueue-client` | [PyPI](https://pypi.org/project/bunqueue-client/) · [source](https://github.com/egeominotti/bunqueue/tree/main/sdk/python) | | PHP ≥ 8.1 | `bunqueue/client` | [Packagist](https://packagist.org/packages/bunqueue/client) · [source](https://github.com/egeominotti/bunqueue/tree/main/sdk/php) | | Go ≥ 1.26.5 | `github.com/egeominotti/bunqueue/sdk/go` | `go get`, [source](https://github.com/egeominotti/bunqueue/tree/main/sdk/go) | | Rust ≥ 1.85 | `bunqueue-client` | `cargo add bunqueue-client` · [API docs](https://docs.rs/bunqueue-client/latest/bunqueue_client/) · [source](https://github.com/egeominotti/bunqueue/tree/main/sdk/rust) | | Elixir ≥ 1.15 | `bunqueue_client` | Hex upcoming, [source](https://github.com/egeominotti/bunqueue/tree/main/sdk/elixir) | | Bun, embedded in process, no server | `bunqueue` | [Quick Start](/guide/quickstart/) | Every SDK is versioned independently from the server and follows semantic versioning; each changelog is linked in [Resources](#resources). The wire protocol is a formal, public, versioned contract (`protocolVersion: 3`, negotiated via `Hello` with the `separate-job-name` capability), so a client written today keeps working across compatible server upgrades, and services in different languages interoperate on the same queues out of the box. ## Getting started ### 1. Run the server The server is the only component that requires [Bun](https://bun.sh), and only if you run it with `bunx`. Docker and the prebuilt binary need nothing at all. ```bash 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. ```bash docker run -d --name bunqueue \ -p 6789:6789 -p 6790:6790 \ -v bunqueue-data:/app/data \ ghcr.io/egeominotti/bunqueue:latest ``` The named volume keeps the SQLite file across container restarts and upgrades. ```yaml # compose.yaml services: bunqueue: image: ghcr.io/egeominotti/bunqueue:latest ports: - '6789:6789' # TCP protocol (SDKs) - '6790:6790' # HTTP API (/health, /metrics) volumes: - bunqueue-data:/app/data # environment: # AUTH_TOKENS: "your-secret-token" volumes: bunqueue-data: ``` ```bash docker compose up -d ``` ```bash BUNQUEUE_STORAGE_DRIVER=postgres \ BUNQUEUE_POSTGRES_URL='postgres://bunqueue:secret@localhost:5432/bunqueue' \ bunx bunqueue start ``` PostgreSQL 15–18 are supported and tested; 18.6 is recommended. Storage is configured only on the bunqueue server. Every SDK still connects to port 6789 and does not need a PostgreSQL driver, URL, or database credentials. Multiple server instances may share that one database and namespace. ```bash # download the binary for your platform from GitHub Releases curl -LO https://github.com/egeominotti/bunqueue/releases/latest/download/bunqueue-darwin-arm64 chmod +x bunqueue-darwin-arm64 ./bunqueue-darwin-arm64 start ``` Prebuilt binaries for macOS and Linux are attached to every [release](https://github.com/egeominotti/bunqueue/releases); no runtime required. 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](/guide/server/). ### 2. Install the client ```bash npm install bunqueue-client ``` ```bash bun add bunqueue-client ``` ```bash deno add npm:bunqueue-client ``` ```bash pip install bunqueue-client ``` ```bash composer require bunqueue/client ``` ```bash go get github.com/egeominotti/bunqueue/sdk/go ``` ```bash cargo add bunqueue-client ``` ```elixir # Hex release upcoming; use sdk/elixir as a path dependency today {:bunqueue_client, path: "../bunqueue/sdk/elixir"} ``` ### 3. Produce jobs ```typescript import { Queue } from 'bunqueue-client'; const queue = new Queue('emails', { embedded: false }); await queue.add('welcome', { to: 'user@example.com' }); ``` ```python from bunqueue import Queue queue = Queue("emails") queue.add("welcome", {"to": "user@example.com"}) ``` ```php use Bunqueue\Queue; $queue = new Queue('emails'); $queue->add('welcome', ['to' => 'user@example.com']); ``` ```go queue := bunqueue.NewQueue("emails", bunqueue.Options{}) defer queue.Close() queue.Add("welcome", map[string]any{"to": "user@example.com"}, nil) ``` ```rust use bunqueue_client::{ConnectionOptions, JobOptions, Queue, Value}; let queue = Queue::new("emails", ConnectionOptions::default()); let data = Value::Map(vec![(Value::from("to"), Value::from("user@example.com"))]); queue.add("welcome", data, JobOptions::default())?; ``` ```elixir queue = Bunqueue.queue("emails") {:ok, _job} = Bunqueue.Queue.add(queue, "welcome", %{to: "user@example.com"}) ``` ### 4. Process jobs ```typescript import { Worker } from 'bunqueue-client'; const worker = new Worker('emails', async (job) => { await sendEmail(job.data.to); return { sent: true }; }, { embedded: false }); worker.on('completed', (job) => console.log('done:', job.id)); worker.on('error', (err) => console.error(err)); // always attach (see Worker semantics) ``` ```python from bunqueue import Worker def process(job): send_email(job.data["to"]) return {"sent": True} Worker("emails", process, concurrency=10).run() ``` ```php use Bunqueue\Worker; $worker = new Worker('emails', function (Bunqueue\Job $job) { sendEmail($job->data()['to']); return ['sent' => true]; }); $worker->installSignalHandlers(); $worker->run(); // blocking loop; or $worker->runOnce() from a cron tick ``` ```go worker := bunqueue.NewWorker("emails", func(job *bunqueue.Job) (any, error) { return sendEmail(job.Data()["to"].(string)) }, bunqueue.WorkerOptions{Concurrency: 8}) worker.Run() ``` ```rust use bunqueue_client::{ProcessError, Value, Worker, WorkerOptions}; let worker = Worker::new( "emails", |job| { deliver(job.data()) .map(|_| Value::from(true)) .map_err(|error| ProcessError::retryable(error.to_string())) }, WorkerOptions::default(), ); worker.run()?; ``` ```elixir worker = Bunqueue.Worker.new("emails", fn job -> send_email(job.data) {:ok, %{sent: true}} end, concurrency: 8) Bunqueue.Worker.run(worker) ``` Run either file with the runtime you already use: ```bash 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 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. **Job names and payloads stay separate**: protocol v3 encodes `add('welcome', {to})` as `name: 'welcome', data: {to}`. Scalars, arrays and `null` remain unchanged in `data`, and a user-owned `data.name` is preserved. Workers receive the separate `job.name` and original `job.data` values. ### Connection options ```typescript const queue = new Queue('emails', { embedded: false, connection: { host: 'queue.example.com', port: 6789, token: process.env.BUNQUEUE_TOKEN, tls: true, // or { caFile } or { rejectUnauthorized: false } commandTimeout: 30_000, // default maxInFlight: 100, // maximum in-flight commands per connection (default) poolSize: 4, // connection pool size (default) }, }); ``` ```python queue = Queue( "emails", host="queue.example.com", port=6789, token=os.environ["BUNQUEUE_TOKEN"], tls=True, # or {"ca_file": "./ca.pem"} or an ssl.SSLContext command_timeout=10.0, # default ) ``` ```php $queue = new Queue('emails', [ 'host' => 'queue.example.com', 'port' => 6789, 'token' => getenv('BUNQUEUE_TOKEN'), 'tls' => true, // or ['caFile' => './ca.pem'] or ['verifyPeer' => false] 'connectTimeout' => 10.0, // seconds, default 'commandTimeout' => 30.0, // seconds, default ]); ``` ```go queue := bunqueue.NewQueue("emails", bunqueue.Options{ Host: "queue.example.com", Port: 6789, Token: os.Getenv("BUNQUEUE_TOKEN"), TLS: &bunqueue.TLSOptions{CAFile: "./ca.pem"}, // or &TLSOptions{} for system CAs ConnectTimeout: 10 * time.Second, // default CommandTimeout: 30 * time.Second, // default }) ``` ```rust use std::{path::PathBuf, time::Duration}; use bunqueue_client::{ConnectionOptions, Queue, TlsOptions}; let queue = Queue::new("emails", ConnectionOptions { host: "queue.example.com".into(), port: 6789, token: std::env::var("BUNQUEUE_TOKEN").ok(), tls: Some(TlsOptions { ca_file: Some(PathBuf::from("./ca.pem")) }), connect_timeout: Duration::from_secs(10), command_timeout: Duration::from_secs(30), ..Default::default() }); ``` ```elixir queue = Bunqueue.queue("emails", host: "queue.example.com", port: 6789, token: System.fetch_env!("BUNQUEUE_TOKEN"), tls: true, ca_file: "./ca.pem", timeout: 30_000 ) ``` 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 Every SDK preserves the same recovery invariants; implementation details follow the runtime: | Mechanism | Availability and behavior | | ------------------------- | ---------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | | Lazy reconnect | All six: a lost connection reconnects on the next call; workers re-register after every connection generation | | Worker retry backoff | All 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 window | Legacy TypeScript and Python throttle repeated failed connection attempts (500 ms → 5 s); canonical TypeScript uses the shared client's reconnect policy | | TCP keepalive | TypeScript and Python enable an approximately 15 s idle probe where the OS exposes the controls | | Timeout-driven teardown | All six discard a stream whose framing state is ambiguous; TypeScript/Python tolerate a configurable consecutive-timeout threshold, PHP/Go/Rust/Elixir tear down immediately | | Backpressure | TypeScript optionally parks callers at `maxInFlight`; the synchronous clients naturally serialize/bound calls | | Auth ordering | All six prevent any command racing ahead of `Auth` after a reconnect | ## Producing jobs ### Job options Every transmitted option is validated server-side. The typed mappers preserve the common option set, with one audited exception: Elixir's `deduplication: %{id: ...}` maps the nested settings but does not derive the owning `uniqueKey`, so supply `uniqueKey` explicitly for now. 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. Prefer typed builders: runtime handling of unknown keys differs by language. | Option | Type | Default | Notes | | -------------------------------------------- | -------------------------------------------- | ------- | ------------------------------------------------------------------------------------------------------- | | `priority` | number | 0 | Higher runs sooner; −1 000 000 … 1 000 000 | | `delay` | ms | 0 | Up to 1 year | | `attempts` | number | 3 | Max attempts including the first; up to 1000 | | `backoff` | number \| `{ type, delay }` | 1000 | `type`: `'fixed'` or `'exponential'`; delay up to 1 day | | `ttl` | ms | - | Expires the job if not processed in time | | `timeout` | ms | - | Per-job processing timeout, up to 1 day | | `jobId` | string | - | 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? }` | - | Persists compatibility metadata; use `deduplication` with `replace: true` for last-write-wins execution | | `dependsOn` | string[] | - | Job ids that must complete first | | `parentId` / `childrenIds` | string / string[] | - | Flow relationships (usually set via `FlowProducer`) | | `tags` / `groupId` | string[] / string | - | Metadata; `groupId` also scopes group rate limits | | `lifo` | boolean | false | At equal priority, LIFO jobs form a newest-first partition ahead of FIFO jobs | | `removeOnComplete` / `removeOnFail` | boolean | false | Drop the job record at the terminal state | | `durable` | boolean | false | SQLite: bypass its write buffer before ACK; PostgreSQL admission is already transactional | | `repeat` | `{ every }` or `{ pattern, tz? }` + `limit?` | - | Repeatable jobs (see [Cron](/guide/cron/)) | | `stallTimeout` | ms | - | Per-job stall detection override | | `stackTraceLimit` / `keepLogs` / `sizeLimit` | number | - | Failure stack cap, retained log lines, payload cap | ```typescript await queue.add('report', data, { priority: 10, delay: 5000, attempts: 5 }); await queue.add('charge', payment, { jobId: `order-${orderId}`, durable: true }); ``` ```python queue.add("report", data, priority=10, delay=5000, attempts=5) queue.add("charge", payment, job_id=f"order-{order_id}", durable=True) ``` ```php $queue->add('report', $data, ['priority' => 10, 'delay' => 5000, 'attempts' => 5]); $queue->add('charge', $payment, ['jobId' => "order-{$orderId}", 'durable' => true]); ``` ```go queue.Add("report", data, bunqueue.JobOptions{"priority": 10, "delay": 5000, "attempts": 5}) queue.Add("charge", payment, bunqueue.JobOptions{"jobId": "order-" + orderID, "durable": true}) ``` Not shown here. The Rust equivalent is in the [Rust SDK reference](https://github.com/egeominotti/bunqueue/tree/main/sdk/rust). Not shown here. The Elixir equivalent is in the [Elixir SDK reference](https://github.com/egeominotti/bunqueue/tree/main/sdk/elixir). ### 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: ```typescript await queue.addBulk( orders.map((o) => ({ name: 'ingest', data: o, opts: { jobId: `order-${o.id}` } })) ); ``` ```python queue.add_bulk([ {"name": "ingest", "data": o, "job_id": f"order-{o['id']}"} for o in orders ]) ``` ```php $queue->addBulk(array_map( fn ($o) => ['name' => 'ingest', 'data' => $o, 'jobId' => "order-{$o['id']}"], $orders )); ``` ```go entries := make([]bunqueue.BulkEntry, 0, len(orders)) for _, o := range orders { entries = append(entries, bunqueue.BulkEntry{ Name: "ingest", Data: o, Opts: bunqueue.JobOptions{"jobId": "order-" + o.ID}, }) } ids, err := queue.AddBulk(entries) ``` Not shown here. The Rust equivalent is in the [Rust SDK reference](https://github.com/egeominotti/bunqueue/tree/main/sdk/rust). Not shown here. The Elixir equivalent is in the [Elixir SDK reference](https://github.com/egeominotti/bunqueue/tree/main/sdk/elixir). ### Producer throughput For high-volume producers the TypeScript SDK can fan commands across a connection pool using the canonical nested connection options: ```typescript const queue = new Queue('ingest', { embedded: false, connection: { poolSize: 4 }, }); ``` Both canonical Queue and Worker use the shared transport implementation. Workers preserve per-job lease tokens and re-register after reconnect. In every SDK, `addBulk` is the first tool for producer throughput: one round-trip for the whole batch. The canonical TypeScript Queue also batches concurrent `add()` calls by default; configure it with `autoBatch`. ## Processing jobs ### Worker options The canonical TypeScript Worker uses the same options as `bunqueue/client`: `concurrency` defaults to 1, `pollTimeout` and `heartbeatInterval` use milliseconds, `lockDuration` controls the lease, and `limiter`, `group`, and processor `batch` are supported. See the [shared Worker reference](/guide/worker/options/). The table below describes the other SDKs and the historical TypeScript `bunqueue-client/legacy` API; its `pollTimeoutMs`, `lockTtlMs`, `heartbeatIntervalS`, and opt-in `ackBatch` names do not apply to the default export. | Option | Default | Notes | | -------------------- | ----------------------------------------------- | --------------------------------------------------------------------------------------- | | `concurrency` | 4 in legacy TypeScript/Python/Go/Rust; 1 in Elixir/PHP | Jobs processed in parallel except PHP, which is sequential by design | | `batchSize` | usually 10; Elixir defaults to concurrency | Jobs fetched per `PULLB`, capped by free slots and the server max (1000) | | `pollTimeoutMs` | usually 5000; Elixir 1000 | Server-side long-poll; max 30 000 | | `lockTtlMs` | 30 000 | Job lease TTL | | `heartbeatIntervalS` | 10; Go defaults to disabled | Worker + per-job lock heartbeats; 0 disables (Go: negative or `DisableHeartbeat: true`) | | `ackBatch` | off | Opt-in ACK batching (below; legacy TypeScript and Python) | | `autorun` | true where exposed | Legacy TypeScript/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 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 ```typescript 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); }, { embedded: false }); ``` ```python from bunqueue import UnrecoverableError, Worker def process(job): if not is_valid(job.data): raise UnrecoverableError("malformed payload") # skip retries -> DLQ return send(job.data) worker = Worker("emails", process) ``` ```php use Bunqueue\UnrecoverableError; use Bunqueue\Worker; $worker = new Worker('emails', function (Bunqueue\Job $job) { if (!isValid($job->data())) { throw new UnrecoverableError('malformed payload'); // skip retries -> DLQ } return send($job->data()); }); ``` ```go worker := bunqueue.NewWorker("emails", func(job *bunqueue.Job) (any, error) { if !isValid(job.Data()) { return nil, bunqueue.NewUnrecoverableError("malformed payload") // skip retries -> DLQ } return send(job.Data()) }, bunqueue.WorkerOptions{}) ``` Panics are recovered, failed with their real stack, and never kill the worker. Not shown here. The Rust equivalent is in the [Rust SDK reference](https://github.com/egeominotti/bunqueue/tree/main/sdk/rust). Not shown here. The Elixir equivalent is in the [Elixir SDK reference](https://github.com/egeominotti/bunqueue/tree/main/sdk/elixir). 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](/guide/dlq/). `UnrecoverableError` bypasses retries entirely. ### 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. ```typescript 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 ``` ```python worker.on("completed", lambda job, result: log.info("done %s", job.id)) worker.on("failed", lambda job, err: log.warning("failed %s: %s", job.id, err)) worker.on("error", lambda err: log.error(err)) ``` ```php $worker->on('completed', fn ($job, $result) => $log->info("done {$job->id()}")); $worker->on('failed', fn ($job, $err) => $log->warning("failed {$job->id()}")); $worker->on('error', fn ($err) => $log->error($err->getMessage())); ``` ```go worker.On("completed", func(args ...any) { job := args[0].(*bunqueue.Job) log.Printf("done %s", job.ID()) }) worker.On("error", func(args ...any) { log.Println(args[0]) }) ``` Rust has no worker lifecycle listeners. Record per-job outcomes where the processor returns them, and pass the connection `telemetry` callback for transport and command lifecycle events. Elixir has no worker lifecycle listeners. Record per-job outcomes where the processor returns them, and pass the connection `:event_handler` callback for transport and command lifecycle events. **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. Terminal outcomes are broker-authoritative in every official SDK. If a job timeout or retired cron lease wins before a processor returns, the successful `already-finalized` ACK/FAIL response settles that handler attempt without emitting a contradictory `completed`/`failed` event or incrementing a terminal counter. Rust does not synthesize terminal events or counters; Elixir applies the same rule to its Worker counters. Malformed outcome evidence is surfaced as a protocol error instead of being treated as completion. ### ACK batching (high volume) Opt-in: coalesce completed-job acknowledgements into `ACKB` round-trips. Available as an explicit option in legacy TypeScript and Python; the PHP, Go, Rust and Elixir workers acknowledge each job individually. The canonical TypeScript Worker manages acknowledgement batching internally and does not expose the legacy `ackBatch` option. The opt-in `ackBatch` option belongs to the historical API. Import it explicitly: ```typescript import { Worker } from 'bunqueue-client/legacy'; const worker = new Worker('ingest', process, { concurrency: 32, ackBatch: { enabled: true, maxSize: 50, maxDelayMs: 5 }, }); ``` ```python worker = Worker( "ingest", process, concurrency=32, ack_batch={"max_size": 50, "max_delay_ms": 5}, ) ``` ACK batching is a TypeScript and Python feature. The PHP worker acknowledges each job individually; there is no batching option to configure. ACK batching is a TypeScript and Python feature. The Go worker acknowledges each job individually; there is no batching option to configure. ACK batching is a TypeScript and Python feature. The Rust worker acknowledges each job individually; there is no batching option to configure. ACK batching is a TypeScript and Python feature. The Elixir worker acknowledges each job individually; there is no batching option to configure. 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. When the broker ignores a retired generation, TypeScript and Python require exact positional `ignoredIndices`; they never infer a position from `ignoredIds`, because one batch can contain two lease generations with the same job ID. ### Graceful shutdown ```typescript await worker.close(); // stop pulling, flush batched ACKs, drain in-flight jobs await worker.close(true); // force: skip the in-flight drain ``` ```python worker.close() # stop pulling, wait for in-flight jobs to drain worker.close(timeout=5) # bound the wait; the drain continues in the background ``` Python has no force flag: `close(timeout=...)` bounds how long the call waits, and `close(timeout=0)` detaches immediately while in-flight jobs finish in the background. ```php $worker->installSignalHandlers(); // SIGTERM / SIGINT -> graceful stop $worker->stop(); // finish the in-flight job, then return from run() $worker->close(); // unregister and close the connection ``` The PHP worker is sequential, so stopping waits for at most one in-flight job; the unprocessed rest of the batch is re-leased by the server. ```go worker.Stop() // stop pulling; in-flight jobs finish worker.Close() // unregister and close the connection ``` Not shown here. The Rust equivalent is in the [Rust SDK reference](https://github.com/egeominotti/bunqueue/tree/main/sdk/rust). Not shown here. The Elixir equivalent is in the [Elixir SDK reference](https://github.com/egeominotti/bunqueue/tree/main/sdk/elixir). ## Query, control and operations `bunqueue-client` builds its default exports from the same source as `bunqueue/client`: the TypeScript APIs, options, job objects, and TCP behavior are identical. Build-time export/signature checks and real-broker tests across Bun, Node.js, Deno, and Cloudflare Workers prevent a separately maintained API from drifting. Embedded storage requires Bun, and a local `SandboxedWorker` thread pool requires a runtime with worker threads. The other SDKs implement their language-specific producer/worker baseline, queries, control, basic DLQ operations, schedulers, flows, auth, and TLS. Their coverage differs from the shared TypeScript surface: | Area | TypeScript | Python | PHP | Go | Rust | Elixir | | --------------------------------------- | ----------------- | ----------------- | ------------------------ | ----------------- | ----------------- | -------------- | | Basic Queue + Worker | Full | Full | Full (sequential worker) | Full | Full | Full | | Full 32-operation Job API | Full | 13 | 4 | 4 | 3 | 2 | | Dependency/waiting-children operations | Full | Partial | Partial | Partial | Missing | Missing | | Limit mutation / readback | Full / full | Partial / missing | Partial / missing | Partial / missing | Partial / missing | Full / missing | | Rich DLQ + selector-aware bulk retry | Full | Partial | Missing | Missing | Missing | Missing | | Flow bulk + fan-in + readback | Full | Full | Read only | Read only | Missing | Missing | | Queue-scoped workers/schedulers | Full | Workers global | Both global | Both global | Missing/partial | Missing/global | | QueueGroup and forwarding | Full | Missing | Missing | Missing | Missing | Missing | | Simple Mode | Full | Full | Missing | Missing | Missing | Missing | Canonical TypeScript supports deduplication-key lookup/removal, rate-limit windows and introspection, `count`/`timestamp` bulk-retry selectors, and exhaustive `end: -1` pagination. Use the `Async` Queue variants for authoritative TCP reads and mutations. The internal [SDK parity audit](https://github.com/egeominotti/bunqueue/blob/main/docs/features/polyglot-sdks.md) tracks remaining differences in the other language SDKs and distinguishes API coverage from runtime capabilities. The historical TypeScript API remains available through the explicit `bunqueue-client/legacy` export; its constructor and result shapes are separate from the canonical default API. TypeScript and PHP use camelCase, Python/Rust/Elixir use snake_case, and Go uses exported Go style (`GetJobCounts`, `RetryDlq`, ...). A representative shared baseline: ```typescript const job = await queue.getJob(id); // null when missing const state = await queue.getJobState(id); const result = await job?.waitUntilFinished(null, 30_000); const counts = await queue.getJobCountsAsync(); await queue.pauseAsync(); const dropped = await queue.drainAsync(); const retried = await queue.retryDlqAsync(); ``` ```python job = queue.get_job(id) # None when missing state = queue.get_state(id) # "waiting" | "active" | ... result = queue.wait_for_job(id, timeout_ms=30000) counts = queue.get_job_counts() # {"waiting": ..., "active": ..., ...} queue.pause() dropped = queue.drain() # number of removed jobs queue.retry_dlq() # re-queue dead-lettered jobs ``` ```php $job = $queue->getJob($id); // null when missing $state = $queue->getState($id); // 'waiting' | 'active' | ... $result = $queue->waitForJob($id, 30000); $counts = $queue->getJobCounts(); // ['waiting' => ..., 'active' => ..., ...] $queue->pause(); $dropped = $queue->drain(); // number of removed jobs $queue->retryDlq(); // re-queue dead-lettered jobs ``` ```go job, _ := queue.GetJob(id) // nil when missing state, _ := queue.GetState(id) // "waiting" | "active" | ... result, _ := queue.WaitForJob(id, 30000) counts, _ := queue.GetJobCounts() // map[string]int{"waiting": ..., ...} _ = queue.Pause() dropped, _ := queue.Drain() // number of removed jobs _, _ = queue.RetryDlq("", 0) // re-queue dead-lettered jobs ``` ```rust let job = queue.get_job(&id)?; // None when missing let state = queue.get_state(&id)?; let result = queue.wait_for_job(&id, 30_000)?; let counts = queue.get_job_counts()?; queue.pause()?; let dropped = queue.drain()?; queue.retry_dlq(None, None)?; ``` ```elixir {:ok, job} = Bunqueue.Queue.get_job(queue, id) # nil when missing {:ok, state} = Bunqueue.Queue.get_state(queue, id) {:ok, result} = Bunqueue.Queue.wait_for_job(queue, id, 30_000) {:ok, counts} = Bunqueue.Queue.get_job_counts(queue) :ok = Bunqueue.Queue.pause(queue) {:ok, dropped} = Bunqueue.Queue.drain(queue) {:ok, _count} = Bunqueue.Queue.retry_dlq(queue) ``` 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 Pipelines and parent/child trees with automatic ordering, children complete before their parent, results are readable from the parent: ```typescript import { FlowProducer } from 'bunqueue-client'; const flow = new FlowProducer({ embedded: false }); // Sequential pipeline await flow.addChain([ { name: 'extract', queueName: 'etl', data: {} }, { name: 'transform', queueName: 'etl', data: {} }, { name: 'load', queueName: 'etl', data: {} }, ]); // Tree: parent waits for its children const node = await flow.add({ name: 'assemble', queueName: 'orders', data: {}, children: [ { name: 'reserve-stock', queueName: 'orders', data: {} }, { name: 'charge-card', queueName: 'orders', data: {} }, ], }); // Fan-in: N parallel jobs converge into one await flow.addBulkThen(parts, { name: 'merge', queueName: 'orders', data: {} }); ``` ```python from bunqueue import FlowProducer flow = FlowProducer() # Sequential pipeline flow.add_chain([ {"name": "extract", "queueName": "etl"}, {"name": "transform", "queueName": "etl"}, {"name": "load", "queueName": "etl"}, ]) # Tree: parent waits for its children node = 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 flow.add_bulk_then(parts, {"name": "merge", "queueName": "orders"}) ``` ```php use Bunqueue\FlowProducer; $flow = new FlowProducer(); // Sequential pipeline $flow->addChain([ ['name' => 'extract', 'queueName' => 'etl'], ['name' => 'transform', 'queueName' => 'etl'], ['name' => 'load', 'queueName' => 'etl'], ]); // Tree: parent waits for its children $node = $flow->add([ 'name' => 'assemble', 'queueName' => 'orders', 'children' => [ ['name' => 'reserve-stock', 'queueName' => 'orders'], ['name' => 'charge-card', 'queueName' => 'orders'], ], ]); ``` ```go flow := bunqueue.NewFlowProducer(bunqueue.Options{}) defer flow.Close() // Sequential pipeline ids, err := flow.AddChain([]bunqueue.ChainStep{ {Name: "extract", QueueName: "etl"}, {Name: "transform", QueueName: "etl"}, {Name: "load", QueueName: "etl"}, }) // Tree: parent waits for its children node, err := flow.Add(bunqueue.FlowJob{ Name: "assemble", QueueName: "orders", Children: []bunqueue.FlowJob{ {Name: "reserve-stock", QueueName: "orders"}, {Name: "charge-card", QueueName: "orders"}, }, }) ``` ```rust use bunqueue_client::{ ChainStep, ConnectionOptions, FlowProducer, JobOptions, Value, }; let flow = FlowProducer::new(ConnectionOptions::default()); let step = |name: &str| ChainStep { name: name.into(), queue_name: "etl".into(), data: Value::Nil, options: JobOptions::default(), }; let ids = flow.add_chain(vec![ step("extract"), step("transform"), step("load"), ])?; ``` ```elixir flow = Bunqueue.FlowProducer.new() {:ok, ids} = Bunqueue.FlowProducer.add_chain(flow, [ %{name: "extract", queue: "etl"}, %{name: "transform", queue: "etl"}, %{name: "load", queue: "etl"} ]) ``` Fan-in (`addBulkThen`, N parallel jobs converging into one final job) is available in TypeScript and Python. Every current external SDK plans the complete graph before I/O and sends one broker-side `PUSHF` command, matching the Bun package's all-or-nothing creation and visibility guarantee. Previously published versions that compose `PUSH`/`UpdateParent` remain server-compatible, but that historical multi-request sequence is not atomic. Where exposed (Python, PHP, and Go), external-SDK `getFlow(id)` returns `null` for a missing root and skips children removed since creation, yielding the surviving partial tree. Rust and Elixir currently create atomic trees/chains but expose no typed flow reader. Both TypeScript packages fails on a missing descendant or malformed/cross-linked topology so corruption cannot masquerade as a valid partial graph. See the [Flow guide](/guide/flow/) for the exact per-client contract. ## 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. These telemetry hooks belong to the historical SDK API, available from the explicit `/legacy` export. The canonical client exposes the Bun client's Worker and QueueEvents events, connection health, and broker metrics. ```typescript import { Queue, consoleLogger, type TelemetryEvent } from 'bunqueue-client/legacy'; 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)); ``` ```python queue = Queue("emails", on_telemetry=lambda event: metrics.observe(event)) ``` ```php $queue = new Queue('emails', [ 'onEvent' => fn (array $event) => $metrics->observe($event), ]); ``` ```go queue := bunqueue.NewQueue("emails", bunqueue.Options{ OnEvent: func(event bunqueue.TelemetryEvent) { metrics.Observe(event) }, }) ``` ```rust use std::sync::Arc; use bunqueue_client::{ConnectionOptions, Queue, TelemetryCallback}; let telemetry: TelemetryCallback = Arc::new(|event| tracing::debug!(?event)); let queue = Queue::new("emails", ConnectionOptions { telemetry: Some(telemetry), ..Default::default() }); ``` ```elixir queue = Bunqueue.queue("emails", event_handler: fn event -> Logger.info("bunqueue", bunqueue: event) end ) ``` 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 Prometheus-text `/prometheus` endpoint for authoritative queue and broker metrics; `/metrics` is the JSON operational snapshot. ## Simple Mode `Bunqueue` bundles a Queue and a Worker into a single object, a 1:1 port of the official client's [Simple Mode](/guide/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. ```typescript import { Bunqueue } from 'bunqueue-client'; const app = new Bunqueue('notifications', { embedded: false, 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' }); ``` ```python from bunqueue import Bunqueue app = Bunqueue( "notifications", routes={ "send-email": lambda job: {"sent": True}, "send-sms": lambda job: {"sent": True}, }, concurrency=10, retry={"max_attempts": 5, "strategy": "jitter"}, circuit_breaker={"threshold": 5, "reset_timeout": 30000}, ) def timing(job, next_fn): result = next_fn() print(f"{job.name} done") return result app.use(timing) app.add("send-email", {"to": "alice@example.com"}) app.cron("daily-digest", "0 9 * * *", {"to": "all"}) ``` Simple Mode ships in the TypeScript and Python SDKs only. In PHP, compose `Queue` and `Worker` directly and route on the job name inside the processor. Simple Mode ships in the TypeScript and Python SDKs only. In Go, compose `Queue` and `Worker` directly and route on the job name inside the processor. Simple Mode ships in the TypeScript and Python SDKs only. In Rust, compose `Queue` and `Worker` directly and route on the job name inside the processor. Simple Mode ships in the TypeScript and Python SDKs only. In Elixir, compose `Queue` and `Worker` directly and route on the job name inside the processor. Scheduler job options are honored end to end for each SDK's supported fields. TypeScript, Python, PHP, and Go expose the broadest scheduler template and repeat flags. Rust omits list plus some flags, while Elixir's scheduler list is server-wide and its deduplication shorthand has the `uniqueKey` caveat above. `embedded: true` requires the Bun runtime, including when importing `bunqueue-client`. On Node.js and Deno, set `embedded: false` and connect to the broker. The client reports a clear error if embedded mode is requested without Bun. ## Cloudflare Workers The same client runs inside Workers. Enable Node.js compatibility and add jobs directly from your fetch handlers: ```toml # wrangler.toml compatibility_flags = ["nodejs_compat"] compatibility_date = "2025-01-01" ``` ```typescript import { Queue } from 'bunqueue-client'; export default { async fetch(req: Request, env: Env): Promise { const queue = new Queue('signups', { embedded: false, connection: { host: env.BQ_HOST, port: 6789, token: env.BQ_TOKEN, tls: true, }, }); try { const job = await queue.add('welcome', await req.json()); return Response.json({ queued: job.id }); } finally { await queue.disconnect(); } }, }; ``` 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](https://developers.cloudflare.com/workers/configuration/cron-triggers/), 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 Authentication uses server-side tokens; transport security uses native TLS. ```typescript const queue = new Queue('emails', { embedded: false, connection: { 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 }, }); ``` ```python queue = Queue( "emails", host="queue.example.com", port=6789, token=os.environ["BUNQUEUE_TOKEN"], tls={"ca_file": "./ca.pem"}, # or True for system CAs, or an ssl.SSLContext ) ``` ```php $queue = new Queue('emails', [ 'host' => 'queue.example.com', 'port' => 6789, 'token' => getenv('BUNQUEUE_TOKEN'), // server started with AUTH_TOKENS=... 'tls' => ['caFile' => './ca.pem'], // or true for system CAs ]); ``` ```go queue := bunqueue.NewQueue("emails", bunqueue.Options{ Host: "queue.example.com", Port: 6789, Token: os.Getenv("BUNQUEUE_TOKEN"), // server started with AUTH_TOKENS=... TLS: &bunqueue.TLSOptions{CAFile: "./ca.pem"}, // or &TLSOptions{} for system CAs }) ``` ```rust use std::path::PathBuf; use bunqueue_client::{ConnectionOptions, Queue, TlsOptions}; let queue = Queue::new("emails", ConnectionOptions { host: "queue.example.com".into(), token: std::env::var("BUNQUEUE_TOKEN").ok(), tls: Some(TlsOptions { ca_file: Some(PathBuf::from("./ca.pem")) }), ..Default::default() }); ``` ```elixir queue = Bunqueue.queue("emails", host: "queue.example.com", token: System.fetch_env!("BUNQUEUE_TOKEN"), tls: true, ca_file: "./ca.pem" ) ``` 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](/guide/deployment/). ## 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): | Error | Meaning | | ------------------------ | ------------------------------------------------------------------------------------- | | Connection | Link lost or server unreachable (in-flight commands reject; the next call reconnects) | | Timeout | No response within the command timeout, including `waitForJob` timeout | | Command | The server answered `ok: false` (validation, failed wait, or a write-side not-found) | | Authentication | Token rejected during the connection handshake | | Protocol / serialization | Invalid map, frame, MessagePack extension, or oversized outgoing body | | Unrecoverable processing | Raised/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 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](https://github.com/egeominotti/bunqueue/blob/main/docs/protocol.md) and certify it with the [conformance suite](https://github.com/egeominotti/bunqueue/tree/main/sdk/conformance): point its runner at your client and it tells you exactly what to fix. The official SDKs are built and certified the same way. CI and the isolated release gate execute all 18 conformance checks for every official SDK against both SQLite and PostgreSQL 18.6. The matrix therefore proves the real producer, worker, retry/DLQ, scheduler, wait-for-result, pause/resume, bulk, Unicode, authentication, and atomic FlowProducer paths through each language driver; it does not infer PostgreSQL compatibility from the TypeScript client alone. 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 - [`bunqueue-client` on npm](https://www.npmjs.com/package/bunqueue-client), the full step by step README, from server start to production - Changelogs: [TypeScript](https://github.com/egeominotti/bunqueue/blob/main/sdk/typescript/CHANGELOG.md) · [Python](https://github.com/egeominotti/bunqueue/blob/main/sdk/python/CHANGELOG.md) · [PHP](https://github.com/egeominotti/bunqueue/blob/main/sdk/php/CHANGELOG.md) · [Go](https://github.com/egeominotti/bunqueue/blob/main/sdk/go/CHANGELOG.md) · [Rust](https://github.com/egeominotti/bunqueue/blob/main/sdk/rust/CHANGELOG.md) · [Elixir](https://github.com/egeominotti/bunqueue/blob/main/sdk/elixir/CHANGELOG.md) - [Queue API](/guide/queue/), every job option explained - [Worker](/guide/worker/), concurrency, events, graceful shutdown - [Simple Mode](/guide/simple-mode/), everything in one object - [Flows](/guide/flow/), pipelines, parent and child jobs, fan in - [Cron Jobs](/guide/cron/), schedules and repeatable jobs - [Dead Letter Queue](/guide/dlq/), what happens when jobs fail - [Deployment guide](/guide/deployment/), Docker, TLS, authentication, monitoring - [Wire protocol specification](https://github.com/egeominotti/bunqueue/blob/main/docs/protocol.md), the normative contract every SDK implements - [Conformance suite](https://github.com/egeominotti/bunqueue/tree/main/sdk/conformance), certify a client in any language - SDK sources: [`sdk/typescript`](https://github.com/egeominotti/bunqueue/tree/main/sdk/typescript) · [`sdk/python`](https://github.com/egeominotti/bunqueue/tree/main/sdk/python) · [`sdk/php`](https://github.com/egeominotti/bunqueue/tree/main/sdk/php) · [`sdk/go`](https://github.com/egeominotti/bunqueue/tree/main/sdk/go) · [`sdk/rust`](https://github.com/egeominotti/bunqueue/tree/main/sdk/rust) · [`sdk/elixir`](https://github.com/egeominotti/bunqueue/tree/main/sdk/elixir)