From 0aba3fe907ee747fed248849f178124433234180 Mon Sep 17 00:00:00 2001 From: Sefa Senturk Date: Wed, 23 Sep 2026 16:53:47 +0200 Subject: [PATCH 1/2] Honour the SES send rate per second and retry throttled sends The "Send Rate" of an SES configuration ("the number of emails to send per second") only set the BullMQ worker's concurrency. A send round trip takes a fraction of a second, so `rate` workers in flight sent several times `rate` mails a second, SES answered "Maximum sending rate exceeded" for the rest, and the worker recorded those mails as FAILED without a retry: they were lost. The worker now carries a limiter of `rate` jobs per second next to the concurrency, so the queue never starts more sends per second than SES allows. Because a limiter is fixed at creation, a changed send rate replaces the worker instead of updating its concurrency; the queue and the jobs waiting in it are untouched, and the old worker closes once its in-flight sends are done. Email jobs get three attempts with exponential backoff. A throttled send (TooManyRequestsException or any error the SDK marks as throttling) is handed back to the queue while attempts remain and only fails the mail on the last one; every other error still fails it at once, as before. --- .../src/server/service/email-queue-service.ts | 101 +++++-- .../service/email-queue-service.unit.test.ts | 272 ++++++++++++++++++ 2 files changed, 356 insertions(+), 17 deletions(-) create mode 100644 apps/web/src/server/service/email-queue-service.unit.test.ts diff --git a/apps/web/src/server/service/email-queue-service.ts b/apps/web/src/server/service/email-queue-service.ts index f7fe3354f..a3f3ff7bf 100644 --- a/apps/web/src/server/service/email-queue-service.ts +++ b/apps/web/src/server/service/email-queue-service.ts @@ -23,24 +23,78 @@ type QueueEmailJob = TeamJob<{ isBulk?: boolean; }>; -function createQueueAndWorker(region: string, quota: number, suffix: string) { +// 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, }); +} + +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 }; } +// 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`, + ); + }); + } +} + +// 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 }, +}; + +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 +122,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 +148,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 +197,7 @@ export class EmailQueueService { isBulk, teamId, }, - { jobId: emailId, delay, ...DEFAULT_QUEUE_OPTIONS } + { jobId: emailId, delay, ...EMAIL_JOB_OPTIONS } ); } @@ -231,7 +288,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) }, })); @@ -461,6 +518,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 000000000..49dce99c0 --- /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(); + }); +}); From 1e125208406c265b8293d0f5e3f828b40ff844f9 Mon Sep 17 00:00:00 2001 From: Sefa Senturk Date: Wed, 23 Sep 2026 17:18:01 +0200 Subject: [PATCH 2/2] Docstrings on the queue's new and touched functions --- .../src/server/service/email-queue-service.ts | 51 ++++++++++++++----- 1 file changed, 38 insertions(+), 13 deletions(-) diff --git a/apps/web/src/server/service/email-queue-service.ts b/apps/web/src/server/service/email-queue-service.ts index a3f3ff7bf..023c79cdd 100644 --- a/apps/web/src/server/service/email-queue-service.ts +++ b/apps/web/src/server/service/email-queue-service.ts @@ -23,12 +23,16 @@ type QueueEmailJob = TeamJob<{ isBulk?: boolean; }>; -// 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. +/** + * 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(); @@ -42,6 +46,9 @@ function createWorker(queueName: string, quota: number) { }); } +/** + * The queue `${region}-${suffix}` and its first worker at `quota` per second. + */ function createQueueAndWorker(region: string, quota: number, suffix: string) { const connection = getRedis(); @@ -54,10 +61,15 @@ function createQueueAndWorker(region: string, quota: number, suffix: string) { return { queue, worker }; } -// 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. +/** + * 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, @@ -76,15 +88,22 @@ function replaceWorker( } } -// 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. +/** + * 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" || @@ -371,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 },