diff --git a/docs/contributing/architecture/data-storage.md b/docs/contributing/architecture/data-storage.md index 7cb85b38e3..fc949b89f4 100644 --- a/docs/contributing/architecture/data-storage.md +++ b/docs/contributing/architecture/data-storage.md @@ -305,13 +305,28 @@ validated derived output or a generated fallback. Raw email MIME payloads live in the `EMAIL_BLOBS` R2 bucket instead of D1. `email_messages` stores an object key in `raw_mime_key` -(`email-raw:v1:{userId}/{messageId}`), and the legacy inline `raw_mime` column -is kept only for residual rows that have not yet been swept (or when R2 put -fails at write time — the write path falls back to inline storage rather than -losing mail, and never throws from that decision). +(`email-raw:v1:{userId}/{messageId}`). **Durability policy (Stage 4a):** R2 is +required for inbound MIME — `insertEmailMessage` puts the payload to +`EMAIL_BLOBS` before the D1 insert and never writes new inline `raw_mime` rows. +On R2 put failure the insert throws `EmailRawMimeStorageError` (a +`RetryableInboundStorageError`; no D1 row). The inbound Worker refunds the daily +receive charge and rethrows only typed pre-commit failures so Cloudflare Email +Routing retries without burning quota. The durable commit boundary is message + +attachment rows: thread prework, R2 put, and D1 message/attachment storage are +pre-commit; `touchEmailThread` / `received` delivery-event writes are +post-commit and are logged without throwing (retry would duplicate mail). If +attachment insert fails but message cleanup cannot remove the row — or the +residual-row probe itself fails (ambiguous commit state) — the handler +acknowledges the already-created message (logged, non-retry) rather than risking +a duplicate. Outbound messages pass `rawMime: null` and are unaffected. If D1 +insert fails after a successful put, the blob is best-effort deleted. Stage 4a +only prevents **new** inline writes — it does not claim residual inline rows are +gone. Dual-read `loadRawMime` and the maintenance/deploy offload sweep stay +until a later column-drop stage, and that stage must wait until deploy logs +verify `remainingInline = 0` (not merely that Stage 4a shipped). - All reads go through `loadRawMime` in `packages/worker/src/email/repo.ts`, - which prefers the inline payload and otherwise fetches the blob by key. + which prefers residual inline payload and otherwise fetches the blob by key. Attachment content extraction re-parses the resolved MIME the same way as before. - Message deletes claim rows first by setting transitional @@ -378,8 +393,10 @@ losing mail, and never throws from that decision). is HTTP 500. Deploy loops until `complete === true` (it does not treat total `remainingInline === 0` as a shortcut) and fails if the secret is missing, on non-2xx / nonzero `failed`, or if `complete` is still false after the attempt - cap. The write-time inline fallback policy and `loadRawMime` dual-read path - stay in place until a later column-drop stage. + cap. Write-time inline fallback is gone (Stage 4a only stops new inline + writes). `loadRawMime` dual-read and this sweep remain; do not treat total + inline as zero until deploy logs verify `remainingInline = 0`, which is the + gate for a later column-drop stage. - Bucket names: `kody-email-blobs` (production), per-preview `{worker}-email-blobs` buckets created and cleaned up by `tools/ci/preview-resources.ts`, and the test env reuses the preview-style diff --git a/docs/contributing/architecture/entitlements.md b/docs/contributing/architecture/entitlements.md index 3457c8108c..e17eeab9c1 100644 --- a/docs/contributing/architecture/entitlements.md +++ b/docs/contributing/architecture/entitlements.md @@ -153,7 +153,13 @@ Rules: a plan so counters reflect real usage the moment a plan is assigned — unless the caller passes `fallbackLimit`, which caps plan-less users with a deployment-level backstop (both email sends and receives do this). Counting - attempts rather than successes keeps the limit abuse-resistant. + attempts rather than successes keeps the limit abuse-resistant for permanent + rejects (parse failures, entitlement/quota rejects). Only typed pre-commit + `RetryableInboundStorageError` failures (thread prework, R2 put, D1 + message/attachment storage after successful cleanup) refund exactly one + `email_receives_per_day` unit via `refundDailyEntitlement` for the same UTC + day that was charged, so Cloudflare Email Routing retries do not burn the + daily receive quota. Post-commit bookkeeping failures do not refund or retry. `incrementDailyEntitlementCounter` remains for raw counter writes (tests, backfills). - **Boolean allowances** (persistent package services) are modeled as limit `0` @@ -245,7 +251,7 @@ The exemplar is job scheduling: `createJob` in | `persistent_package_services` | `service_start` for services declared `mode: 'persistent'` | | `repo_sessions` | `repo_open_session` before creating a new session | | `email_sends_per_day` | `sendOutboundEmail` (atomic `consumeDailyEntitlement`, NULL-plan backstop) | -| `email_receives_per_day` | `handleInboundEmail` (atomic `consumeDailyEntitlement`, NULL-plan fallback) | +| `email_receives_per_day` | `handleInboundEmail` (atomic `consumeDailyEntitlement`, NULL-plan fallback; refund only on `RetryableInboundStorageError`) | | `stored_email_messages` | `handleInboundEmail` before storage (NULL-plan fallback) | | `email_message_bytes` | `handleInboundEmail` before quota/parse (per-message raw size, NULL-plan fallback) | | `secrets` | new-entry branch of `saveSecret` in `packages/worker/src/mcp/secrets/service.ts` | diff --git a/docs/use/email-primitives.md b/docs/use/email-primitives.md index 682c0da8a4..640857653b 100644 --- a/docs/use/email-primitives.md +++ b/docs/use/email-primitives.md @@ -75,7 +75,10 @@ Inbound storage is quota-gated per user: with a generic "over quota" response to the sender, and the detailed reason is recorded as a `rejected` delivery event. Oversize mail is rejected before it consumes any daily receive quota, and mail to unverified accounts (which can - never receive) is rejected without consuming any quota at all. + never receive) is rejected without consuming any quota at all. Transient + storage failures (for example an R2 outage while saving raw MIME) do not keep + the daily receive charge — the attempt is refunded so delivery retries are not + blocked by quota. - Plan users get their plan's limits; users without a plan get conservative deployment fallbacks (they are not unlimited for inbound mail). - Quota, size, and unverified-account rejections store at most five detailed diff --git a/packages/worker/src/email/inbound.ts b/packages/worker/src/email/inbound.ts index 333643bf46..999921e43a 100644 --- a/packages/worker/src/email/inbound.ts +++ b/packages/worker/src/email/inbound.ts @@ -12,6 +12,7 @@ import { assertWithinStorageBytesEntitlement, consumeDailyEntitlement, estimateEntitlementStorageEntryBytes, + refundDailyEntitlement, } from '#worker/entitlements/service.ts' import { recordUsage } from '#worker/usage/record-usage.ts' import { @@ -34,6 +35,7 @@ import { insertEmailDeliveryEvent, insertEmailMessageWithAttachments, recordBoundedEmailRejectionEvent, + RetryableInboundStorageError, touchEmailThread, } from './repo.ts' import { @@ -45,11 +47,27 @@ import { countStoredSystemEmailMessages, ensureSystemEmailInbox, isSystemEmailLocal, + refundSystemEmailDailyReceive, systemEmailLimits, systemEmailOwnerId, type SystemEmailLocal, } from './system-email.ts' +/** + * Best-effort refund of a pre-parse daily receive charge after a retryable + * storage failure. Logs refund failures but never masks the original error. + */ +async function refundReceiveQuotaAfterStorageFailure(input: { + refund: () => Promise + logLabel: string +}): Promise { + try { + await input.refund() + } catch (refundError) { + console.error(input.logLabel, refundError) + } +} + type ParsedInboundEmail = Awaited< ReturnType > @@ -104,82 +122,104 @@ async function parseAndStoreInboundEmail(input: { } const now = new Date().toISOString() const subjectNormalized = normalizeSubject(parsed.subject) - const existingThread = await findEmailThreadForInboundMessage({ - db: input.db, - userId: input.userId, - inboxId: input.inboxId, - references: parsed.references, - inReplyToHeader: parsed.inReplyTo, - }) - const thread = - existingThread ?? - (await createEmailThread({ + // Durable commit boundary: thread prework + message/attachment storage are + // pre-commit (RetryableInboundStorageError → refund + Email Routing retry). + // touchEmailThread / received delivery-event writes are post-commit and must + // not throw after the message row is durable (retry would duplicate mail). + let thread + let stored + try { + const existingThread = await findEmailThreadForInboundMessage({ db: input.db, userId: input.userId, inboxId: input.inboxId, - subjectNormalized, - rootMessageIdHeader: parsed.messageId, + references: parsed.references, + inReplyToHeader: parsed.inReplyTo, + }) + thread = + existingThread ?? + (await createEmailThread({ + db: input.db, + userId: input.userId, + inboxId: input.inboxId, + subjectNormalized, + rootMessageIdHeader: parsed.messageId, + lastMessageAt: now, + })) + stored = await insertEmailMessageWithAttachments({ + db: input.db, + blobs: input.blobs, + message: { + direction: 'inbound', + userId: input.userId, + inboxId: input.inboxId, + threadId: thread.id, + senderIdentityId: null, + fromAddress: parsed.headerFrom, + envelopeFrom: parsed.envelopeFrom, + toAddresses: parsed.to.map((entry) => entry.address), + ccAddresses: parsed.cc.map((entry) => entry.address), + bccAddresses: parsed.bcc.map((entry) => entry.address), + replyToAddresses: parsed.replyTo.map((entry) => entry.address), + subject: parsed.subject, + messageIdHeader: parsed.messageId, + inReplyToHeader: parsed.inReplyTo, + references: parsed.references, + headers: parsed.headers, + authResults: parsed.authResults, + textBody: parsed.textBody, + htmlBody: parsed.htmlBody, + rawMime: parsed.rawMime, + rawSize: parsed.rawSize, + processingStatus: 'stored', + providerMessageId: null, + error: null, + receivedAt: now, + sentAt: null, + }, + attachments: parsed.attachments.map((attachment) => ({ + filename: attachment.filename, + contentType: attachment.contentType, + contentId: attachment.contentId, + disposition: attachment.disposition, + size: attachment.size, + storageKind: 'raw-mime', + storageKey: null, + })), + }) + } catch (error) { + if (error instanceof RetryableInboundStorageError) throw error + throw new RetryableInboundStorageError( + 'Failed to store inbound email before durable commit; delivery should be retried.', + error, + ) + } + try { + await touchEmailThread({ + db: input.db, + threadId: thread.id, lastMessageAt: now, - })) - const stored = await insertEmailMessageWithAttachments({ - db: input.db, - blobs: input.blobs, - message: { - direction: 'inbound', + }) + await insertEmailDeliveryEvent({ + db: input.db, + messageId: stored.id, userId: input.userId, inboxId: input.inboxId, - threadId: thread.id, - senderIdentityId: null, - fromAddress: parsed.headerFrom, - envelopeFrom: parsed.envelopeFrom, - toAddresses: parsed.to.map((entry) => entry.address), - ccAddresses: parsed.cc.map((entry) => entry.address), - bccAddresses: parsed.bcc.map((entry) => entry.address), - replyToAddresses: parsed.replyTo.map((entry) => entry.address), - subject: parsed.subject, - messageIdHeader: parsed.messageId, - inReplyToHeader: parsed.inReplyTo, - references: parsed.references, - headers: parsed.headers, - authResults: parsed.authResults, - textBody: parsed.textBody, - htmlBody: parsed.htmlBody, - rawMime: parsed.rawMime, - rawSize: parsed.rawSize, - processingStatus: 'stored', - providerMessageId: null, - error: null, - receivedAt: now, - sentAt: null, - }, - attachments: parsed.attachments.map((attachment) => ({ - filename: attachment.filename, - contentType: attachment.contentType, - contentId: attachment.contentId, - disposition: attachment.disposition, - size: attachment.size, - storageKind: 'raw-mime', - storageKey: null, - })), - }) - await touchEmailThread({ - db: input.db, - threadId: thread.id, - lastMessageAt: now, - }) - await insertEmailDeliveryEvent({ - db: input.db, - messageId: stored.id, - userId: input.userId, - inboxId: input.inboxId, - eventType: 'received', - provider: 'cloudflare-email-routing', - detail: { - recipient: input.recipient, - envelope_from: parsed.envelopeFrom, - from_address: parsed.headerFrom, - }, - }) + eventType: 'received', + provider: 'cloudflare-email-routing', + detail: { + recipient: input.recipient, + envelope_from: parsed.envelopeFrom, + from_address: parsed.headerFrom, + }, + }) + } catch (error) { + console.error( + 'inbound-email-post-commit-bookkeeping-failed', + stored.id, + error, + ) + } return stored } @@ -338,6 +378,8 @@ export async function handleInboundEmail( account.plan, 'email_message_bytes', ) + // Captured at consume time so a storage-failure refund uses the same UTC day. + let receiveQuotaNow: Date | null = null try { // Size first: an oversize message is rejected without consuming any // of the owner's daily receive quota (griefing resistance) and @@ -357,12 +399,14 @@ export async function handleInboundEmail( email: account.email, requested: estimateInboundEmailStorageBytes({ message, recipient }), }) + receiveQuotaNow = new Date() await consumeDailyEntitlement({ db: env.APP_DB, userId, email: account.email, resource: 'email_receives_per_day', fallbackLimit: nullPlanEmailFallbackLimits.email_receives_per_day, + now: receiveQuotaNow, }) // Check-then-insert: a concurrent burst can overshoot the stored cap // by a few rows, which is the documented row-count-limit trade-off @@ -397,34 +441,55 @@ export async function handleInboundEmail( return } - const stored = await parseAndStoreInboundEmail({ - db: env.APP_DB, - blobs: env.EMAIL_BLOBS, - message, - recipient, - userId, - inboxId: inbox.id, - maxMessageBytes: maxMessageBytes ?? maxInlineRawMimeBytes, - async onParseRejected(reason) { - // Parse failures keep one event per attempt: unlike quota/size - // rejections they are bounded by the daily receive quota (the - // counter was already consumed above), and the per-attempt detail - // is useful for the owner to debug a misbehaving sender. - await insertEmailDeliveryEvent({ - db: env.APP_DB, - userId, - inboxId: inbox.id, - eventType: 'rejected', - provider: 'cloudflare-email-routing', - detail: { - recipient, - reason, - phase: 'parse', - }, - }).catch(warnRejectionAuditWriteFailed) - await recordReceiveUsage({ outcome: 'error' }) - }, - }) + let stored + try { + stored = await parseAndStoreInboundEmail({ + db: env.APP_DB, + blobs: env.EMAIL_BLOBS, + message, + recipient, + userId, + inboxId: inbox.id, + maxMessageBytes: maxMessageBytes ?? maxInlineRawMimeBytes, + async onParseRejected(reason) { + // Parse failures keep one event per attempt: unlike quota/size + // rejections they are bounded by the daily receive quota (the + // counter was already consumed above), and the per-attempt detail + // is useful for the owner to debug a misbehaving sender. + await insertEmailDeliveryEvent({ + db: env.APP_DB, + userId, + inboxId: inbox.id, + eventType: 'rejected', + provider: 'cloudflare-email-routing', + detail: { + recipient, + reason, + phase: 'parse', + }, + }).catch(warnRejectionAuditWriteFailed) + await recordReceiveUsage({ outcome: 'error' }) + }, + }) + } catch (error) { + // Only typed pre-commit failures refund + retry. Post-commit + // bookkeeping is swallowed inside parseAndStoreInboundEmail. + if (!(error instanceof RetryableInboundStorageError)) throw error + const chargedAt = receiveQuotaNow + if (chargedAt) { + await refundReceiveQuotaAfterStorageFailure({ + logLabel: 'email-receives-daily-refund-failed', + refund: () => + refundDailyEntitlement({ + db: env.APP_DB, + userId, + resource: 'email_receives_per_day', + now: chargedAt, + }), + }) + } + throw error + } if (!stored) return await recordReceiveUsage({ entityId: stored.id, outcome: 'success' }) const dispatchPromise = dispatchInboundEmailSubscriptionEvents({ @@ -495,9 +560,12 @@ async function handleSystemInboundEmail(input: { return } + // Same-day key for consume + possible storage-failure refund. + const receiveQuotaNow = new Date() const receivesToday = await consumeSystemEmailDailyReceive({ db: input.env.APP_DB, localPart: input.localPart, + now: receiveQuotaNow, }) if (receivesToday === null) { input.message.setReject('Recipient mailbox is over quota.') @@ -530,30 +598,45 @@ async function handleSystemInboundEmail(input: { return } - const stored = await parseAndStoreInboundEmail({ - db: input.env.APP_DB, - blobs: input.env.EMAIL_BLOBS, - message: input.message, - recipient: input.recipient, - userId: systemEmailOwnerId, - inboxId: inbox.id, - maxMessageBytes: systemEmailLimits.maxMessageBytes, - async onParseRejected(reason) { - await insertEmailDeliveryEvent({ - db: input.env.APP_DB, - userId: systemEmailOwnerId, - inboxId: inbox.id, - eventType: 'rejected', - provider: 'cloudflare-email-routing', - detail: { - recipient: input.recipient, - reason, - phase: 'parse', - }, - }).catch(warnRejectionAuditWriteFailed) - await recordReceiveUsage({ outcome: 'error' }) - }, - }) + let stored + try { + stored = await parseAndStoreInboundEmail({ + db: input.env.APP_DB, + blobs: input.env.EMAIL_BLOBS, + message: input.message, + recipient: input.recipient, + userId: systemEmailOwnerId, + inboxId: inbox.id, + maxMessageBytes: systemEmailLimits.maxMessageBytes, + async onParseRejected(reason) { + await insertEmailDeliveryEvent({ + db: input.env.APP_DB, + userId: systemEmailOwnerId, + inboxId: inbox.id, + eventType: 'rejected', + provider: 'cloudflare-email-routing', + detail: { + recipient: input.recipient, + reason, + phase: 'parse', + }, + }).catch(warnRejectionAuditWriteFailed) + await recordReceiveUsage({ outcome: 'error' }) + }, + }) + } catch (error) { + if (!(error instanceof RetryableInboundStorageError)) throw error + await refundReceiveQuotaAfterStorageFailure({ + logLabel: 'system-email-receives-daily-refund-failed', + refund: () => + refundSystemEmailDailyReceive({ + db: input.env.APP_DB, + localPart: input.localPart, + now: receiveQuotaNow, + }), + }) + throw error + } if (!stored) return await recordReceiveUsage({ entityId: stored.id, outcome: 'success' }) // System mail fans out to packages saved by admin users on the dedicated diff --git a/packages/worker/src/email/inbound.workers.test.ts b/packages/worker/src/email/inbound.workers.test.ts index dfdc5f63e3..2e0b72520f 100644 --- a/packages/worker/src/email/inbound.workers.test.ts +++ b/packages/worker/src/email/inbound.workers.test.ts @@ -6,7 +6,9 @@ import { createEmailThread, deleteEmailMessageById, emailRawMimeKey, + EmailRawMimeStorageError, getEmailMessageById, + insertEmailMessage, insertEmailMessageWithAttachments, getEmailAttachmentById, listEmailInboxesForUser, @@ -14,11 +16,16 @@ import { listEmailMessages, listEmailAttachmentsForMessage, loadRawMime, + RetryableInboundStorageError, } from './repo.ts' import { createForwardableEmailMessage } from './test-fixtures.ts' import { ensureEmailTestSchema } from './test-schema.ts' import { ensureUsageRollupsTestSchema } from '#worker/usage/test-schema.ts' import { buildPublishedSourceManifestSnapshotKvKey } from '#worker/package-runtime/published-runtime-artifacts.ts' +import { + consoleError, + silenceExpectedConsoleErrors, +} from '#worker/test-support/console-spies.ts' import { silenceIncidentalRuntimeWarnings } from '#worker/test-support/incidental-runtime-warnings.ts' import { createStableUserIdFromEmail } from '#worker/user-id.ts' @@ -735,10 +742,102 @@ test('inbound email offloads raw MIME to R2 and readers and deletes follow the b expect(await env.EMAIL_BLOBS.get(stored.rawMimeKey!)).toBeNull() }) -test('raw MIME stays inline when the R2 put fails', async () => { +async function readUserDailyReceiveCount(userId: string) { + const row = await env.APP_DB.prepare( + `SELECT count FROM entitlement_daily_counters + WHERE user_id = ? AND resource = 'email_receives_per_day' AND day = ?`, + ) + .bind(userId, new Date().toISOString().slice(0, 10)) + .first<{ count: number }>() + return Number(row?.count ?? 0) +} + +function createFailingEmailBlobs() { + return new Proxy(env.EMAIL_BLOBS, { + get(target, property, receiver) { + if (property === 'put') { + return async () => { + throw new Error('simulated R2 outage') + } + } + const value = Reflect.get(target, property, receiver) + return typeof value === 'function' ? value.bind(target) : value + }, + }) +} + +function createPreCommitD1FailureDb() { + return new Proxy(env.APP_DB, { + get(target, property, receiver) { + if (property === 'prepare') { + return (query: string) => { + const statement = target.prepare(query) + if (!query.includes('INSERT INTO email_messages')) { + return statement + } + return { + bind: () => ({ + run: async () => { + throw new Error('simulated D1 insert failure') + }, + }), + } + } + } + const value = Reflect.get(target, property, receiver) + return typeof value === 'function' ? value.bind(target) : value + }, + }) as D1Database +} + +function createPostCommitBookkeepingFailureDb() { + let messageCommitted = false + return new Proxy(env.APP_DB, { + get(target, property, receiver) { + if (property === 'prepare') { + return (query: string) => { + const statement = target.prepare(query) + if (query.includes('INSERT INTO email_messages')) { + return { + bind(...params: Array) { + const bound = statement.bind(...params) + return { + run: async () => { + const result = await bound.run() + messageCommitted = true + return result + }, + } + }, + } + } + if ( + messageCommitted && + (query.includes('UPDATE email_threads') || + query.includes('INSERT INTO email_delivery_events')) + ) { + return { + bind: () => ({ + run: async () => { + throw new Error('simulated post-commit bookkeeping failure') + }, + }), + } + } + return statement + } + } + const value = Reflect.get(target, property, receiver) + return typeof value === 'function' ? value.bind(target) : value + }, + }) as D1Database +} + +test('inbound pre-commit R2/D1 failures refund receive quota and rethrow; retry consumes one', async () => { + silenceIncidentalRuntimeWarnings() await ensureEmailTestSchema(env.APP_DB) - const username = `inline-${crypto.randomUUID().slice(0, 8)}` - const accountEmail = `inline-${crypto.randomUUID()}@example.com` + const username = `r2fail-${crypto.randomUUID().slice(0, 8)}` + const accountEmail = `r2fail-${crypto.randomUUID()}@example.com` const userId = await createStableUserIdFromEmail(accountEmail) const address = `${username}@${platformDomain}` await seedVerifiedAccount({ @@ -750,47 +849,493 @@ test('raw MIME stays inline when the R2 put fails', async () => { const raw = [ 'From: Sender ', `To: ${address}`, - 'Subject: Inline fallback mail', - 'Message-ID: ', + 'Subject: R2 failure mail', + 'Message-ID: ', '', - 'Inline body.', + 'Should not persist.', ].join('\r\n') - const message = createForwardableEmailMessage({ + const r2FailingEnv = { + ...createInboundEnv(), + EMAIL_BLOBS: createFailingEmailBlobs(), + } as Parameters[1] + const d1FailingEnv = { + ...createInboundEnv(), + APP_DB: createPreCommitD1FailureDb(), + } as Parameters[1] + + for (const failingEnv of [r2FailingEnv, d1FailingEnv]) { + for (let attempt = 0; attempt < 2; attempt += 1) { + const message = createForwardableEmailMessage({ + from: 'sender@example.net', + to: address, + raw, + }) + await expect( + handleInboundEmail(message, failingEnv), + ).rejects.toBeInstanceOf(RetryableInboundStorageError) + expect(message.rejectedReason).toBeNull() + expect(await readUserDailyReceiveCount(userId)).toBe(0) + } + } + + expect( + await listEmailMessages({ + db: env.APP_DB, + userId, + limit: 10, + }), + ).toEqual([]) + + const retryMessage = createForwardableEmailMessage({ from: 'sender@example.net', to: address, raw, }) - const failingBlobs = new Proxy(env.EMAIL_BLOBS, { + await handleInboundEmail(retryMessage, createInboundEnv()) + expect(retryMessage.rejectedReason).toBeNull() + expect(await readUserDailyReceiveCount(userId)).toBe(1) + expect( + await listEmailMessages({ + db: env.APP_DB, + userId, + limit: 10, + }), + ).toHaveLength(1) +}) + +test('inbound post-commit bookkeeping failure keeps one stored row without refund or retry throw', async () => { + silenceIncidentalRuntimeWarnings() + silenceExpectedConsoleErrors(['inbound-email-post-commit-bookkeeping-failed']) + await ensureEmailTestSchema(env.APP_DB) + const username = `postcommit-${crypto.randomUUID().slice(0, 8)}` + const accountEmail = `postcommit-${crypto.randomUUID()}@example.com` + const userId = await createStableUserIdFromEmail(accountEmail) + const address = `${username}@${platformDomain}` + await seedVerifiedAccount({ + db: env.APP_DB, + email: accountEmail, + username, + }) + const message = createForwardableEmailMessage({ + from: 'sender@example.net', + to: address, + raw: [ + 'From: Sender ', + `To: ${address}`, + 'Subject: Post-commit bookkeeping', + 'Message-ID: ', + '', + 'Body', + ].join('\r\n'), + }) + const failingEnv = { + ...createInboundEnv(), + APP_DB: createPostCommitBookkeepingFailureDb(), + } as Parameters[1] + + await handleInboundEmail(message, failingEnv) + expect(message.rejectedReason).toBeNull() + expect(await readUserDailyReceiveCount(userId)).toBe(1) + expect( + await listEmailMessages({ + db: env.APP_DB, + userId, + limit: 10, + }), + ).toHaveLength(1) +}) + +test('inbound parse rejection still consumes daily receive quota', async () => { + silenceIncidentalRuntimeWarnings() + await ensureEmailTestSchema(env.APP_DB) + const username = `parse-quota-${crypto.randomUUID().slice(0, 8)}` + const accountEmail = `parse-quota-${crypto.randomUUID()}@example.com` + const userId = await createStableUserIdFromEmail(accountEmail) + const address = `${username}@${platformDomain}` + await seedVerifiedAccount({ + db: env.APP_DB, + email: accountEmail, + username, + }) + + const unreadableMessage = createForwardableEmailMessage({ + from: 'sender@example.net', + to: address, + raw: 'Subject: Unreadable\r\n\r\nBody', + }) + Object.defineProperty(unreadableMessage, 'raw', { + value: new ReadableStream({ + pull() { + throw new Error('raw stream read failed') + }, + }), + }) + await handleInboundEmail(unreadableMessage, createInboundEnv()) + expect(unreadableMessage.rejectedReason).toMatch(/raw stream read failed/) + expect(await readUserDailyReceiveCount(userId)).toBe(1) +}) + +test('inbound storage refund failure is logged and original storage error is rethrown', async () => { + silenceIncidentalRuntimeWarnings() + consoleError.mockImplementation(() => {}) + await ensureEmailTestSchema(env.APP_DB) + const username = `refund-fail-${crypto.randomUUID().slice(0, 8)}` + const accountEmail = `refund-fail-${crypto.randomUUID()}@example.com` + await seedVerifiedAccount({ + db: env.APP_DB, + email: accountEmail, + username, + }) + const address = `${username}@${platformDomain}` + const failingDb = new Proxy(env.APP_DB, { get(target, property, receiver) { - if (property === 'put') { - return async () => { - throw new Error('simulated R2 outage') + if (property === 'prepare') { + return (query: string) => { + if (query.includes('UPDATE entitlement_daily_counters')) { + return { + bind: () => ({ + run: async () => { + throw new Error('simulated refund failure') + }, + }), + } + } + return target.prepare(query) } } - return Reflect.get(target, property, receiver) + const value = Reflect.get(target, property, receiver) + return typeof value === 'function' ? value.bind(target) : value }, - }) - const envWithFailingBlobs = { + }) as D1Database + const failingEnv = { ...createInboundEnv(), - EMAIL_BLOBS: failingBlobs, + APP_DB: failingDb, + EMAIL_BLOBS: createFailingEmailBlobs(), } as Parameters[1] - await handleInboundEmail(message, envWithFailingBlobs) - expect(message.rejectedReason).toBeNull() + const message = createForwardableEmailMessage({ + from: 'sender@example.net', + to: address, + raw: [ + 'From: Sender ', + `To: ${address}`, + 'Subject: Refund failure', + 'Message-ID: ', + '', + 'Body', + ].join('\r\n'), + }) - const [stored] = await listEmailMessages({ + await expect(handleInboundEmail(message, failingEnv)).rejects.toBeInstanceOf( + EmailRawMimeStorageError, + ) + expect(consoleError).toHaveBeenCalledWith( + 'email-receives-daily-refund-failed', + expect.anything(), + ) +}) + +test('insertEmailMessage never writes inline raw_mime on successful R2 put', async () => { + await ensureEmailTestSchema(env.APP_DB) + const userId = `user-${crypto.randomUUID()}` + const messageId = crypto.randomUUID() + const rawMime = 'From: a@example.net\r\nTo: b@example.net\r\n\r\nbody' + const stored = await insertEmailMessage({ db: env.APP_DB, - userId, - limit: 1, + blobs: env.EMAIL_BLOBS, + message: { + id: messageId, + direction: 'inbound', + userId, + rawMime, + processingStatus: 'stored', + }, + }) + expect(stored.rawMime).toBeNull() + expect(stored.rawMimeKey).toBe(emailRawMimeKey(userId, messageId)) + const row = await env.APP_DB.prepare( + `SELECT raw_mime, raw_mime_key FROM email_messages WHERE id = ?`, + ) + .bind(messageId) + .first<{ raw_mime: string | null; raw_mime_key: string | null }>() + expect(row).toEqual({ + raw_mime: null, + raw_mime_key: emailRawMimeKey(userId, messageId), }) - expect(stored).toBeDefined() - if (!stored) throw new Error('Expected stored inbound message') - - // Legacy inline behavior: the offload never throws and keeps mail. - expect(stored.rawMime).toBe(raw) - expect(stored.rawMimeKey).toBeNull() - // The read helper prefers the inline payload over the bucket. expect(await loadRawMime({ blobs: env.EMAIL_BLOBS, message: stored })).toBe( - raw, + rawMime, + ) +}) + +test('insertEmailMessage best-effort deletes R2 blob when D1 insert fails', async () => { + await ensureEmailTestSchema(env.APP_DB) + const userId = `user-${crypto.randomUUID()}` + const messageId = crypto.randomUUID() + const key = emailRawMimeKey(userId, messageId) + const deletes: Array = [] + const blobs = new Proxy(env.EMAIL_BLOBS, { + get(target, property, receiver) { + if (property === 'delete') { + return async (objectKey: string) => { + deletes.push(objectKey) + return target.delete(objectKey) + } + } + const value = Reflect.get(target, property, receiver) + return typeof value === 'function' ? value.bind(target) : value + }, + }) + const failingDb = new Proxy(env.APP_DB, { + get(target, property, receiver) { + if (property === 'prepare') { + return (query: string) => { + const statement = target.prepare(query) + if (!query.includes('INSERT INTO email_messages')) { + return statement + } + return { + bind: () => ({ + run: async () => { + throw new Error('simulated D1 insert failure') + }, + }), + } + } + } + const value = Reflect.get(target, property, receiver) + return typeof value === 'function' ? value.bind(target) : value + }, + }) as D1Database + + await expect( + insertEmailMessage({ + db: failingDb, + blobs, + message: { + id: messageId, + direction: 'inbound', + userId, + rawMime: 'orphan-candidate', + processingStatus: 'stored', + }, + }), + ).rejects.toThrow('simulated D1 insert failure') + + expect(deletes).toEqual([key]) + expect(await env.EMAIL_BLOBS.get(key)).toBeNull() + expect( + await env.APP_DB.prepare(`SELECT id FROM email_messages WHERE id = ?`) + .bind(messageId) + .first(), + ).toBeNull() +}) + +test('insertEmailMessageWithAttachments cleans message and blob when attachment insert fails', async () => { + await ensureEmailTestSchema(env.APP_DB) + const userId = `user-${crypto.randomUUID()}` + const messageId = crypto.randomUUID() + const key = emailRawMimeKey(userId, messageId) + // Fail only the first batch (attachment insert). Message cleanup also + // uses db.batch and must still run. + let attachmentBatchFailed = false + const failingDb = new Proxy(env.APP_DB, { + get(target, property, receiver) { + if (property === 'batch') { + return async (statements: Parameters[0]) => { + if (!attachmentBatchFailed) { + attachmentBatchFailed = true + throw new Error('simulated attachment insert failure') + } + return target.batch(statements) + } + } + const value = Reflect.get(target, property, receiver) + return typeof value === 'function' ? value.bind(target) : value + }, + }) as D1Database + + await expect( + insertEmailMessageWithAttachments({ + db: failingDb, + blobs: env.EMAIL_BLOBS, + message: { + id: messageId, + direction: 'inbound', + userId, + rawMime: 'attachment-cleanup-bytes', + processingStatus: 'stored', + }, + attachments: [ + { + filename: 'note.txt', + contentType: 'text/plain', + contentId: null, + disposition: 'attachment', + size: 4, + storageKind: 'raw-mime', + storageKey: null, + }, + ], + }), + ).rejects.toBeInstanceOf(RetryableInboundStorageError) + + expect( + await getEmailMessageById({ + db: env.APP_DB, + userId, + messageId, + }), + ).toBeNull() + expect(await env.EMAIL_BLOBS.get(key)).toBeNull() +}) + +test('attachment cleanup failure with remaining row is acknowledged without retryable throw', async () => { + silenceExpectedConsoleErrors(['inbound-email-attachment-cleanup-failed']) + await ensureEmailTestSchema(env.APP_DB) + const userId = `user-${crypto.randomUUID()}` + const messageId = crypto.randomUUID() + const key = emailRawMimeKey(userId, messageId) + const failingDb = new Proxy(env.APP_DB, { + get(target, property, receiver) { + if (property === 'batch') { + return async () => { + throw new Error('simulated attachment and cleanup batch failure') + } + } + const value = Reflect.get(target, property, receiver) + return typeof value === 'function' ? value.bind(target) : value + }, + }) as D1Database + + const stored = await insertEmailMessageWithAttachments({ + db: failingDb, + blobs: env.EMAIL_BLOBS, + message: { + id: messageId, + direction: 'inbound', + userId, + rawMime: 'attachment-orphan-bytes', + processingStatus: 'stored', + }, + attachments: [ + { + filename: 'note.txt', + contentType: 'text/plain', + contentId: null, + disposition: 'attachment', + size: 4, + storageKind: 'raw-mime', + storageKey: null, + }, + ], + }) + + expect(stored.id).toBe(messageId) + expect( + await getEmailMessageById({ + db: env.APP_DB, + userId, + messageId, + }), + ).not.toBeNull() + expect(await env.EMAIL_BLOBS.get(key)).not.toBeNull() + expect(consoleError).toHaveBeenCalledWith( + 'inbound-email-attachment-cleanup-failed', + messageId, + expect.anything(), + expect.anything(), + ) +}) + +test('attachment cleanup probe failure acknowledges message without retry or quota refund', async () => { + silenceIncidentalRuntimeWarnings() + silenceExpectedConsoleErrors([ + 'inbound-email-attachment-cleanup-probe-failed', + ]) + await ensureEmailTestSchema(env.APP_DB) + const username = `probe-fail-${crypto.randomUUID().slice(0, 8)}` + const accountEmail = `probe-fail-${crypto.randomUUID()}@example.com` + const userId = await createStableUserIdFromEmail(accountEmail) + const address = `${username}@${platformDomain}` + await seedVerifiedAccount({ + db: env.APP_DB, + email: accountEmail, + username, + }) + const failingDb = new Proxy(env.APP_DB, { + get(target, property, receiver) { + if (property === 'batch') { + return async () => { + throw new Error('simulated attachment and cleanup batch failure') + } + } + if (property === 'prepare') { + return (query: string) => { + const statement = target.prepare(query) + // Residual probe after attachment cleanup (getEmailMessageById). + if ( + query.includes('FROM email_messages') && + query.includes('WHERE id = ?') && + query.includes('AND user_id = ?') && + query.includes('LIMIT 1') + ) { + return { + bind: () => ({ + first: async () => { + throw new Error('simulated residual probe failure') + }, + }), + } + } + return statement + } + } + const value = Reflect.get(target, property, receiver) + return typeof value === 'function' ? value.bind(target) : value + }, + }) as D1Database + const message = createForwardableEmailMessage({ + from: 'sender@example.net', + to: address, + raw: [ + 'From: Sender ', + `To: ${address}`, + 'Subject: Probe failure mail', + 'Message-ID: ', + 'Content-Type: multipart/mixed; boundary="probe-boundary"', + '', + '--probe-boundary', + 'Content-Type: text/plain; charset="utf-8"', + '', + 'Body', + '--probe-boundary', + 'Content-Type: text/plain; name="note.txt"', + 'Content-Disposition: attachment; filename="note.txt"', + '', + 'Note', + '--probe-boundary--', + ].join('\r\n'), + }) + const failingEnv = { + ...createInboundEnv(), + APP_DB: failingDb, + } as Parameters[1] + + await handleInboundEmail(message, failingEnv) + expect(message.rejectedReason).toBeNull() + expect(await readUserDailyReceiveCount(userId)).toBe(1) + expect( + await listEmailMessages({ + db: env.APP_DB, + userId, + limit: 10, + }), + ).toHaveLength(1) + expect(consoleError).toHaveBeenCalledWith( + 'inbound-email-attachment-cleanup-probe-failed', + expect.any(String), + expect.anything(), + expect.anything(), + expect.anything(), ) }) diff --git a/packages/worker/src/email/repo.ts b/packages/worker/src/email/repo.ts index 16f222507d..c86d19b6b2 100644 --- a/packages/worker/src/email/repo.ts +++ b/packages/worker/src/email/repo.ts @@ -193,37 +193,60 @@ export function emailAttachmentBlobKey( } /** - * Best-effort raw-MIME offload to R2. Returns the stored object key, or - * null when the payload must stay inline in D1 because the put failed - * (for example a transient R2 outage). Never throws: falling back to - * the legacy inline raw_mime column must not lose mail. + * Pre-commit inbound storage failure. Thrown only before the message and + * attachment rows are durably committed. The inbound handler refunds daily + * receive quota and rethrows so Cloudflare Email Routing retries delivery. + * Post-commit bookkeeping failures must not use this type. */ -async function offloadRawMimeToBlobs(input: { +export class RetryableInboundStorageError extends Error { + override name = 'RetryableInboundStorageError' + constructor(message: string, cause?: unknown) { + super(message, { cause }) + } +} + +/** + * Retryable EMAIL_BLOBS put failure for inbound raw MIME (pre-commit). + * Callers must not fall back to inline D1 `raw_mime`; the inbound email + * handler lets this propagate so Cloudflare Email Routing treats delivery + * as a temporary failure (throw) rather than a permanent `setReject`. + */ +export class EmailRawMimeStorageError extends RetryableInboundStorageError { + override name = 'EmailRawMimeStorageError' + constructor(messageId: string, cause?: unknown) { + super( + `Failed to store email raw MIME in EMAIL_BLOBS (message ${messageId}); delivery should be retried.`, + cause, + ) + } +} + +/** + * Persist inbound raw MIME to EMAIL_BLOBS before the D1 insert. Returns the + * object key on success. Throws EmailRawMimeStorageError on put failure — + * there is no inline D1 fallback (Stage 4a; column drop follows once + * residuals are swept). + */ +async function putRawMimeToBlobs(input: { blobs: R2Bucket userId: string messageId: string rawMime: string -}): Promise { +}): Promise { const key = emailRawMimeKey(input.userId, input.messageId) try { await input.blobs.put(key, input.rawMime) return key } catch (error) { - console.debug( - 'email-raw-mime-inline-fallback', - 'R2 put failed', - input.messageId, - error, - ) - return null + throw new EmailRawMimeStorageError(input.messageId, error) } } /** - * Resolve a message's raw MIME regardless of where it is stored: legacy - * rows keep the payload inline in raw_mime, offloaded rows store an R2 - * key in raw_mime_key. Returns null when the message has no raw MIME or - * the blob is unreachable. + * Resolve a message's raw MIME regardless of where it is stored. Dual-read + * for the Stage 4a rollout: residual legacy rows may still have inline + * raw_mime; new inbound rows store only an R2 key in raw_mime_key. + * Returns null when the message has no raw MIME or the blob is unreachable. */ export async function loadRawMime(input: { blobs: R2Bucket @@ -768,9 +791,11 @@ export async function touchEmailThread(input: { export async function insertEmailMessage(input: { db: D1Database /** - * EMAIL_BLOBS bucket. Raw MIME is offloaded to R2 and only - * raw_mime_key is persisted; if the put fails the payload stays - * inline in the legacy raw_mime column rather than losing mail. + * EMAIL_BLOBS bucket. Inbound raw MIME must be written here before the + * D1 insert; put failure throws EmailRawMimeStorageError (no inline + * raw_mime fallback). Outbound messages pass rawMime null and skip + * the put. If D1 insert fails after a successful put, the blob is + * best-effort deleted. */ blobs: R2Bucket message: { @@ -805,16 +830,16 @@ export async function insertEmailMessage(input: { }) { const timestamp = nowIso() const messageId = input.message.id ?? crypto.randomUUID() - let rawMime = input.message.rawMime ?? null + // New rows never write inline raw_mime (Stage 4a). Inbound MIME goes to + // R2 first; outbound leaves both columns null. let rawMimeKey: string | null = null - if (rawMime != null) { - rawMimeKey = await offloadRawMimeToBlobs({ + if (input.message.rawMime != null) { + rawMimeKey = await putRawMimeToBlobs({ blobs: input.blobs, userId: input.message.userId, messageId, - rawMime, + rawMime: input.message.rawMime, }) - if (rawMimeKey != null) rawMime = null } const row = { id: messageId, @@ -841,7 +866,7 @@ export async function insertEmailMessage(input: { auth_results: input.message.authResults ?? null, text_body: input.message.textBody ?? null, html_body: input.message.htmlBody ?? null, - raw_mime: rawMime, + raw_mime: null, raw_mime_key: rawMimeKey, raw_size: input.message.rawSize ?? 0, processing_status: input.message.processingStatus, @@ -1271,7 +1296,17 @@ export async function insertEmailMessageWithAttachments( attachments: Parameters[0]['attachments'] }, ) { - const message = await insertEmailMessage(input) + let message + try { + message = await insertEmailMessage(input) + } catch (error) { + if (error instanceof RetryableInboundStorageError) throw error + throw new RetryableInboundStorageError( + 'Failed to store inbound email message; delivery should be retried.', + error, + ) + } + if (input.attachments.length === 0) return message try { await insertEmailAttachments({ db: input.db, @@ -1279,13 +1314,51 @@ export async function insertEmailMessageWithAttachments( attachments: input.attachments, }) return message - } catch (error) { - await deleteEmailMessageById({ - db: input.db, - blobs: input.blobs, - messageId: message.id, - }).catch(() => undefined) - throw error + } catch (attachmentError) { + let cleanupError: unknown + try { + await deleteEmailMessageById({ + db: input.db, + blobs: input.blobs, + messageId: message.id, + }) + } catch (error) { + cleanupError = error + } + let remaining: Awaited> + try { + remaining = await getEmailMessageById({ + db: input.db, + userId: message.userId, + messageId: message.id, + }) + } catch (probeError) { + // Probe failed: commit state is ambiguous. Do not retry/refund — + // that risks duplicates if the row is still durable. + console.error( + 'inbound-email-attachment-cleanup-probe-failed', + message.id, + attachmentError, + cleanupError, + probeError, + ) + return message + } + if (remaining) { + // Message row is durable; retrying would duplicate mail. Acknowledge + // the commit and leave operators the cleanup/attachment failure logs. + console.error( + 'inbound-email-attachment-cleanup-failed', + message.id, + attachmentError, + cleanupError, + ) + return remaining + } + throw new RetryableInboundStorageError( + 'Failed to store inbound email attachments; message cleaned up and delivery should be retried.', + attachmentError, + ) } } diff --git a/packages/worker/src/email/system-email.ts b/packages/worker/src/email/system-email.ts index ccea43c40d..985ced3df8 100644 --- a/packages/worker/src/email/system-email.ts +++ b/packages/worker/src/email/system-email.ts @@ -172,6 +172,30 @@ export async function consumeSystemEmailDailyReceive(input: { return row ? Number(row.count) : null } +/** + * Atomically refund one previously consumed system-inbox daily receive for + * the given local_part/UTC day (floors at zero). Scoped by `local_part` so a + * refund cannot touch another system inbox's counter. Pass the same `now` + * (day key) as the matching `consumeSystemEmailDailyReceive` call. + */ +export async function refundSystemEmailDailyReceive(input: { + db: D1Database + localPart: SystemEmailLocal + now?: Date +}): Promise { + const now = input.now ?? new Date() + await input.db + .prepare( + `UPDATE system_email_daily_counters + SET count = MAX(0, count - 1), + updated_at = ? + WHERE local_part = ? + AND day = ?`, + ) + .bind(now.toISOString(), input.localPart, systemEmailDayKey(now)) + .run() +} + export async function countStoredSystemEmailMessages(input: { db: D1Database }) { diff --git a/packages/worker/src/email/system-email.workers.test.ts b/packages/worker/src/email/system-email.workers.test.ts index 3d1b8aa299..c48ba32067 100644 --- a/packages/worker/src/email/system-email.workers.test.ts +++ b/packages/worker/src/email/system-email.workers.test.ts @@ -6,14 +6,20 @@ import { listEmailMessages, maxDetailedEmailRejectionEventsPerDay, } from './repo.ts' +import { RetryableInboundStorageError } from './repo.ts' import { pruneSystemEmailRetention, + refundSystemEmailDailyReceive, + systemEmailDayKey, systemEmailLimits, systemEmailOwnerId, } from './system-email.ts' import { createForwardableEmailMessage } from './test-fixtures.ts' import { ensureEmailTestSchema } from './test-schema.ts' -import { consoleWarn } from '#worker/test-support/console-spies.ts' +import { + consoleWarn, + silenceExpectedConsoleErrors, +} from '#worker/test-support/console-spies.ts' import { silenceIncidentalRuntimeWarnings } from '#worker/test-support/incidental-runtime-warnings.ts' import { ensureUsageRollupsTestSchema } from '#worker/usage/test-schema.ts' import { createStableUserIdFromEmail } from '#worker/user-id.ts' @@ -200,6 +206,192 @@ test('non-system reserved locals still reject while username addresses are unaff ).toHaveLength(1) }) +async function readSystemDailyReceiveCount(localPart: string) { + const row = await env.APP_DB.prepare( + `SELECT count FROM system_email_daily_counters + WHERE local_part = ? AND day = ?`, + ) + .bind(localPart, systemEmailDayKey()) + .first<{ count: number }>() + return Number(row?.count ?? 0) +} + +test('refundSystemEmailDailyReceive decrements local/day counter and floors at zero', async () => { + await ensureEmailTestSchema(env.APP_DB) + const now = new Date('2026-07-05T12:00:00.000Z') + const day = systemEmailDayKey(now) + const readCount = async (localPart: string) => + Number( + ( + await env.APP_DB.prepare( + `SELECT count FROM system_email_daily_counters + WHERE local_part = ? AND day = ?`, + ) + .bind(localPart, day) + .first<{ count: number }>() + )?.count ?? 0, + ) + await env.APP_DB.prepare( + `INSERT INTO system_email_daily_counters (local_part, day, count, updated_at) + VALUES ('abuse', ?, 1, ?), ('support', ?, 2, ?)`, + ) + .bind(day, now.toISOString(), day, now.toISOString()) + .run() + + await refundSystemEmailDailyReceive({ + db: env.APP_DB, + localPart: 'abuse', + now, + }) + expect(await readCount('abuse')).toBe(0) + expect(await readCount('support')).toBe(2) + + await refundSystemEmailDailyReceive({ + db: env.APP_DB, + localPart: 'abuse', + now, + }) + expect(await readCount('abuse')).toBe(0) +}) + +test('system inbox pre-commit R2/D1 failures refund daily receive quota; retry consumes one', async () => { + silenceIncidentalRuntimeWarnings() + await ensureEmailTestSchema(env.APP_DB) + await ensureUsageRollupsTestSchema(env.APP_DB) + const r2FailingEnv = { + ...createInboundEnv(), + EMAIL_BLOBS: new Proxy(env.EMAIL_BLOBS, { + get(target, property, receiver) { + if (property === 'put') { + return async () => { + throw new Error('simulated R2 outage') + } + } + const value = Reflect.get(target, property, receiver) + return typeof value === 'function' ? value.bind(target) : value + }, + }), + } as Parameters[1] + const d1FailingEnv = { + ...createInboundEnv(), + APP_DB: new Proxy(env.APP_DB, { + get(target, property, receiver) { + if (property === 'prepare') { + return (query: string) => { + const statement = target.prepare(query) + if (!query.includes('INSERT INTO email_messages')) { + return statement + } + return { + bind: () => ({ + run: async () => { + throw new Error('simulated D1 insert failure') + }, + }), + } + } + } + const value = Reflect.get(target, property, receiver) + return typeof value === 'function' ? value.bind(target) : value + }, + }) as D1Database, + } as Parameters[1] + + for (const failingEnv of [r2FailingEnv, d1FailingEnv]) { + for (let attempt = 0; attempt < 2; attempt += 1) { + const message = buildInboundMessage({ + to: `abuse@${systemDomain}`, + messageId: `system-precommit-fail-${attempt}-${crypto.randomUUID().slice(0, 6)}`, + }) + await expect( + handleInboundEmail(message, failingEnv), + ).rejects.toBeInstanceOf(RetryableInboundStorageError) + expect(message.rejectedReason).toBeNull() + expect(await readSystemDailyReceiveCount('abuse')).toBe(0) + } + } + + const retry = buildInboundMessage({ + to: `abuse@${systemDomain}`, + messageId: 'system-r2-retry-ok', + }) + await handleInboundEmail(retry, createInboundEnv()) + expect(retry.rejectedReason).toBeNull() + expect(await readSystemDailyReceiveCount('abuse')).toBe(1) + expect( + await listEmailMessages({ + db: env.APP_DB, + userId: systemEmailOwnerId, + limit: 10, + }), + ).toHaveLength(1) +}) + +test('system inbox post-commit bookkeeping failure keeps one stored row without refund or retry throw', async () => { + silenceIncidentalRuntimeWarnings() + silenceExpectedConsoleErrors(['inbound-email-post-commit-bookkeeping-failed']) + await ensureEmailTestSchema(env.APP_DB) + await ensureUsageRollupsTestSchema(env.APP_DB) + let messageCommitted = false + const failingEnv = { + ...createInboundEnv(), + APP_DB: new Proxy(env.APP_DB, { + get(target, property, receiver) { + if (property === 'prepare') { + return (query: string) => { + const statement = target.prepare(query) + if (query.includes('INSERT INTO email_messages')) { + return { + bind(...params: Array) { + const bound = statement.bind(...params) + return { + run: async () => { + const result = await bound.run() + messageCommitted = true + return result + }, + } + }, + } + } + if ( + messageCommitted && + (query.includes('UPDATE email_threads') || + query.includes('INSERT INTO email_delivery_events')) + ) { + return { + bind: () => ({ + run: async () => { + throw new Error('simulated post-commit bookkeeping failure') + }, + }), + } + } + return statement + } + } + const value = Reflect.get(target, property, receiver) + return typeof value === 'function' ? value.bind(target) : value + }, + }) as D1Database, + } as Parameters[1] + + const message = buildInboundMessage({ + to: `support@${systemDomain}`, + messageId: 'system-post-commit', + }) + await handleInboundEmail(message, failingEnv) + expect(message.rejectedReason).toBeNull() + expect(await readSystemDailyReceiveCount('support')).toBe(1) + expect( + await listEmailMessages({ + db: env.APP_DB, + userId: systemEmailOwnerId, + limit: 10, + }), + ).toHaveLength(1) +}) + test('system email size and daily caps reject before storage with bounded events', async () => { await ensureEmailTestSchema(env.APP_DB) await ensureUsageRollupsTestSchema(env.APP_DB) diff --git a/packages/worker/src/entitlements/entitlements.node.test.ts b/packages/worker/src/entitlements/entitlements.node.test.ts index 8683ac378a..ae95da3104 100644 --- a/packages/worker/src/entitlements/entitlements.node.test.ts +++ b/packages/worker/src/entitlements/entitlements.node.test.ts @@ -21,6 +21,7 @@ import { estimateEntitlementStorageEntryBytes, getUserPlan, incrementDailyEntitlementCounter, + refundDailyEntitlement, } from './service.ts' import { utcDayKey } from '@kody-internal/shared/date-keys.ts' @@ -169,6 +170,20 @@ function createEntitlementsTestDb( } return { meta: { changes: 1 } } } + if (query.includes('UPDATE entitlement_daily_counters')) { + // bind(updated_at, user_id, resource, day) + const existing = counters.find( + (counter) => + counter.user_id === params[1] && + counter.resource === params[2] && + counter.day === params[3], + ) + if (existing) { + existing.count = Math.max(0, existing.count - 1) + return { meta: { changes: 1 } } + } + return { meta: { changes: 0 } } + } throw new Error(`Unsupported run query: ${query}`) }, } @@ -485,6 +500,62 @@ test('plan user daily entitlements increment, enforce at limit, and reset on a n expect(counters[1]?.count).toBe(1) }) +test('refundDailyEntitlement decrements the user/day counter and floors at zero', async () => { + const { db, counters } = createEntitlementsTestDb() + const now = new Date('2026-07-05T15:00:00.000Z') + await incrementDailyEntitlementCounter({ + db, + userId: 'user-1', + resource: 'email_receives_per_day', + amount: 2, + now, + }) + await incrementDailyEntitlementCounter({ + db, + userId: 'user-2', + resource: 'email_receives_per_day', + amount: 3, + now, + }) + await refundDailyEntitlement({ + db, + userId: 'user-1', + resource: 'email_receives_per_day', + now, + }) + expect( + counters.find( + (row) => + row.user_id === 'user-1' && row.resource === 'email_receives_per_day', + )?.count, + ).toBe(1) + expect( + counters.find( + (row) => + row.user_id === 'user-2' && row.resource === 'email_receives_per_day', + )?.count, + ).toBe(3) + + await refundDailyEntitlement({ + db, + userId: 'user-1', + resource: 'email_receives_per_day', + now, + }) + await refundDailyEntitlement({ + db, + userId: 'user-1', + resource: 'email_receives_per_day', + now, + }) + expect( + counters.find( + (row) => + row.user_id === 'user-1' && row.resource === 'email_receives_per_day', + )?.count, + ).toBe(0) +}) + test('plan-less users count uncapped sends but honor fallback receive limits', async () => { const { db, counters } = createEntitlementsTestDb() const sendLimit = planLimits.free.maxEmailSendsPerDay diff --git a/packages/worker/src/entitlements/plans.ts b/packages/worker/src/entitlements/plans.ts index b0e770ecdc..f1a1197d97 100644 --- a/packages/worker/src/entitlements/plans.ts +++ b/packages/worker/src/entitlements/plans.ts @@ -76,9 +76,10 @@ export type PlanLimits = { maxStoredEmailMessages: number | null /** * Maximum raw MIME bytes for a single stored email message. Hard - * platform bound: raw MIME is stored inline in the email_messages row - * next to the extracted bodies (worst case ~2x raw), and D1 caps rows - * at 2 MB — so keep this well under ~1 MB regardless of plan. + * platform bound: raw MIME lives in EMAIL_BLOBS, but extracted text/html + * bodies are still stored on the email_messages row (worst case ~2x + * raw), and D1 caps rows at 2 MB — so keep this well under ~1 MB + * regardless of plan. */ maxEmailMessageBytes: number | null /** Maximum stored secret entries across non-expired buckets. */ diff --git a/packages/worker/src/entitlements/service.ts b/packages/worker/src/entitlements/service.ts index e1486d63d9..a3a237ded7 100644 --- a/packages/worker/src/entitlements/service.ts +++ b/packages/worker/src/entitlements/service.ts @@ -767,6 +767,33 @@ export async function consumeDailyEntitlement(input: { } } +/** + * Atomically refund one previously consumed daily entitlement unit for the + * given user/resource/UTC day (floors at zero). Scoped by `user_id` so a + * refund can never touch another user's counter. Used when inbound receive + * quota was charged before a retryable storage failure — the same `now` + * (day key) as the matching `consumeDailyEntitlement` call must be passed. + */ +export async function refundDailyEntitlement(input: { + db: D1Database + userId: string + resource: EntitlementResource + now?: Date +}): Promise { + const now = input.now ?? new Date() + await input.db + .prepare( + `UPDATE entitlement_daily_counters + SET count = MAX(0, count - 1), + updated_at = ? + WHERE user_id = ? + AND resource = ? + AND day = ?`, + ) + .bind(now.toISOString(), input.userId, input.resource, utcDayKey(now)) + .run() +} + export const defaultWorkflowConcurrencyBackstop = 100 /** diff --git a/packages/worker/src/index.ts b/packages/worker/src/index.ts index 42ea91fcd0..c3be2bfb27 100644 --- a/packages/worker/src/index.ts +++ b/packages/worker/src/index.ts @@ -507,6 +507,9 @@ const workerHandler = { env: Env, ctx: ExecutionContext, ) { + // Let storage/transient failures throw so Email Routing does not + // acknowledge the message (retryable). Permanent rejects use + // message.setReject inside handleInboundEmail. await handleInboundEmail(message, env, ctx) }, async queue(batch: MessageBatch, env: Env, ctx: ExecutionContext) {