fix(workflows): keep queue and priority across email queue retries

docs/internal/workflows-email-sending-tiers.md

@@ -65,6 +65,7 @@ Decay, suspension drops, admin recomputes, and the backfill stay silent.
 
 - The batch audience cap is decided when the batch is dispatched. Adding an email step to the workflow while a batch is queued does not re-cap it; the send-time buckets still cap every email at execution. This is why enforcement requires the worker caps to be deployed (see the rollout order).
 - Test-panel sends bypass the team buckets on purpose, matching the per-workflow rate limit.
+- Rescheduled emails keep their email queue priority while waiting. After a team-cap, workflow-pacing, or SES-throttle retry, the workflow restores its origin queue and priority when it next leaves the email queue. A push notification retried while the workflow is on the email queue behaves the same way and waits at the priority of its own message category.
 - The buckets are token buckets: a full idle bucket plus refill allows up to roughly twice the stated cap in the very first period. The bucket TTLs exceed the refill periods so this does not recur from idling.
 - A denied send parks until every short bucket has refilled enough to cover it, instead of retrying on a fixed few-minute cadence.
   The computed wait is capped at one hour and then jittered 1x to 2x, so a parked send can wait just under two hours between attempts.

nodejs/src/cdp/services/hog-executor-async.service.ts

@@ -30,7 +30,12 @@ import { mergeSecretHeaders, resolveSecretHeaders } from '../utils/secret-header
 import { resolveStandardWebhooksKey, signStandardWebhooksRequest } from '../utils/standard-webhooks'
 import { HogExecutorExecuteOptions, HogExecutorPreviousResult, HogExecutorService } from './hog-executor.service'
 import { HogInputsService } from './hog-inputs.service'
-import { EMAIL_QUEUE_PRIORITY, cdpEmailQueuedTotal, getEmailQueuePriorityClass } from './messaging/email-priority'
+import {
+    EMAIL_QUEUE_PRIORITY,
+    cdpEmailQueuedTotal,
+    getEmailQueuePriority,
+    getEmailQueuePriorityClass,
+} from './messaging/email-priority'
 import { EmailService } from './messaging/email.service'
 import { PushNotificationService } from './messaging/push-notification.service'
 import { RecipientTokensService, preferencesTokenSourceForInvocation } from './messaging/recipient-tokens.service'
@@ -163,6 +168,9 @@ export class HogExecutorAsyncService {
                         nextInvocation,
                         options?.isTest ?? false
                     )
+                    if (invocation.queue === 'email' && !result.finished) {
+                        this.keepPushRetryInEmailQueueOrder(result, invocation)
+                    }
                 } else if (queueParamsType === 'email') {
                     // Route to the email queue unless this is a test run: tests execute in-process and
                     // never enqueue, so routing would leave the job unworked.
@@ -179,8 +187,8 @@ export class HogExecutorAsyncService {
                         // the current action's metadata so a throttle retry re-queues under this
                         // send's class rather than the previous email action's.
                         if (invocation.queue === 'email') {
-                            nextInvocation.queuePriority =
-                                EMAIL_QUEUE_PRIORITY[getEmailQueuePriorityClass(nextInvocation.hogFunction.metadata)]
+                            nextInvocation.queueMetadata = invocation.queueMetadata
+                            nextInvocation.queuePriority = getEmailQueuePriority(nextInvocation.hogFunction.metadata)
                         }
                         // isTest is forwarded so a test send stays out of the email's engagement tracking.
                         result = await this.deps.emailService.executeSendEmail(nextInvocation, options?.isTest ?? false)
@@ -324,6 +332,15 @@ export class HogExecutorAsyncService {
         return result
     }
 
+    private keepPushRetryInEmailQueueOrder(
+        result: CyclotronJobInvocationResult<CyclotronJobInvocationHogFunction>,
+        entryInvocation: CyclotronJobInvocationHogFunction
+    ): void {
+        const { originQueue, originPriority } = entryInvocation.queueMetadata ?? {}
+        result.invocation.queueMetadata = { ...result.invocation.queueMetadata, originQueue, originPriority }
+        result.invocation.queuePriority = getEmailQueuePriority(result.invocation.hogFunction.metadata)
+    }
+
     private routeToQueue(
         invocation: CyclotronJobInvocationHogFunction,
         targetQueue: string,

nodejs/src/cdp/services/hogflows/hogflow-executor.service.ts

@@ -76,6 +76,19 @@ const TRIGGER_MISMATCH_EXITS: ReadonlySet<string> = new Set([
     'exit_on_trigger_not_matched_or_conversion',
 ])
 
+// The email service sets teamCapped when the team sending cap delays one send, and that send reads it from the
+// job's stored metadata. The actions after it in the same dequeue start from this executor's result, so the
+// flag stays off them. Otherwise their queue waits would also count as cap delays.
+function withoutTeamCapped(
+    metadata: CyclotronJobInvocationHogFlow['queueMetadata']
+): CyclotronJobInvocationHogFlow['queueMetadata'] {
+    if (!metadata || !('teamCapped' in metadata)) {
+        return metadata
+    }
+    const { teamCapped: _teamCapped, ...rest } = metadata
+    return rest
+}
+
 export function createHogFlowInvocation(
     globals: HogFunctionInvocationGlobals,
     hogFlow: HogFlow,
@@ -686,6 +699,7 @@ export class HogFlowExecutorService {
         // run whose email action isn't the first action executed.
         const result = createInvocationResult<CyclotronJobInvocationHogFlow>(invocation, {
             queuePriority: invocation.queuePriority,
+            queueMetadata: invocation.queue === 'email' ? withoutTeamCapped(invocation.queueMetadata) : undefined,
         })
         result.finished = false // Typically we are never finished unless we error or exit
 

nodejs/src/cdp/services/messaging/email-priority.ts

@@ -39,6 +39,10 @@ export function getEmailQueuePriorityClass(metadata: HogFunctionType['metadata']
     return metadata?.message_category_type === 'transactional' ? 'fast' : 'bulk'
 }
 
+export function getEmailQueuePriority(metadata: HogFunctionType['metadata']): number {
+    return EMAIL_QUEUE_PRIORITY[getEmailQueuePriorityClass(metadata)]
+}
+
 export const cdpEmailQueuedTotal = new Counter({
     name: 'cdp_email_queued_total',
     help: 'Total emails routed to the dedicated email queue',

nodejs/src/cdp/services/messaging/email.service.test.ts

@@ -240,6 +240,7 @@ describe('EmailService', () => {
                 stack: [],
             } as any
             invocation.queueParameters = createEmailParams({ from: { integrationId: 1 } })
+            invocation.queueMetadata = { originQueue: 'hogflow', originPriority: 2 }
 
             // Mock SES v2 send to avoid actual AWS calls
             sendEmailSpy = jest.spyOn(service.sesV2Client!, 'send') as any
@@ -508,6 +509,7 @@ describe('EmailService', () => {
                 expect(scheduledMs).toBeGreaterThanOrEqual(before + 400)
                 expect(scheduledMs).toBeLessThan(before + 2000)
                 expect(result.invocation.queueParameters).toEqual(invocation.queueParameters)
+                expect(result.invocation.queueMetadata).toEqual(invocation.queueMetadata)
                 // No business metric emitted on throttle — the eventual retry
                 // will produce email_sent.
                 expect(result.metrics ?? []).toEqual([])
@@ -702,6 +704,7 @@ describe('EmailService', () => {
                 // The reschedule must carry the email payload forward: without queueParameters the
                 // retry has nothing to send and the throttled email is dropped rather than delayed.
                 expect(result.invocation.queueParameters).toEqual(invocation.queueParameters)
+                expect(result.invocation.queueMetadata).toEqual(invocation.queueMetadata)
                 // Bracketed against both ends of the call, so the bounds hold the delay itself
                 // and not the time the call took.
                 const scheduledMs = result.invocation.queueScheduledAt!.toMillis()
@@ -940,7 +943,7 @@ describe('EmailService', () => {
                 // The reschedule must carry the email payload forward, same as the workflow limit.
                 expect(result.invocation.queueParameters).toEqual(invocation.queueParameters)
                 // The flag labels the eventual send's queue wait, so alerts can leave out the cap delay.
-                expect(result.invocation.queueMetadata).toMatchObject({ teamCapped: true })
+                expect(result.invocation.queueMetadata).toEqual({ ...invocation.queueMetadata, teamCapped: true })
                 const scheduledMs = result.invocation.queueScheduledAt!.toMillis()
                 expect(scheduledMs).toBeGreaterThanOrEqual(before + minMs)
                 expect(scheduledMs).toBeLessThan(before + maxMs + 5000)

nodejs/src/cdp/services/messaging/email.service.ts

@@ -689,6 +689,7 @@ export class EmailService {
                     // send. Mirrors the fetch-retry (`result.invocation.queueParameters = params`) and
                     // queue-routing paths, which re-attach the same way.
                     result.invocation.queueParameters = params
+                    result.invocation.queueMetadata = invocation.queueMetadata
                     const retryDelayMs = pickReservedRetryDelayMs(claim.retryAfterMs, refillPerSecond, claim.reserved)
                     emailReservedParkMs.labels('workflow-email').observe(retryDelayMs)
                     result.invocation.queueScheduledAt = DateTime.utc().plus({ milliseconds: retryDelayMs })
@@ -792,6 +793,7 @@ export class EmailService {
                 throttled = true
                 result.finished = false
                 result.invocation.queueParameters = params
+                result.invocation.queueMetadata = invocation.queueMetadata
                 result.invocation.queueScheduledAt = DateTime.utc().plus({ milliseconds: error.retryAfterMs })
                 addLog('warn', `SES rate-limited (${error.errorCode}); rescheduling email in ${error.retryAfterMs}ms`)
             } else {

nodejs/tests/fixtures/hog-bytecode/1b3f99defea3027176062c76e92e8779a3d2d4c39e28f7d002d66cdbfb8cd8e2.json

@@ -0,0 +1 @@
+["_H",1,32,"integrationIds",32,"integrationId",32,"inputs",1,2,43,1,32,"distinctId",32,"distinct-id",32,"payload",32,"title",32,"Hi",32,"body",32,"Hello",42,2,42,3,2,"sendPushNotification",1,36,0,32,"success",45,5,40,9,36,0,32,"error",45,2,"Error",1,49,35]