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..4df643e09d7c 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 } 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, IssuedClaim, 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'; claim: IssuedClaim; 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', claim, 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.claim, grant.buckets) + } + public async withIdentificationFooter( params: CyclotronInvocationQueueParametersEmailType, senderName: string, 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 490f0fee49ef..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 @@ -4,9 +4,12 @@ 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 { RateLimiterService } from './rate-limiter.service' +import { SandboxEmailSender } from '../messaging/sandbox-email-sender' +import { createSesRateLimiterValkeyPool } from './rate-limiter-valkey-pool' +import { ClaimRequest, IssuedClaim, RateLimiterService } from './rate-limiter.service' const KEY = `@posthog-test/ses-rate-limiter/${randomUUID()}/bucket` @@ -413,4 +416,136 @@ 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])).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( + { + 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', () => { + 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 } + const claim = await grantedClaim(bucket) + await grantedClaim(bucket) + + await limiter.returnClaim(claim, [bucket]) + await limiter.returnClaim(claim, [bucket]) + + expect(await limiter.claimAllOrNothing([bucket])).toMatchObject({ granted: true }) + expect(await limiter.claimAllOrNothing([bucket])).toEqual({ granted: false, deniedIndexes: [0] }) + }) + + 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 } + const claim = await grantedClaim(bucket) + + 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 1d4b178a0574..f5d72eee2efd 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,11 +250,84 @@ 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 } +export type IssuedClaim = { granted: true; claimId: string; returnBy: number } + export interface ClaimRequest { key: string requested: number @@ -373,6 +447,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 { + 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, claimId, returnBy: deadline } + } + 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 +578,31 @@ export class RateLimiterService { endTimer() } } + + public async returnClaim( + claim: Pick, + 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/${claim.claimId}`, + ...buckets.map((bucket) => String(bucket.requested)), + String(claim.returnBy), + 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..d623943478fe 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,19 @@ 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', + { optional: true } + ), + } + : {}), + idempotencyValkey: createIdempotencyValkeyPool(this.config), + }, { hogQueue: kafkaQueue, hogflowQueue: postgresV2Queue }, batchResolverProducer ) @@ -456,9 +469,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. *