From f410b3f200b2e0766a71b5f1f6bbc8fcc61c3e2e Mon Sep 17 00:00:00 2001 From: Vlad Bisceanu <7993591+vladbisceanu@users.noreply.github.com> Date: Wed, 26 Aug 2026 20:03:36 +0200 Subject: [PATCH] fix: preserve email retries until attempts exhaust --- .../jobs/__tests__/email-processor.test.ts | 250 ++++++++++++++++++ apps/api/src/jobs/email-processor.ts | 225 ++++++++++++---- packages/types/src/jobs/email.ts | 5 + 3 files changed, 429 insertions(+), 51 deletions(-) diff --git a/apps/api/src/jobs/__tests__/email-processor.test.ts b/apps/api/src/jobs/__tests__/email-processor.test.ts index 881b762bb..e1c1a4ba6 100644 --- a/apps/api/src/jobs/__tests__/email-processor.test.ts +++ b/apps/api/src/jobs/__tests__/email-processor.test.ts @@ -1,7 +1,19 @@ import {beforeEach, describe, expect, it, vi} from 'vitest'; import {EmailSourceType, EmailStatus, TrackingMode} from '@plunk/db'; import {toPrismaJson} from '@plunk/types'; +import {Job} from 'bullmq'; import {createServiceMocks, factories, getPrismaClient} from '../../../../../test/helpers'; +import {prisma as runtimePrisma} from '../../database/prisma.js'; +import {EventService} from '../../services/EventService.js'; +import {emailQueue} from '../../services/QueueService.js'; +import {createEmailWorker} from '../email-processor'; + +const sesMocks = vi.hoisted(() => ({ + getSendingQuota: vi.fn(), + sendRawEmail: vi.fn(), +})); + +vi.mock('../../services/SESService.js', () => sesMocks); // Mock MeterService vi.mock('../../services/MeterService.js', () => ({ @@ -10,12 +22,37 @@ vi.mock('../../services/MeterService.js', () => ({ }, })); +async function waitForEmailStatus(emailId: string, status: EmailStatus) { + for (let attempt = 0; attempt < 200; attempt++) { + const email = await getPrismaClient().email.findUniqueOrThrow({where: {id: emailId}}); + if (email.status === status) return email; + await new Promise(resolve => setTimeout(resolve, 25)); + } + + throw new Error(`Email ${emailId} did not reach ${status}`); +} + +async function waitForJobState(job: {getState(): Promise}, state: string) { + for (let attempt = 0; attempt < 200; attempt++) { + if ((await job.getState()) === state) return; + await new Promise(resolve => setTimeout(resolve, 25)); + } + + throw new Error(`Job did not reach ${state}`); +} + describe('Email Processor', () => { let projectId: string; const prisma = getPrismaClient(); const _serviceMocks = createServiceMocks(); beforeEach(async () => { + sesMocks.getSendingQuota.mockReset().mockResolvedValue({ + maxSendRate: 14, + sentLast24Hours: 0, + max24HourSend: 200, + }); + sesMocks.sendRawEmail.mockReset().mockResolvedValue({messageId: 'mock-message-id'}); const {project} = await factories.createUserWithProject({}, {tracking: TrackingMode.ENABLED}); projectId = project.id; }); @@ -124,6 +161,219 @@ describe('Email Processor', () => { expect(template.type).toBe('TRANSACTIONAL'); }); + + it('should retry SES after the first failure and succeed on the second attempt', async () => { + const contact = await factories.createContact({projectId}); + const email = await factories.createEmail(projectId, contact.id, { + sourceType: EmailSourceType.TRANSACTIONAL, + status: EmailStatus.PENDING, + }); + const retryableSesError = Object.assign(new Error('transient SES failure'), { + $metadata: {httpStatusCode: 503}, + }); + sesMocks.sendRawEmail + .mockRejectedValueOnce(retryableSesError) + .mockResolvedValueOnce({messageId: 'ses-retry-success'}); + const worker = await createEmailWorker(); + + try { + await emailQueue.add( + 'send-email', + {emailId: email.id}, + { + jobId: `retry-success-${email.id}`, + attempts: 2, + backoff: {type: 'fixed', delay: 10}, + }, + ); + + await expect(waitForEmailStatus(email.id, EmailStatus.SENT)).resolves.toMatchObject({ + status: EmailStatus.SENT, + messageId: 'ses-retry-success', + error: null, + }); + } finally { + await worker.close(); + } + + expect(sesMocks.sendRawEmail).toHaveBeenCalledTimes(2); + }); + + it('should record FAILED only when SES attempts are exhausted', async () => { + const contact = await factories.createContact({projectId}); + const email = await factories.createEmail(projectId, contact.id, { + sourceType: EmailSourceType.TRANSACTIONAL, + status: EmailStatus.PENDING, + }); + sesMocks.sendRawEmail.mockRejectedValue( + Object.assign(new Error('terminal SES failure'), {$metadata: {httpStatusCode: 503}}), + ); + const worker = await createEmailWorker(); + + try { + await emailQueue.add( + 'send-email', + {emailId: email.id}, + { + jobId: `retry-exhausted-${email.id}`, + attempts: 2, + backoff: {type: 'fixed', delay: 10}, + }, + ); + + await waitForEmailStatus(email.id, EmailStatus.FAILED); + } finally { + await worker.close(); + } + + expect(sesMocks.sendRawEmail).toHaveBeenCalledTimes(2); + await expect(prisma.email.findUniqueOrThrow({where: {id: email.id}})).resolves.toMatchObject({ + status: EmailStatus.FAILED, + error: 'terminal SES failure', + }); + }); + + it('should finalize a checkpointed SES acceptance after the database recovers', async () => { + const contact = await factories.createContact({projectId}); + const email = await factories.createEmail(projectId, contact.id, { + sourceType: EmailSourceType.TRANSACTIONAL, + status: EmailStatus.PENDING, + }); + sesMocks.sendRawEmail.mockImplementationOnce(async () => { + // The SENDING write already succeeded. Fail both attempts to persist the + // accepted message so the BullMQ retry must recover it from job data. + vi.spyOn(runtimePrisma.email, 'updateMany').mockRejectedValueOnce(new Error('database unavailable')); + vi.spyOn(runtimePrisma.email, 'update').mockRejectedValueOnce(new Error('database still unavailable')); + return {messageId: 'ses-checkpointed'}; + }); + const worker = await createEmailWorker(); + + try { + await emailQueue.add( + 'send-email', + {emailId: email.id}, + { + jobId: `accepted-checkpoint-${email.id}`, + attempts: 2, + backoff: {type: 'fixed', delay: 10}, + }, + ); + + await expect(waitForEmailStatus(email.id, EmailStatus.SENT)).resolves.toMatchObject({ + status: EmailStatus.SENT, + messageId: 'ses-checkpointed', + }); + } finally { + await worker.close(); + } + + expect(sesMocks.sendRawEmail).toHaveBeenCalledOnce(); + }); + + it('should fail visibly when neither the acceptance checkpoint nor database writes succeed', async () => { + const contact = await factories.createContact({projectId}); + const email = await factories.createEmail(projectId, contact.id, { + sourceType: EmailSourceType.TRANSACTIONAL, + status: EmailStatus.PENDING, + }); + vi.spyOn(Job.prototype, 'updateData') + .mockRejectedValueOnce(new Error('redis unavailable')) + .mockRejectedValueOnce(new Error('redis still unavailable')); + sesMocks.sendRawEmail.mockImplementationOnce(async () => { + vi.spyOn(runtimePrisma.email, 'updateMany').mockRejectedValueOnce(new Error('database unavailable')); + vi.spyOn(runtimePrisma.email, 'update').mockRejectedValueOnce(new Error('database still unavailable')); + return {messageId: 'ses-uncheckpointed'}; + }); + const worker = await createEmailWorker(); + + try { + await emailQueue.add( + 'send-email', + {emailId: email.id}, + { + jobId: `uncheckpointed-acceptance-${email.id}`, + attempts: 2, + backoff: {type: 'fixed', delay: 10}, + }, + ); + + await waitForEmailStatus(email.id, EmailStatus.FAILED); + } finally { + await worker.close(); + } + + expect(sesMocks.sendRawEmail).toHaveBeenCalledOnce(); + await expect(prisma.email.findUniqueOrThrow({where: {id: email.id}})).resolves.toMatchObject({ + status: EmailStatus.FAILED, + error: 'Previous attempt ended without an SES acceptance checkpoint; not retried to avoid a duplicate', + }); + }); + + it('should not retry an ambiguous SES transport failure', async () => { + const contact = await factories.createContact({projectId}); + const email = await factories.createEmail(projectId, contact.id, { + sourceType: EmailSourceType.TRANSACTIONAL, + status: EmailStatus.PENDING, + }); + sesMocks.sendRawEmail.mockRejectedValue(new Error('socket closed before the response')); + const worker = await createEmailWorker(); + + try { + const job = await emailQueue.add( + 'send-email', + {emailId: email.id}, + { + jobId: `ambiguous-ses-${email.id}`, + attempts: 2, + backoff: {type: 'fixed', delay: 10}, + }, + ); + + await waitForJobState(job, 'failed'); + } finally { + await worker.close(); + } + + expect(sesMocks.sendRawEmail).toHaveBeenCalledOnce(); + await expect(prisma.email.findUniqueOrThrow({where: {id: email.id}})).resolves.toMatchObject({ + status: EmailStatus.FAILED, + error: 'SES outcome is unknown; not retried to avoid a duplicate: socket closed before the response', + }); + }); + + it('should not resubmit an email when post-send processing fails', async () => { + const contact = await factories.createContact({projectId}); + const email = await factories.createEmail(projectId, contact.id, { + sourceType: EmailSourceType.TRANSACTIONAL, + status: EmailStatus.PENDING, + }); + sesMocks.sendRawEmail.mockResolvedValue({messageId: 'ses-already-accepted'}); + vi.spyOn(EventService, 'trackEvent').mockRejectedValueOnce(new Error('event persistence failed')); + const worker = await createEmailWorker(); + + try { + const job = await emailQueue.add( + 'send-email', + {emailId: email.id}, + { + jobId: `post-send-failure-${email.id}`, + attempts: 2, + backoff: {type: 'fixed', delay: 10}, + }, + ); + + await waitForJobState(job, 'failed'); + } finally { + await worker.close(); + } + + expect(sesMocks.sendRawEmail).toHaveBeenCalledOnce(); + await expect(prisma.email.findUniqueOrThrow({where: {id: email.id}})).resolves.toMatchObject({ + status: EmailStatus.SENT, + messageId: 'ses-already-accepted', + error: 'Post-send processing failed: event persistence failed', + }); + }); }); describe('Email Status Transitions', () => { diff --git a/apps/api/src/jobs/email-processor.ts b/apps/api/src/jobs/email-processor.ts index 6e17449fc..5daba4fed 100644 --- a/apps/api/src/jobs/email-processor.ts +++ b/apps/api/src/jobs/email-processor.ts @@ -5,14 +5,13 @@ * This is the only send path. `EmailService.sendEmail` used to hold a second copy of it, tested * while this one was not, and the two had drifted; it has been removed. The behaviour those tests * claimed to cover -- PENDING → SENDING → SENT, the failure transition, send idempotency, and - * attachments reaching SES -- is implemented here and is currently untested, because the job body - * is inline in `createEmailWorker` and cannot be called without a queue. Extracting it is worth - * doing before this logic is next changed. + * attachments reaching SES -- is implemented here. Integration tests exercise it through real + * queue jobs; keep new assertions on this path rather than reviving a second send implementation. */ import {EmailStatus} from '@plunk/db'; import type {SendEmailJobData} from '@plunk/types'; -import {type Job, Worker} from 'bullmq'; +import {type Job, UnrecoverableError, Worker} from 'bullmq'; import signale from 'signale'; import { @@ -83,6 +82,45 @@ function deriveWorkerConcurrency(rateLimit: number): number { return Math.max(MIN_CONCURRENCY, Math.min(derived, EMAIL_WORKER_MAX_CONCURRENCY)); } +function isExplicitlyRetryableSesFailure(error: unknown): boolean { + if (!error || typeof error !== 'object') return false; + + const {name, $metadata} = error as { + name?: string; + $metadata?: {httpStatusCode?: number}; + }; + const status = $metadata?.httpStatusCode; + + // A signed SES error response establishes that SES rejected this attempt. + // Transport errors without a response are ambiguous and must not be retried, + // because SES may have accepted the message before the connection failed. + return ( + status === 429 || + (status !== undefined && status >= 500) || + name === 'Throttling' || + name === 'ThrottlingException' || + name === 'TooManyRequestsException' + ); +} + +type SesAcceptance = {messageId: string; sentAt: Date}; + +async function checkpointSesAcceptance(job: Job, accepted: SesAcceptance): Promise { + try { + await job.updateData({ + ...job.data, + acceptedBySes: { + messageId: accepted.messageId, + sentAt: accepted.sentAt.toISOString(), + }, + }); + return true; + } catch (error) { + signale.error(`[EMAIL-PROCESSOR] Failed to checkpoint SES acceptance for ${job.data.emailId}:`, error); + return false; + } +} + export async function createEmailWorker() { // Fetch the rate limit (from env, AWS, or default) const rateLimit = await getEmailRateLimit(); @@ -109,12 +147,28 @@ export async function createEmailWorker() { throw new Error(`Email ${emailId} not found`); } - if (email.status !== EmailStatus.PENDING) { + const recoveredAcceptance = job.data.acceptedBySes + ? { + messageId: job.data.acceptedBySes.messageId, + sentAt: new Date(job.data.acceptedBySes.sentAt), + } + : undefined; + + if (email.status === EmailStatus.SENDING && !recoveredAcceptance) { + const message = 'Previous attempt ended without an SES acceptance checkpoint; not retried to avoid a duplicate'; + await prisma.email.update({ + where: {id: emailId}, + data: {status: EmailStatus.FAILED, error: message}, + }); + throw new UnrecoverableError(message); + } + + if (email.status !== EmailStatus.PENDING && !(email.status === EmailStatus.SENDING && recoveredAcceptance)) { return; } // Check if project is disabled - if (email.project.disabled) { + if (email.project.disabled && !recoveredAcceptance) { signale.warn(`[EMAIL-PROCESSOR] Project ${email.projectId} is disabled, cancelling email ${emailId}`); await prisma.email.update({ where: {id: emailId}, @@ -132,6 +186,11 @@ export async function createEmailWorker() { return; } + let acceptedBySes: SesAcceptance | undefined = recoveredAcceptance; + let acceptedPersisted = email.sentAt !== null; + let acceptanceCheckpointed = recoveredAcceptance !== undefined; + let sesSubmissionStarted = false; + try { // Update status to sending await prisma.email.update({ @@ -211,53 +270,60 @@ export async function createEmailWorker() { // Determine tracking based on project settings and email type const shouldTrack = EmailService.shouldTrackEmail(email.project.tracking, email.sourceType); - // Check for phishing/dangerous content before sending - const phishingCheck = await SecurityService.checkPhishingContent( - email.projectId, - email.project.name, - email.from, - formattedEmail.subject, - compiledHtml, - ); - - if (phishingCheck.shouldDisable) { - // Disable project immediately - await SecurityService.disableProjectForPhishing( + if (!acceptedBySes) { + // Check for phishing/dangerous content before sending + const phishingCheck = await SecurityService.checkPhishingContent( email.projectId, + email.project.name, + email.from, formattedEmail.subject, - phishingCheck.confidence, - 'Phishing content detected', + compiledHtml, ); - // Mark email as failed - await prisma.email.update({ - where: {id: emailId}, - data: { - status: EmailStatus.FAILED, - error: 'This email could not be sent. The project has been disabled. Please contact support.', + if (phishingCheck.shouldDisable) { + // Disable project immediately + await SecurityService.disableProjectForPhishing( + email.projectId, + formattedEmail.subject, + phishingCheck.confidence, + 'Phishing content detected', + ); + + // Mark email as failed + await prisma.email.update({ + where: {id: emailId}, + data: { + status: EmailStatus.FAILED, + error: 'This email could not be sent. The project has been disabled. Please contact support.', + }, + }); + + throw new UnrecoverableError(`Project ${email.projectId} has been disabled due to a policy violation`); + } + + // Send via AWS SES, then checkpoint acceptance in Redis before any + // database work. If Postgres is unavailable, the retry can finalize + // this exact message without submitting it again. + sesSubmissionStarted = true; + const result = await sendRawEmail({ + from: { + name: fromName, + email: fromEmail, }, + to: typeof recipient === 'string' ? [recipient] : [{name: recipient.name, email: recipient.email}], + content: { + subject: formattedEmail.subject, + html: compiledHtml, + }, + reply: email.replyTo || undefined, + headers: outboundHeaders, + tracking: shouldTrack, + attachments: email.attachments as {filename: string; content: string; contentType: string}[] | null, }); - - throw new Error(`Project ${email.projectId} has been disabled due to a policy violation`); + acceptedBySes = {messageId: result.messageId, sentAt: new Date()}; + acceptanceCheckpointed = await checkpointSesAcceptance(job, acceptedBySes); } - // Send via AWS SES - const result = await sendRawEmail({ - from: { - name: fromName, - email: fromEmail, - }, - to: typeof recipient === 'string' ? [recipient] : [{name: recipient.name, email: recipient.email}], - content: { - subject: formattedEmail.subject, - html: compiledHtml, - }, - reply: email.replyTo || undefined, - headers: outboundHeaders, - tracking: shouldTrack, - attachments: email.attachments as {filename: string; content: string; contentType: string}[] | null, - }); - // Mark as sent with SES message ID. // // Guarded on `sentAt` still being null so the campaign counter below is only @@ -267,10 +333,12 @@ export async function createEmailWorker() { where: {id: emailId, sentAt: null}, data: { status: EmailStatus.SENT, - sentAt: new Date(), - messageId: result.messageId, + sentAt: acceptedBySes.sentAt, + messageId: acceptedBySes.messageId, + error: null, }, }); + acceptedPersisted = true; // Zero rows means another run already stamped this email -- SES accepted the // message, then the job was retried. Everything below sends a second signal @@ -305,7 +373,7 @@ export async function createEmailWorker() { subject: formattedEmail.subject, from: email.from, fromName: email.fromName, - messageId: result.messageId, + messageId: acceptedBySes.messageId, emailId: email.id, templateId: email.templateId, campaignId: email.campaignId, @@ -319,15 +387,70 @@ export async function createEmailWorker() { } catch (error) { signale.error(`[EMAIL-PROCESSOR] Failed to send email ${emailId}:`, error); - // Mark as failed + if (acceptedBySes) { + const message = error instanceof Error ? error.message : 'Unknown error'; + + // SES accepted the message, so another attempt must never submit it + // again. Best-effort persistence keeps the delivery truthful even when + // a later billing/event/finalization step failed. + try { + await prisma.email.update({ + where: {id: emailId}, + data: { + status: EmailStatus.SENT, + sentAt: acceptedBySes.sentAt, + messageId: acceptedBySes.messageId, + error: `Post-send processing failed: ${message}`, + }, + }); + acceptedPersisted = true; + + if (email.campaignId) { + await CampaignService.finalizeIfDone(email.campaignId); + } + } catch (persistenceError) { + signale.error(`[EMAIL-PROCESSOR] Failed to persist accepted SES message ${emailId}:`, persistenceError); + } + + if (!acceptedPersisted && !acceptanceCheckpointed) { + acceptanceCheckpointed = await checkpointSesAcceptance(job, acceptedBySes); + } + + if (!acceptedPersisted) { + // A checkpointed retry finalizes the known SES message. Without one, + // the retry turns the stranded SENDING row into an explicit FAILED + // state rather than silently completing or risking a second send. + throw error; + } + + throw new UnrecoverableError(`SES accepted email ${emailId}, but post-send processing failed: ${message}`); + } + + const configuredAttempts = Math.max(1, job.opts.attempts ?? 1); + const attemptsExhausted = job.attemptsMade + 1 >= configuredAttempts; + const hasAmbiguousSesOutcome = + sesSubmissionStarted && !acceptedBySes && !isExplicitlyRetryableSesFailure(error); + const isTerminal = error instanceof UnrecoverableError || attemptsExhausted || hasAmbiguousSesOutcome; + const errorMessage = error instanceof Error ? error.message : 'Unknown error'; + const persistedError = hasAmbiguousSesOutcome + ? `SES outcome is unknown; not retried to avoid a duplicate: ${errorMessage}` + : errorMessage; + + // Keep retryable failures eligible for the next BullMQ attempt. The + // worker's entry guard only processes PENDING rows, so writing FAILED + // before attempts are exhausted silently turns the retry into a no-op. await prisma.email.update({ where: {id: emailId}, data: { - status: EmailStatus.FAILED, - error: error instanceof Error ? error.message : 'Unknown error', + status: isTerminal ? EmailStatus.FAILED : EmailStatus.PENDING, + error: persistedError, }, }); + if (hasAmbiguousSesOutcome) { + throw new UnrecoverableError(persistedError); + } + throw error; // Re-throw to trigger retry } }, diff --git a/packages/types/src/jobs/email.ts b/packages/types/src/jobs/email.ts index a3a57d5c9..b5a3c4b26 100644 --- a/packages/types/src/jobs/email.ts +++ b/packages/types/src/jobs/email.ts @@ -8,6 +8,11 @@ */ export interface SendEmailJobData { emailId: string; + /** SES acceptance persisted in Redis before database finalization. */ + acceptedBySes?: { + messageId: string; + sentAt: string; + }; } /**