examples · sdk · queues and workers

N queues. N workers. One truth.

Put producers on broker A, workers on B and C, and observers on C. PostgreSQL keeps jobs, policies, lifecycle state, events, logs, and results coherent across the fleet.

Connection factory

Every SDK object receives one normal TCP connection. The SDK does not receive the PostgreSQL URL and does not need database credentials.

examples/postgres-multibroker/shared.ts

import type { ConnectionOptions } from 'bunqueue/client';

export type BrokerName = 'a' | 'b' | 'c';
export type CleanupTask = () => void | Promise<void>;

const defaults: Record<BrokerName, { host: string; port: number }> = {
  a: { host: '127.0.0.1', port: 16789 },
  b: { host: '127.0.0.1', port: 17789 },
  c: { host: '127.0.0.1', port: 18789 },
};

export function connection(name: BrokerName): ConnectionOptions {
  const key = name.toUpperCase();
  return {
    commandTimeout: 15_000,
    host: Bun.env[`BROKER_${key}_HOST`] ?? defaults[name].host,
    pingInterval: 0,
    poolSize: 2,
    port: Number(Bun.env[`BROKER_${key}_PORT`] ?? defaults[name].port),
    token: Bun.env.BUNQUEUE_TOKEN ?? 'demo-token',
  };
}

export function httpUrl(name: BrokerName): string {
  const key = name.toUpperCase();
  return (
    Bun.env[`BROKER_${key}_HTTP_URL`] ?? `http://${defaults[name].host}:${defaults[name].port + 1}`
  );
}

export function uniqueQueue(label: string): string {
  return `example-${label}-${crypto.randomUUID()}`;
}

export function invariant(condition: unknown, message: string): asserts condition {
  if (!condition) throw new Error(message);
}

export async function withTimeout<T>(
  label: string,
  operation: () => T | Promise<T>,
  timeoutMs: number
): Promise<T> {
  if (!Number.isFinite(timeoutMs) || timeoutMs <= 0) {
    throw new Error(`Timeout for ${label} must be a positive finite number`);
  }

  let timer: ReturnType<typeof setTimeout> | undefined;
  const timeout = new Promise<never>((_, reject) => {
    timer = setTimeout(
      () => reject(new Error(`Timed out after ${Math.ceil(timeoutMs)}ms while ${label}`)),
      timeoutMs
    );
  });

  try {
    return await Promise.race([Promise.resolve().then(operation), timeout]);
  } finally {
    if (timer !== undefined) clearTimeout(timer);
  }
}

export async function waitFor(
  label: string,
  predicate: () => boolean | Promise<boolean>,
  timeoutMs = 20_000
): Promise<void> {
  const deadline = performance.now() + timeoutMs;
  while (performance.now() < deadline) {
    const remaining = deadline - performance.now();
    if (await withTimeout(`waiting for ${label}`, predicate, remaining)) return;
    await Bun.sleep(Math.min(25, Math.max(0, deadline - performance.now())));
  }
  throw new Error(`Timed out waiting for ${label}`);
}

export async function settleCleanup(
  ...phases: ReadonlyArray<ReadonlyArray<CleanupTask>>
): Promise<void> {
  const errors: Error[] = [];
  for (const tasks of phases) {
    const results = await Promise.allSettled(tasks.map((task) => Promise.resolve().then(task)));
    for (const result of results) {
      if (result.status === 'rejected') {
        errors.push(
          result.reason instanceof Error ? result.reason : new Error(String(result.reason))
        );
      }
    }
  }
  if (errors.length > 0) throw new AggregateError(errors, 'Example cleanup failed');
}

The example deliberately names three endpoints so cross-broker behavior is observable. In production, pass a service or TCP load-balancer address instead when clients do not need deterministic placement.

Realistic multi-queue scenario

examples/postgres-multibroker/multi-queue.ts

import { Queue, QueueEvents, Worker } from 'bunqueue/client';
import { connection, invariant, settleCleanup, uniqueQueue, waitFor } from './shared';

interface EmailJob {
  orderId: string;
  to: string;
  transient?: boolean;
}

interface MediaJob {
  assetId: string;
  width: number;
}

interface AuditJob {
  action: string;
  orderId: string;
}

export async function runMultiQueueExample(): Promise<void> {
  const emailName = uniqueQueue('emails');
  const mediaName = uniqueQueue('media');
  const auditName = uniqueQueue('audit');
  const emails = new Queue<EmailJob>(emailName, { connection: connection('a') });
  const media = new Queue<MediaJob>(mediaName, { connection: connection('a') });
  const audit = new Queue<AuditJob>(auditName, { connection: connection('a') });
  const emailObserver = new Queue<EmailJob>(emailName, { connection: connection('c') });
  const events = new QueueEvents<{ delivered: string }, { stage: string }>(emailName, {
    connection: connection('c'),
  });
  const retried = new Set<string>();
  const completedEvents = new Set<string>();
  const progressEvents = new Set<string>();
  const emailStarts: string[] = [];
  const workerErrors: Error[] = [];

  events.on('completed', ({ jobId }) => completedEvents.add(jobId));
  events.on('progress', ({ jobId }) => progressEvents.add(jobId));

  const emailWorker = new Worker<EmailJob, { delivered: string }>(
    emailName,
    async (job) => {
      emailStarts.push(job.id);
      await job.updateProgress(50, 'rendering-template');
      await job.log(`sending order ${job.data.orderId}`);
      if (job.data.transient && !retried.has(job.id)) {
        retried.add(job.id);
        throw new Error('simulated provider timeout');
      }
      await job.updateProgress(100, 'delivered');
      return { delivered: job.data.to };
    },
    { batchSize: 1, concurrency: 1, connection: connection('b') }
  );
  const mediaWorker = new Worker<MediaJob, { output: string }>(
    mediaName,
    (job) => ({ output: `${job.data.assetId}-${job.data.width}.webp` }),
    { concurrency: 2, connection: connection('c') }
  );
  const auditWorker = new Worker<AuditJob, { recorded: boolean }>(
    auditName,
    () => ({ recorded: true }),
    { concurrency: 2, connection: connection('b') }
  );
  for (const worker of [emailWorker, mediaWorker, auditWorker]) {
    worker.on('error', (error) => workerErrors.push(error));
  }

  try {
    await Promise.all([
      emails.waitUntilReady(),
      media.waitUntilReady(),
      audit.waitUntilReady(),
      events.waitUntilReady(),
      emailWorker.waitUntilReady(),
      mediaWorker.waitUntilReady(),
      auditWorker.waitUntilReady(),
    ]);
    await emails.pauseAsync();
    await waitFor('the email pause to reach broker C', () => emailObserver.isPausedAsync());

    const emailJobs = await emails.addBulk([
      {
        name: 'order-confirmation',
        data: { orderId: 'ord-100', to: 'customer@example.com' },
        opts: { durable: true, jobId: `${emailName}-normal` },
      },
      {
        name: 'vip-confirmation',
        data: { orderId: 'ord-101', to: 'vip@example.com' },
        opts: { durable: true, jobId: `${emailName}-vip`, priority: 100 },
      },
      {
        name: 'provider-retry',
        data: { orderId: 'ord-102', to: 'retry@example.com', transient: true },
        opts: { attempts: 3, backoff: 25, durable: true, jobId: `${emailName}-retry` },
      },
      {
        name: 'delayed-follow-up',
        data: { orderId: 'ord-103', to: 'later@example.com' },
        opts: {
          delay: 300_000,
          durable: true,
          jobId: `${emailName}-delayed`,
          priority: 200,
        },
      },
    ]);
    const mediaJob = await media.add(
      'thumbnail',
      { assetId: 'asset-7', width: 640 },
      { durable: true }
    );
    const auditJob = await audit.add(
      'order-created',
      { action: 'created', orderId: 'ord-100' },
      { durable: true }
    );

    const delayedJob = emailJobs[3];
    invariant(
      (await emailObserver.getJobState(delayedJob.id)) === 'delayed',
      'the delayed email became eligible before promotion'
    );
    invariant(emailStarts.length === 0, 'the paused email queue started work');
    await emailObserver.resumeAsync();

    await waitFor('all ready emails to complete in priority order', async () => {
      const states = await Promise.all(emailJobs.slice(0, 3).map((job) => job.getState()));
      return states.every((state) => state === 'completed');
    });
    invariant(emailStarts[0] === `${emailName}-vip`, 'the highest-priority ready email ran late');
    invariant(!emailStarts.includes(delayedJob.id), 'the delayed email ran before promotion');
    invariant(
      (await emailObserver.getJobState(delayedJob.id)) === 'delayed',
      'the delayed email did not remain delayed'
    );
    const beforePromotion = await emailObserver.getJobCountsAsync();
    invariant(
      beforePromotion.completed === 3 && beforePromotion.delayed === 1,
      'ready and delayed email counts diverged before promotion'
    );
    await delayedJob.promote();

    await waitFor('all queues to complete across three brokers', async () => {
      const [emailCounts, mediaCounts, auditCounts] = await Promise.all([
        emailObserver.getJobCountsAsync(),
        media.getJobCountsAsync(),
        audit.getJobCountsAsync(),
      ]);
      return (
        emailCounts.completed === emailJobs.length &&
        mediaCounts.completed === 1 &&
        auditCounts.completed === 1
      );
    });
    await waitFor(
      'cross-broker completion and progress events',
      () => completedEvents.size === emailJobs.length && progressEvents.size === emailJobs.length
    );

    invariant(retried.has(`${emailName}-retry`), 'the transient email was not retried');
    invariant(workerErrors.length === 0, `worker error: ${workerErrors[0]?.message}`);
    invariant(await mediaJob.isCompleted(), 'the media job did not complete');
    invariant(await auditJob.isCompleted(), 'the audit job did not complete');
    invariant((await emailObserver.getWorkersCount()) >= 1, 'no email worker was registered');

    const logs = await emailObserver.getJobLogs(`${emailName}-normal`);
    invariant(logs.logs.includes('[info] sending order ord-100'), 'the email log was not retained');
    const retained = await emailObserver.getJob(`${emailName}-normal`);
    invariant(
      retained?.returnvalue &&
        (retained.returnvalue as { delivered: string }).delivered === 'customer@example.com',
      'the cross-broker result was not retained'
    );
  } finally {
    await settleCleanup(
      [
        () => emailWorker.close(true),
        () => mediaWorker.close(true),
        () => auditWorker.close(true),
        () => events.close(),
      ],
      [
        () => emails.obliterateAsync(),
        () => media.obliterateAsync(),
        () => audit.obliterateAsync(),
      ],
      [() => emails.close(), () => media.close(), () => audit.close(), () => emailObserver.close()]
    );
  }
}

What the assertions prove

Surface	Broker path	Assertion
Email producer	A	Four durable jobs accepted with bulk, priority, delay, and retry options
Email worker	B	One-slot priority order, delayed hold/promotion, retry, progress, logs
Media worker	C	Independent queue and typed result
Audit worker	B	Third queue processed independently
QueueEvents	C	Completion and progress events converge for all email jobs
Queue observer	C	Counts, worker registration, logs, state, and return value are visible

Auto-batching and addBulk

Network queues auto-batch nearby add() calls by default. This scenario calls addBulk() explicitly because the batch is already known. PostgreSQL commits the admission transaction before the promise resolves; durable: true is retained for portable intent but PostgreSQL does not use SQLite’s write buffer.

Scale workers independently

Queue names define the work stream; worker processes do not need to live with their producer. Add replicas according to the workload:

• more emails workers for network-bound delivery;
• fewer media workers with lower concurrency for CPU or memory-heavy work;
• separate audit workers with stricter retention and access controls.

The broker-side queue concurrency limit is global across all brokers and workers. Worker concurrency controls only that worker instance. Use both when you need local capacity and a fleet-wide safety ceiling.

Scheduling proof

The email queue is paused before addBulk(), and broker C confirms that shared state before admission. The worker uses concurrency: 1 and batchSize: 1, so the VIP job must be the first eligible start. The delayed follow-up has an even higher priority but remains in delayed; three ready emails complete while the counts stay at three completed and one delayed. Only an explicit delayedJob.promote() makes the fourth job eligible. This proves priority among ready work and proves that priority never bypasses a delay.

Lifecycle rule

Close workers first so in-flight outcomes can settle, then event subscriptions, then queues. The example obliterates its UUID-suffixed queues only because it is a disposable demonstration. A production shutdown must not obliterate queues.

Next: idempotency, retries, DLQ, and shared limits.