diff --git a/specs/mail/sending.md b/specs/mail/sending.md index 62348de..e4d48ab 100644 --- a/specs/mail/sending.md +++ b/specs/mail/sending.md @@ -115,7 +115,13 @@ send that outlives its own lease can be re-claimed and retried by another attempt while the original call is still in flight — a genuine concurrent double-send, not merely a race over which of two callers marks the outcome. Every `EmailSender` used behind these retry paths must therefore bound its -own call time well below this lease (see §4). +own call time well below this lease (see §4). This is enforced mechanically, +not by convention: the `EmailSender` contract requires each implementation to +declare the bound it itself enforces (`maxSendMs`, +`src/providers/email-sender.ts`), and both retry paths assert +`maxSendMs < leaseMs` before claiming a row +(`assertLeaseExceedsSenderBound`, `src/mail/send.ts`) — a violating +lease/timeout combination throws up front, claiming and sending nothing. **Delivery is at-least-once, not at-most-once — and nothing above changes that.** The idempotency key, the envelope snapshot, and the lease all close @@ -188,7 +194,12 @@ too.** §3a's lease only holds "at most one attempt in flight per row" if the provider's `send()` reliably returns well inside the lease window — an adapter whose HTTP call has no timeout (or one comparable to or longer than the lease) can outlive its own claim and collide with a re-claimed retry. -See each adapter's own timeout documentation for its bound. +The contract makes this precondition checkable: every `EmailSender` declares +`maxSendMs` — the bound it really enforces (a mechanical timeout on its own +call, e.g. the Gmail adapter's `timeoutMs` feeding `AbortSignal.timeout`), +not an estimate — and the retry paths refuse to claim under a lease that +does not strictly exceed it (§3a). See each adapter's own timeout +documentation for its bound. ## 5. Scope diff --git a/src/api/index.test.ts b/src/api/index.test.ts index 910112b..7f3744a 100644 --- a/src/api/index.test.ts +++ b/src/api/index.test.ts @@ -21,6 +21,7 @@ function createFakeSender(): { sender: EmailSender; sent: OutboundEmail[] } { const sent: OutboundEmail[] = [] return { sender: { + maxSendMs: 30_000, async send(email) { sent.push(email) return {} @@ -33,6 +34,7 @@ function createFakeSender(): { sender: EmailSender; sent: OutboundEmail[] } { /** An `EmailSender` that always rejects — for exercising the `502 send_failed` path. */ function createThrowingSender(): EmailSender { return { + maxSendMs: 30_000, async send() { throw new Error('provider rejected the message (must never leak to the client)') }, diff --git a/src/mail/delivery-worker.test.ts b/src/mail/delivery-worker.test.ts index bbabba1..77d0686 100644 --- a/src/mail/delivery-worker.test.ts +++ b/src/mail/delivery-worker.test.ts @@ -22,6 +22,7 @@ function fakeSender(): EmailSender & { sent: OutboundEmail[] } { const sent: OutboundEmail[] = [] return { sent, + maxSendMs: 30_000, async send(email) { sent.push(email) return { providerMessageId: 'provider-1' } @@ -268,6 +269,7 @@ describe('runDeliveryWorker', () => { const errorSpy = vi.spyOn(console, 'error').mockImplementation(() => {}) const sender: EmailSender = { + maxSendMs: 30_000, async send() { throw new Error('still down') }, @@ -304,6 +306,7 @@ describe('runDeliveryWorker', () => { const seedDeps: SendReplyDeps = { store, sender: { + maxSendMs: 30_000, async send() { throw new Error('boom') }, @@ -333,6 +336,7 @@ describe('runDeliveryWorker', () => { }) let workerSendCalls = 0 const workerSender: EmailSender = { + maxSendMs: 30_000, async send() { workerSendCalls++ await workerSendGate @@ -363,4 +367,21 @@ describe('runDeliveryWorker', () => { expect(thread?.deliveryStatus).toBe('sent') expect(thread?.messageId).toBe(seeded.messageId) }) + + it("a leaseMs that does not strictly exceed the sender's maxSendMs throws up front — before anything is listed or claimed", async () => { + const { store } = await freshStore() + const listSpy = vi.spyOn(store, 'listDeliverableThreads') + const claimSpy = vi.spyOn(store, 'claimThreadForDelivery') + const sender = fakeSender() // maxSendMs: 30_000 + + // Equality is a violation too — the lease must STRICTLY exceed the + // sender's enforced bound (specs/mail/sending.md §3a). + await expect( + runDeliveryWorker({ store, sender }, { leaseMs: sender.maxSendMs }), + ).rejects.toThrow(/must strictly exceed/) + + expect(listSpy).not.toHaveBeenCalled() + expect(claimSpy).not.toHaveBeenCalled() + expect(sender.sent).toHaveLength(0) + }) }) diff --git a/src/mail/delivery-worker.ts b/src/mail/delivery-worker.ts index 24f45a5..4f7ce04 100644 --- a/src/mail/delivery-worker.ts +++ b/src/mail/delivery-worker.ts @@ -40,7 +40,11 @@ import type { EmailSender } from '../providers/index.js' import type { ConversationStore } from '../store/conversations.js' -import { attemptDeliveryOfClaimedThread, DEFAULT_LEASE_MS } from './send.js' +import { + assertLeaseExceedsSenderBound, + attemptDeliveryOfClaimedThread, + DEFAULT_LEASE_MS, +} from './send.js' /** Default age a `'pending'` row must reach before this worker considers it stuck rather than merely in flight. */ const DEFAULT_STALE_AFTER_MS = 5 * 60_000 @@ -58,7 +62,13 @@ export interface DeliveryWorkerDeps { export interface DeliveryWorkerOptions { /** How old a `'pending'` row must be before it's a retry candidate (default {@link DEFAULT_STALE_AFTER_MS}). */ staleAfterMs?: number - /** Lease duration held while a candidate is being attempted (default {@link DEFAULT_LEASE_MS}, shared with `sendReply`'s own retry-claim). */ + /** + * Lease duration held while a candidate is being attempted (default + * {@link DEFAULT_LEASE_MS}, shared with `sendReply`'s own retry-claim). + * Must strictly exceed the sender's enforced per-`send()` bound + * (`EmailSender.maxSendMs`) — asserted up front, before anything is + * claimed (see `assertLeaseExceedsSenderBound`, `src/mail/send.ts`). + */ leaseMs?: number /** Hard cap on rows attempted in this one call (default {@link DEFAULT_BATCH_SIZE}). */ batchSize?: number @@ -91,6 +101,10 @@ export async function runDeliveryWorker( const leaseMs = options?.leaseMs ?? DEFAULT_LEASE_MS const batchSize = options?.batchSize ?? DEFAULT_BATCH_SIZE + // Fail loudly on a lease/sender-timeout misconfiguration BEFORE listing or + // claiming anything — see this helper's doc comment in send.ts. + assertLeaseExceedsSenderBound(deps.sender, leaseMs) + const candidates = await deps.store.listDeliverableThreads({ staleAfterMs, batchSize }) let sent = 0 diff --git a/src/mail/send.test.ts b/src/mail/send.test.ts index fd00fcb..1517488 100644 --- a/src/mail/send.test.ts +++ b/src/mail/send.test.ts @@ -5,7 +5,7 @@ import type { EmailSender, OutboundEmail } from '../providers/index.js' import { type ConversationStore, createConversationStore } from '../store/conversations.js' import type { ParsedEmail } from './parse.js' import type { Keyring, SigningKey } from './reply-token.js' -import { type SendReplyDeps, sendReply } from './send.js' +import { DEFAULT_LEASE_MS, type SendReplyDeps, sendReply } from './send.js' import { decideThreading } from './thread.js' // --- fixtures ---------------------------------------------------------------- @@ -21,6 +21,7 @@ function fakeSender(): EmailSender & { sent: OutboundEmail[] } { const sent: OutboundEmail[] = [] return { sent, + maxSendMs: 30_000, async send(email) { sent.push(email) return { providerMessageId: 'provider-1' } @@ -31,6 +32,7 @@ function fakeSender(): EmailSender & { sent: OutboundEmail[] } { /** Always throws — simulates a provider transport failure. */ function failingSender(): EmailSender { return { + maxSendMs: 30_000, async send() { throw new Error('boom: provider unreachable') }, @@ -475,6 +477,7 @@ describe('sendReply idempotency (HT-16)', () => { }) let sendCallCount = 0 const sender: EmailSender = { + maxSendMs: 30_000, async send() { sendCallCount++ await sendGate @@ -563,3 +566,72 @@ describe('sendReply idempotency (HT-16)', () => { expect(sender.sent).toHaveLength(0) // never re-sent — already delivered }) }) + +describe('lease / sender-bound coupling', () => { + let db: Db | undefined + + afterEach(async () => { + await db?.close() + db = undefined + }) + + async function freshStore(): Promise<{ db: Db; store: ConversationStore }> { + db = await createPgliteDb() + await migrate(db) + return { db, store: createConversationStore(db) } + } + + async function seedConversation(store: ConversationStore) { + return store.createConversation({ + subject: 'Help with my order', + customerEmail: 'customer@example.test', + firstMessage: { + direction: 'inbound', + messageId: '', + fromAddress: 'customer@example.test', + bodyText: 'Where is my order?', + }, + }) + } + + const input = (conversationId: string) => ({ + conversationId, + from: 'support@example.test', + to: ['customer@example.test'], + subject: 'Re: Help with my order', + text: "We're looking into it!", + }) + + it('keyed path: a sender whose maxSendMs does not stay strictly below the lease throws BEFORE claiming or sending', async () => { + const { store } = await freshStore() + const { conversationId } = await seedConversation(store) + // Equality is deliberately a violation too — the lease must STRICTLY + // exceed the bound (specs/mail/sending.md §3a). + const sender = { ...fakeSender(), maxSendMs: DEFAULT_LEASE_MS } + const sendSpy = vi.spyOn(sender, 'send') + const claimSpy = vi.spyOn(store, 'claimThreadForDelivery') + const deps: SendReplyDeps = { store, sender, keyring, mailDomain } + + await expect( + sendReply({ ...input(conversationId), idempotencyKey: 'k-violating' }, deps), + ).rejects.toThrow(/must strictly exceed/) + + expect(claimSpy).not.toHaveBeenCalled() + expect(sendSpy).not.toHaveBeenCalled() + }) + + it('no-key path: the assertion does not apply — there is no lease to violate', async () => { + // A fresh no-key send never claims a lease, so a sender whose bound + // exceeds DEFAULT_LEASE_MS is not a misconfiguration ON THIS PATH; the + // retry paths (keyed claim above, worker sweep — see + // delivery-worker.test.ts) are where the invariant is enforced. + const { store } = await freshStore() + const { conversationId } = await seedConversation(store) + const sender = { ...fakeSender(), maxSendMs: DEFAULT_LEASE_MS * 2 } + const deps: SendReplyDeps = { store, sender, keyring, mailDomain } + + const result = await sendReply(input(conversationId), deps) + expect(result).toMatchObject({ ok: true, delivery: 'sent' }) + expect(sender.sent).toHaveLength(1) + }) +}) diff --git a/src/mail/send.ts b/src/mail/send.ts index 72859d5..5067418 100644 --- a/src/mail/send.ts +++ b/src/mail/send.ts @@ -132,12 +132,43 @@ import { type Keyring, mintReplyMessageId } from './reply-token.js' * measured worst case, because none has been measured here. Any * `EmailSender` used behind these retry paths (§4) MUST bound its own * `send()` call well below this lease — via its own request timeout — so - * this margin is never actually spent. Raising this constant without also - * checking every adapter's timeout against it re-opens the hole it exists - * to close. + * this margin is never actually spent. + * + * This relationship is enforced mechanically, not by convention: every + * `EmailSender` declares the bound it enforces (`maxSendMs`, + * `src/providers/email-sender.ts`), and both retry paths assert + * `maxSendMs < leaseMs` via {@link assertLeaseExceedsSenderBound} before + * claiming a row. Changing this constant, an adapter's timeout, or a + * worker's `leaseMs` option into a violating combination therefore throws + * at the call site instead of silently re-opening the hole. */ export const DEFAULT_LEASE_MS = 120_000 +/** + * Assert the invariant {@link DEFAULT_LEASE_MS}'s doc comment exists to + * hold: the delivery lease strictly exceeds the sender's own enforced + * per-`send()` bound (`EmailSender.maxSendMs`). Called by both retry paths + * — `sendReply`'s keyed claim and `runDeliveryWorker`'s sweep + * (`src/mail/delivery-worker.ts`) — BEFORE any row is claimed, so a + * violating configuration fails loudly up front: nothing is claimed, + * nothing is sent, and the throw names both numbers. + * + * A violation is a wiring bug (an adapter timeout raised to/past the lease, + * or a lease tuned down below an adapter's timeout), never an expected + * runtime outcome — hence a throw, not a discriminated result, matching + * `sendReply`'s "only throw on genuinely unexpected faults" contract. + */ +export function assertLeaseExceedsSenderBound(sender: EmailSender, leaseMs: number): void { + if (!(sender.maxSendMs < leaseMs)) { + throw new Error( + `delivery lease (${leaseMs}ms) must strictly exceed the sender's enforced send() bound ` + + `(maxSendMs: ${sender.maxSendMs}ms), or a re-claimed retry can race a still-in-flight ` + + `send into a concurrent double-send (specs/mail/sending.md §3a) — ` + + `raise the lease or lower the sender's timeout`, + ) + } +} + /** Dependencies `sendReply` needs, injected so it stays testable against fakes/in-memory stores. */ export interface SendReplyDeps { store: ConversationStore @@ -293,6 +324,7 @@ export async function sendReply( // pre-existing from an earlier attempt, both converge here: claim the // delivery lease before sending, so a concurrent duplicate call (same key) // or the delivery worker cannot also be sending this row right now. + assertLeaseExceedsSenderBound(sender, DEFAULT_LEASE_MS) const claimed = await store.claimThreadForDelivery(thread.id, DEFAULT_LEASE_MS) if (claimed === null) { // The claim can fail for two different reasons, and conflating them diff --git a/src/providers/adapters/gmail/sender.test.ts b/src/providers/adapters/gmail/sender.test.ts index 2a7a286..4885392 100644 --- a/src/providers/adapters/gmail/sender.test.ts +++ b/src/providers/adapters/gmail/sender.test.ts @@ -54,6 +54,22 @@ describe('createGmailEmailSender', () => { await expect(sender.send(email)).rejects.toThrow(/timeout|timed out|aborted/i) }) + it('declares maxSendMs equal to the timeout it actually enforces (default and custom)', () => { + // `maxSendMs` is what the engine's retry paths assert against the + // delivery lease (see EmailSender.maxSendMs's doc) — it must be the SAME + // number as the AbortSignal.timeout bound, or the assertion checks a + // fiction. The abort test above proves timeoutMs is really enforced; + // this pins the declaration to it. + const { fetchImpl } = fakeFetch(200, { id: 'gmail-123' }) + const getAccessToken = async () => 'token' + + const defaulted = createGmailEmailSender({ getAccessToken, fetchImpl }) + expect(defaulted.maxSendMs).toBe(30_000) + + const custom = createGmailEmailSender({ getAccessToken, fetchImpl, timeoutMs: 5_000 }) + expect(custom.maxSendMs).toBe(5_000) + }) + it('happy path: POSTs the encoded raw MIME to the send endpoint and returns providerMessageId', async () => { const { fetchImpl, calls } = fakeFetch(200, { id: 'gmail-123' }) const getAccessToken = vi.fn(async () => 'token-abc-123') diff --git a/src/providers/adapters/gmail/sender.ts b/src/providers/adapters/gmail/sender.ts index d011b56..f2665a5 100644 --- a/src/providers/adapters/gmail/sender.ts +++ b/src/providers/adapters/gmail/sender.ts @@ -71,6 +71,12 @@ export interface GmailEmailSenderOptions { * delivered-but-reported-failed window every network sender has; the * HT-16 idempotency work is where retry-safety lands). This is the safe * direction: never report a delivery that can't be confirmed. + * + * This value is also declared as the adapter's `EmailSender.maxSendMs`, + * which the engine's retry paths assert is strictly below the delivery + * lease (`DEFAULT_LEASE_MS`, `src/mail/send.ts`) before claiming a row — + * so raising it to or past the lease makes those paths throw rather than + * silently risk a concurrent double-send. */ timeoutMs?: number } @@ -117,6 +123,11 @@ export function createGmailEmailSender(options: GmailEmailSenderOptions): EmailS const endpoint = `${GMAIL_API_BASE}/users/${encodeURIComponent(userId)}/messages/send` return { + // The same value that bounds the HTTP call below (`AbortSignal.timeout`) + // — one variable, so the declared bound and the enforced one cannot + // drift apart. See `EmailSender.maxSendMs`'s doc for what consumes this. + maxSendMs: timeoutMs, + async send(email: OutboundEmail): Promise { const raw = buildRawMessage(email) const encoded = Buffer.from(raw, 'utf8').toString('base64url') diff --git a/src/providers/email-sender.ts b/src/providers/email-sender.ts index a39faa1..492a81f 100644 --- a/src/providers/email-sender.ts +++ b/src/providers/email-sender.ts @@ -74,12 +74,32 @@ export interface EmailSendResult { * `Message-ID` contract every implementation must uphold. */ export interface EmailSender { + /** + * The upper bound, in milliseconds, that this implementation ITSELF + * enforces on one `send()` call — a real, mechanical timeout (e.g. an + * `AbortSignal.timeout` on the underlying HTTP request), not an estimate + * or an aspiration. Every `send()` call MUST settle (resolve or reject) + * within this many milliseconds. + * + * Why the contract carries this: the delivery lease (`DEFAULT_LEASE_MS`, + * `src/mail/send.ts`) must strictly exceed the worst-case `send()` + * duration, or a re-claimed retry can race a still-in-flight send into a + * concurrent double-send (specs/mail/sending.md §3a, §4). The engine's + * retry paths assert `maxSendMs < leaseMs` before claiming a row, so an + * adapter whose bound is missing or too large fails loudly at the call + * site instead of silently re-opening that hole. Declaring a value the + * implementation does not actually enforce defeats the check — set it + * from the same variable that configures the real timeout (see the Gmail + * adapter's `timeoutMs`). + */ + readonly maxSendMs: number + /** * Send `email`. Resolves with an {@link EmailSendResult} on success. * Rejects (throws) on any failure to hand the message to the provider — * `src/mail/send.ts`'s `sendReply` treats a rejection as a delivery * failure and marks the outbound thread `'failed'` accordingly - * (specs/mail/sending.md §3). + * (specs/mail/sending.md §3). Must settle within {@link maxSendMs}. */ send(email: OutboundEmail): Promise }