@bumail/queue

The outbound queue of a mail server: it keeps every message your server sends to another one, delivers it through @bumail/smtp/client, retries what failed for now, and sends a delivery status notification (RFC 3464) back when it gives up. Each recipient has its own state, the queue survives a restart on bun:sqlite, PostgreSQL or Redis, and several workers — on one machine, or on many with PostgreSQL or Redis — can share one queue. No runtime dependency: only peers.

Install

bun add @bumail/queue @bumail/smtp @bumail/mime
bun add @bumail/mta-sts   # a required peer (its types are in the API), used when you set mtaSts
bun add @bumail/dns   # for direct delivery by MX: the resolver

@bumail/smtp (its client), @bumail/mime (for the DSNs) and @bumail/mta-sts (for mtaSts) are required peers. @bumail/dns is optional: any resolver of its shape will do, and a queue that only uses a smarthost needs none.

Subpaths

Usage

import { nodeResolver } from '@bumail/dns';
import { createQueue } from '@bumail/queue';
import { SqliteQueueStore } from '@bumail/queue/sqlite';

const queue = createQueue({
    store: SqliteQueueStore.open({ directory: '/var/lib/bumail/queue' }),
    hostname: 'mail.example.net', // your public name: EHLO, and the DSN's Reporting-MTA
    resolver: nodeResolver(),
});
queue.start();

// The message as it leaves: whole, headers first, CRLF, already DKIM-signed
// (with @bumail/auth's signDkim, say).
const signedMessage = await Bun.file('outgoing.eml').bytes();
const item = await queue.enqueue(signedMessage, {
    from: '[email protected]',
    to: ['[email protected]', '[email protected]'],
});
item.recipients; // [{ address: '[email protected]', status: 'pending' }, …]

process.on('SIGTERM', () => queue.stop()); // lets deliveries under way end

The message is a Uint8Array, a string or a ReadableStream<Uint8Array>, with CRLF line ends: a bare CR or LF is refused, as sendMail would refuse it. Every address is checked as one sendMail takes (@bumail/smtp/client's isMailbox: an RFC 5321 Mailbox with no source route, no control character (C0, DEL or C1), no >, no U+2028 or U+2029, no Unicode format character (\p{Cf}), no lone surrogate and no IPv4 literal octet above 255), so a bad one is refused at enqueue, never at delivery. Delivery groups the recipients by domain: one session per domain per attempt, with STARTTLS when offered.

Retries and bounces

A 4xx, a connection error or a timeout leaves the recipient deferred; the queue tries it again after 30 minutes, then 1 hour, 2, 4, and every 4 hours, each plus up to 10% of jitter, and gives up after 5 days (RFC 5321 §4.5.4.1). A 5xx fails the recipient at once.

import { nodeResolver } from '@bumail/dns';
import { createQueue } from '@bumail/queue';
import { MemoryQueueStore } from '@bumail/queue/memory';

const queue = createQueue({
    store: new MemoryQueueStore(),
    hostname: 'mail.example.net',
    resolver: nodeResolver(),
    retry: { first: 15 * 60_000, giveUpAfter: 3 * 24 * 3_600_000 },
    dsn: { delayAfter: 4 * 3_600_000, returnContent: 'headers' },
});

A failed recipient gets a DSN back to the sender, a multipart/report with the original's header fields; a recipient still deferred after delayAfter gets one "delayed" warning. A DSN is enqueued from the null sender <>, and a message from <> never causes one.

The envelope can carry the sender's DSN parameters (RFC 3461), which the SMTP server hands over as they are and the queue passes on to the next hop:

await queue.enqueue(message, {
    from: '[email protected]',
    to: ['[email protected]'],
    ret: 'HDRS', // what a DSN returns: the headers only (FULL: the whole message)
    envid: 'QQ314159', // echoed as Original-Envelope-Id
    recipients: [
        { address: '[email protected]', notify: ['SUCCESS', 'FAILURE'], orcpt: { type: 'rfc822', address: '[email protected]' } },
    ],
});

notify: 'NEVER' sends no DSN; absent, failure and delay as always; a list, only those events. Upgrading a shared store: stop every old worker before starting a new one (an old worker that keeps running ignores the DSN parameters and drops them from the items it completes). SUCCESS with a next hop that does not offer DSN sends a relayed DSN. For a message the application delivered itself, deliveredDsn() writes the delivered one, and failedDsn() the failed one for a recipient it could not deliver.

Routing

import { createQueue } from '@bumail/queue';
import { MemoryQueueStore } from '@bumail/queue/memory';

const smtpPassword = (await Bun.file('/run/secrets/smtp').text()).trim(); // from your secrets
createQueue({
    store: new MemoryQueueStore(),
    hostname: 'mail.example.net',
    // Port 25 blocked? Everything through a provider's submission port:
    route: {
        host: 'smtp.provider.example',
        port: 587,
        auth: { username: '[email protected]', password: smtpPassword },
    },
    // …or only some domains, the rest by MX (which needs the resolver):
    routes: { 'partner.example': { host: 'relay.partner.example' } },
    // …or by MX, with a smarthost only as the way out when MX fails:
    // fallback: { host: 'smtp.provider.example', port: 587, auth: { username: '[email protected]', password: smtpPassword } },
});

route defaults to 'mx': each recipient domain's own mail hosts. Credentials go only over TLS whose certificate checked out: createQueue refuses auth with a tls other than 'required', and checks the ports, the TLS modes and the timeouts as sendMail would. A route sendMail still refuses (INVALID_OPTION) defers its recipients as 4.3.5 and says so on the error event; it does not bounce them until retry.giveUpAfter, when they fail as 4.4.7 with a DSN like any deferred recipient.

With fallback (a smarthost), a recipient whose session by MX failed by a connection failure, a timeout or a reputation refusal goes through that smarthost at once (not after the final dot of DATA, where the MX may have taken the message: every recipient the MX did not refuse at RCPT is marked mayHaveReachedMx and never falls back on its own; a refused one may). If the smarthost's own session ends after the final dot, the recipient is marked mayHaveReachedFallback and later attempts go to the smarthost again, not to the MX; see Falling back to a smarthost.

MTA-STS

Opt in with mtaSts (off unless given; the smarthost routes and the fallback never use it). For each recipient domain on the MX route the queue reads _mta-sts.<domain> and fetches the policy (RFC 8461) through @bumail/mta-sts, keeping it in a PolicyCache until its max_age ends. With mode enforce it sends with tls: 'required' to the MX hosts the policy names only; with testing it sends as usual and only reports; a failure under enforce is deferred as 4.7.5 and bounced as 5.7.5 only at retry.giveUpAfter, after the policy is fetched once more. Under enforce nothing goes to the fallback: a connection failure defers. mtaSts.skip(item, domain) exempts a session (a mailed TLS-RPT report, a domain delivered in-process). Every try is told on the tls event, for RFC 8460 reports.

import { nodeResolver } from '@bumail/dns';
import { createMemoryPolicyCache, createQueue } from '@bumail/queue';
import { MemoryQueueStore } from '@bumail/queue/memory';

const resolver = nodeResolver(); // has txt
const queue = createQueue({
    store: new MemoryQueueStore(),
    hostname: 'mail.example.net',
    resolver,
    mtaSts: { cache: createMemoryPolicyCache() }, // fetch: globalThis.fetch; resolver: the queue's
});
queue.on('tls', ({ domain, policy, mxHost, ok, tlsResult }) => {});

The tls event also carries kind ('try' or 'policy'), at (the queue's clock, ms), sendingAddress and reused. A send of your own receives mxNames, tls, onAttempt and onSkip in its options and has to pass them to sendMail. See MTA-STS.

Reusing connections

Without a send of your own, the queue sends through a pool of SMTP connections it owns: several messages to one server share a connection (100 at most, idle 5 seconds), and stop() closes the pool. reuse tunes it, reuse: false opens one connection per session, and a send you give is never wrapped. Events carry reused when the sender says so. The bumail serve queue of @bumail/server sends through the same pool, with no setting. See createSmtpPool in the @bumail/smtp README.

createQueue({
    store: new MemoryQueueStore(),
    hostname: 'mail.example.net',
    resolver: nodeResolver(),
    reuse: { maxMessages: 50, idleSeconds: 10 }, // default { 100, 5 }; or false
});

See Reusing connections.

Several workers

import { nodeResolver } from '@bumail/dns';
import { createQueue } from '@bumail/queue';
import { SqliteQueueStore } from '@bumail/queue/sqlite';

// In each process, on the same directory: a claim leases an item to one worker.
const queue = createQueue({
    store: SqliteQueueStore.open({ directory: '/var/lib/bumail/queue' }),
    hostname: 'mail.example.net',
    resolver: nodeResolver(),
});
queue.start();

A worker claims a due item with a lease, renews it while it delivers, and lets go of it with the outcome; a renewal that fails is told on error and tried again. If it crashes, another worker claims the item once the lease expires (leaseMs, 10 minutes by default). A directory the store makes is 0700; one that exists keeps its mode. concurrency (20) bounds the items one worker delivers at once — each item opens one session per recipient domain — and perDomain (2) its sessions to one recipient domain.

PostgreSQL, for several machines

import { nodeResolver } from '@bumail/dns';
import { createQueue } from '@bumail/queue';
import { PostgresQueueStore } from '@bumail/queue/postgres';

// In each instance, on the same database. A Bun.SQL client of your own works too:
// PostgresQueueStore.open({ sql: new Bun.SQL({ url, max: 10 }) }).
const store = PostgresQueueStore.open({
    sql: Bun.env['DATABASE_URL'] ?? 'postgres://bumail@localhost:5432/mail',
    tablePrefix: 'bumail_queue_', // the default: bumail_queue_items, …_messages, …_schema
});
await store.migrate(); // makes the tables; the first call would anyway
const queue = createQueue({ store, hostname: 'mail.example.net', resolver: nodeResolver() });
queue.start();

process.on('SIGTERM', async () => {
    await queue.stop();
    await store.close(); // closes the client it opened for the URL, never yours
});

A claim is one UPDATE … RETURNING whose item a SELECT … FOR UPDATE SKIP LOCKED picks: two instances never take the same item, and none waits on one another is taking. An instance that crashes loses its items when their leases expire, as with bun:sqlite. Every write runs at READ COMMITTED, so a client whose sessions default to repeatable read or serializable serves the queue too. No driver to install: Bun.sql is Bun's.

Redis, for several machines

import { nodeResolver } from '@bumail/dns';
import { createQueue } from '@bumail/queue';
import { RedisQueueStore } from '@bumail/queue/redis';

// In each instance, on the same Redis. A Bun.RedisClient of your own works too:
// RedisQueueStore.open({ client: new Bun.RedisClient(url) }).
const store = RedisQueueStore.open({
    url: Bun.env['REDIS_URL'] ?? 'redis://localhost:6379',
    keyPrefix: 'bumail:queue:', // the default: bumail:queue:items, …:item:<id>, …
});
const queue = createQueue({ store, hostname: 'mail.example.net', resolver: nodeResolver() });
queue.start();

process.on('SIGTERM', async () => {
    await queue.stop();
    await store.close(); // closes the client it opened for the URL, never yours
});

Every operation that writes is one Lua script, which Redis runs whole: two instances never take the same item, and a crashed instance's items are claimed again once their leases expire. The message is kept byte for byte. One Redis, or a primary with replicas — not Redis Cluster. Redis acknowledges a write once it is in memory: with appendfsync everysec a crash loses up to a second of writes, and a failover to a replica can lose acknowledged ones, so keep appendonly yes, appendfsync always and maxmemory-policy noeviction where a lost message matters (the guide's Redis section). No driver to install: Bun.redis is Bun's.

Events and admin

// queue: the one created under Usage.
queue.on('delivered', ({ id, recipient, reply }) => console.info({ id, recipient, reply }));
queue.on('deferred', ({ recipient, reply, nextAttemptAt }) => {});
queue.on('failed', ({ recipient, reply }) => {}); // reply: { code?, status?, text, host? }
queue.on('dsn', ({ kind, of, to }) => {});
queue.on('tls', ({ domain, policy, mxHost, address, ok, tlsResult }) => {}); // with mtaSts
queue.on('error', ({ error, id }) => console.error(id, error)); // the store, a lost lease, a bad route

const [next] = await queue.list({ limit: 50 }); // the next due first
if (next) await queue.retryNow(next.id); // false while a worker delivers it
if (next) await queue.cancel(next.id); // no DSN

A store of your own

QueueStore may grow in a 0.x minor version (methods, fields); a store of your own follows it.

Hold a QueueStore you wrote to the same tests the stores here pass. The spec runs with bun test (the suite imports bun:test):

// my-store.spec.ts
import { describeQueueStore } from '@bumail/queue/conformance';
import { MyQueueStore } from './my-store';

describeQueueStore('MyQueueStore', {
    create: () => MyQueueStore.open({ url: process.env.MY_STORE_URL }),
    share: (store) => (store as MyQueueStore).connectAgain(), // a second handle on the same queue
    dispose: (store) => (store as MyQueueStore).close(),
});

Without share the two several-workers tests are skipped, and a store shared over a network should pass a real second connection. See Holding your own store to the contract.

Testing

import { createQueue, type Sender } from '@bumail/queue';
import { MemoryQueueStore } from '@bumail/queue/memory';

let now = Date.UTC(2026, 0, 1);
const send: Sender = async (_message, options) => ({
    accepted: [options.to].flat().map((recipient) => ({ recipient, reply: { code: 250, text: 'OK' } })),
    rejected: [],
    reply: { code: 250, text: 'Queued' },
    host: 'mx.test',
    port: 25,
    tls: false,
    authenticated: false,
});
const queue = createQueue({
    store: new MemoryQueueStore(),
    hostname: 'mail.test',
    route: { host: 'mx.test' },
    clock: { now: () => now },
    send,
});
await queue.enqueue('Subject: hi\r\n\r\nhello\r\n', { from: 'a@test', to: 'b@test' });
await queue.deliverDue(); // one pass, by hand: no timer, no network
now += 30 * 60_000;

Limits

import { nodeResolver } from '@bumail/dns';
import { createQueue } from '@bumail/queue';
import { MemoryQueueStore } from '@bumail/queue/memory';

createQueue({
    store: new MemoryQueueStore(),
    hostname: 'mail.example.net',
    resolver: nodeResolver(),
    limits: { maxMessageSize: 10 * 1024 * 1024, maxRecipients: 50, maxItems: 100_000 },
});

Everything is bounded, through limits: the message (25 MiB), the recipients per message (100), the items in the store (none by default), the reply text kept per recipient (512 characters, control characters replaced by spaces), and the original a DSN returns (64 KiB, each line cut at 998 bytes). Nothing a remote server says reaches a DSN's header fields with a CR, an LF or a control character in it.

Traps

• Bun only. The stores use bun:sqlite, Bun.sql and Bun.redis, so the package runs on Bun 1.4.2 or later, not on Node.

• It sends what you enqueue, to anyone. The queue is not a relay policy: enqueue only what an authenticated user submitted, or what your own server writes, never what an unauthenticated client handed you.

• A custom QueueStore must port complete (0.11, breaking). complete(id, owner, result, dsns) takes the DSN items the outcome calls for and returns { item, dsns }, or undefined for a lost lease. Store the outcome and the DSNs in one step, so a crash cannot leave one without the other; a store that returns the bare item still works, but the queue reports Queue store outdated on the error event and adds each DSN with add, outside that step. The stores of this package are atomic. DSNs ignore limits.maxItems: a full queue still bounces, and the count may exceed the limit by the DSNs it holds. They are delivered at least once; see Writing a store. The this.* calls below are your store's own helpers, a sketch.

  async complete(id, owner, result, dsns) {
    // refuse a bad DSN (INVALID) before opening a connection or a transaction
    for (const dsn of dsns) this.check(dsn);
    return this.transaction(async () => {
        const item = await this.apply(id, owner, result); // undefined on a lost lease
        if (!item) return undefined; // writes no DSN
        const stored = [];
        for (const dsn of dsns) stored.push(await this.insert(dsn)); // in order
        return { item, dsns: stored };
    });
  }
  

Upgrading

With MTA-STS (the release that adds mtaSts):

• @bumail/mta-sts is a required peer. Install it even when mtaSts is unused: bun add @bumail/mta-sts.
• QueueEvents has a new tls event kind, for the TLS-RPT counts. A listener that switches on the event kinds should handle it or ignore it.
• SmtpErrorCode has MX_NOT_ALLOWED (X.7.5, from @bumail/smtp), for an MX host an enforce policy does not allow. A switch over the codes needs the case.
• A custom send must pass mxNames, tls and onAttempt to sendMail, or an enforce policy is not applied and the tls events stay silent.
• QueueStore is unchanged.

API

Documentation

These pages ship in the package, under docs/.

• Index: the pages, and when to read each.
• Guide: enqueuing, delivery and routing, the retry schedule, DSNs, several workers and leases, events and admin, the stores (PostgreSQL and Redis included), testing, and writing a store of your own.
• Troubleshooting: every QueueError, and what to do about it.
• Roadmap: what is coming, and what is not planned.