Archived
修复轮1:凭证解不开时明文经 SyntaxError 泄进 last_poll_error;同键在飞的那封会被当失败行重发;发成功不清上一次的错误话;卡死的 dispatching 给人手一条路
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01TM4kUBvekALu9qqmp7D3wr
This commit is contained in:
@@ -195,6 +195,8 @@ export function memoryOutbound(): OutboundStore & { rows: Map<string, OutboundRe
|
||||
rows.set(id, {
|
||||
...r,
|
||||
state,
|
||||
// 与真 store 同一条不变量:发成功的那一行不留上一次的错误话(替身与真身分叉,测试就会连错的点)
|
||||
...(state === 'sent' ? { error: null } : {}),
|
||||
...(patch.error === undefined ? {} : { error: patch.error }),
|
||||
...(patch.providerRef === undefined ? {} : { providerRef: patch.providerRef }),
|
||||
...(patch.sentAt === undefined ? {} : { sentAt: patch.sentAt }),
|
||||
|
||||
@@ -0,0 +1,172 @@
|
||||
/**
|
||||
* 这一份钉的是 Task 9 评审逮出的三条 Important 与一条设计律 ——
|
||||
* 它们都在拉取 / 发信 / 分发那几条**当时还没有测试**的路上。
|
||||
*/
|
||||
import { describe, expect, it } from 'vitest';
|
||||
import { mailSendInputSchema } from '@contracts';
|
||||
import { createMailService } from '../internal/usecase/mail-service.ts';
|
||||
import { buildInboundRecord } from '../internal/domain/inbound-record.ts';
|
||||
import { createPostalMimeParser } from '../internal/adapter/postal-mime-parser.ts';
|
||||
import type { OutboundRecord, ParsedMail } from '../port.ts';
|
||||
import { account, fakeProvider, fixedClock, memoryAccounts, memoryAttachments, memoryInbound, memoryOutbound, plainSecrets, seqIds } from './fakes.ts';
|
||||
|
||||
function harness() {
|
||||
const clock = fixedClock();
|
||||
const accounts = memoryAccounts(clock);
|
||||
const inbound = memoryInbound();
|
||||
const outbound = memoryOutbound();
|
||||
const imap = fakeProvider('imap-smtp');
|
||||
const wecom = fakeProvider('wecom');
|
||||
const changes: string[] = [];
|
||||
const service = createMailService({
|
||||
accounts,
|
||||
inbound,
|
||||
outbound,
|
||||
attachments: memoryAttachments(),
|
||||
providers: { 'imap-smtp': imap, wecom },
|
||||
mime: createPostalMimeParser(),
|
||||
secrets: plainSecrets,
|
||||
clock,
|
||||
ids: seqIds('out'),
|
||||
notify: { changed: (r) => void changes.push(r) },
|
||||
});
|
||||
return { service, accounts, inbound, outbound, imap, wecom, clock, changes };
|
||||
}
|
||||
|
||||
const sendInput = mailSendInputSchema.parse({ to: ['buyer@acme.com'], subject: '报价', text: '见附', requestedBy: { kind: 'ops' } });
|
||||
|
||||
function outboundRow(over: Partial<OutboundRecord> = {}): OutboundRecord {
|
||||
return {
|
||||
id: 'out-9',
|
||||
accountId: 'acc-1',
|
||||
messageId: '<out-9@we.com>',
|
||||
idempotencyKey: 'k-1',
|
||||
to: ['buyer@acme.com'],
|
||||
cc: [],
|
||||
bcc: [],
|
||||
subject: '报价',
|
||||
text: '见附',
|
||||
html: null,
|
||||
inReplyTo: null,
|
||||
references: [],
|
||||
attachments: [],
|
||||
state: 'sending',
|
||||
error: null,
|
||||
providerRef: null,
|
||||
requestedBy: { kind: 'ops' },
|
||||
createdAt: new Date('2026-08-25T10:00:00Z'),
|
||||
sentAt: null,
|
||||
...over,
|
||||
};
|
||||
}
|
||||
|
||||
describe('凭证解不开', () => {
|
||||
it('密文解得开但不是 JSON:错误话里没有明文,只说「未配置凭证」', async () => {
|
||||
const { service, accounts, imap } = harness();
|
||||
// 解得开(plainSecrets 会回出 `pw-1234`),但那不是 JSON —— JSON.parse 的 SyntaxError 把原文嵌在 message 里
|
||||
await accounts.save(account({ secretCipher: 'sealed:pw-1234' }));
|
||||
const outcome = await service.pollNow('acc-1');
|
||||
expect(outcome.error).not.toContain('pw-1234');
|
||||
expect(outcome.error).toContain('未配置凭证');
|
||||
expect(accounts.rows.get('acc-1')?.lastPollError).not.toContain('pw-1234');
|
||||
expect(accounts.rows.get('acc-1')?.lastPollError).toContain('未配置凭证');
|
||||
// 凭证都没打开,一次都不该去够厂商
|
||||
expect(imap.pullCalls).toEqual([]);
|
||||
});
|
||||
});
|
||||
|
||||
describe('幂等键', () => {
|
||||
it('同键的上一封还在飞:不发第二封,如实回没发出去', async () => {
|
||||
const { service, accounts, outbound, imap, clock } = harness();
|
||||
await accounts.save(account());
|
||||
await outbound.insert(outboundRow({ createdAt: clock.now() }));
|
||||
const result = await service.send({ ...sendInput, idempotencyKey: 'k-1' });
|
||||
expect(result.state).toBe('failed');
|
||||
expect(result.error).toContain('仍在发送中');
|
||||
expect(imap.sent).toHaveLength(0);
|
||||
expect(outbound.rows.size).toBe(1);
|
||||
expect(outbound.rows.get('out-9')?.state).toBe('sending');
|
||||
});
|
||||
|
||||
it('同键的 sending 停太久:当上次崩在半路,复用那一行重发', async () => {
|
||||
const { service, accounts, outbound, imap, clock } = harness();
|
||||
await accounts.save(account());
|
||||
await outbound.insert(outboundRow({ createdAt: new Date(clock.now().getTime() - 6 * 60_000) }));
|
||||
const result = await service.send({ ...sendInput, idempotencyKey: 'k-1' });
|
||||
expect(result.state).toBe('sent');
|
||||
expect(imap.sent).toHaveLength(1);
|
||||
expect(outbound.rows.get('out-9')?.state).toBe('sent');
|
||||
// 复用,不是新开一行
|
||||
expect(outbound.rows.size).toBe(1);
|
||||
expect(result.outboundId).toBe('out-9');
|
||||
});
|
||||
|
||||
it('失败重发成功之后,那一行不留上一次的错误话', async () => {
|
||||
const { service, accounts, outbound, imap } = harness();
|
||||
await accounts.save(account());
|
||||
imap.failSendWith(new Error('暂时不行'));
|
||||
const first = await service.send({ ...sendInput, idempotencyKey: 'k-1' });
|
||||
expect(first.state).toBe('failed');
|
||||
expect(outbound.rows.get(first.outboundId)?.error).toContain('暂时不行');
|
||||
|
||||
imap.failSendWith(undefined);
|
||||
const second = await service.send({ ...sendInput, idempotencyKey: 'k-1' });
|
||||
expect(second.state).toBe('sent');
|
||||
expect(outbound.rows.get(first.outboundId)?.error).toBeNull();
|
||||
expect((await service.listOutbound({}))[0]?.error).toBeNull();
|
||||
|
||||
// 第三次同键走 resultOf(prior):既说成功又带错误的结果会让调用方按 error 分支判反
|
||||
const third = await service.send({ ...sendInput, idempotencyKey: 'k-1' });
|
||||
expect(third.state).toBe('sent');
|
||||
expect(third.error).toBeUndefined();
|
||||
expect(imap.sent).toHaveLength(1);
|
||||
});
|
||||
});
|
||||
|
||||
describe('卡死的 dispatching', () => {
|
||||
const parsed: ParsedMail = {
|
||||
messageId: '<m-1@acme.com>',
|
||||
references: [],
|
||||
from: { address: 'buyer@acme.com', name: '买家' },
|
||||
to: [{ address: 'sales@we.com', name: '销售' }],
|
||||
cc: [],
|
||||
subject: '询价',
|
||||
text: '请报价',
|
||||
attachments: [],
|
||||
};
|
||||
|
||||
it('自动路径不进(墙),人按的重新投递能接管(路)', async () => {
|
||||
const { service, accounts, inbound, clock } = harness();
|
||||
await accounts.save(account());
|
||||
await inbound.insert(
|
||||
buildInboundRecord({
|
||||
id: 'in-1',
|
||||
accountId: 'acc-1',
|
||||
address: 'sales@we.com',
|
||||
providerRef: 'uid-1',
|
||||
parsed,
|
||||
eml: new Uint8Array(),
|
||||
receivedAt: clock.now(),
|
||||
attachments: [],
|
||||
now: clock.now(),
|
||||
}),
|
||||
);
|
||||
let calls = 0;
|
||||
service.subscribe('channel', async () => {
|
||||
calls += 1;
|
||||
});
|
||||
|
||||
// 进程崩在写 dispatching 与写结果之间:这条留痕永远停在 dispatching
|
||||
const stuck = { mailId: 'in-1', subscriber: 'channel', state: 'dispatching' as const, error: null, finishedAt: null };
|
||||
await inbound.setDelivery({ ...stuck, startedAt: new Date(clock.now().getTime() - 11 * 60_000) });
|
||||
const reclaimed = await service.redeliver('in-1');
|
||||
expect(reclaimed[0]?.state).toBe('done');
|
||||
expect(calls).toBe(1);
|
||||
|
||||
// 新鲜的那条仍是墙:人按也不抢,免得两条线同时改一件事
|
||||
await inbound.setDelivery({ ...stuck, startedAt: clock.now() });
|
||||
const skipped = await service.redeliver('in-1');
|
||||
expect(skipped[0]?.state).toBe('dispatching');
|
||||
expect(calls).toBe(1);
|
||||
});
|
||||
});
|
||||
@@ -342,6 +342,12 @@ export function createOutboundStore(db: Db): OutboundStore {
|
||||
.update(mailOutboundTable)
|
||||
.set({
|
||||
state,
|
||||
/*
|
||||
* **发成功的那一行不留上一次的错误话**:同一行可能先 failed、再被幂等键复用重发成功。
|
||||
* 不清的话,`listOutbound` 会摆出「state: sent 而 error: 上次那句」,
|
||||
* `resultOf` 也会回出一个既说成功又带错误的结果 —— 调用方按 error 分支就判反了。
|
||||
*/
|
||||
...(state === 'sent' ? { error: null } : {}),
|
||||
...(patch.error === undefined ? {} : { error: patch.error }),
|
||||
...(patch.providerRef === undefined ? {} : { providerRef: patch.providerRef }),
|
||||
...(patch.sentAt === undefined ? {} : { sentAt: patch.sentAt }),
|
||||
|
||||
@@ -65,6 +65,20 @@ export type MailServiceDeps = {
|
||||
const DEFAULT_LIST_LIMIT = 50;
|
||||
const MAX_LIST_LIMIT = 200;
|
||||
|
||||
/**
|
||||
* 同键的上一封停在 `sending` 多久之后,算作「上次崩在半路」而不是「正在飞」。
|
||||
* 这两种局面在库里长得一模一样,只能靠时限分开:短于此值一律不发(宁可让调用方再试一次,
|
||||
* 也不能发出第二封真邮件);长于此值就当那一行没人管了,复用它重发。
|
||||
*/
|
||||
const SEND_STALE_MS = 5 * 60_000;
|
||||
|
||||
/**
|
||||
* 一条投递停在 `dispatching` 多久之后,允许**人手**的「重新投递」把它接管过去。
|
||||
* 自动路径永远不重投(那是防双投的墙);但进程若崩在写 dispatching 与写结果之间,
|
||||
* 那条留痕就永远卡着 —— 墙要配路:人按的那一下得有门进。
|
||||
*/
|
||||
const DISPATCH_STALE_MS = 10 * 60_000;
|
||||
|
||||
type PollTally = { accountId: string; pulled: number; stored: number; dispatched: number; more: boolean; error?: string };
|
||||
|
||||
function tallyToOutcome(t: PollTally): MailPollOutcome {
|
||||
@@ -112,7 +126,19 @@ export function createMailService(deps: MailServiceDeps): MailService {
|
||||
/** 明文只在这里打开、交给 provider 那一段;不进任何留痕 */
|
||||
function openSecret(record: Pick<MailAccountRecord, 'id' | 'secretCipher'>): Record<string, string> {
|
||||
if (record.secretCipher === null || record.secretCipher === '') throw new MailSecretMissingError(record.id);
|
||||
const parsed: unknown = JSON.parse(deps.secrets.open(record.secretCipher));
|
||||
/*
|
||||
* 解密与解析都包进来:`JSON.parse` 抛的 SyntaxError **把被解析的原文嵌在 message 里**
|
||||
* (V8:`Unexpected token 'p', "pw-1234" is not valid JSON`),而这一抛发生在调用方把 `secret`
|
||||
* 赋值**之前** —— 脱敏那一层此刻手里是空的、一个字都洗不掉,明文会经 `savePoll({error})`
|
||||
* 落进 `last_poll_error`,再由账号视图上到运营台。
|
||||
* 解不开就是解不开:一律换成「未配置凭证」,原文一个字不外传。
|
||||
*/
|
||||
let parsed: unknown;
|
||||
try {
|
||||
parsed = JSON.parse(deps.secrets.open(record.secretCipher));
|
||||
} catch {
|
||||
throw new MailSecretMissingError(record.id);
|
||||
}
|
||||
if (typeof parsed !== 'object' || parsed === null) throw new MailSecretMissingError(record.id);
|
||||
return Object.fromEntries(Object.entries(parsed as Record<string, unknown>).map(([k, v]) => [k, String(v)]));
|
||||
}
|
||||
@@ -141,14 +167,23 @@ export function createMailService(deps: MailServiceDeps): MailService {
|
||||
|
||||
/* ── 分发 ────────────────────────────────────────────────────────── */
|
||||
|
||||
async function dispatch(record: InboundRecord, ctx: { accountId: string; channelId: string | null }, only?: string): Promise<DeliveryRecord[]> {
|
||||
async function dispatch(
|
||||
record: InboundRecord,
|
||||
ctx: { accountId: string; channelId: string | null },
|
||||
only?: string,
|
||||
opts: { readonly reclaimStale?: boolean } = {},
|
||||
): Promise<DeliveryRecord[]> {
|
||||
const existing = (await deps.inbound.get(record.id))?.deliveries ?? [];
|
||||
const out: DeliveryRecord[] = [];
|
||||
for (const [name, handler] of subscribers) {
|
||||
if (only !== undefined && name !== only) continue;
|
||||
const current = existing.find((d) => d.subscriber === name);
|
||||
// 正在投的那条不重投:进程还没回话,再投一遍就是两条线同时改一件事
|
||||
if (!canRedeliver(current?.state)) {
|
||||
const stale =
|
||||
opts.reclaimStale === true &&
|
||||
current?.state === 'dispatching' &&
|
||||
deps.clock.now().getTime() - current.startedAt.getTime() >= DISPATCH_STALE_MS;
|
||||
if (!canRedeliver(current?.state) && !stale) {
|
||||
if (current !== undefined) out.push(current);
|
||||
continue;
|
||||
}
|
||||
@@ -395,16 +430,35 @@ export function createMailService(deps: MailServiceDeps): MailService {
|
||||
if (hit === undefined) throw new MailInboundNotFoundError(mailId);
|
||||
if (subscriber !== undefined && !subscribers.has(subscriber)) throw new MailSubscriberUnknownError(subscriber);
|
||||
const owner = await deps.accounts.get(hit.record.accountId);
|
||||
const deliveries = await dispatch(hit.record, { accountId: hit.record.accountId, channelId: owner?.channelId ?? null }, subscriber);
|
||||
const deliveries = await dispatch(hit.record, { accountId: hit.record.accountId, channelId: owner?.channelId ?? null }, subscriber, {
|
||||
reclaimStale: true,
|
||||
});
|
||||
deps.notify.changed('inbound');
|
||||
return deliveries.map(deliveryViewOf);
|
||||
},
|
||||
|
||||
async send(input) {
|
||||
const account = pickAccount(await deps.accounts.list(), { accountId: input.accountId, from: input.from });
|
||||
// 幂等键:放行重放不发两封;上一次失败 / 中断的那一行复用,重发
|
||||
// 幂等键:放行重放不发两封
|
||||
const prior = input.idempotencyKey === undefined ? undefined : await deps.outbound.byIdempotencyKey(input.idempotencyKey);
|
||||
if (prior !== undefined && prior.state === 'sent') return resultOf(prior);
|
||||
/*
|
||||
* **同键不发两封**。`sending` 有两种局面而库里长得一样:正在飞 / 上次崩在半路 —— 用时限分开。
|
||||
* 没超时的:不发,如实回「没发出去」并说清缘由(调用方再试一次即可;第一封落地后同键就短路成 sent 了)。
|
||||
* 超时的、以及 `failed` 的:复用那一行重发。
|
||||
*/
|
||||
if (prior !== undefined && prior.state === 'sending') {
|
||||
const ageMs = deps.clock.now().getTime() - prior.createdAt.getTime();
|
||||
if (ageMs < SEND_STALE_MS) {
|
||||
return {
|
||||
outboundId: prior.id,
|
||||
accountId: prior.accountId,
|
||||
messageId: prior.messageId,
|
||||
state: 'failed',
|
||||
error: `同一幂等键的上一封仍在发送中(${Math.round(ageMs / 1000)}s),本次未重复发送`,
|
||||
};
|
||||
}
|
||||
}
|
||||
|
||||
const files: ParsedAttachment[] = [];
|
||||
for (const a of input.attachments) {
|
||||
|
||||
Reference in New Issue
Block a user