From d122703ae84271f9368e59f42013eb24575a79ea Mon Sep 17 00:00:00 2001 From: "Reichenbach, Michael" <755327+Silthus@users.noreply.github.com> Date: Tue, 27 Oct 2026 06:42:04 +0000 Subject: [PATCH 1/2] feat(workflows): cap sandbox sends per project and recipient each day Sandbox sends claim a daily allowance per project and per recipient address across the organization right before delivery. A reached allowance skips the send. SES throttles give the claim back. Co-Authored-By: Claude Opus 5.5 --- .../internal/workflows-email-sending-tiers.md | 8 +- nodejs/src/cdp/cdp-services.ts | 26 +- nodejs/src/cdp/config.ts | 5 + .../services/messaging/email.service.test.ts | 285 ++++++++++++++++-- .../cdp/services/messaging/email.service.ts | 71 ++++- .../messaging/sandbox-email-sender.test.ts | 8 +- .../messaging/sandbox-email-sender.ts | 90 +++++- .../rate-limiter/rate-limiter.service.test.ts | 109 +++++++ .../rate-limiter/rate-limiter.service.ts | 151 ++++++++++ nodejs/src/server.ts | 23 +- 10 files changed, 722 insertions(+), 54 deletions(-) diff --git a/docs/internal/workflows-email-sending-tiers.md b/docs/internal/workflows-email-sending-tiers.md index 3f828a01807f..c615d9bdd65d 100644 --- a/docs/internal/workflows-email-sending-tiers.md +++ b/docs/internal/workflows-email-sending-tiers.md @@ -89,7 +89,13 @@ The worker parses and normalizes the HTML, then appends a footer that names the It overrides root clipping and visibility styles and keeps the template's typography, so malformed markup may render differently from an own-domain send. A template whose HTML cannot keep the footer fails the send. The visible footer is best effort across email clients and template CSS; the signed tracking header still attributes the message. +Right before delivery, every sandbox send, test sends included, claims a daily allowance per project (`WORKFLOWS_SANDBOX_DAILY_TEAM_CAP`, one token per recipient copy) and per recipient address across the organization (`WORKFLOWS_SANDBOX_DAILY_RECIPIENT_CAP`). +The allowances are token buckets that refill their capacity over 24 hours, not calendar-day counters. +A reached allowance skips the send instead of delaying it. A limiter error, or an empty or invalid cap value, skips every sandbox send. +The worker checks members after the claim and gives the claim back when that check skips the send. +An SES throttle response also gives the claim back before the delivery retry. Returns are deduplicated, and Valkey ignores a return that arrives more than five minutes after it was issued. +Test sends run inline in the CDP API, so a sandbox-enabled CDP API opens the SES Valkey pool (`SES_RATE_LIMITER_VALKEY_*`) for the allowances. Without it, every sandbox test send skips. Each delivered send emits a `workflows sandbox email sent` event. Test sends never record `email_sent`, so they do not count toward the sending tier. Live workflow sends still count as `email_sent`, which the sending tier reads, until sandbox sends get their own metric. -Keep the switch off in production until the sandbox caps are deployed, and use it for test sends only until that metric is deployed. +Use the switch in production for test sends only until that metric is deployed. diff --git a/nodejs/src/cdp/cdp-services.ts b/nodejs/src/cdp/cdp-services.ts index 91ea747695ef..cf9ca2accf6d 100644 --- a/nodejs/src/cdp/cdp-services.ts +++ b/nodejs/src/cdp/cdp-services.ts @@ -174,6 +174,8 @@ export type CdpCoreServicesConfig = Pick< | 'SES_SANDBOX_TENANT_NAME' | 'SES_SANDBOX_CONFIGURATION_SET' | 'SES_SANDBOX_FROM_ADDRESS' + | 'WORKFLOWS_SANDBOX_DAILY_TEAM_CAP' + | 'WORKFLOWS_SANDBOX_DAILY_RECIPIENT_CAP' | 'EMAIL_SUPPRESSION_TRANSIENT_BOUNCE_THRESHOLD' | 'EMAIL_TEAM_SENDING_CAP_MODE' | 'EMAIL_TEAM_SENDING_CAP_HOURLY_BY_TIER' @@ -215,13 +217,15 @@ export interface CdpCoreServicesDeps { /** * SES Valkey pool shared with the SES rate limiter, opened only on pods whose * capabilities actually execute email actions (hogflow/email cyclotron workers). - * `null` on every other CDP consumer and on cdp-api so idle pods don't hold - * open connections against the SES Valkey instance. + * Sandbox-enabled cdp-api opens it too, for the sandbox daily caps of test sends. + * `null` on every other CDP consumer so idle pods don't hold open connections + * against the SES Valkey instance. */ emailValidationValkey: RedisV2 | null /** - * Dedicated frequency cap Valkey pool, opened only on the same pods as `emailValidationValkey`. - * `null` elsewhere, and when no host is configured, which turns the cap off. + * Dedicated frequency cap Valkey pool, opened only on the email workers. `null` elsewhere, + * cdp-api included because its test sends skip the cap, and when no host is configured, + * which turns the cap off. */ frequencyCapValkey: RedisV2 | null } @@ -459,6 +463,9 @@ export function createCdpCoreServices( const teamEmailRateLimiter = deps.emailValidationValkey ? new RateLimiterService(deps.emailValidationValkey, { name: 'team-email' }) : null + const sandboxEmailRateLimiter = deps.emailValidationValkey + ? new RateLimiterService(deps.emailValidationValkey, { name: 'sandbox-email' }) + : null const emailService = new EmailService( { sesAccessKeyId: config.SES_ACCESS_KEY_ID, @@ -487,9 +494,12 @@ export function createCdpCoreServices( tenantName: config.SES_SANDBOX_TENANT_NAME, configurationSetName: config.SES_SANDBOX_CONFIGURATION_SET, fromAddress: config.SES_SANDBOX_FROM_ADDRESS, + dailyTeamCap: config.WORKFLOWS_SANDBOX_DAILY_TEAM_CAP, + dailyRecipientCap: config.WORKFLOWS_SANDBOX_DAILY_RECIPIENT_CAP, }, deps.teamManager, - redis + redis, + sandboxEmailRateLimiter ), new OrganizationMembersService(deps.postgres) ) @@ -557,9 +567,9 @@ export function createCdpCoreServices( : undefined ) // MX verdicts live on the dedicated SES Valkey (same instance as the SES rate - // limiter, separate pool). The pool is created by the server only on pods - // whose capabilities execute email actions; everywhere else this is null - // and EmailValidationService degrades to the local cache + DNS. + // limiter, separate pool). The server creates the pool only on pods whose + // capabilities execute email actions and on sandbox-enabled cdp-api; everywhere + // else this is null and EmailValidationService degrades to the local cache + DNS. const emailValidationService = new EmailValidationService(deps.emailValidationValkey) const cohortMembershipRepository = new PostgresCohortMembershipRepository(deps.postgres) // Observer writes to both stores; the read source decides which verdict drives the metric. diff --git a/nodejs/src/cdp/config.ts b/nodejs/src/cdp/config.ts index 33d291e3b187..1e09831724e0 100644 --- a/nodejs/src/cdp/config.ts +++ b/nodejs/src/cdp/config.ts @@ -171,6 +171,9 @@ export type CdpConfig = ClickhouseConfig & { SES_SANDBOX_TENANT_NAME: string SES_SANDBOX_CONFIGURATION_SET: string SES_SANDBOX_FROM_ADDRESS: string + // Daily sandbox allowances per project and per recipient address. An empty or invalid value skips every sandbox send. + WORKFLOWS_SANDBOX_DAILY_TEAM_CAP: string + WORKFLOWS_SANDBOX_DAILY_RECIPIENT_CAP: string // Comma-separated allowlist of SNS Topic ARNs the SES webhook accepts events from. Empty string // means no restriction (dev/test); production should set this to the workflow SES topic ARN(s). SES_ALLOWED_SNS_TOPIC_ARNS: string @@ -416,6 +419,8 @@ export function getDefaultCdpConfig(): CdpConfig { SES_SANDBOX_TENANT_NAME: '', SES_SANDBOX_CONFIGURATION_SET: '', SES_SANDBOX_FROM_ADDRESS: '', + WORKFLOWS_SANDBOX_DAILY_TEAM_CAP: '25', + WORKFLOWS_SANDBOX_DAILY_RECIPIENT_CAP: '5', SES_ALLOWED_SNS_TOPIC_ARNS: '', EMAIL_SUPPRESSION_TRANSIENT_BOUNCE_THRESHOLD: 5, diff --git a/nodejs/src/cdp/services/messaging/email.service.test.ts b/nodejs/src/cdp/services/messaging/email.service.test.ts index 52bf4b1a1d73..d84b5280a370 100644 --- a/nodejs/src/cdp/services/messaging/email.service.test.ts +++ b/nodejs/src/cdp/services/messaging/email.service.test.ts @@ -14,7 +14,8 @@ import { } from '~/cdp/schema/cyclotron' import { CyclotronJobInvocationHogFunction, CyclotronJobInvocationResult } from '~/cdp/types' import { IngestionOutputs } from '~/common/outputs/ingestion-outputs' -import { RedisClient, createRedisV2PoolFromConfig } from '~/common/redis/redis-v2' +import { deleteKeysWithPrefix } from '~/common/redis/_tests/redis' +import { RedisClient, RedisV2, createRedisV2PoolFromConfig } from '~/common/redis/redis-v2' import { closeHub, createHub } from '~/common/utils/db/hub' import { PostgresRouter, PostgresUse } from '~/common/utils/db/postgres' import * as posthog from '~/common/utils/posthog' @@ -260,12 +261,32 @@ describe('EmailService', () => { ) return id } + const sandboxRedis: RedisV2 = { + useClient: async (_options, callback) => { + const client = await hub.redisPool.acquire() + try { + return await callback(client as unknown as RedisClient) + } finally { + await hub.redisPool.release(client) + } + }, + usePipeline: () => Promise.reject(new Error('The sandbox tests do not use pipelines')), + } const createSandboxService = ( enabled: boolean, - membersPostgres: PostgresRouter = hub.postgres, - routeSettings: Partial = {}, - teamEmailRateLimiter: RateLimiterService | null = null, - messageAssetsService?: MessageAssetsService + { + membersPostgres = hub.postgres, + senderSettings = {}, + teamEmailRateLimiter = null, + messageAssetsService, + dailyCapLimiter = new RateLimiterService(sandboxRedis, { name: 'sandbox-email' }), + }: { + membersPostgres?: PostgresRouter + senderSettings?: Partial + teamEmailRateLimiter?: RateLimiterService | null + messageAssetsService?: MessageAssetsService + dailyCapLimiter?: RateLimiterService | null + } = {} ): EmailService => { const workflowsConfig = new TeamWorkflowsConfigService(hub.postgres, hub.pubSub) jest.spyOn(workflowsConfig, 'getEmailSendingTier').mockResolvedValue(0) @@ -297,19 +318,13 @@ describe('EmailService', () => { tenantName: 'sandbox-tenant', configurationSetName: 'sandbox-email', fromAddress: 'fixed-sandbox@example.com', - ...routeSettings, + dailyTeamCap: '100', + dailyRecipientCap: '100', + ...senderSettings, }, hub.teamManager, - { - useClient: async (_options, callback) => { - const client = await hub.redisPool.acquire() - try { - return await callback(client as unknown as RedisClient) - } finally { - await hub.redisPool.release(client) - } - }, - } + sandboxRedis, + dailyCapLimiter ), new OrganizationMembersService(membersPostgres) ) @@ -373,15 +388,8 @@ describe('EmailService', () => { [extraOrganizationIds], 'test:delete-sandbox-organizations' ) - const client = await hub.redisPool.acquire() - try { - const keys = await client.keys(`@posthog/workflows-sandbox-blocked/${team.id}/*`) - if (keys.length) { - await client.del(...keys) - } - } finally { - await hub.redisPool.release(client) - } + await deleteKeysWithPrefix(sandboxRedis, `@posthog/workflows-sandbox-blocked/${team.id}/`) + await deleteKeysWithPrefix(sandboxRedis, `@posthog/workflows-sandbox-daily/{${team.organization_id}}`) capture.mockRestore() }) @@ -394,7 +402,7 @@ describe('EmailService', () => { ] as const)( 'skips when the sandbox sender is switched off because %s (isTest=%s)', async (_reason, isTest, enabled, routeSettings: Partial) => { - service = createSandboxService(enabled, hub.postgres, routeSettings) + service = createSandboxService(enabled, { senderSettings: routeSettings }) const result = await service.executeSendEmail(invocation, isTest) @@ -441,7 +449,7 @@ describe('EmailService', () => { const outputs = { produce: jest.fn().mockResolvedValue(undefined), } as unknown as IngestionOutputs<'message_assets'> - service = createSandboxService(true, hub.postgres, {}, null, new MessageAssetsService(outputs)) + service = createSandboxService(true, { messageAssetsService: new MessageAssetsService(outputs) }) invocation.state.actionId = 'send-email' invocation.queueParameters = createSandboxParams({ from: { integrationId: 4, email: 'override@example.com', name: 'Custom sender' }, @@ -513,9 +521,9 @@ describe('EmailService', () => { const claimAllOrNothingPair = jest .fn() .mockResolvedValue({ granted: false, deniedIndex: 0, retryAfterMs: 60_000, reserved: false }) - service = createSandboxService(true, hub.postgres, {}, { - claimAllOrNothingPair, - } as unknown as RateLimiterService) + service = createSandboxService(true, { + teamEmailRateLimiter: { claimAllOrNothingPair } as unknown as RateLimiterService, + }) const result = await service.executeSendEmail(invocation) @@ -680,7 +688,7 @@ describe('EmailService', () => { it('fails closed when the member check fails', async () => { const membersPostgres = new PostgresRouter(hub) await membersPostgres.end() - service = createSandboxService(true, membersPostgres) + service = createSandboxService(true, { membersPostgres }) invocation.queueParameters = createSandboxParams({ cc: memberEmail('cc') }) const result = await service.executeSendEmail(invocation) @@ -748,6 +756,221 @@ describe('EmailService', () => { expect(capture).toHaveBeenCalledTimes(3) expectBlockedEvent('switch_off', false, 0) }) + + describe('daily caps', () => { + const TEAM_CAP_REACHED = + "Skipping send: this email would go over the sandbox sender's daily limit for this project. Verify your own domain to send more." + const recipientCapReached = (addresses: string): string => + `Skipping send: these addresses reached the sandbox sender's daily limit per address: ${addresses}. Verify your own domain to send more.` + const CAP_CHECK_FAILED = + "Skipping send: could not check the sandbox sender's daily limit. Try again later, or verify your own domain to send more." + + const send = async ( + isTest: boolean, + params: Partial = {} + ): Promise> => { + invocation.state.vmState = { stack: [] } as any + invocation.queueParameters = createSandboxParams(params) + return await service.executeSendEmail(invocation, isTest) + } + const expectSent = (result: CyclotronJobInvocationResult): void => { + expect(result.error).toBeUndefined() + expect(result.skipped).not.toBe(true) + expect(result.invocation.state.vmState?.stack).toEqual([{ success: true }]) + } + const expectCapSkip = ( + result: CyclotronJobInvocationResult, + isTest: boolean, + message: string, + reason: 'cap_reached' | 'check_failed' + ): void => { + expect(result).toMatchObject({ finished: true, skipped: true, metrics: [], messageAssets: [] }) + expect(result.error).toBeUndefined() + expect(result.invocation.queueScheduledAt).toBeUndefined() + expect(result.invocation.state.vmState?.stack).toEqual([{ success: false }]) + expect(result.logs).toContainEqual(expect.objectContaining({ level: 'info', message })) + expect(capture).toHaveBeenCalledWith( + expect.objectContaining({ id: team.id }), + 'workflows sandbox email blocked', + expect.objectContaining({ reason, is_test: isTest }) + ) + } + + it.each([false, true])( + 'holds the project to its daily cap without charging denied or own-sender sends (isTest=%s)', + async (isTest) => { + service = createSandboxService(true, { + senderSettings: { dailyTeamCap: '3', dailyRecipientCap: '1' }, + }) + + expectSent(await send(isTest, { from: { integrationId: 1 } })) + expectSent(await send(isTest)) + expectCapSkip( + await send(isTest, { cc: memberEmail('cc'), bcc: memberEmail('bcc') }), + isTest, + TEAM_CAP_REACHED, + 'cap_reached' + ) + expectSent(await send(isTest, { to: { email: memberEmail('cc') } })) + expectSent(await send(isTest, { to: { email: memberEmail('bcc') } })) + expectCapSkip(await send(isTest), isTest, TEAM_CAP_REACHED, 'cap_reached') + expect(sendEmailSpy).toHaveBeenCalledTimes(4) + } + ) + + it.each([false, true])( + 'skips a send with recipients at the daily cap and leaves the others their budget (isTest=%s)', + async (isTest) => { + service = createSandboxService(true, { + senderSettings: { dailyTeamCap: '5', dailyRecipientCap: '1' }, + }) + + expectSent(await send(isTest, { to: { email: memberEmail('cc') }, cc: memberEmail('bcc') })) + expectCapSkip( + await send(isTest, { cc: memberEmail('cc').toUpperCase(), bcc: memberEmail('bcc') }), + isTest, + recipientCapReached(`${memberEmail('cc').toUpperCase()}, ${memberEmail('bcc')}`), + 'cap_reached' + ) + expectSent(await send(isTest)) + expect(sendEmailSpy).toHaveBeenCalledTimes(2) + } + ) + + it('charges a repeated address once per address and once per copy for the project', async () => { + service = createSandboxService(true, { + senderSettings: { dailyTeamCap: '3', dailyRecipientCap: '1' }, + }) + + expectSent(await send(false, { cc: memberEmail().toUpperCase(), bcc: ` ${memberEmail()} ` })) + expectCapSkip( + await send(false, { to: { email: memberEmail('cc') } }), + false, + TEAM_CAP_REACHED, + 'cap_reached' + ) + expect(sendEmailSpy).toHaveBeenCalledTimes(1) + }) + + it('refills a spent allowance over a day rather than seconds', async () => { + service = createSandboxService(true, { + senderSettings: { dailyTeamCap: '1', dailyRecipientCap: '1' }, + }) + expectSent(await send(true)) + const ageBuckets = async (ageMs: number): Promise => { + await sandboxRedis.useClient({ name: 'test:age-sandbox-daily-caps' }, async (client) => { + const [seconds, microseconds] = await client.time() + const redisNow = Number(seconds) * 1000 + Math.floor(Number(microseconds) / 1000) + const keys = await client.keys( + `@posthog/workflows-sandbox-daily/{${team.organization_id}}/*` + ) + for (const key of keys.filter((key) => !key.includes('/claimed/'))) { + await client.hset(key, 'ts', redisNow - ageMs) + } + }) + } + + await ageBuckets(12 * 60 * 60 * 1000) + expectCapSkip(await send(true), true, TEAM_CAP_REACHED, 'cap_reached') + await ageBuckets(24 * 60 * 60 * 1000 + 1000) + expectSent(await send(true)) + expect(sendEmailSpy).toHaveBeenCalledTimes(2) + }) + + it.each([false, true])( + 'spends the daily caps once for a send that SES throttles before accepting it (isTest=%s)', + async (isTest) => { + service = createSandboxService(true, { + senderSettings: { dailyTeamCap: '2', dailyRecipientCap: '1' }, + }) + sendEmailSpy.mockRejectedValueOnce(new ThrottlingException('Rate exceeded')) + + const throttled = await send(isTest, { cc: memberEmail('cc') }) + expect(throttled.finished).toBe(false) + expectSent(await service.executeSendEmail(throttled.invocation, isTest)) + expectCapSkip( + await send(isTest, { to: { email: memberEmail('bcc') } }), + isTest, + TEAM_CAP_REACHED, + 'cap_reached' + ) + expect(sendEmailSpy).toHaveBeenCalledTimes(2) + } + ) + + it('checks members after a slow cap claim and gives the claim back when one left', async () => { + const now = Date.now() + const clock = jest.spyOn(Date, 'now').mockReturnValue(now) + const dailyCapLimiter = new RateLimiterService(sandboxRedis, { name: 'sandbox-email' }) + const claim = dailyCapLimiter.claimAllOrNothing.bind(dailyCapLimiter) + jest.spyOn(dailyCapLimiter, 'claimAllOrNothing').mockImplementationOnce(async (buckets) => { + const grant = await claim(buckets) + await hub.postgres.query( + PostgresUse.COMMON_WRITE, + 'UPDATE posthog_user SET is_email_verified = false WHERE id = $1', + [memberId], + 'test:revoke-sandbox-verification-during-cap-claim' + ) + clock.mockReturnValue(now + 61_000) + return grant + }) + try { + service = createSandboxService(true, { + dailyCapLimiter, + senderSettings: { dailyTeamCap: '1', dailyRecipientCap: '1' }, + }) + + const blocked = await send(false) + expect(blocked).toMatchObject({ finished: true, skipped: true, metrics: [] }) + expect(sendEmailSpy).not.toHaveBeenCalled() + expectBlockedEvent('recipient_unverified', false, 1) + + await hub.postgres.query( + PostgresUse.COMMON_WRITE, + 'UPDATE posthog_user SET is_email_verified = true WHERE id = $1', + [memberId], + 'test:restore-sandbox-verification' + ) + clock.mockReturnValue(now + 122_000) + expectSent(await send(false)) + } finally { + clock.mockRestore() + } + }) + + it.each([ + [ + 'the limiter is unreachable', + false, + { + dailyCapLimiter: new RateLimiterService( + { + useClient: () => Promise.reject(new Error('Connection is closed.')), + usePipeline: () => Promise.reject(new Error('Connection is closed.')), + }, + { name: 'sandbox-email' } + ), + }, + ], + ['no limiter is configured', true, { dailyCapLimiter: null }], + ['the project cap is empty', false, { senderSettings: { dailyTeamCap: '' } }], + ['the recipient cap is blank', true, { senderSettings: { dailyRecipientCap: ' ' } }], + ['the project cap is zero', true, { senderSettings: { dailyTeamCap: '0' } }], + ['the recipient cap is fractional', false, { senderSettings: { dailyRecipientCap: '1.5' } }], + ['the project cap has trailing text', true, { senderSettings: { dailyTeamCap: '25 per day' } }], + ['the recipient cap is negative', false, { senderSettings: { dailyRecipientCap: '-5' } }], + ] as const)('skips every send when %s (isTest=%s)', async (_name, isTest, options) => { + service = createSandboxService(true, options) + + expectCapSkip( + await send(isTest, { cc: memberEmail('cc') }), + isTest, + CAP_CHECK_FAILED, + 'check_failed' + ) + expect(sendEmailSpy).not.toHaveBeenCalled() + }) + }) }) describe('integration validation', () => { beforeEach(async () => { diff --git a/nodejs/src/cdp/services/messaging/email.service.ts b/nodejs/src/cdp/services/messaging/email.service.ts index 23c3b61aa1ea..10efa7f9b92b 100644 --- a/nodejs/src/cdp/services/messaging/email.service.ts +++ b/nodejs/src/cdp/services/messaging/email.service.ts @@ -40,7 +40,7 @@ import { RecipientTokensService, preferencesTokenSourceForInvocation, } from './recipient-tokens.service' -import { SandboxEmailBlockReason, SandboxEmailSender } from './sandbox-email-sender' +import { SandboxDailyCapGrant, SandboxEmailBlockReason, SandboxEmailSender } from './sandbox-email-sender' const sesThrottleResponsesTotal = new Counter({ name: 'cdp_ses_throttle_responses_total', @@ -438,8 +438,20 @@ const SANDBOX_RECIPIENT_BLOCK_COPY: Record< label: ' for these addresses: ', guidance: 'Try again, or verify your own domain to send to anyone.', }, + cap_reached: { + explanation: "Skipping send: these addresses reached the sandbox sender's daily limit per address", + label: ': ', + guidance: 'Verify your own domain to send more.', + }, } +const SANDBOX_CAP_SKIP_MESSAGES = { + project_cap_reached: + "Skipping send: this email would go over the sandbox sender's daily limit for this project. Verify your own domain to send more.", + check_failed: + "Skipping send: could not check the sandbox sender's daily limit. Try again later, or verify your own domain to send more.", +} as const + export class EmailService { sesV2Client: SESv2Client | null @@ -813,10 +825,15 @@ export class EmailService { ...(route.ccAddresses ?? []), ...(route.bccAddresses ?? []), ] + const capGrant = await this.claimSandboxDailyCaps(result, deliveryRecipients, isTest) + if (!capGrant) { + return result + } if (!(await this.checkSandboxRecipients(result, deliveryRecipients, isTest))) { + await this.sandboxSender!.returnDailyCaps(capGrant) return result } - await this.sendEmailWithSES(result, deliveryParams, from, false, isTest, route) + await this.sendSandboxEmail(result, deliveryParams, from, isTest, route, capGrant) await this.sandboxSender!.captureSent(invocation.teamId, isTest, deliveryRecipients.length) break } @@ -1075,10 +1092,56 @@ export class EmailService { if (!blockedRecipients.length) { return true } + await this.blockSandboxRecipients(result, blockedRecipients, reason, isTest) + return false + } + + private async blockSandboxRecipients( + result: CyclotronJobInvocationResult, + blockedRecipients: string[], + reason: SandboxRecipientBlockReason, + isTest: boolean + ): Promise { const { explanation, label, guidance } = SANDBOX_RECIPIENT_BLOCK_COPY[reason] this.skipSandboxSend(result, `${explanation}${label}${blockedRecipients.join(', ')}. ${guidance}`) - await this.sandboxSender?.captureBlocked(teamId, isTest, reason, blockedRecipients.length) - return false + await this.sandboxSender?.captureBlocked(result.invocation.teamId, isTest, reason, blockedRecipients.length) + } + + private async claimSandboxDailyCaps( + result: CyclotronJobInvocationResult, + recipients: string[], + isTest: boolean + ): Promise { + const claim = await this.sandboxSender!.claimDailyCaps(result.invocation.teamId, recipients) + if (claim.type === 'granted') { + return claim + } + if (claim.type === 'recipient_cap_reached') { + await this.blockSandboxRecipients(result, claim.addresses, 'cap_reached', isTest) + return null + } + this.skipSandboxSend(result, SANDBOX_CAP_SKIP_MESSAGES[claim.type]) + const reason = claim.type === 'check_failed' ? 'check_failed' : 'cap_reached' + await this.sandboxSender!.captureBlocked(result.invocation.teamId, isTest, reason, recipients.length) + return null + } + + private async sendSandboxEmail( + result: CyclotronJobInvocationResult, + params: CyclotronInvocationQueueParametersEmailType, + from: { email: string; name: string }, + isTest: boolean, + route: SesRoute, + capGrant: SandboxDailyCapGrant + ): Promise { + try { + await this.sendEmailWithSES(result, params, from, false, isTest, route) + } catch (error) { + if (error instanceof SESThrottleError) { + await this.sandboxSender!.returnDailyCaps(capGrant) + } + throw error + } } // Per-send tracking decision, combining two independent controls: diff --git a/nodejs/src/cdp/services/messaging/sandbox-email-sender.test.ts b/nodejs/src/cdp/services/messaging/sandbox-email-sender.test.ts index 5224c728a867..08ace91b93af 100644 --- a/nodejs/src/cdp/services/messaging/sandbox-email-sender.test.ts +++ b/nodejs/src/cdp/services/messaging/sandbox-email-sender.test.ts @@ -17,9 +17,12 @@ const sender = new SandboxEmailSender( tenantName: 'sandbox-tenant', configurationSetName: 'sandbox-email', fromAddress: 'fixed-sandbox@example.com', + dailyTeamCap: '25', + dailyRecipientCap: '5', }, teamManager, - blockedEventRedis + blockedEventRedis, + null ) const params = (html: string, text?: string, preheader?: string): CyclotronInvocationQueueParametersEmailType => ({ type: 'email', @@ -68,7 +71,8 @@ describe('sandbox email content', () => { fromAddress: ' fixed-sandbox@example.com ', }, teamManager, - blockedEventRedis + blockedEventRedis, + null ) const result = await paddedSender.withIdentificationFooter( params('

Hello

'), diff --git a/nodejs/src/cdp/services/messaging/sandbox-email-sender.ts b/nodejs/src/cdp/services/messaging/sandbox-email-sender.ts index a3bc0457604c..604858b86682 100644 --- a/nodejs/src/cdp/services/messaging/sandbox-email-sender.ts +++ b/nodejs/src/cdp/services/messaging/sandbox-email-sender.ts @@ -1,3 +1,4 @@ +import { createHash, randomUUID } from 'crypto' import { type DefaultTreeAdapterMap, defaultTreeAdapter, parse, parseFragment, serialize } from 'parse5' import { CyclotronInvocationQueueParametersEmailType } from '~/cdp/schema/cyclotron' @@ -6,13 +7,19 @@ import { logger } from '~/common/utils/logger' import { captureTeamEvent } from '~/common/utils/posthog' import { TeamManager } from '~/common/utils/team-manager' +import { ClaimRequest, RateLimiterService } from '../rate-limiter/rate-limiter.service' import { maybeAddPreheaderToEmail } from './helpers/preheader' +const SECONDS_PER_DAY = 86400 +const DAILY_CAP_TTL_SECONDS = 2 * SECONDS_PER_DAY + export interface SandboxEmailSenderConfig { enabled: boolean tenantName: string configurationSetName: string fromAddress: string + dailyTeamCap: string + dailyRecipientCap: string } const ROUTE_SETTING_NAMES = { @@ -27,7 +34,49 @@ function emptyRouteSettings(config: SandboxEmailSenderConfig): string[] { .map((setting) => ROUTE_SETTING_NAMES[setting]) } -export type SandboxEmailBlockReason = 'switch_off' | 'recipient_not_member' | 'recipient_unverified' | 'check_failed' +export type SandboxEmailBlockReason = + | 'switch_off' + | 'recipient_not_member' + | 'recipient_unverified' + | 'check_failed' + | 'cap_reached' + +export type SandboxDailyCapGrant = { type: 'granted'; claimId: string; buckets: ClaimRequest[] } + +export type SandboxDailyCapClaim = + | SandboxDailyCapGrant + | { type: 'project_cap_reached' } + | { type: 'recipient_cap_reached'; addresses: string[] } + | { type: 'check_failed' } + +function dailyBucket(key: string, requested: number, capacity: number): ClaimRequest { + return { key, requested, capacity, refillPerSecond: capacity / SECONDS_PER_DAY, ttlSeconds: DAILY_CAP_TTL_SECONDS } +} + +function parseDailyCap(value: string): number | null { + const cap = /^\d+$/.test(value) ? Number(value) : null + return cap !== null && Number.isSafeInteger(cap) && cap > 0 ? cap : null +} + +// The organization hash tag keeps the project and recipient buckets in one Valkey cluster slot. +function dailyCapKeyPrefix(organizationId: string): string { + return `@posthog/workflows-sandbox-daily/{${organizationId}}` +} + +function addressDigest(address: string): string { + return createHash('sha256').update(address.toLowerCase()).digest('hex') +} + +function distinctAddresses(recipients: string[]): string[] { + const addresses = new Map() + for (const recipient of recipients) { + const trimmed = recipient.trim() + if (!addresses.has(trimmed.toLowerCase())) { + addresses.set(trimmed.toLowerCase(), trimmed) + } + } + return [...addresses.values()] +} const BLOCKED_EVENT_WINDOW_SECONDS = 3600 @@ -118,7 +167,8 @@ export class SandboxEmailSender { constructor( public readonly config: SandboxEmailSenderConfig, private teamManager: TeamManager, - private blockedEventRedis: Pick + private blockedEventRedis: Pick, + private dailyCapLimiter: RateLimiterService | null ) { this.config = { ...config, @@ -135,6 +185,42 @@ export class SandboxEmailSender { this.canSend = config.enabled && !emptySettings.length } + public async claimDailyCaps(teamId: number, recipients: string[]): Promise { + const dailyTeamCap = parseDailyCap(this.config.dailyTeamCap) + const dailyRecipientCap = parseDailyCap(this.config.dailyRecipientCap) + if (!this.dailyCapLimiter || dailyTeamCap === null || dailyRecipientCap === null) { + logger.error('Sandbox email daily caps are not configured correctly', { teamId }) + return { type: 'check_failed' } + } + const team = await this.teamManager.getTeam(teamId) + if (!team) { + return { type: 'check_failed' } + } + const prefix = dailyCapKeyPrefix(team.organization_id) + const addresses = distinctAddresses(recipients) + const buckets = [ + dailyBucket(`${prefix}/project/${teamId}`, recipients.length, dailyTeamCap), + ...addresses.map((address) => + dailyBucket(`${prefix}/recipient/${addressDigest(address)}`, 1, dailyRecipientCap) + ), + ] + const claim = await this.dailyCapLimiter.claimAllOrNothing(buckets) + if (claim.granted) { + return { type: 'granted', claimId: randomUUID(), buckets } + } + if (claim.deniedIndexes === null) { + return { type: 'check_failed' } + } + if (claim.deniedIndexes.includes(0)) { + return { type: 'project_cap_reached' } + } + return { type: 'recipient_cap_reached', addresses: claim.deniedIndexes.map((index) => addresses[index - 1]) } + } + + public async returnDailyCaps(grant: SandboxDailyCapGrant): Promise { + await this.dailyCapLimiter?.returnClaim(grant.claimId, grant.buckets) + } + public async withIdentificationFooter( params: CyclotronInvocationQueueParametersEmailType, senderName: string, diff --git a/nodejs/src/cdp/services/rate-limiter/rate-limiter.service.test.ts b/nodejs/src/cdp/services/rate-limiter/rate-limiter.service.test.ts index 490f0fee49ef..effd952bdee7 100644 --- a/nodejs/src/cdp/services/rate-limiter/rate-limiter.service.test.ts +++ b/nodejs/src/cdp/services/rate-limiter/rate-limiter.service.test.ts @@ -4,8 +4,10 @@ import { register } from 'prom-client' import { deleteKeysWithPrefix } from '~/common/redis/_tests/redis' import { RedisV2, createRedisV2PoolFromConfig } from '~/common/redis/redis-v2' import { closeHub, createHub } from '~/common/utils/db/hub' +import { TeamManager } from '~/common/utils/team-manager' import { Hub } from '~/types' +import { SandboxEmailSender } from '../messaging/sandbox-email-sender' import { RateLimiterService } from './rate-limiter.service' const KEY = `@posthog-test/ses-rate-limiter/${randomUUID()}/bucket` @@ -413,4 +415,111 @@ describe('RateLimiterService', () => { expect([first.reserved, second.reserved, third.reserved]).toEqual([true, false, false]) }) }) + + it('shares recipient allowances across projects in an organization while keeping project allowances separate', async () => { + const organizations = { 1: randomUUID(), 3: randomUUID() } + const teamManager = { + getTeam: (id: number) => + Promise.resolve({ organization_id: id === 3 ? organizations[3] : organizations[1] }), + } as unknown as TeamManager + const sender = new SandboxEmailSender( + { + enabled: true, + tenantName: 'sandbox', + configurationSetName: 'sandbox', + fromAddress: 'sender@example.com', + dailyTeamCap: '1', + dailyRecipientCap: '1', + }, + teamManager, + redis, + limiter + ) + + try { + expect(await sender.claimDailyCaps(1, ['member@example.com'])).toMatchObject({ type: 'granted' }) + expect(await sender.claimDailyCaps(2, ['member@example.com'])).toEqual({ + type: 'recipient_cap_reached', + addresses: ['member@example.com'], + }) + expect(await sender.claimDailyCaps(2, ['other@example.com'])).toMatchObject({ type: 'granted' }) + expect(await sender.claimDailyCaps(3, ['member@example.com'])).toMatchObject({ type: 'granted' }) + expect(await sender.claimDailyCaps(2, ['third@example.com'])).toEqual({ type: 'project_cap_reached' }) + } finally { + for (const organizationId of Object.values(organizations)) { + await deleteKeysWithPrefix(redis, `@posthog/workflows-sandbox-daily/{${organizationId}}`) + } + } + }) + + describe('claimAllOrNothing', () => { + it('spends an unanswered claim once when the same EVAL is replayed', async () => { + const bucket = { key: `${KEY}/claim-replay`, requested: 1, capacity: 2, refillPerSecond: 0 } + const replayingRedis: RedisV2 = { + ...redis, + useClient: async (options, callback) => + redis.useClient(options, async (client) => { + const evaluate = client.eval.bind(client) + const replay = jest.spyOn(client, 'eval').mockImplementation(async (...args) => { + await evaluate(...args) + return evaluate(...args) + }) + try { + return await callback(client) + } finally { + replay.mockRestore() + } + }), + } + const replayingLimiter = new RateLimiterService(replayingRedis, { name: 'ses-rate-limiter' }) + + expect(await replayingLimiter.claimAllOrNothing([bucket])).toEqual({ granted: true }) + expect(await limiter.claimAllOrNothing([bucket])).toEqual({ granted: true }) + expect(await limiter.claimAllOrNothing([bucket])).toEqual({ granted: false, deniedIndexes: [0] }) + }) + + it('reports a limiter error without naming a denied bucket', async () => { + const unreachable = new RateLimiterService( + { + useClient: () => Promise.reject(new Error('Connection is closed.')), + usePipeline: () => Promise.reject(new Error('Connection is closed.')), + }, + { name: 'ses-rate-limiter' } + ) + + expect( + await unreachable.claimAllOrNothing([ + { key: `${KEY}/unreachable`, requested: 1, capacity: 1, refillPerSecond: 0 }, + ]) + ).toEqual({ granted: false, deniedIndexes: null }) + }) + }) + + describe('returnClaim', () => { + it('gives a claim back once, even when the return is replayed', async () => { + const bucket = { key: `${KEY}/return`, requested: 1, capacity: 2, refillPerSecond: 0 } + expect(await limiter.claimAllOrNothing([bucket])).toEqual({ granted: true }) + expect(await limiter.claimAllOrNothing([bucket])).toEqual({ granted: true }) + + await limiter.returnClaim('claim-1', [bucket]) + await limiter.returnClaim('claim-1', [bucket]) + + expect(await limiter.claimAllOrNothing([bucket])).toEqual({ granted: true }) + expect(await limiter.claimAllOrNothing([bucket])).toEqual({ granted: false, deniedIndexes: [0] }) + }) + + it('ignores a return that reaches Valkey after its window', async () => { + const bucket = { key: `${KEY}/late-return`, requested: 1, capacity: 1, refillPerSecond: 0 } + expect(await limiter.claimAllOrNothing([bucket])).toEqual({ granted: true }) + + const issuedLongAgo = jest.spyOn(Date, 'now').mockReturnValue(Date.now() - 60 * 60 * 1000) + try { + await limiter.returnClaim('claim-1', [bucket]) + } finally { + issuedLongAgo.mockRestore() + } + + expect(await limiter.claimAllOrNothing([bucket])).toEqual({ granted: false, deniedIndexes: [0] }) + }) + }) }) diff --git a/nodejs/src/cdp/services/rate-limiter/rate-limiter.service.ts b/nodejs/src/cdp/services/rate-limiter/rate-limiter.service.ts index 1d4b178a0574..3e09da89b2cb 100644 --- a/nodejs/src/cdp/services/rate-limiter/rate-limiter.service.ts +++ b/nodejs/src/cdp/services/rate-limiter/rate-limiter.service.ts @@ -1,3 +1,4 @@ +import { randomUUID } from 'crypto' import { Counter, Histogram } from 'prom-client' import { RedisV2 } from '~/common/redis/redis-v2' @@ -249,6 +250,77 @@ end return {1, 0, 0, 0} ` +const CLAIM_ALL_OR_NOTHING_LUA = ` +local time = redis.call('TIME') +local now = tonumber(time[1]) * 1000 + math.floor(tonumber(time[2]) / 1000) +local claimKey = KEYS[#KEYS] +local previous = redis.call('GET', claimKey) +if previous then + return cjson.decode(previous) +end +if now > tonumber(ARGV[#ARGV - 1]) then + return {0} +end +local replayTtlMs = tonumber(ARGV[#ARGV]) +local available = {} +local denied = {0} +for i = 1, #KEYS - 1 do + local offset = (i - 1) * 4 + local requested = tonumber(ARGV[offset + 1]) + local capacity = tonumber(ARGV[offset + 2]) + local refillPerSecond = tonumber(ARGV[offset + 3]) + local existing = redis.call('hmget', KEYS[i], 'ts', 'pool') + local avail + if existing[1] == false then + avail = capacity + else + local elapsedMs = math.max(0, now - tonumber(existing[1])) + local currentTokens = capacity + if existing[2] ~= false then + currentTokens = tonumber(existing[2]) + end + avail = math.min(capacity, currentTokens + (elapsedMs / 1000.0) * refillPerSecond) + end + available[i] = avail + if avail < requested then + table.insert(denied, i) + end +end +if #denied > 1 then + redis.call('SET', claimKey, cjson.encode(denied), 'NX', 'PX', replayTtlMs) + return denied +end +for i = 1, #KEYS - 1 do + local offset = (i - 1) * 4 + redis.call('hset', KEYS[i], 'ts', now, 'pool', available[i] - tonumber(ARGV[offset + 1])) + redis.call('expire', KEYS[i], tonumber(ARGV[offset + 4])) +end +local granted = {1} +redis.call('SET', claimKey, cjson.encode(granted), 'NX', 'PX', replayTtlMs) +return granted +` + +// ioredis can replay unanswered commands, so returns need deduplication and an expiry. +const RETURN_CLAIM_LUA = ` +local time = redis.call('TIME') +local now = tonumber(time[1]) * 1000 + math.floor(tonumber(time[2]) / 1000) +local buckets = #KEYS - 1 +if now > tonumber(ARGV[buckets + 1]) then + return 0 +end +if not redis.call('set', KEYS[#KEYS], 1, 'NX', 'PX', ARGV[buckets + 2]) then + return 0 +end +for i = 1, buckets do + if redis.call('hexists', KEYS[i], 'pool') == 1 then + redis.call('hincrbyfloat', KEYS[i], 'pool', ARGV[i]) + end +end +return 1 +` + +const RETURN_WINDOW_MS = 5 * 60 * 1000 + export interface RateLimiterConfig { /** Logical name for metrics/logging only (e.g. 'ses'). */ name: string @@ -373,6 +445,61 @@ export class RateLimiterService { } } + // All bucket keys must share a hash tag because clustered Valkey rejects cross-slot Lua calls. + public async claimAllOrNothing( + buckets: ClaimRequest[] + ): Promise<{ granted: true } | { granted: false; deniedIndexes: number[] | null }> { + const endTimer = claimLatency.startTimer({ limiter: this.config.name }) + const keys = buckets.map((bucket) => bucket.key) + const claimId = randomUUID() + const deadline = Date.now() + RETURN_WINDOW_MS + try { + const result = await this.valkey.useClient( + { name: `rate-limiter:${this.config.name}:claimAllOrNothing`, timeout: 1000 }, + (client) => + client.eval( + CLAIM_ALL_OR_NOTHING_LUA, + buckets.length + 1, + ...keys, + `${keys[0]}/claimed/${claimId}`, + ...buckets.flatMap((bucket) => + [bucket.requested, bucket.capacity, bucket.refillPerSecond, bucket.ttlSeconds ?? 3600].map( + String + ) + ), + String(deadline), + String(2 * RETURN_WINDOW_MS) + ) + ) + if (Array.isArray(result) && result.length === 1 && result[0] === 1) { + claimCounter.inc({ limiter: this.config.name, result: 'granted_full' }) + return { granted: true } + } + if ( + Array.isArray(result) && + result[0] === 0 && + result.length > 1 && + result.slice(1).every((index) => Number.isInteger(index) && index >= 1 && index <= buckets.length) && + new Set(result.slice(1)).size === result.length - 1 + ) { + claimCounter.inc({ limiter: this.config.name, result: 'denied' }) + return { granted: false, deniedIndexes: result.slice(1).map((index) => index - 1) } + } + logger.warn('🪙', `RateLimiterService(${this.config.name}) claim returned invalid result`, { + keys, + raw: result, + }) + claimCounter.inc({ limiter: this.config.name, result: 'valkey_error' }) + return { granted: false, deniedIndexes: null } + } catch (err) { + logger.warn('🪙', `RateLimiterService(${this.config.name}) claim threw`, { keys, error: String(err) }) + claimCounter.inc({ limiter: this.config.name, result: 'valkey_error' }) + return { granted: false, deniedIndexes: null } + } finally { + endTimer() + } + } + /** * Atomically claim `requested` tokens from BOTH buckets, or neither. A denial consumes no * tokens, so a caller that retries a multi-token claim cannot drain the buckets while never @@ -449,4 +576,28 @@ export class RateLimiterService { endTimer() } } + + public async returnClaim(claimId: string, buckets: Pick[]): Promise { + const keys = buckets.map((bucket) => bucket.key) + try { + await this.valkey.useClient( + { name: `rate-limiter:${this.config.name}:returnClaim`, timeout: 1000 }, + (client) => + client.eval( + RETURN_CLAIM_LUA, + buckets.length + 1, + ...keys, + `${keys[0]}/returned/${claimId}`, + ...buckets.map((bucket) => String(bucket.requested)), + String(Date.now() + RETURN_WINDOW_MS), + String(2 * RETURN_WINDOW_MS) + ) + ) + } catch (err) { + logger.warn('🪙', `RateLimiterService(${this.config.name}) claim return threw`, { + keys, + error: String(err), + }) + } + } } diff --git a/nodejs/src/server.ts b/nodejs/src/server.ts index 81391973e04b..a95a45ab3c39 100644 --- a/nodejs/src/server.ts +++ b/nodejs/src/server.ts @@ -137,8 +137,9 @@ export class PluginServer implements NodeServer { // Build typed deps objects for consumers. `emailValidationValkey` and `frequencyCapValkey` // are null in the shared deps and set only by the cyclotron workers that run the hogflow - // email action (see `withEmailActionValkeys` at their loaders below) — no - // other consumer touches the SES or frequency cap Valkey. + // email action (see `withEmailActionValkeys` at their loaders below). Sandbox-enabled + // cdp-api gets only the SES pool, because its test sends skip the frequency cap. + // No other consumer touches the SES or frequency cap Valkey. const cdpDeps: CdpConsumerBaseDeps | undefined = needsCdp ? { postgres: this.postgres!, @@ -245,7 +246,18 @@ export class PluginServer implements NodeServer { : null const api = new CdpApi( this.config, - { ...cdpDeps!, idempotencyValkey: createIdempotencyValkeyPool(this.config) }, + { + ...cdpDeps!, + ...(this.config.WORKFLOWS_SANDBOX_SENDER_ENABLED + ? { + emailValidationValkey: createSesRateLimiterValkeyPool( + this.config, + 'email-mx-validation' + ), + } + : {}), + idempotencyValkey: createIdempotencyValkeyPool(this.config), + }, { hogQueue: kafkaQueue, hogflowQueue: postgresV2Queue }, batchResolverProducer ) @@ -456,9 +468,8 @@ export class PluginServer implements NodeServer { /** * Grants the SES Valkey pool that backs the shared MX-verdict cache to a worker's - * deps. Only the cyclotron workers that run the hogflow email action call this, so - * the pool is never opened on other CDP consumers or cdp-api — an idle Valkey sized - * for the SES rate limiter shouldn't hold connections from pods that never validate. + * deps. Only the email workers call this. Sandbox-enabled cdp-api opens the SES pool on its + * own, without the frequency cap pool, and the pool stays closed on other CDP consumers. * Null when no SES Valkey host is configured (local dev), in which case * EmailValidationService degrades to its local cache + DNS. * From e189abdf88635e69921b9c6c2e2faca2b97bddc7 Mon Sep 17 00:00:00 2001 From: "Reichenbach, Michael" <755327+Silthus@users.noreply.github.com> Date: Tue, 27 Oct 2026 07:27:17 +0000 Subject: [PATCH 2/2] fix(workflows): keep sandbox cap failures inside the send The CDP API's SES Valkey pool no longer stops the process on repeated errors, and it bounds connection and command waits, so an outage skips sandbox sends instead. A claim now carries its own return deadline, so an old claim cannot refund into a refilled bucket. Refs Silthus/personal#199 Co-Authored-By: Claude Opus 5.5 --- .../messaging/sandbox-email-sender.ts | 10 ++-- .../rate-limiter/rate-limiter-valkey-pool.ts | 10 +++- .../rate-limiter/rate-limiter.service.test.ts | 58 ++++++++++++++----- .../rate-limiter/rate-limiter.service.ts | 15 +++-- nodejs/src/server.ts | 3 +- 5 files changed, 67 insertions(+), 29 deletions(-) diff --git a/nodejs/src/cdp/services/messaging/sandbox-email-sender.ts b/nodejs/src/cdp/services/messaging/sandbox-email-sender.ts index 604858b86682..4df643e09d7c 100644 --- a/nodejs/src/cdp/services/messaging/sandbox-email-sender.ts +++ b/nodejs/src/cdp/services/messaging/sandbox-email-sender.ts @@ -1,4 +1,4 @@ -import { createHash, randomUUID } from 'crypto' +import { createHash } from 'crypto' import { type DefaultTreeAdapterMap, defaultTreeAdapter, parse, parseFragment, serialize } from 'parse5' import { CyclotronInvocationQueueParametersEmailType } from '~/cdp/schema/cyclotron' @@ -7,7 +7,7 @@ import { logger } from '~/common/utils/logger' import { captureTeamEvent } from '~/common/utils/posthog' import { TeamManager } from '~/common/utils/team-manager' -import { ClaimRequest, RateLimiterService } from '../rate-limiter/rate-limiter.service' +import { ClaimRequest, IssuedClaim, RateLimiterService } from '../rate-limiter/rate-limiter.service' import { maybeAddPreheaderToEmail } from './helpers/preheader' const SECONDS_PER_DAY = 86400 @@ -41,7 +41,7 @@ export type SandboxEmailBlockReason = | 'check_failed' | 'cap_reached' -export type SandboxDailyCapGrant = { type: 'granted'; claimId: string; buckets: ClaimRequest[] } +export type SandboxDailyCapGrant = { type: 'granted'; claim: IssuedClaim; buckets: ClaimRequest[] } export type SandboxDailyCapClaim = | SandboxDailyCapGrant @@ -206,7 +206,7 @@ export class SandboxEmailSender { ] const claim = await this.dailyCapLimiter.claimAllOrNothing(buckets) if (claim.granted) { - return { type: 'granted', claimId: randomUUID(), buckets } + return { type: 'granted', claim, buckets } } if (claim.deniedIndexes === null) { return { type: 'check_failed' } @@ -218,7 +218,7 @@ export class SandboxEmailSender { } public async returnDailyCaps(grant: SandboxDailyCapGrant): Promise { - await this.dailyCapLimiter?.returnClaim(grant.claimId, grant.buckets) + await this.dailyCapLimiter?.returnClaim(grant.claim, grant.buckets) } public async withIdentificationFooter( diff --git a/nodejs/src/cdp/services/rate-limiter/rate-limiter-valkey-pool.ts b/nodejs/src/cdp/services/rate-limiter/rate-limiter-valkey-pool.ts index 7e9eaed1011b..e6dfdb3842fc 100644 --- a/nodejs/src/cdp/services/rate-limiter/rate-limiter-valkey-pool.ts +++ b/nodejs/src/cdp/services/rate-limiter/rate-limiter-valkey-pool.ts @@ -24,9 +24,12 @@ export type RateLimiterValkeyConfig = Pick< * callers treat that as "feature disabled in this environment" and proceed * without it. */ +const OPTIONAL_POOL_TIMEOUT_MS = 1000 + export function createSesRateLimiterValkeyPool( config: RateLimiterValkeyConfig, - name = 'ses-rate-limiter' + name = 'ses-rate-limiter', + { optional = false }: { optional?: boolean } = {} ): RedisV2 | null { if (!config.SES_RATE_LIMITER_VALKEY_HOST) { logger.info('🪙', `[${name}] no SES Valkey host configured — feature disabled`) @@ -47,10 +50,13 @@ export function createSesRateLimiterValkeyPool( port: config.SES_RATE_LIMITER_VALKEY_PORT, password: config.SES_RATE_LIMITER_VALKEY_PASSWORD, tls, + ...(optional ? { commandTimeout: OPTIONAL_POOL_TIMEOUT_MS } : {}), }, name, + exitOnRepeatedErrors: !optional, }, - poolMinSize: config.REDIS_POOL_MIN_SIZE, + poolMinSize: optional ? 0 : config.REDIS_POOL_MIN_SIZE, poolMaxSize: config.REDIS_POOL_MAX_SIZE, + ...(optional ? { acquireTimeoutMillis: OPTIONAL_POOL_TIMEOUT_MS } : {}), }) } diff --git a/nodejs/src/cdp/services/rate-limiter/rate-limiter.service.test.ts b/nodejs/src/cdp/services/rate-limiter/rate-limiter.service.test.ts index effd952bdee7..6ef291ebd57b 100644 --- a/nodejs/src/cdp/services/rate-limiter/rate-limiter.service.test.ts +++ b/nodejs/src/cdp/services/rate-limiter/rate-limiter.service.test.ts @@ -8,7 +8,8 @@ import { TeamManager } from '~/common/utils/team-manager' import { Hub } from '~/types' import { SandboxEmailSender } from '../messaging/sandbox-email-sender' -import { RateLimiterService } from './rate-limiter.service' +import { createSesRateLimiterValkeyPool } from './rate-limiter-valkey-pool' +import { ClaimRequest, IssuedClaim, RateLimiterService } from './rate-limiter.service' const KEY = `@posthog-test/ses-rate-limiter/${randomUUID()}/bucket` @@ -473,11 +474,33 @@ describe('RateLimiterService', () => { } const replayingLimiter = new RateLimiterService(replayingRedis, { name: 'ses-rate-limiter' }) - expect(await replayingLimiter.claimAllOrNothing([bucket])).toEqual({ granted: true }) - expect(await limiter.claimAllOrNothing([bucket])).toEqual({ granted: true }) + expect(await replayingLimiter.claimAllOrNothing([bucket])).toMatchObject({ granted: true }) + expect(await limiter.claimAllOrNothing([bucket])).toMatchObject({ granted: true }) expect(await limiter.claimAllOrNothing([bucket])).toEqual({ granted: false, deniedIndexes: [0] }) }) + it('fails closed through an optional pool whose Valkey is unreachable', async () => { + const unreachableValkey = createSesRateLimiterValkeyPool( + { + SES_RATE_LIMITER_VALKEY_HOST: '127.0.0.1', + SES_RATE_LIMITER_VALKEY_PORT: 1, + SES_RATE_LIMITER_VALKEY_PASSWORD: '', + SES_RATE_LIMITER_VALKEY_TLS: false, + REDIS_POOL_MIN_SIZE: 1, + REDIS_POOL_MAX_SIZE: 2, + }, + 'unreachable-test', + { optional: true } + )! + const unreachable = new RateLimiterService(unreachableValkey, { name: 'sandbox-email' }) + + expect( + await unreachable.claimAllOrNothing([ + { key: `${KEY}/optional-pool`, requested: 1, capacity: 1, refillPerSecond: 0 }, + ]) + ).toEqual({ granted: false, deniedIndexes: null }) + }) + it('reports a limiter error without naming a denied bucket', async () => { const unreachable = new RateLimiterService( { @@ -496,28 +519,31 @@ describe('RateLimiterService', () => { }) describe('returnClaim', () => { + const grantedClaim = async (bucket: ClaimRequest): Promise => { + const claim = await limiter.claimAllOrNothing([bucket]) + if (!claim.granted) { + throw new Error('Expected the claim to be granted.') + } + return claim + } + it('gives a claim back once, even when the return is replayed', async () => { const bucket = { key: `${KEY}/return`, requested: 1, capacity: 2, refillPerSecond: 0 } - expect(await limiter.claimAllOrNothing([bucket])).toEqual({ granted: true }) - expect(await limiter.claimAllOrNothing([bucket])).toEqual({ granted: true }) + const claim = await grantedClaim(bucket) + await grantedClaim(bucket) - await limiter.returnClaim('claim-1', [bucket]) - await limiter.returnClaim('claim-1', [bucket]) + await limiter.returnClaim(claim, [bucket]) + await limiter.returnClaim(claim, [bucket]) - expect(await limiter.claimAllOrNothing([bucket])).toEqual({ granted: true }) + expect(await limiter.claimAllOrNothing([bucket])).toMatchObject({ granted: true }) expect(await limiter.claimAllOrNothing([bucket])).toEqual({ granted: false, deniedIndexes: [0] }) }) - it('ignores a return that reaches Valkey after its window', async () => { + it('ignores the return of a claim issued longer ago than its window', async () => { const bucket = { key: `${KEY}/late-return`, requested: 1, capacity: 1, refillPerSecond: 0 } - expect(await limiter.claimAllOrNothing([bucket])).toEqual({ granted: true }) + const claim = await grantedClaim(bucket) - const issuedLongAgo = jest.spyOn(Date, 'now').mockReturnValue(Date.now() - 60 * 60 * 1000) - try { - await limiter.returnClaim('claim-1', [bucket]) - } finally { - issuedLongAgo.mockRestore() - } + await limiter.returnClaim({ ...claim, returnBy: Date.now() - 1000 }, [bucket]) expect(await limiter.claimAllOrNothing([bucket])).toEqual({ granted: false, deniedIndexes: [0] }) }) diff --git a/nodejs/src/cdp/services/rate-limiter/rate-limiter.service.ts b/nodejs/src/cdp/services/rate-limiter/rate-limiter.service.ts index 3e09da89b2cb..f5d72eee2efd 100644 --- a/nodejs/src/cdp/services/rate-limiter/rate-limiter.service.ts +++ b/nodejs/src/cdp/services/rate-limiter/rate-limiter.service.ts @@ -326,6 +326,8 @@ export interface RateLimiterConfig { name: string } +export type IssuedClaim = { granted: true; claimId: string; returnBy: number } + export interface ClaimRequest { key: string requested: number @@ -448,7 +450,7 @@ export class RateLimiterService { // All bucket keys must share a hash tag because clustered Valkey rejects cross-slot Lua calls. public async claimAllOrNothing( buckets: ClaimRequest[] - ): Promise<{ granted: true } | { granted: false; deniedIndexes: number[] | null }> { + ): Promise { const endTimer = claimLatency.startTimer({ limiter: this.config.name }) const keys = buckets.map((bucket) => bucket.key) const claimId = randomUUID() @@ -473,7 +475,7 @@ export class RateLimiterService { ) if (Array.isArray(result) && result.length === 1 && result[0] === 1) { claimCounter.inc({ limiter: this.config.name, result: 'granted_full' }) - return { granted: true } + return { granted: true, claimId, returnBy: deadline } } if ( Array.isArray(result) && @@ -577,7 +579,10 @@ export class RateLimiterService { } } - public async returnClaim(claimId: string, buckets: Pick[]): Promise { + public async returnClaim( + claim: Pick, + buckets: Pick[] + ): Promise { const keys = buckets.map((bucket) => bucket.key) try { await this.valkey.useClient( @@ -587,9 +592,9 @@ export class RateLimiterService { RETURN_CLAIM_LUA, buckets.length + 1, ...keys, - `${keys[0]}/returned/${claimId}`, + `${keys[0]}/returned/${claim.claimId}`, ...buckets.map((bucket) => String(bucket.requested)), - String(Date.now() + RETURN_WINDOW_MS), + String(claim.returnBy), String(2 * RETURN_WINDOW_MS) ) ) diff --git a/nodejs/src/server.ts b/nodejs/src/server.ts index a95a45ab3c39..d623943478fe 100644 --- a/nodejs/src/server.ts +++ b/nodejs/src/server.ts @@ -252,7 +252,8 @@ export class PluginServer implements NodeServer { ? { emailValidationValkey: createSesRateLimiterValkeyPool( this.config, - 'email-mx-validation' + 'email-mx-validation', + { optional: true } ), } : {}),