feat(workflows): skip sandbox sends while the shared ses tenant is paused

.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",

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.

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)
     )

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)
+    })
+})

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<SandboxSenderState>
+    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<boolean> {
+        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<Record<string, SandboxSenderState>> {
+        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 }]))
+    }
+}

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<SandboxEmailSenderConfig>
                     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<void> => {
+                    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<void>((resolve) => {
+                        snapshotLoaded = resolve
+                    })
+                    const heldRead = new Promise<void>((resolve) => {
+                        releaseRead = resolve
+                    })
+                    let acknowledgeReload!: () => void
+                    const reload = new Promise<void>((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."