@ bumail/ queue (英语)

邮件服务器的外出队列: 它保存您服务器的每封信 送去另一个,送过去 @bumail/smtp/client重试 现在的失败,并发出交货状态通知(RFC 3464) 当它放弃的时候 每个收件人都有自己的状态,队列 在重新启动中幸存下来 bun:sqlite, PostgreSQL 或 Redis, 以及多个 工人——用一台机器,或用PostgreSQL或Redis的许多人——可以 共享一个队列。 没有运行时间依赖性: 只有对等者.

安装

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 (客户), @bumail/mime (针对DSN)和 @bumail/mta-sts (计: mtaSts)是要求的同龄人. @bumail/dns 是可选的:其形状的任何解析器都会起作用,并且 a 只使用智能主机的队列不需要.

子路径

使用量

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

信息是一个 Uint8Array,字符串或 ReadableStream<Uint8Array>, (中文(简体) ), 带有 CRLF 线末端: 拒绝使用空 CR 或 LF , 如 sendMail 将 拒绝 每个地址都是一个的 sendMail 带 (单位:千美元)@bumail/smtp/client是因为 isMailbox: 无源的 RFC 5321 邮箱 路径,没有控制字符(C0,DEL或C1),没有 >,没有U+2028或U+2029, 无 Unicode 格式字符 (E)\p{Cf}),没有单独代孕和IPv4 255以上字节,所以一个坏字被拒绝 enqueue,从来没有 交货. 交付按域分类:每个域一次 尝试,提供STATLS.

反弹和反弹

4xx, 连接出错或超时离开收件人 deferred· ; 30分钟后队列再试一次 1小时 2 4 每4次 5天后放弃 (RFC 5321 第4.5.4.1节)。 A 5xx 一次失败接收者.

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' },
});

失败的接收者会得到一个 DSN 返回发送者, a multipart/report 与原始标题字段; 接收者在 delayAfter 得到一个"延迟"警告。 一个 DSN 被从空中排出 发送者 <>,以及来自 <> 从来没有造成一个.

信封可以携带发送者的DSN参数(RFC 3461),即: SMTP 服务器按原样交接,队列传递到下一个 跳动:

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' 不发送 DSN; 一如既往的缺席、 失败和延迟; a 列表,只有那些事件。 升级共享商店:停止所有老工人 在开始新的工作之前(一个老工人不断跑动,无视 DSN参数并从它完成的项目中掉出. SUCCESS 带有不提供 DSN 的下个 hop 发送一个 relayed DSN (英语). 对于所递交的电文, deliveredDsn() 写入 delivered 一个,和 failedDsn() 页:1 failed 给接收者一个它无法提供.

路线

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 默认为 'mx':每个收件人域自己的邮件主机. 证书只通过 TLS 检查: createQueue 拒绝 auth 带有 tls 其他 'required'检查港口, tLS 模式和超时 sendMail 会的 路线 sendMail 仍然拒绝(INVALID_OPTION(b) 推迟接收者,因为 4.3.5 财务报告和审定财务报表 这么说的 error 事件,它不反弹,直到 retry.giveUpAfter当它们失败时 4.4.7 和任意的DSN一样 推迟接收者.

与 fallback (a smarthost),一个被 MX 的会话失败的收受者 连接失败, 超时或拒绝名声都会经历 一次智能主机( 不在最后点之后 ) DATA,其中MX可能接收到消息:每个接收者MX没有拒绝 RCPT 标记 mayHaveReachedMx 绝不退缩,被拒绝者可以退缩。 如果智能主机自己的会话在最后点后结束,则收件人会被标记 mayHaveReachedFallback 而后来的尝试再次去 聪明的宿主,而不是MX 回到一个聪明的宿主。 。 。 .

技术

打开 mtaSts (关闭,除非给出;智能主线和 fallback 永远不要用它). MX 路线上的每个收件人域都读取队列 _mta-sts.<domain> 并获取保险单(RFC 8461) @bumail/mta-sts保持它在一个 PolicyCache 直至 max_age 结束。 带有模式 enforce 它发送与 tls: 'required' 到 MX 仅主机政策名称 testing 它一如既往地只发送报告 enforce 推迟到 4.7.5 弹出为 5.7.5 仅在 retry.giveUpAfter,在再次获取保险单之后。 下 enforce 什么都没有 fallback: 连接失败延迟 . mtaSts.skip(item, domain) 免除一个会话(一个被邮寄的TLS-RPT报告,一个在进程中发送的域). 每一个尝试都被告知 tls 活动,用于 RFC 8460 报告.

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 }) => {});

这个 tls 事件还携带 kind (单位:千美元)'try' 或者说 'policy', (中文(简体) ), at (队列的时钟,ms), sendingAddress 财务报告和审定财务报表 reusedA级 send 自己接收的 mxNames, (中文(简体) ), tls, (中文(简体) ), onAttempt 财务报告和审定财务报表 onSkip 并且必须传递给 sendMail。见 技术。 。 。 .

重用连接

没有 send ,队列通过 SMTP 池发送 连接它拥有的: 多个信件到一个服务器共享连接 (最多100个,闲出5秒),和 stop() 关闭了游泳池. reuse 曲调 这, reuse: false 每个会话打开一个连接, a send 你给 从来没有包裹。 活动 reused 当发送者说. 这个 bumail serve 队列 @bumail/server 从同一个游泳池发出 没有设置。 见 createSmtpPool 输入 @bumail/smtp 读取.

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

见 重用连接。 。 。 .

若干工人

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();

工人要求有租约的到期物品,在交货时延长租约 随其结果而去;失败的续约将被告知 error 并再次尝试。 如果撞倒了, 另一工人一旦抢走 租赁期满(leaseMs,默认为10分钟). 存储目录 makes is 0700; 一个存在它的模式. concurrency (20) 将一个工人一并交付的物品捆绑起来——每人 条目打开每个收件人域的会话——和 perDomain (二) 联合国 会话到一个收件人域.

PostgreSQL, 用于多台机器

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
});

一项索赔就是一项 UPDATE … RETURNING 其项目 a SELECT … FOR UPDATE SKIP LOCKED 选择:两例从不采用同一项, 无 等待着对方正在承受。 撞出东西的事故 当他们的租约到期时 bun:sqlite。每封信都运行在 READ COMMITTED,因此其会话默认为 repeatable read 或者说 serializable 服务队列。 无驱动程序安装 : Bun.sql 这是邦的.

Redis, 用于几台机器

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
});

每个写作的操作都是一个Lua脚本,Redis运行整个: 两起案件从不带走同一物品,一起坠毁的案件的物品 一旦租约到期,将再次提出索赔。 信件为 字节。 一个Redis,或一个有复制品的初级-不是Redis集群. 重写 在记忆中承认写入: appendfsync everysec 崩溃会损失一秒钟的写作, 和失败的复制品可以 失去承认的,所以保持 appendonly yes, (中文(简体) ), appendfsync always 财务报告和审定财务报表 maxmemory-policy noeviction 如果丢失的信息很重要 指南的 Redis 部分)。 无驱动程序安装 : Bun.redis 这是邦的.

事件和管理

// 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

你自己的店铺

QueueStore 可能以 0.x 的小型版本(方法,字段)生长;您自己的商店也遵循它.

拿着 QueueStore 你给同样的测试 店铺在这里通过. 光谱运行于 bun test (套房进口) 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(),
});

无 share 两个数个工人的测试被跳过 一个商店 在一个网络上共享,应该通过真正的第二次连接。 见 拥有自己的商店与合同。 。 。 .

测试

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;

限制

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 },
});

万事俱备,万事俱备 limits: 信息(25个米B) 每个信件的收件人(100),存储中的项目(默认无), 每个收件人保存的回复文本( 512个字符, 控制字符) 替换为空格),原始的 DSN 返回(64 KiB,每行剪接 以998字节表示). 没什么 远程服务器说用 CR 、 LF 或 a 到达 DSN 头字段 控制其中的字符.

陷阱

• 仅存一文. 仓库用 bun:sqlite, (中文(简体) ), Bun.sql 财务报告和审定财务报表 Bun.redis, (中文(简体) ), 所以包裹在1.4.2号机上运行 或晚于节点运行.

• 它向任何人发出你所发出的声音. 队列不是中继 策略:只排入经认证的用户提交的内容,或您的 自己的服务器写,从来没有什么 未经认证的客户端给你.

• 习惯 QueueStore 必须移植 complete (0.11,断裂). complete(id, owner, result, dsns) 带 DSN 项 结果 请求和返回 { item, dsns },或 undefined 失去的租约. 将结果和 DSN 保存在一个步骤中, 所以崩溃不能留下一个 没有另一个;返回裸体物品的商店仍然工作,但是 队列报告 Queue store outdated 编辑 error 事件和添加 每个 DSN 有 add在楼梯外 这个软件包的存储是 原子 DSN 忽略 limits.maxItems: 整队仍会弹出 计数可能超过它持有的DSN的限制. 它们至少交付一次;见 写入 a 存储编辑 this.* 电话: 你店里的帮手,一张素描.

  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 };
    });
  }
  

升级

与MTA-STS(添加的版本) mtaSts:

• @bumail/mta-sts 是需要的对等点. 安装它即使当 mtaSts 未使用 : bun add @bumail/mta-sts。 。 。 .
• QueueEvents 有一个新的 tls 事件类型为TLS -RPT计数。 A 类 在事件类型上切换的收听器应该处理或者忽略它.
• SmtpErrorCode 已经 MX_NOT_ALLOWED (单位:千美元)X.7.5从 @bumail/smtp),用于 mX 主机 a enforce 政策不允许。 A 类 switch 超过代码 需要这个案子.
• 习惯 send 必须通过 mxNames, (中文(简体) ), tls 财务报告和审定财务报表 onAttempt 改为 sendMail,或一个 enforce 未执行政策 tls 事件 保持安静.
• QueueStore 不变.

API 密码

文档

这些页面在包里,在下 docs/。 。 。 .

• 指数:各页,何时读出.
• 指南:排队,送货和出行,重试时间表,DSN,几个工人和租赁,事件和管理员,商店(包括PostgreSQL和Redis),测试,并写出自有的商店.
• 解决问题: 每一个 QueueError,怎么办呢.
• 路线图:即将到来的,以及没有计划的.