diff --git a/apps/web/src/server/service/email-queue-service.ts b/apps/web/src/server/service/email-queue-service.ts index f7fe3354..023c79cd 100644 --- a/apps/web/src/server/service/email-queue-service.ts +++ b/apps/web/src/server/service/email-queue-service.ts @@ -23,24 +23,97 @@ type QueueEmailJob = TeamJob<{ isBulk?: boolean; }>; -function createQueueAndWorker(region: string, quota: number, suffix: string) { +/** + * A worker for one email queue, limited to `quota` jobs per second. + * + * SES enforces its maximum send rate per second. Concurrency alone does not + * honour it: a send round trip takes a fraction of a second, so `quota` + * workers in flight send several times `quota` mails a second and SES + * throttles the rest. The limiter caps the jobs a queue's workers start per + * second (BullMQ keeps it in Redis, shared by every worker of the queue); + * the concurrency keeps that many sends in flight. + */ +function createWorker(queueName: string, quota: number) { const connection = getRedis(); - const queueName = `${region}-${suffix}`; - - const queue = new Queue(queueName, { connection, prefix: BULL_PREFIX, skipVersionCheck: true }); - // TODO: Add team context to job data when queueing - const worker = new Worker(queueName, createWorkerHandler(executeEmail), { + return new Worker(queueName, createWorkerHandler(executeEmail), { concurrency: quota, + limiter: { max: quota, duration: 1000 }, connection, prefix: BULL_PREFIX, skipVersionCheck: true, }); +} + +/** + * The queue `${region}-${suffix}` and its first worker at `quota` per second. + */ +function createQueueAndWorker(region: string, quota: number, suffix: string) { + const connection = getRedis(); + + const queueName = `${region}-${suffix}`; + + const queue = new Queue(queueName, { connection, prefix: BULL_PREFIX, skipVersionCheck: true }); + + const worker = createWorker(queueName, quota); return { queue, worker }; } +/** + * Puts a worker at the new `quota` on the region's queue in place of the + * current one. + * + * A worker's limiter is fixed when it is created, so a new send rate means a + * new worker on the same queue. The old one stops taking jobs and closes + * once its in-flight sends are done; the queue and the jobs waiting in it + * are untouched. + */ +function replaceWorker( + workers: Map, + region: string, + queueName: string, + quota: number, +) { + const previous = workers.get(region); + workers.set(region, createWorker(queueName, quota)); + if (previous) { + previous.close().catch((error) => { + logger.error( + { err: error, queue: queueName }, + `[EmailQueueService]: Failed to close the replaced worker`, + ); + }); + } +} + +/** + * Job options for every email: a send SES throttled ("Maximum sending rate + * exceeded") is retried with backoff instead of failing the mail; + * executeEmail decides per error whether to hand the job back to the queue + * (isThrottled) or to fail it. + */ +const EMAIL_JOB_OPTIONS = { + ...DEFAULT_QUEUE_OPTIONS, + attempts: 3, + backoff: { type: "exponential", delay: 1000 }, +}; + +/** + * Whether a send failed because SES refused it for the account's send rate: + * the SES error names for it, or any error the AWS SDK marks as throttling. + */ +function isThrottled(error: any): boolean { + return ( + error?.name === "TooManyRequestsException" || + error?.name === "ThrottlingException" || + error?.name === "Throttling" || + error?.$retryable?.throttling === true || + error?.$metadata?.httpStatusCode === 429 + ); +} + export class EmailQueueService { private static initialized = false; public static transactionalQueue = new Map>(); @@ -68,11 +141,12 @@ export class EmailQueueService { { region, transactionalQuota }, `[EmailQueueService]: Updating transactional quota for region` ); - const transactionalWorker = this.transactionalWorker.get(region); - if (transactionalWorker) { - transactionalWorker.concurrency = - transactionalQuota !== 0 ? transactionalQuota : 1; - } + replaceWorker( + this.transactionalWorker, + region, + `${region}-transaction`, + transactionalQuota !== 0 ? transactionalQuota : 1, + ); } else { logger.info( { region, transactionalQuota }, @@ -93,10 +167,12 @@ export class EmailQueueService { { region, marketingQuota }, `[EmailQueueService]: Updating marketing quota for region` ); - const marketingWorker = this.marketingWorker.get(region); - if (marketingWorker) { - marketingWorker.concurrency = marketingQuota !== 0 ? marketingQuota : 1; - } + replaceWorker( + this.marketingWorker, + region, + `${region}-marketing`, + marketingQuota !== 0 ? marketingQuota : 1, + ); } else { logger.info( { region, marketingQuota }, @@ -140,7 +216,7 @@ export class EmailQueueService { isBulk, teamId, }, - { jobId: emailId, delay, ...DEFAULT_QUEUE_OPTIONS } + { jobId: emailId, delay, ...EMAIL_JOB_OPTIONS } ); } @@ -231,7 +307,7 @@ export class EmailQueueService { opts: { jobId: job.emailId, // Use emailId as jobId delay: job.delay, - ...DEFAULT_QUEUE_OPTIONS, // Apply default options (attempts, backoff) + ...EMAIL_JOB_OPTIONS, // Apply default options (attempts, backoff) }, })); @@ -314,6 +390,12 @@ export class EmailQueueService { } } +/** + * Sends one queued email through SES and records the outcome on the email. + * A throttled send is thrown back to the queue while the job has attempts + * left, so the queue retries it after its backoff; any other error, and a + * throttle on the last attempt, marks the email FAILED. + */ async function executeEmail(job: QueueEmailJob) { logger.info( { emailId: job.data.emailId, elapsed: Date.now() - job.data.timestamp }, @@ -461,6 +543,16 @@ async function executeEmail(job: QueueEmailJob) { data: { sesEmailId: messageId, text, attachments: null, headers: null }, }); } catch (error: any) { + const attempt = job.attemptsMade + 1; + const attempts = job.opts.attempts ?? 1; + if (isThrottled(error) && attempt < attempts) { + logger.warn( + { emailId: email.id, attempt, attempts }, + `[EmailQueueService]: SES throttled the send, the queue retries it`, + ); + // rethrowing fails this attempt, so BullMQ retries the job after its backoff + throw error; + } await db.emailEvent.create({ data: { emailId: email.id, diff --git a/apps/web/src/server/service/email-queue-service.unit.test.ts b/apps/web/src/server/service/email-queue-service.unit.test.ts new file mode 100644 index 00000000..49dce99c --- /dev/null +++ b/apps/web/src/server/service/email-queue-service.unit.test.ts @@ -0,0 +1,272 @@ +import { beforeEach, describe, expect, it, vi } from "vitest"; +import type { Processor } from "bullmq"; + +const { + mockDb, + mockLogger, + mockQueueAdd, + mockQueueAddBulk, + mockSendRawEmail, + mockCheckEmailLimit, + workers, +} = vi.hoisted(() => ({ + mockDb: { + sesSetting: { findMany: vi.fn() }, + email: { findUnique: vi.fn(), update: vi.fn() }, + emailEvent: { create: vi.fn() }, + domain: { findUnique: vi.fn() }, + contact: { findUnique: vi.fn() }, + }, + mockLogger: { + debug: vi.fn(), + error: vi.fn(), + info: vi.fn(), + warn: vi.fn(), + }, + mockQueueAdd: vi.fn(), + mockQueueAddBulk: vi.fn(async () => []), + mockSendRawEmail: vi.fn(), + mockCheckEmailLimit: vi.fn(), + workers: [] as Array<{ + name: string; + processor: Processor; + opts: any; + close: ReturnType; + }>, +})); + +vi.mock("bullmq", () => ({ + Queue: class { + public name: string; + public add = mockQueueAdd; + public addBulk = mockQueueAddBulk; + constructor(name: string) { + this.name = name; + } + }, + Worker: class { + public close = vi.fn(async () => undefined); + constructor(name: string, processor: Processor, opts: any) { + workers.push({ name, processor, opts, close: this.close }); + } + }, +})); + +vi.mock("~/server/db", () => ({ db: mockDb })); +vi.mock("~/server/logger/log", () => ({ logger: mockLogger })); +vi.mock("~/server/redis", () => ({ + getRedis: () => ({}), + BULL_PREFIX: "test", +})); +vi.mock("~/server/aws/ses", () => ({ sendRawEmail: mockSendRawEmail })); +vi.mock("~/utils/ses-utils", () => ({ + getConfigurationSetName: vi.fn(async () => "config-set"), +})); +vi.mock("~/server/service/limit-service", () => ({ + LimitService: { checkEmailLimit: mockCheckEmailLimit }, +})); +vi.mock("~/server/queue/bullmq-context", () => ({ + createWorkerHandler: (handler: any) => handler, +})); +vi.mock("~/env", () => ({ + env: { AWS_DEFAULT_REGION: "us-east-1" }, +})); + +import { EmailQueueService } from "./email-queue-service"; + +const email = { + id: "email-1", + teamId: 7, + domainId: null, + to: ["someone@example.com"], + from: "Sender ", + subject: "Hello", + html: "

Hello

", + text: "Hello", + replyTo: null, + cc: [], + bcc: [], + attachments: null, + headers: null, + campaignId: null, + contactId: null, + inReplyToId: null, +}; + +// what the AWS SDK throws when SES refuses a send for its rate +const throttleError = Object.assign( + new Error("Maximum sending rate exceeded."), + { name: "TooManyRequestsException", $retryable: { throttling: true } }, +); + +function resetQueues() { + workers.length = 0; + EmailQueueService.transactionalQueue.clear(); + EmailQueueService.marketingQueue.clear(); + (EmailQueueService as any).transactionalWorker.clear(); + (EmailQueueService as any).marketingWorker.clear(); + (EmailQueueService as any).initialized = false; +} + +describe("EmailQueueService rate limiting", () => { + beforeEach(() => { + resetQueues(); + }); + + it("limits each worker to the send rate per second, not only in concurrency", () => { + EmailQueueService.initializeQueue("us-east-1", 14, 100); + + const transactional = workers.find( + (w) => w.name === "us-east-1-transaction", + ); + const marketing = workers.find((w) => w.name === "us-east-1-marketing"); + expect(transactional?.opts).toMatchObject({ + concurrency: 14, + limiter: { max: 14, duration: 1000 }, + }); + // a zero share still gets one worker slot, as before + expect(marketing?.opts).toMatchObject({ + concurrency: 1, + limiter: { max: 1, duration: 1000 }, + }); + }); + + it("replaces the workers when the send rate changes, since a limiter is fixed at creation", () => { + EmailQueueService.initializeQueue("us-east-1", 14, 100); + const [firstTransactional, firstMarketing] = workers; + + EmailQueueService.initializeQueue("us-east-1", 2, 50); + + expect(firstTransactional?.close).toHaveBeenCalled(); + expect(firstMarketing?.close).toHaveBeenCalled(); + const replaced = workers.slice(2); + expect(replaced.map((w) => [w.name, w.opts.limiter])).toEqual([ + ["us-east-1-transaction", { max: 1, duration: 1000 }], + ["us-east-1-marketing", { max: 1, duration: 1000 }], + ]); + // the queues themselves are kept: jobs already waiting stay where they are + expect(EmailQueueService.transactionalQueue.get("us-east-1")?.name).toBe( + "us-east-1-transaction", + ); + }); + + it("queues every email with retries and backoff", async () => { + mockDb.sesSetting.findMany.mockResolvedValue([]); + EmailQueueService.initializeQueue("us-east-1", 14, 100); + + await EmailQueueService.queueEmail("email-1", 7, "us-east-1", true); + await EmailQueueService.queueBulk([ + { + emailId: "email-2", + teamId: 7, + region: "us-east-1", + transactional: false, + }, + ]); + + const retryOptions = { + attempts: 3, + backoff: { type: "exponential", delay: 1000 }, + }; + expect(mockQueueAdd).toHaveBeenCalledWith( + "email-1", + expect.objectContaining({ emailId: "email-1" }), + expect.objectContaining({ jobId: "email-1", ...retryOptions }), + ); + expect(mockQueueAddBulk).toHaveBeenCalledWith([ + expect.objectContaining({ + opts: expect.objectContaining({ jobId: "email-2", ...retryOptions }), + }), + ]); + }); +}); + +describe("EmailQueueService throttled sends", () => { + beforeEach(() => { + resetQueues(); + mockDb.email.findUnique.mockResolvedValue(email); + mockCheckEmailLimit.mockResolvedValue({ isLimitReached: false }); + EmailQueueService.initializeQueue("us-east-1", 14, 100); + }); + + function process(job: { attemptsMade: number; attempts: number }) { + const worker = workers.find((w) => w.name === "us-east-1-transaction")!; + return worker.processor({ + id: email.id, + data: { emailId: email.id, timestamp: Date.now(), teamId: email.teamId }, + attemptsMade: job.attemptsMade, + opts: { attempts: job.attempts }, + } as unknown as Parameters[0]); + } + + it("hands a throttled send back to the queue while attempts remain", async () => { + mockSendRawEmail.mockRejectedValueOnce(throttleError); + + await expect(process({ attemptsMade: 0, attempts: 3 })).rejects.toBe( + throttleError, + ); + + // not failed: the job's backoff retries it + expect(mockDb.emailEvent.create).not.toHaveBeenCalled(); + expect(mockDb.email.update).not.toHaveBeenCalled(); + expect(mockLogger.warn).toHaveBeenCalledWith( + expect.objectContaining({ emailId: email.id, attempt: 1, attempts: 3 }), + expect.stringContaining("throttled"), + ); + }); + + it("fails a throttled send on its last attempt", async () => { + mockSendRawEmail.mockRejectedValueOnce(throttleError); + + await expect( + process({ attemptsMade: 2, attempts: 3 }), + ).resolves.toBeUndefined(); + + expect(mockDb.emailEvent.create).toHaveBeenCalledWith({ + data: expect.objectContaining({ + emailId: email.id, + status: "FAILED", + data: { error: throttleError.toString() }, + }), + }); + expect(mockDb.email.update).toHaveBeenCalledWith({ + where: { id: email.id }, + data: { latestStatus: "FAILED" }, + }); + }); + + it("still fails any other error on the first attempt", async () => { + const rejected = Object.assign( + new Error("Email address is not verified."), + { + name: "MessageRejected", + }, + ); + mockSendRawEmail.mockRejectedValueOnce(rejected); + + await expect( + process({ attemptsMade: 0, attempts: 3 }), + ).resolves.toBeUndefined(); + + expect(mockDb.emailEvent.create).toHaveBeenCalledWith({ + data: expect.objectContaining({ status: "FAILED" }), + }); + }); + + it("records the SES message id on a successful send", async () => { + mockSendRawEmail.mockResolvedValueOnce("ses-message-id"); + + await process({ attemptsMade: 0, attempts: 3 }); + + expect(mockDb.email.update).toHaveBeenCalledWith({ + where: { id: email.id }, + data: { + sesEmailId: "ses-message-id", + text: email.text, + attachments: null, + headers: null, + }, + }); + expect(mockDb.emailEvent.create).not.toHaveBeenCalled(); + }); +});