From 43154bff064ddd5190a85e25babdccc6598296a0 Mon Sep 17 00:00:00 2001 From: "Reichenbach, Michael" <755327+Silthus@users.noreply.github.com> Date: Tue, 27 Oct 2026 06:54:44 +0000 Subject: [PATCH 1/2] feat(workflows): skip sandbox sends while the shared ses tenant is paused Django mirrors the shared sandbox SES tenant state from the tenant-state webhook and the daily reconciliation. The worker skips every sandbox send while that tenant is paused, and checks again after the daily allowance claim. Co-Authored-By: Claude Opus 5.5 --- .github/scripts/check-idor-model-coverage.py | 1 + .../internal/workflows-email-sending-tiers.md | 4 + nodejs/src/cdp/cdp-services.ts | 4 +- .../sandbox-sender-state.service.test.ts | 24 ++ .../managers/sandbox-sender-state.service.ts | 52 +++++ .../services/messaging/email.service.test.ts | 211 +++++++++++++++++- .../cdp/services/messaging/email.service.ts | 29 ++- .../messaging/sandbox-email-sender.test.ts | 7 +- .../messaging/sandbox-email-sender.ts | 14 +- .../rate-limiter/rate-limiter.service.test.ts | 3 +- posthog/plugins/plugin_server_api.py | 5 + posthog/settings/ses.py | 2 + .../0039_sandbox_sender_tenant_state.py | 32 +++ .../backend/migrations/max_migration.txt | 2 +- products/workflows/backend/models/__init__.py | 2 + .../models/sandbox_sender_tenant_state.py | 9 + products/workflows/backend/providers/ses.py | 79 ++++--- .../backend/services/ses_tenant_events.py | 34 ++- .../backend/services/ses_tenant_state.py | 49 +++- .../backend/tasks/ses_tenant_state.py | 19 +- .../backend/test/test_ses_tenant_events.py | 29 +++ .../backend/test/test_ses_tenant_state.py | 123 ++++++++++ 22 files changed, 684 insertions(+), 50 deletions(-) create mode 100644 nodejs/src/cdp/services/managers/sandbox-sender-state.service.test.ts create mode 100644 nodejs/src/cdp/services/managers/sandbox-sender-state.service.ts create mode 100644 products/workflows/backend/migrations/0039_sandbox_sender_tenant_state.py create mode 100644 products/workflows/backend/models/sandbox_sender_tenant_state.py diff --git a/.github/scripts/check-idor-model-coverage.py b/.github/scripts/check-idor-model-coverage.py index ffe2c08f1681..aa7d0b87aa9c 100644 --- a/.github/scripts/check-idor-model-coverage.py +++ b/.github/scripts/check-idor-model-coverage.py @@ -324,6 +324,7 @@ def get_scoped_models() -> tuple[dict[str, set[str]], set[str], set[str], set[st "DevStackImageTarget", "DevStackImageAttempt", "License", + "SandboxSenderTenantState", # instance-global mirror of the shared sandbox SES tenant state # --- Deprecated --- "Prompt", "PromptSequence", diff --git a/docs/internal/workflows-email-sending-tiers.md b/docs/internal/workflows-email-sending-tiers.md index 539917773f47..fa96812f88c0 100644 --- a/docs/internal/workflows-email-sending-tiers.md +++ b/docs/internal/workflows-email-sending-tiers.md @@ -87,6 +87,10 @@ A reached allowance skips the send instead of delaying it. A limiter error, or a 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. +Every sandbox send, test sends included, skips while the shared sandbox SES tenant is paused (`DISABLED`), or when the pause state cannot be read. A tenant Django has never synced counts as open, and SES enforces its own state. +The worker checks the pause again after the daily allowance claim and gives the claim back when the pause blocks delivery. +Django mirrors the tenant named by `SES_SANDBOX_TENANT_NAME` into `SandboxSenderTenantState` from the SES tenant-state webhook and the daily reconciliation, without emailing project admins. Set the same `SES_SANDBOX_TENANT_NAME` on Django web, Celery, and the worker. +A committed state change reaches the workers through the `reload-sandbox-sender-state` message. Own-sender sends do not read this state. Each delivered send emits a `workflows sandbox email sent` event. Successful sandbox 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 that metric is deployed. diff --git a/nodejs/src/cdp/cdp-services.ts b/nodejs/src/cdp/cdp-services.ts index 7950d0666c12..c9e73bfa3f7a 100644 --- a/nodejs/src/cdp/cdp-services.ts +++ b/nodejs/src/cdp/cdp-services.ts @@ -34,6 +34,7 @@ import { IntegrationManagerService } from './services/managers/integration-manag import { OrganizationMembersService } from './services/managers/organization-members.service' import { PersonsManagerService } from './services/managers/persons-manager.service' import { RecipientsManagerService } from './services/managers/recipients-manager.service' +import { SandboxSenderStateService } from './services/managers/sandbox-sender-state.service' import { TeamWorkflowsConfigService } from './services/managers/team-workflows-config.service' import { EmailSuppressionService } from './services/messaging/email-suppression.service' import { EmailValidationService } from './services/messaging/email-validation.service' @@ -498,7 +499,8 @@ export function createCdpCoreServices( }, deps.teamManager, redis, - sandboxEmailRateLimiter + sandboxEmailRateLimiter, + new SandboxSenderStateService(deps.postgres, deps.pubSub) ), new OrganizationMembersService(deps.postgres) ) diff --git a/nodejs/src/cdp/services/managers/sandbox-sender-state.service.test.ts b/nodejs/src/cdp/services/managers/sandbox-sender-state.service.test.ts new file mode 100644 index 000000000000..6bdde7cfa59e --- /dev/null +++ b/nodejs/src/cdp/services/managers/sandbox-sender-state.service.test.ts @@ -0,0 +1,24 @@ +import { DependencyUnavailableError } from '~/common/utils/db/error' +import { PostgresRouter } from '~/common/utils/db/postgres' +import { PubSub } from '~/common/utils/pubsub' + +import { SandboxSenderStateService } from './sandbox-sender-state.service' + +describe('SandboxSenderStateService', () => { + it('rides out a transient database error instead of failing the sandbox send', async () => { + const query = jest + .fn() + .mockRejectedValueOnce( + new DependencyUnavailableError('server closed the connection unexpectedly', 'Postgres', new Error()) + ) + .mockResolvedValueOnce({ rows: [{ tenant_name: 'sandbox-tenant', sending_status: 'DISABLED' }] }) + const service = new SandboxSenderStateService( + { query } as unknown as PostgresRouter, + { + on: jest.fn(), + } as unknown as PubSub + ) + + await expect(service.isPaused('sandbox-tenant')).resolves.toBe(true) + }) +}) diff --git a/nodejs/src/cdp/services/managers/sandbox-sender-state.service.ts b/nodejs/src/cdp/services/managers/sandbox-sender-state.service.ts new file mode 100644 index 000000000000..3a82caf91e31 --- /dev/null +++ b/nodejs/src/cdp/services/managers/sandbox-sender-state.service.ts @@ -0,0 +1,52 @@ +import { PostgresRouter, PostgresUse } from '~/common/utils/db/postgres' +import { DEFAULT_LOADER_RETRY, LazyLoader } from '~/common/utils/lazy-loader' +import { PubSub } from '~/common/utils/pubsub' + +type SandboxSenderState = { sendingStatus: string | undefined; generation: number } + +export class SandboxSenderStateService { + private lazyLoader: LazyLoader + private generation = 0 + + constructor( + private postgres: PostgresRouter, + pubSub: PubSub + ) { + this.lazyLoader = new LazyLoader({ + name: 'sandbox_sender_state', + refreshAgeMs: 30_000, + refreshJitterMs: 5_000, + loader: async (tenantNames) => await this.fetchStates(tenantNames), + loaderRetry: DEFAULT_LOADER_RETRY, + }) + pubSub.on<{ tenantName: string }>('reload-sandbox-sender-state', ({ tenantName }): void => { + this.generation++ + this.lazyLoader.markForRefresh(tenantName) + }) + } + + public async isPaused(tenantName: string): Promise { + for (let attempt = 0; attempt < 2; attempt++) { + const state = await this.lazyLoader.get(tenantName) + if (state?.generation === this.generation) { + // No row means never synced; SES stays the final word. + return state.sendingStatus === 'DISABLED' + } + this.lazyLoader.markForRefresh(tenantName) + } + throw new Error('Sandbox sender state changed during lookup') + } + + private async fetchStates(tenantNames: string[]): Promise> { + const generation = this.generation + const { rows } = await this.postgres.query<{ tenant_name: string; sending_status: string }>( + // The reload message fires right after the commit, so a lagging replica would cache the old state again. + PostgresUse.COMMON_WRITE, + 'SELECT tenant_name, sending_status FROM workflows_sandboxsendertenantstate WHERE tenant_name = ANY($1)', + [tenantNames], + 'fetch-sandbox-sender-state' + ) + const statuses = new Map(rows.map((row) => [row.tenant_name, row.sending_status])) + return Object.fromEntries(tenantNames.map((name) => [name, { sendingStatus: statuses.get(name), generation }])) + } +} diff --git a/nodejs/src/cdp/services/messaging/email.service.test.ts b/nodejs/src/cdp/services/messaging/email.service.test.ts index c787988f003c..92870928aa44 100644 --- a/nodejs/src/cdp/services/messaging/email.service.test.ts +++ b/nodejs/src/cdp/services/messaging/email.service.test.ts @@ -31,6 +31,7 @@ import { INVALID_TEAM_ID_KEYS } from '~/tests/helpers/team' import { Hub, Team } from '../../../types' import { OrganizationMembersService } from '../managers/organization-members.service' import { RecipientsManagerService } from '../managers/recipients-manager.service' +import { SandboxSenderStateService } from '../managers/sandbox-sender-state.service' import { TeamWorkflowsConfigService } from '../managers/team-workflows-config.service' import { RateLimiterService } from '../rate-limiter/rate-limiter.service' import { selectEmailSenderIntegrationId } from './email-sender-selection' @@ -279,12 +280,14 @@ describe('EmailService', () => { teamEmailRateLimiter = null, messageAssetsService, dailyCapLimiter = new RateLimiterService(sandboxRedis, { name: 'sandbox-email' }), + senderState = new SandboxSenderStateService(hub.postgres, hub.pubSub), }: { membersPostgres?: PostgresRouter senderSettings?: Partial teamEmailRateLimiter?: RateLimiterService | null messageAssetsService?: MessageAssetsService dailyCapLimiter?: RateLimiterService | null + senderState?: SandboxSenderStateService } = {} ): EmailService => { const workflowsConfig = new TeamWorkflowsConfigService(hub.postgres, hub.pubSub) @@ -323,7 +326,8 @@ describe('EmailService', () => { }, hub.teamManager, sandboxRedis, - dailyCapLimiter + dailyCapLimiter, + senderState ), new OrganizationMembersService(membersPostgres) ) @@ -756,6 +760,211 @@ describe('EmailService', () => { expectBlockedEvent('switch_off', false, 0) }) + describe('pause gate', () => { + const pausedMessage = + 'Skipping send: the sandbox sender is paused right now. Verify your own domain to keep sending.' + const tenantName = (): string => `sandbox-tenant-${team.id}` + const setTenantStatus = async (status: string): Promise => { + await hub.postgres.query( + PostgresUse.COMMON_WRITE, + `INSERT INTO workflows_sandboxsendertenantstate (tenant_name, sending_status, synced_at) + VALUES ($1, $2, now()) + ON CONFLICT (tenant_name) DO UPDATE SET sending_status = $2`, + [tenantName(), status], + 'test:set-sandbox-tenant-status' + ) + } + + afterEach(async () => { + await hub.postgres.query( + PostgresUse.COMMON_WRITE, + 'DELETE FROM workflows_sandboxsendertenantstate WHERE tenant_name = $1', + [tenantName()], + 'test:delete-sandbox-tenant-status' + ) + }) + + it.each([false, true])('skips while the sandbox tenant is paused (isTest=%s)', async (isTest) => { + await setTenantStatus('DISABLED') + service = createSandboxService(true, { senderSettings: { tenantName: tenantName() } }) + + const result = await service.executeSendEmail(invocation, isTest) + + expect(sendEmailSpy).not.toHaveBeenCalled() + expect(result).toMatchObject({ finished: true, skipped: true, metrics: [] }) + expect(result.error).toBeUndefined() + expect(result.invocation.state.vmState?.stack).toEqual([{ success: false }]) + expect(result.logs).toEqual( + expect.arrayContaining([expect.objectContaining({ level: 'info', message: pausedMessage })]) + ) + expect(capture).toHaveBeenCalledWith( + expect.objectContaining({ id: team.id }), + 'workflows sandbox email blocked', + { reason: 'paused', is_test: isTest, blocked_recipient_count: 0 } + ) + }) + + it.each(['ENABLED', 'REINSTATED'])('sends while the sandbox tenant is %s', async (status) => { + await setTenantStatus(status) + service = createSandboxService(true, { senderSettings: { tenantName: tenantName() } }) + + const result = await service.executeSendEmail(invocation) + + expect(result.error).toBeUndefined() + expect(result.skipped).toBeFalsy() + expect(sendEmailSpy).toHaveBeenCalledTimes(1) + }) + + it('skips when the pause state cannot be read', async () => { + const brokenPostgres = new PostgresRouter(hub) + await brokenPostgres.end() + service = createSandboxService(true, { + senderSettings: { tenantName: tenantName() }, + senderState: new SandboxSenderStateService(brokenPostgres, hub.pubSub), + }) + + const result = await service.executeSendEmail(invocation) + + expect(sendEmailSpy).not.toHaveBeenCalled() + expect(result).toMatchObject({ finished: true, skipped: true, metrics: [] }) + expect(result.invocation.state.vmState?.stack).toEqual([{ success: false }]) + expect(result.logs).toEqual( + expect.arrayContaining([ + expect.objectContaining({ + level: 'info', + message: + 'Skipping send: could not check whether the sandbox sender is paused. Try again later, or verify your own domain to keep sending.', + }), + ]) + ) + expect(capture).toHaveBeenCalledWith( + expect.objectContaining({ id: team.id }), + 'workflows sandbox email blocked', + { reason: 'check_failed', is_test: false, blocked_recipient_count: 0 } + ) + }) + + it('leaves own-sender sends alone while the sandbox tenant is paused', async () => { + await setTenantStatus('DISABLED') + service = createSandboxService(true, { senderSettings: { tenantName: tenantName() } }) + invocation.queueParameters = createEmailParams({ from: { integrationId: 1 } }) + + const result = await service.executeSendEmail(invocation) + + expect(result.error).toBeUndefined() + expect(sendEmailSpy).toHaveBeenCalledTimes(1) + }) + + it('rechecks a pause after a delayed cap claim and returns the grant', async () => { + await setTenantStatus('ENABLED') + const now = Date.now() + const clock = jest.spyOn(Date, 'now').mockReturnValue(now) + const dailyCapLimiter = new RateLimiterService(sandboxRedis, { name: 'sandbox-email' }) + service = createSandboxService(true, { + dailyCapLimiter, + senderSettings: { tenantName: tenantName(), dailyTeamCap: '1', dailyRecipientCap: '1' }, + }) + const claim = dailyCapLimiter.claimAllOrNothing.bind(dailyCapLimiter) + const delayedClaim = jest + .spyOn(dailyCapLimiter, 'claimAllOrNothing') + .mockImplementationOnce(async (buckets) => { + const grant = await claim(buckets) + await setTenantStatus('DISABLED') + clock.mockReturnValue(now + 40_000) + return grant + }) + try { + const blocked = await service.executeSendEmail(invocation) + expect(blocked).toMatchObject({ finished: true, skipped: true, metrics: [] }) + expect(blocked.error).toBeUndefined() + expect(sendEmailSpy).not.toHaveBeenCalled() + expect(blocked.logs).toContainEqual(expect.objectContaining({ message: pausedMessage })) + + await setTenantStatus('ENABLED') + clock.mockReturnValue(now + 80_000) + invocation.state.vmState = { stack: [] } as any + const resumed = await service.executeSendEmail(invocation) + expect(resumed.error).toBeUndefined() + expect(resumed.skipped).toBeFalsy() + expect(sendEmailSpy).toHaveBeenCalledTimes(1) + } finally { + delayedClaim.mockRestore() + clock.mockRestore() + } + }) + + it('skips when a pause invalidates a pending state read', async () => { + await setTenantStatus('ENABLED') + service = createSandboxService(true, { senderSettings: { tenantName: tenantName() } }) + let snapshotLoaded!: () => void + let releaseRead!: () => void + const snapshot = new Promise((resolve) => { + snapshotLoaded = resolve + }) + const heldRead = new Promise((resolve) => { + releaseRead = resolve + }) + let acknowledgeReload!: () => void + const reload = new Promise((resolve) => { + acknowledgeReload = resolve + }) + hub.pubSub.on<{ tenantName: string }>('reload-sandbox-sender-state', ({ tenantName: name }) => { + if (name === tenantName()) { + acknowledgeReload() + } + }) + const query = hub.postgres.query.bind(hub.postgres) + let held = false + const delayedRead = jest.spyOn(hub.postgres, 'query').mockImplementation(async (...args) => { + const result = await query(...args) + if (args[3] === 'fetch-sandbox-sender-state' && !held) { + held = true + snapshotLoaded() + await heldRead + } + return result + }) + try { + const pendingSend = service.executeSendEmail(invocation) + await snapshot + await setTenantStatus('DISABLED') + await hub.pubSub.publish( + 'reload-sandbox-sender-state', + JSON.stringify({ tenantName: tenantName() }) + ) + await reload + releaseRead() + const result = await pendingSend + expect(result).toMatchObject({ finished: true, skipped: true, metrics: [] }) + expect(result.error).toBeUndefined() + expect(result.logs).toContainEqual(expect.objectContaining({ message: pausedMessage })) + expect(sendEmailSpy).not.toHaveBeenCalled() + } finally { + releaseRead() + delayedRead.mockRestore() + } + }) + + it('picks up a pause announced while the state is cached', async () => { + await setTenantStatus('ENABLED') + service = createSandboxService(true, { senderSettings: { tenantName: tenantName() } }) + expect((await service.executeSendEmail(invocation)).skipped).toBeFalsy() + + await setTenantStatus('DISABLED') + + await waitForExpect(async () => { + await hub.pubSub.publish( + 'reload-sandbox-sender-state', + JSON.stringify({ tenantName: tenantName() }) + ) + const result = await service.executeSendEmail(invocation) + expect(result.logs).toEqual( + expect.arrayContaining([expect.objectContaining({ message: pausedMessage })]) + ) + }, 3000) + }) + }) + 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." diff --git a/nodejs/src/cdp/services/messaging/email.service.ts b/nodejs/src/cdp/services/messaging/email.service.ts index 1c1cebaf3b5c..be25934f7f80 100644 --- a/nodejs/src/cdp/services/messaging/email.service.ts +++ b/nodejs/src/cdp/services/messaging/email.service.ts @@ -410,7 +410,7 @@ function sandboxRecipientsOf(params: CyclotronInvocationQueueParametersEmailType return [params.to.email.trim(), ...sandboxAddressList(params.cc), ...sandboxAddressList(params.bcc)] } -type SandboxRecipientBlockReason = Exclude +type SandboxRecipientBlockReason = Exclude type SesRoute = { tenantName: string @@ -448,6 +448,12 @@ const SANDBOX_RECIPIENT_BLOCK_COPY: Record< }, } +const SANDBOX_PAUSE_SKIP_MESSAGES = { + paused: 'Skipping send: the sandbox sender is paused right now. Verify your own domain to keep sending.', + check_failed: + 'Skipping send: could not check whether the sandbox sender is paused. Try again later, or verify your own domain to keep sending.', +} as const + 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.", @@ -615,6 +621,9 @@ export class EmailService { await this.sandboxSender?.captureBlocked(invocation.teamId, isTest, 'switch_off', 0) return result } + if (isSandbox && !(await this.passesSandboxPauseGate(result, isTest))) { + return result + } const from = this.resolveFromSender(integration, params.from, addLog) @@ -805,7 +814,10 @@ export class EmailService { if (!capGrant) { return result } - if (!(await this.checkSandboxRecipients(result, deliveryRecipients, isTest))) { + if ( + !(await this.passesSandboxPauseGate(result, isTest)) || + !(await this.checkSandboxRecipients(result, deliveryRecipients, isTest)) + ) { await this.sandboxSender!.returnDailyCaps(capGrant) return result } @@ -1072,6 +1084,19 @@ export class EmailService { return false } + private async passesSandboxPauseGate( + result: CyclotronJobInvocationResult, + isTest: boolean + ): Promise { + const gate = await this.sandboxSender!.pauseGate() + if (gate === 'open') { + return true + } + this.skipSandboxSend(result, SANDBOX_PAUSE_SKIP_MESSAGES[gate]) + await this.sandboxSender!.captureBlocked(result.invocation.teamId, isTest, gate, 0) + return false + } + private async blockSandboxRecipients( result: CyclotronJobInvocationResult, blockedRecipients: string[], 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 08ace91b93af..08a78499abd0 100644 --- a/nodejs/src/cdp/services/messaging/sandbox-email-sender.test.ts +++ b/nodejs/src/cdp/services/messaging/sandbox-email-sender.test.ts @@ -11,6 +11,7 @@ const teamManager = { getTeam: jest.fn().mockResolvedValue({ organization_id: '00000000-0000-0000-0000-000000000001' }), } as unknown as TeamManager const blockedEventRedis = { useClient: jest.fn() } +const senderState = { isPaused: jest.fn().mockResolvedValue(false) } const sender = new SandboxEmailSender( { enabled: true, @@ -22,7 +23,8 @@ const sender = new SandboxEmailSender( }, teamManager, blockedEventRedis, - null + null, + senderState ) const params = (html: string, text?: string, preheader?: string): CyclotronInvocationQueueParametersEmailType => ({ type: 'email', @@ -72,7 +74,8 @@ describe('sandbox email content', () => { }, teamManager, blockedEventRedis, - null + null, + senderState ) 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 4df643e09d7c..0bb188858895 100644 --- a/nodejs/src/cdp/services/messaging/sandbox-email-sender.ts +++ b/nodejs/src/cdp/services/messaging/sandbox-email-sender.ts @@ -7,6 +7,7 @@ import { logger } from '~/common/utils/logger' import { captureTeamEvent } from '~/common/utils/posthog' import { TeamManager } from '~/common/utils/team-manager' +import { SandboxSenderStateService } from '../managers/sandbox-sender-state.service' import { ClaimRequest, IssuedClaim, RateLimiterService } from '../rate-limiter/rate-limiter.service' import { maybeAddPreheaderToEmail } from './helpers/preheader' @@ -40,6 +41,7 @@ export type SandboxEmailBlockReason = | 'recipient_unverified' | 'check_failed' | 'cap_reached' + | 'paused' export type SandboxDailyCapGrant = { type: 'granted'; claim: IssuedClaim; buckets: ClaimRequest[] } @@ -168,7 +170,8 @@ export class SandboxEmailSender { public readonly config: SandboxEmailSenderConfig, private teamManager: TeamManager, private blockedEventRedis: Pick, - private dailyCapLimiter: RateLimiterService | null + private dailyCapLimiter: RateLimiterService | null, + private senderState: Pick ) { this.config = { ...config, @@ -185,6 +188,15 @@ export class SandboxEmailSender { this.canSend = config.enabled && !emptySettings.length } + public async pauseGate(): Promise<'open' | 'paused' | 'check_failed'> { + try { + return (await this.senderState.isPaused(this.config.tenantName)) ? 'paused' : 'open' + } catch (error) { + logger.warn('Could not check the sandbox sender pause state', { error }) + return 'check_failed' + } + } + public async claimDailyCaps(teamId: number, recipients: string[]): Promise { const dailyTeamCap = parseDailyCap(this.config.dailyTeamCap) const dailyRecipientCap = parseDailyCap(this.config.dailyRecipientCap) 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 6ef291ebd57b..15095c35f6ab 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 @@ -434,7 +434,8 @@ describe('RateLimiterService', () => { }, teamManager, redis, - limiter + limiter, + { isPaused: () => Promise.resolve(false) } ) try { diff --git a/posthog/plugins/plugin_server_api.py b/posthog/plugins/plugin_server_api.py index 4176b4f87395..7cbb333e652b 100644 --- a/posthog/plugins/plugin_server_api.py +++ b/posthog/plugins/plugin_server_api.py @@ -82,6 +82,11 @@ def reload_team_workflows_config_on_workers(team_id: int): publish_message("reload-team-workflows-config", {"teamId": team_id}) +def reload_sandbox_sender_state_on_workers(tenant_name: str) -> None: + logger.info(f"Reloading sandbox sender state for tenant {tenant_name} on workers") + publish_message("reload-sandbox-sender-state", {"tenantName": tenant_name}) + + def populate_plugin_capabilities_on_workers(plugin_id: str): logger.info(f"Populating plugin capabilities for plugin {plugin_id} on workers") publish_message("populate-plugin-capabilities", {"pluginId": plugin_id}) diff --git a/posthog/settings/ses.py b/posthog/settings/ses.py index 6cd9beaff4c7..578442af8cda 100644 --- a/posthog/settings/ses.py +++ b/posthog/settings/ses.py @@ -29,6 +29,8 @@ if cs.strip() ] +SES_SANDBOX_TENANT_NAME: str = os.getenv("SES_SANDBOX_TENANT_NAME", "").strip() + # SNS topics allowed to deliver SES tenant reputation events (EventBridge -> SNS -> webhook). # Empty (the default) leaves the webhook inert: the SNS signature proves a message came from AWS, # but only the allowlist proves it came from *our* topic. diff --git a/products/workflows/backend/migrations/0039_sandbox_sender_tenant_state.py b/products/workflows/backend/migrations/0039_sandbox_sender_tenant_state.py new file mode 100644 index 000000000000..e88cfac2bca9 --- /dev/null +++ b/products/workflows/backend/migrations/0039_sandbox_sender_tenant_state.py @@ -0,0 +1,32 @@ +# Generated by Django 5.2.18 on 2026-10-27 07:00 + +from django.db import migrations, models + + +class Migration(migrations.Migration): + dependencies = [ + ("workflows", "0038_hogflow_team_source_record_idx"), + ] + + operations = [ + migrations.CreateModel( + name="SandboxSenderTenantState", + fields=[ + ( + "id", + models.BigAutoField( + auto_created=True, + primary_key=True, + serialize=False, + verbose_name="ID", + ), + ), + ("tenant_name", models.CharField(max_length=64, unique=True)), + ( + "sending_status", + models.CharField(blank=True, db_default="", default="", max_length=32), + ), + ("synced_at", models.DateTimeField(blank=True, null=True)), + ], + ), + ] diff --git a/products/workflows/backend/migrations/max_migration.txt b/products/workflows/backend/migrations/max_migration.txt index 2ce22c5893e0..c3daea1fde11 100644 --- a/products/workflows/backend/migrations/max_migration.txt +++ b/products/workflows/backend/migrations/max_migration.txt @@ -1 +1 @@ -0038_hogflow_team_source_record_idx +0039_sandbox_sender_tenant_state diff --git a/products/workflows/backend/models/__init__.py b/products/workflows/backend/models/__init__.py index bbc6b3a22a6c..fdf755e421f6 100644 --- a/products/workflows/backend/models/__init__.py +++ b/products/workflows/backend/models/__init__.py @@ -4,6 +4,7 @@ from .hog_flow_optimization import HogFlowOptimization from .hog_flow_revision import HogFlowRevision from .hog_flow_schedule.hog_flow_schedule import HogFlowSchedule +from .sandbox_sender_tenant_state import SandboxSenderTenantState from .team_workflows_config import TeamWorkflowsConfig from .workflow_idea import WorkflowIdea from .workflow_proposal import WorkflowProposal @@ -15,6 +16,7 @@ "HogFlowRevision", "HogFlowSchedule", "HogFlowTemplate", + "SandboxSenderTenantState", "TeamWorkflowsConfig", "WorkflowIdea", "WorkflowProposal", diff --git a/products/workflows/backend/models/sandbox_sender_tenant_state.py b/products/workflows/backend/models/sandbox_sender_tenant_state.py new file mode 100644 index 000000000000..f6b9172850bc --- /dev/null +++ b/products/workflows/backend/models/sandbox_sender_tenant_state.py @@ -0,0 +1,9 @@ +from django.db import models + + +class SandboxSenderTenantState(models.Model): + """Instance-level mirror of the shared sandbox SES tenant; platform state, not tenant data, so no team_id.""" + + tenant_name = models.CharField(max_length=64, unique=True) + sending_status = models.CharField(max_length=32, blank=True, default="", db_default="") + synced_at = models.DateTimeField(null=True, blank=True) diff --git a/products/workflows/backend/providers/ses.py b/products/workflows/backend/providers/ses.py index 4ecb6490fda4..763abd7a8495 100644 --- a/products/workflows/backend/providers/ses.py +++ b/products/workflows/backend/providers/ses.py @@ -395,6 +395,13 @@ def _is_throttling(error: Exception) -> bool: return isinstance(error, ClientError) and error.response["Error"]["Code"] in ("Throttling", "ThrottlingException") +@frozen +class TenantSendingState: + sending_status: str + reputation_impact: str | None + tenant_arn: str | None + + class SESProvider: ses_client: "SESClient" ses_v2_client: "SESV2Client" @@ -469,7 +476,17 @@ def get_tenant_reputation(self, team_id: int) -> dict[str, Any] | None: provider feedback loops, third-party listings), so this is the authoritative health source; our own app metrics only provide the per-workflow diagnosis. """ - tenant_name = self._tenant_name_for_team(team_id) + state = self.get_tenant_sending_state(self._tenant_name_for_team(team_id)) + if state is None: + return None + return { + "sending_status": state.sending_status, + "reputation_impact": state.reputation_impact, + "findings": self._open_findings(state.tenant_arn) if state.tenant_arn else [], + } + + def get_tenant_sending_state(self, tenant_name: str) -> TenantSendingState | None: + """Sending status and reputation impact of the named SES tenant, or None when it doesn't exist.""" try: tenant = self.ses_v2_client.get_tenant(TenantName=tenant_name)["Tenant"] except ClientError as e: @@ -479,40 +496,38 @@ def get_tenant_reputation(self, team_id: int) -> dict[str, Any] | None: sending_status: str = tenant.get("SendingStatus", "ENABLED") tenant_arn = tenant.get("TenantArn") - reputation_impact: str | None = None - findings: list[dict[str, Any]] = [] + if not tenant_arn: + return TenantSendingState(sending_status=sending_status, reputation_impact=None, tenant_arn=None) - if tenant_arn: - try: - entity = self.ses_v2_client.get_reputation_entity( - ReputationEntityReference=tenant_arn, ReputationEntityType="RESOURCE" - )["ReputationEntity"] - except ClientError as e: - # A tenant with no attributed sends yet has no reputation entity. - if e.response["Error"]["Code"] != "NotFoundException": - raise - entity = {} - reputation_impact = entity.get("ReputationImpact") + entity = self._reputation_entity(tenant_arn) + return TenantSendingState( # The aggregate folds in both AWS-managed and customer-managed pauses. - sending_status = entity.get("SendingStatusAggregate", sending_status) - - # RESOURCE_ARN is the only filter key AWS documents as usable on its own with this - # scoping (see _iter_open_recommendations for why STATUS is filtered locally). - findings.extend( - { - "finding_type": recommendation.get("Type", ""), - "impact": recommendation.get("Impact", "LOW"), - "description": recommendation.get("Description", ""), - "last_updated_at": recommendation.get("LastUpdatedTimestamp"), - } - for recommendation in self._iter_open_recommendations({"RESOURCE_ARN": tenant_arn}) - ) + sending_status=entity.get("SendingStatusAggregate", sending_status), + reputation_impact=entity.get("ReputationImpact"), + tenant_arn=tenant_arn, + ) - return { - "sending_status": sending_status, - "reputation_impact": reputation_impact, - "findings": findings, - } + def _reputation_entity(self, tenant_arn: str) -> Mapping[str, Any]: + try: + return self.ses_v2_client.get_reputation_entity( + ReputationEntityReference=tenant_arn, ReputationEntityType="RESOURCE" + )["ReputationEntity"] + except ClientError as e: + # A tenant with no attributed sends yet has no reputation entity. + if e.response["Error"]["Code"] != "NotFoundException": + raise + return {} + + def _open_findings(self, tenant_arn: str) -> list[dict[str, Any]]: + return [ + { + "finding_type": recommendation.get("Type", ""), + "impact": recommendation.get("Impact", "LOW"), + "description": recommendation.get("Description", ""), + "last_updated_at": recommendation.get("LastUpdatedTimestamp"), + } + for recommendation in self._iter_open_recommendations({"RESOURCE_ARN": tenant_arn}) + ] def _iter_open_recommendations( self, finding_filter: dict[Any, str] | None, client: "SESV2Client | None" = None diff --git a/products/workflows/backend/services/ses_tenant_events.py b/products/workflows/backend/services/ses_tenant_events.py index 77672ddc31cf..423cd5837aa5 100644 --- a/products/workflows/backend/services/ses_tenant_events.py +++ b/products/workflows/backend/services/ses_tenant_events.py @@ -15,21 +15,22 @@ from collections.abc import Mapping from typing import Any +from django.conf import settings + import requests import structlog from posthog.ingress.contracts import WebhookDelivery from posthog.ingress.verify.sns_signature import is_valid_sns_url -from products.workflows.backend.tasks.ses_tenant_state import sync_ses_tenant_state_task +from products.workflows.backend.tasks.ses_tenant_state import sync_sandbox_tenant_state_task, sync_ses_tenant_state_task logger = structlog.get_logger(__name__) _TENANT_TEAM_RE = re.compile(r"\bteam-(\d+)\b") -def _extract_team_id(event: dict[str, Any]) -> int | None: - """Pull the `team-` tenant name out of an EventBridge event, wherever AWS put it.""" +def _tenant_candidates(event: dict[str, Any]) -> list[str]: detail = event.get("detail") or {} candidates: list[Any] = [ detail.get("tenantName"), @@ -39,8 +40,24 @@ def _extract_team_id(event: dict[str, Any]) -> int | None: resources = event.get("resources") if isinstance(resources, list): candidates.extend(resources) - for candidate in candidates: - if isinstance(candidate, str) and (match := _TENANT_TEAM_RE.search(candidate)): + return [candidate for candidate in candidates if isinstance(candidate, str)] + + +def _names_sandbox_tenant(event: dict[str, Any]) -> bool: + tenant_name = settings.SES_SANDBOX_TENANT_NAME + if not tenant_name: + return False + tenant_path = re.compile(rf"(?:/|:)tenant/{re.escape(tenant_name)}(?:/|$)") + return any( + candidate == tenant_name or (candidate.startswith("arn:") and tenant_path.search(candidate) is not None) + for candidate in _tenant_candidates(event) + ) + + +def _extract_team_id(event: dict[str, Any]) -> int | None: + """Pull the `team-` tenant name out of an EventBridge event, wherever AWS put it.""" + for candidate in _tenant_candidates(event): + if match := _TENANT_TEAM_RE.search(candidate): return int(match.group(1)) return None @@ -61,7 +78,7 @@ def _confirm_subscription(message: Mapping[str, Any]) -> None: def _handle_notification(message: Mapping[str, Any]) -> None: - """Enqueue a tenant-state sync for the team the EventBridge event names. + """Enqueue a tenant-state sync for the sandbox tenant or the team the EventBridge event names. Every unusable payload is dropped rather than raised, because a redelivery would carry the same unusable payload. @@ -73,6 +90,11 @@ def _handle_notification(message: Mapping[str, Any]) -> None: if not isinstance(event, dict) or event.get("source") != "aws.ses": return + if _names_sandbox_tenant(event): + logger.info("ses_tenant_events_webhook_accepted", tenant="sandbox", detail_type=event.get("detail-type")) + sync_sandbox_tenant_state_task.delay() + return + team_id = _extract_team_id(event) if team_id is None: logger.warning("ses_tenant_events_webhook_no_tenant", detail_type=event.get("detail-type")) diff --git a/products/workflows/backend/services/ses_tenant_state.py b/products/workflows/backend/services/ses_tenant_state.py index 99f9f5c3c3de..82da157a6dc6 100644 --- a/products/workflows/backend/services/ses_tenant_state.py +++ b/products/workflows/backend/services/ses_tenant_state.py @@ -1,21 +1,26 @@ from collections.abc import Callable from typing import Any +from django.conf import settings from django.db import transaction from django.utils import timezone import structlog from posthog.models.team import Team -from posthog.plugins.plugin_server_api import reload_team_workflows_config_on_workers +from posthog.plugins.plugin_server_api import ( + reload_sandbox_sender_state_on_workers, + reload_team_workflows_config_on_workers, +) from posthog.tasks.email import ( send_email_sending_reputation_finding, send_email_sending_suspended, send_email_sending_unsuspended, ) +from products.workflows.backend.models.sandbox_sender_tenant_state import SandboxSenderTenantState from products.workflows.backend.models.team_workflows_config import TeamWorkflowsConfig -from products.workflows.backend.providers.ses import SESProvider +from products.workflows.backend.providers.ses import SESProvider, TenantSendingState logger = structlog.get_logger(__name__) @@ -58,6 +63,46 @@ def _announce_state_change(team_id: int) -> None: logger.exception("Failed to announce SES tenant state change to workers", team_id=team_id) +def _announce_sandbox_state_change(tenant_name: str) -> None: + try: + reload_sandbox_sender_state_on_workers(tenant_name) + except Exception: + logger.exception("Failed to announce sandbox SES tenant state change to workers", tenant_name=tenant_name) + + +def _apply_sandbox_tenant_state(state: SandboxSenderTenantState, tenant: TenantSendingState) -> None: + previous_status = state.sending_status + state.synced_at = timezone.now() + if previous_status == tenant.sending_status: + state.save(update_fields=["synced_at"]) + logger.info("Sandbox SES tenant state unchanged", tenant_name=state.tenant_name, status=previous_status) + return + + state.sending_status = tenant.sending_status + state.save(update_fields=["sending_status", "synced_at"]) + logger.info( + "Sandbox SES tenant state changed", + tenant_name=state.tenant_name, + from_status=previous_status, + to_status=tenant.sending_status, + ) + tenant_name = state.tenant_name + transaction.on_commit(lambda: _announce_sandbox_state_change(tenant_name)) + + +def sync_sandbox_tenant_state(provider: SESProvider | None = None) -> None: + # Serialize SES reads so a stale sweep cannot overwrite a newer pause. + tenant_name = settings.SES_SANDBOX_TENANT_NAME + if not tenant_name: + return + SandboxSenderTenantState.objects.get_or_create(tenant_name=tenant_name) + with transaction.atomic(): + state = SandboxSenderTenantState.objects.select_for_update().get(tenant_name=tenant_name) + tenant = (provider or SESProvider()).get_tenant_sending_state(tenant_name) + if tenant is not None: + _apply_sandbox_tenant_state(state, tenant) + + def sync_ses_tenant_state(team_id: int, provider: SESProvider | None = None, *, verify_team: bool = True) -> None: """ Fetch the authoritative AWS SES tenant state for a team and apply it. Called from the diff --git a/products/workflows/backend/tasks/ses_tenant_state.py b/products/workflows/backend/tasks/ses_tenant_state.py index 55b41d59978f..d8c01d8c355f 100644 --- a/products/workflows/backend/tasks/ses_tenant_state.py +++ b/products/workflows/backend/tasks/ses_tenant_state.py @@ -6,7 +6,7 @@ from posthog.tasks.utils import CeleryQueue from products.workflows.backend.providers.ses import SESProvider -from products.workflows.backend.services.ses_tenant_state import sync_ses_tenant_state +from products.workflows.backend.services.ses_tenant_state import sync_sandbox_tenant_state, sync_ses_tenant_state logger = get_logger(__name__) @@ -27,6 +27,19 @@ def sync_ses_tenant_state_task(team_id: int) -> None: sync_ses_tenant_state(team_id) +@shared_task( + ignore_result=True, + queue=CeleryQueue.DEFAULT.value, + autoretry_for=(Exception,), + retry_backoff=60, + retry_backoff_max=600, + max_retries=5, + retry_jitter=True, +) +def sync_sandbox_tenant_state_task() -> None: + sync_sandbox_tenant_state() + + @shared_task(ignore_result=True, queue=CeleryQueue.LONG_RUNNING.value) @skip_team_scope_audit def reconcile_ses_tenant_states() -> None: @@ -44,6 +57,10 @@ def reconcile_ses_tenant_states() -> None: # One provider for the whole sweep: a fresh SESProvider per team would rebuild its boto3 # clients (and re-resolve credentials) thousands of times. provider = SESProvider() + try: + sync_sandbox_tenant_state(provider=provider) + except Exception: + logger.exception("SES sandbox tenant reconciliation failed") synced = 0 failed = 0 for team_id in team_ids.iterator(chunk_size=500): diff --git a/products/workflows/backend/test/test_ses_tenant_events.py b/products/workflows/backend/test/test_ses_tenant_events.py index 8b4f7aa2d9e8..fb4fe7f6aa6a 100644 --- a/products/workflows/backend/test/test_ses_tenant_events.py +++ b/products/workflows/backend/test/test_ses_tenant_events.py @@ -23,6 +23,7 @@ from products.workflows.backend.facade.api import accept_ses_event TOPIC = "arn:aws:sns:us-east-1:123456789012:ses-tenant-events" +SANDBOX_TENANT = "sandbox-team-1" WEBHOOK_PATH = "/webhooks/workflows/ses-events" SUBSCRIBE_URL = "https://sns.us-east-1.amazonaws.com/?Action=ConfirmSubscription&Token=tok" _CERT_URL = "https://sns.us-east-1.amazonaws.com/SimpleNotificationService-01d088a6f77103d0fe307c0069e40ed6.pem" @@ -106,6 +107,9 @@ def setUp(self) -> None: sync = patch("products.workflows.backend.services.ses_tenant_events.sync_ses_tenant_state_task") self.sync_mock = sync.start() self.addCleanup(sync.stop) + sandbox_sync = patch("products.workflows.backend.services.ses_tenant_events.sync_sandbox_tenant_state_task") + self.sandbox_sync_mock = sandbox_sync.start() + self.addCleanup(sandbox_sync.stop) def test_enqueues_a_sync_for_the_tenant_named_in_the_event(self) -> None: accept_ses_event(_delivery(_notification(_eventbridge_event()))) @@ -119,6 +123,31 @@ def test_finds_the_tenant_in_resource_arns_when_detail_has_no_name(self) -> None self.sync_mock.delay.assert_called_once_with(7) + @parameterized.expand( + [ + ("sandbox_tenant_by_name", {"tenantName": SANDBOX_TENANT}, [], "sandbox"), + ( + "sandbox_tenant_by_arn", + {}, + [f"arn:aws:ses:us-east-1:123456789012:tenant/{SANDBOX_TENANT}/abc"], + "sandbox", + ), + ("team_tenant", {"tenantName": "team-42"}, [], "team"), + ] + ) + @override_settings(SES_SANDBOX_TENANT_NAME=SANDBOX_TENANT) + def test_routes_the_sandbox_tenant_to_its_own_sync( + self, _name: str, detail: dict[str, Any], resources: list[str], expected_sync: str + ) -> None: + accept_ses_event(_delivery(_notification(_eventbridge_event(detail=detail, resources=resources)))) + + if expected_sync == "sandbox": + self.sandbox_sync_mock.delay.assert_called_once_with() + self.sync_mock.delay.assert_not_called() + else: + self.sync_mock.delay.assert_called_once_with(42) + self.sandbox_sync_mock.delay.assert_not_called() + def test_ignores_events_from_other_sources(self) -> None: accept_ses_event(_delivery(_notification(_eventbridge_event(source="aws.health")))) diff --git a/products/workflows/backend/test/test_ses_tenant_state.py b/products/workflows/backend/test/test_ses_tenant_state.py index e00b17a1df33..214aa4f1f8fb 100644 --- a/products/workflows/backend/test/test_ses_tenant_state.py +++ b/products/workflows/backend/test/test_ses_tenant_state.py @@ -1,10 +1,18 @@ +from datetime import UTC, datetime + +import time_machine from posthog.test.base import BaseTest from unittest.mock import MagicMock, patch +from django.test import override_settings from django.utils import timezone +from botocore.exceptions import ClientError from parameterized import parameterized +from posthog.models.integration import Integration + +from products.workflows.backend.models.sandbox_sender_tenant_state import SandboxSenderTenantState from products.workflows.backend.models.team_workflows_config import TeamWorkflowsConfig from products.workflows.backend.services.ses_tenant_state import ( PROVIDER_PAUSE_REASON, @@ -12,6 +20,10 @@ apply_ses_tenant_state, sync_ses_tenant_state, ) +from products.workflows.backend.tasks.ses_tenant_state import ( + reconcile_ses_tenant_states, + sync_sandbox_tenant_state_task, +) class TestApplySesTenantState(BaseTest): @@ -183,3 +195,114 @@ def test_unknown_team_is_skipped_without_calling_ses(self) -> None: provider.get_tenant_reputation.assert_not_called() assert not TeamWorkflowsConfig.objects.filter(team_id=self.team.id + 99_999).exists() + + +SANDBOX_TENANT = "workflows-sandbox" +SANDBOX_TENANT_ARN = f"arn:aws:ses:us-east-1:123456789012:tenant/{SANDBOX_TENANT}/abc" + + +@override_settings(SES_SANDBOX_TENANT_NAME=SANDBOX_TENANT) +class TestSandboxTenantStateSync(BaseTest): + def setUp(self) -> None: + super().setUp() + boto3_client = patch("products.workflows.backend.providers.ses.boto3.client") + self.ses = boto3_client.start().return_value + self.addCleanup(boto3_client.stop) + publish = patch("posthog.plugins.plugin_server_api.publish_message") + self.publish = publish.start() + self.addCleanup(publish.stop) + self.ses.list_recommendations.return_value = {"Recommendations": []} + + def _aws_tenant(self, sending_status: str, impact: str) -> None: + self.ses.get_tenant.return_value = {"Tenant": {"TenantName": SANDBOX_TENANT, "TenantArn": SANDBOX_TENANT_ARN}} + self.ses.get_reputation_entity.return_value = { + "ReputationEntity": {"ReputationImpact": impact, "SendingStatusAggregate": sending_status} + } + + def _sync(self) -> None: + with self.captureOnCommitCallbacks(execute=True): + sync_sandbox_tenant_state_task() + + def test_a_changed_state_is_stored_and_announced_after_commit(self) -> None: + self._aws_tenant("ENABLED", "NONE") + self._sync() + self.publish.reset_mock() + self._aws_tenant("DISABLED", "HIGH") + + with self.captureOnCommitCallbacks(execute=True): + sync_sandbox_tenant_state_task() + self.publish.assert_not_called() + + state = SandboxSenderTenantState.objects.get(tenant_name=SANDBOX_TENANT) + assert state.sending_status == "DISABLED" + self.ses.get_tenant.assert_called_with(TenantName=SANDBOX_TENANT) + self.publish.assert_called_once_with("reload-sandbox-sender-state", {"tenantName": SANDBOX_TENANT}) + + @parameterized.expand([("same_impact", "HIGH"), ("only_the_impact_changed", "LOW")]) + def test_an_unchanged_sending_status_refreshes_synced_at_without_announcing(self, _name: str, impact: str) -> None: + self._aws_tenant("DISABLED", "HIGH") + with time_machine.travel("2026-10-05T10:00:00Z", tick=False): + self._sync() + self.publish.reset_mock() + self._aws_tenant("DISABLED", impact) + + with time_machine.travel("2026-10-05T11:00:00Z", tick=False): + self._sync() + + state = SandboxSenderTenantState.objects.get(tenant_name=SANDBOX_TENANT) + assert state.sending_status == "DISABLED" + assert state.synced_at == datetime(2026, 10, 5, 11, tzinfo=UTC) + self.publish.assert_not_called() + + def test_a_pause_is_stored_even_when_findings_cannot_be_listed(self) -> None: + self._aws_tenant("DISABLED", "HIGH") + self.ses.list_recommendations.side_effect = ClientError( + {"Error": {"Code": "TooManyRequestsException"}}, "ListRecommendations" + ) + + self._sync() + + assert SandboxSenderTenantState.objects.get(tenant_name=SANDBOX_TENANT).sending_status == "DISABLED" + + def test_an_unknown_sandbox_tenant_is_never_paused(self) -> None: + self.ses.get_tenant.side_effect = ClientError({"Error": {"Code": "NotFoundException"}}, "GetTenant") + + self._sync() + + state = SandboxSenderTenantState.objects.get(tenant_name=SANDBOX_TENANT) + assert (state.sending_status, state.synced_at) == ("", None) + self.publish.assert_not_called() + + def test_the_daily_sweep_syncs_the_sandbox_tenant_once_beside_team_tenants(self) -> None: + Integration.objects.create(team=self.team, kind="email", config={"provider": "ses"}) + self._aws_tenant("ENABLED", "LOW") + + with self.captureOnCommitCallbacks(execute=True): + reconcile_ses_tenant_states() + + tenant_names = [call.kwargs["TenantName"] for call in self.ses.get_tenant.call_args_list] + assert sorted(tenant_names) == sorted([SANDBOX_TENANT, f"team-{self.team.id}"]) + assert SandboxSenderTenantState.objects.get(tenant_name=SANDBOX_TENANT).sending_status == "ENABLED" + + def test_a_failed_sandbox_sync_does_not_stop_the_team_sweep(self) -> None: + Integration.objects.create(team=self.team, kind="email", config={"provider": "ses"}) + + def get_tenant(TenantName: str) -> dict: + if TenantName == SANDBOX_TENANT: + raise ClientError({"Error": {"Code": "TooManyRequestsException"}}, "GetTenant") + return {"Tenant": {"TenantName": TenantName, "SendingStatus": "ENABLED"}} + + self.ses.get_tenant.side_effect = get_tenant + + with self.captureOnCommitCallbacks(execute=True): + reconcile_ses_tenant_states() + + assert TeamWorkflowsConfig.objects.get(team=self.team).ses_tenant_sending_status == "ENABLED" + + @override_settings(SES_SANDBOX_TENANT_NAME="") + def test_nothing_is_synced_while_no_sandbox_tenant_is_configured(self) -> None: + sync_sandbox_tenant_state_task() + reconcile_ses_tenant_states() + + self.ses.get_tenant.assert_not_called() + assert not SandboxSenderTenantState.objects.exists() From ddb1a9eeb6a9f7e03cf723d50a848371481c3655 Mon Sep 17 00:00:00 2001 From: "Reichenbach, Michael" <755327+Silthus@users.noreply.github.com> Date: Tue, 27 Oct 2026 07:31:46 +0000 Subject: [PATCH 2/2] fix(workflows): announce sandbox tenant changes with a robust on_commit Refs Silthus/personal#199 Co-Authored-By: Claude Opus 5.5 --- products/workflows/backend/services/ses_tenant_state.py | 9 +-------- 1 file changed, 1 insertion(+), 8 deletions(-) diff --git a/products/workflows/backend/services/ses_tenant_state.py b/products/workflows/backend/services/ses_tenant_state.py index 82da157a6dc6..d713c4671204 100644 --- a/products/workflows/backend/services/ses_tenant_state.py +++ b/products/workflows/backend/services/ses_tenant_state.py @@ -63,13 +63,6 @@ def _announce_state_change(team_id: int) -> None: logger.exception("Failed to announce SES tenant state change to workers", team_id=team_id) -def _announce_sandbox_state_change(tenant_name: str) -> None: - try: - reload_sandbox_sender_state_on_workers(tenant_name) - except Exception: - logger.exception("Failed to announce sandbox SES tenant state change to workers", tenant_name=tenant_name) - - def _apply_sandbox_tenant_state(state: SandboxSenderTenantState, tenant: TenantSendingState) -> None: previous_status = state.sending_status state.synced_at = timezone.now() @@ -87,7 +80,7 @@ def _apply_sandbox_tenant_state(state: SandboxSenderTenantState, tenant: TenantS to_status=tenant.sending_status, ) tenant_name = state.tenant_name - transaction.on_commit(lambda: _announce_sandbox_state_change(tenant_name)) + transaction.on_commit(lambda: reload_sandbox_sender_state_on_workers(tenant_name), robust=True) def sync_sandbox_tenant_state(provider: SESProvider | None = None) -> None: