Repository navigation
feat(cdp): give transactional email its own queue and worker - #127614
meikelmosby wants to merge 1 commit into
Conversation
|
Risk: Medium · 1 medium This PR splits transactional email onto its own Sentinel reviewed |
🤖 CI report✅ Trunk lane — non-backend laneThis PR is assigned to the non-backend lane. It does not run backend Python tests and may merge in parallel with PRs in other lanes.
|
| First copy | Second copy | Lines | Tokens |
|---|---|---|---|
products/warehouse_sources/sources/folk/folk.py:144 |
products/warehouse_sources/sources/select_star/select_star.py:156 |
18 | 90 |
products/warehouse_sources/sources/ruddr/ruddr.py:127 |
products/warehouse_sources/sources/tempo/tempo.py:136 |
20 | 72 |
⚠️ Duplication (TypeScript) — 11 new duplicated blocks (worst 195 tokens)
New TypeScript code duplication introduced by this branch. Fails at 70+ tokens in app code, or 150+ tokens when both copies live in test files. Advisory while the gate proves itself: extract a shared helper instead of copying.
| First copy | Second copy | Lines | Tokens |
|---|---|---|---|
packages/agent/packages/agent/src/adapters/claude/claude-agent.slash-command.test.ts:49 |
packages/agent/packages/agent/src/adapters/claude/claude-agent.task-notification.test.ts:53 |
38 | 195 |
products/replay_vision/frontend/components/QuillScanList.tsx:31 |
products/replay_vision/frontend/components/ScanList.tsx:30 |
34 | 191 |
products/replay_vision/frontend/components/ChapterFilmstrip.tsx:26 |
products/replay_vision/frontend/components/QuillChapterFilmstrip.tsx:27 |
31 | 174 |
products/replay_vision/frontend/components/QuillScanList.tsx:203 |
products/replay_vision/frontend/components/ScanList.tsx:179 |
15 | 172 |
frontend/src/scenes/session-recordings/player/sidebar/PlayerSidebarOverviewDeviceTab.tsx:11 |
frontend/src/scenes/session-recordings/player/watch/QuillOverviewDeviceCard.tsx:11 |
14 | 144 |
products/replay_vision/frontend/components/QuillScanList.tsx:87 |
products/replay_vision/frontend/components/ScanList.tsx:81 |
27 | 138 |
frontend/src/scenes/session-recordings/player/sidebar/PlayerSidebarOverviewSessionTab.tsx:14 |
frontend/src/scenes/session-recordings/player/watch/QuillOverviewSessionCard.tsx:14 |
12 | 102 |
products/replay_vision/frontend/components/QuillScanList.tsx:257 |
products/replay_vision/frontend/components/ScanList.tsx:231 |
24 | 101 |
products/replay_vision/frontend/components/ObservationProgressBar.tsx:12 |
products/replay_vision/frontend/components/QuillObservationProgressBar.tsx:10 |
19 | 87 |
frontend/src/scenes/session-recordings/player/sidebar/PlayerSidebarOverviewPersonTab.tsx:15 |
frontend/src/scenes/session-recordings/player/watch/QuillOverviewPersonCard.tsx:16 |
14 | 84 |
products/replay_vision/frontend/components/ObservationProgressBar.tsx:45 |
products/replay_vision/frontend/components/QuillObservationProgressBar.tsx:43 |
13 | 70 |
🚨 Comment density — 16% of added code lines are comments (34 of 213)
This section warns when comments are more than 3% of the code lines a PR adds, and alerts above 6%. Before agent-assisted PRs, the typical share was about 2%. Only full-line comments count. Docstrings, generated files, snapshots, migrations, and workflow files are left out.
Comments that restate the code, record how the change came about, or narrate the next line add noise for the next reader. Keep the comments that explain a reason the code cannot show, and remove the rest. See .agents/skills/writing-code-comments/SKILL.md for the house rules.
Files with the most added comment lines:
| File | Comment lines | Added lines |
|---|---|---|
nodejs/src/cdp/services/messaging/email-queues.ts |
11 | 28 |
nodejs/src/cdp/services/hog-executor-async.service.ts |
10 | 38 |
nodejs/src/server.ts |
8 | 39 |
nodejs/src/cdp/services/job-queue/job-queue-postgres-v2.ts |
3 | 5 |
nodejs/src/cdp/cdp-api.ts |
2 | 13 |
This check does not block merging. It updates on every push and clears when the share drops.
|
|
||
| import { JobQueue } from '../services/job-queue/job-queue.interface' | ||
| import { EMAIL_TRANSACTIONAL_QUEUE } from '../services/messaging/email-queues' | ||
| import { CdpConsumerBaseDeps } from './cdp-base.consumer' |
There was a problem hiding this comment.
Transactional-email runs on the new lane are never counted, so rolled-out teams send unbilled and outside usage limits
The transactional send API sets invocation.state.triggerMetricPending = true (nodejs/src/cdp/cdp-api.ts:1238) because it skips the trigger pipeline; the marketing-lane worker is the only code that turns that flag into the triggered app metric (nodejs/src/cdp/consumers/cdp-cyclotron-worker-email.consumer.ts:24-43, countPendingTrigger). The new CdpCyclotronWorkerEmailTransactional extends CdpCyclotronWorkerHogFlow directly and has no such override, so every team enabled in CDP_EMAIL_TRANSACTIONAL_QUEUE_TEAMS gets API-triggered transactional sends that never emit the triggered metric and never clear the flag — those runs are missing from the app-metrics usage/billing counts (the triggered metric is what the trigger pipeline emits for every other workflow run, nodejs/src/cdp/services/hog-flow-invocation-pipeline.service.ts:261-270).
How:
- Ops enables a team in
CDP_EMAIL_TRANSACTIONAL_QUEUE_TEAMS(the rollout this PR exists to perform). - That team calls the transactional send API;
getEmailQueueroutes the job toemailtransactional. - The transactional worker processes the batch via plain
CdpCyclotronWorkerHogFlow.processInvocations, which never readstriggerMetricPending. - No
triggeredmetric is recorded, so the run is absent from usage metering and product-usage limits, no matter how many sends the team makes.
Fix: Make the transactional consumer reuse the email worker's counting (extend CdpCyclotronWorkerEmail and only override the queue, or extract countPendingTrigger into the shared path both workers call).
React with 👍 if useful or 👎 if not
There was a problem hiding this comment.
PostHog Review · Standard · reviewed c874aee: 4 Should fix.
See this review and its findings in PostHog Review. Comment @posthog review --deep for a Deep review. A Deep review takes longer and looks deeper than a Standard review.
| if (isEmailQueue(invocation.queue)) { | ||
| nextInvocation.queue = this.pickEmailQueue(nextInvocation) | ||
| } | ||
| // 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) |
There was a problem hiding this comment.
P2 · Reschedule onto the selected lane before sending
For an enabled team with consecutive transactional and marketing email actions, pickEmailQueue selects the marketing lane, but executeSendEmail still runs inline on the transactional worker. Assigning nextInvocation.queue does not transfer execution, so marketing sends bypass the reserve floor and fleet isolation; reschedule before sending when the selected lane differs.
| queue, | ||
| queueParameters: invocation.queueParameters, | ||
| queueMetadata: { | ||
| ...invocation.queueMetadata, | ||
| originQueue: invocation.queue, |
There was a problem hiding this comment.
P2 · Preserve the origin priority across email routing
When a workflow enters at priority 1 or 2, removing the explicit priority handling makes createInvocationResult reset it to 0. Returning through routeToQueue also resets it, and queueInvocationResults persists that promotion, defeating origin-queue ordering. The updated test expecting 2 fails; preserve the entry priority across both routing helpers.
| import { CdpConsumerBaseDeps } from './cdp-base.consumer' | ||
| import { CdpCyclotronWorkerHogFlow } from './cdp-cyclotron-worker-hogflow.consumer' | ||
|
|
||
| export class CdpCyclotronWorkerEmailTransactional extends CdpCyclotronWorkerHogFlow { |
There was a problem hiding this comment.
P2 · Retain trigger counting in the transactional worker
For API sends routed to emailtransactional, inheriting directly from CdpCyclotronWorkerHogFlow skips CdpCyclotronWorkerEmail.processInvocations, the only consumer of triggerMetricPending. These runs never emit triggered, so workflow trigger totals omit enabled teams, breaking end-to-end tracking (AGENTS.md:286); inherit or share the email worker's counting hook.
| .labels({ | ||
| source: queueWait.source, | ||
| priority_class: getEmailQueuePriorityClass(result.invocation.hogFunction.metadata), | ||
| priority_class: emailPriorityClassLabel(result.invocation.hogFunction.metadata), |
There was a problem hiding this comment.
P2 · Include the transactional lane in queue-wait observations
Once a team routes sends to emailtransactional, executeSendEmail leaves queueWait undefined because its guard still checks invocation.queue === 'email'. Consequently this histogram records no observations for the new lane, hiding transactional latency from existing dashboards and alerts; use isEmailQueue in the queue-wait guard.
c874aee to
4cb696c
Compare
Transactional sends move to an `emailtransactional` lane with its own worker fleet, so a marketing backlog, a bad rollout or a stuck pod on the marketing lane can no longer hold them up. Priority classes and the originPriority stash stay: a team outside the rollout still shares the `email` queue, where transactional has to dequeue first. A send that finds itself on the other lane is moved, not sent, so the marketing lane never spends the transactional reserve. The marketing lane claims SES capacity above the reserve floor, leaving the bottom of the bucket for the transactional lane. Rollout is per team through CDP_EMAIL_TRANSACTIONAL_QUEUE_TEAMS, empty on deploy, so nothing routes over until a team is named. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
4cb696c to
5e1104f
Compare
|
Closing while the transactional email work is replanned. The branch stays for reference. |
Problem
emailqueue, worked by one fleet.Changes
Transactional sends move to their own queue and worker fleet.
flowchart LR H[hogflow] --> Q[(email queue)] --> W{{email worker}} --> S[SES] class H,W phBlue; class Q phGray; class S phRed; classDef phBlue fill:#1d4aff,stroke:#1d4aff,color:#fff; classDef phRed fill:#f54e00,stroke:#f54e00,color:#fff; classDef phGray fill:#e5e7eb,stroke:#c7ccd1,color:#000;flowchart LR H[hogflow] --> Q[(email queue)] --> W{{email worker}} --> S[SES] H --> QT[(emailtransactional queue)] --> WT{{transactional worker}} --> S class H,W,WT phBlue; class Q,QT phGray; class S phRed; classDef phBlue fill:#1d4aff,stroke:#1d4aff,color:#fff; classDef phRed fill:#f54e00,stroke:#f54e00,color:#fff; classDef phGray fill:#e5e7eb,stroke:#c7ccd1,color:#000;CDP_EMAIL_TRANSACTIONAL_QUEUE_TEAMS, empty on deploy. Rollback reverses the edit.emailqueue, where transactional has to dequeue first.isEmailQueue()replaces=== 'email', in a newemail-queues.tsbesideemail-priority.ts.cdp-cyclotron-worker-email-transactionaldeployment in charts first.Warning
Name no team until #127615 lands. This layer adds no dequeue index for the lane on
cyclotron_jobs.How did you test this code?
Test rationale:
email-queues.test.tscovers lane choice and the team gate. Two cases inhog-executor.service.test.tscover a transactional send landing on its lane in the fast class, and a send on the wrong lane moving with its origin stash. The existing priority assertions stay unchanged.Ran locally:
hog-executor.service.test.ts,hogflow-executor.service.test.ts,hog_function.test.ts,email-queues.test.ts. Not run:cdp-api.serial.test.ts.Release status
Docs update
🤖 Agent context
Autonomy: Human-driven (agent-assisted)
Agent: Claude Code, Opus 5; restacked and finished by Claude Code, Opus 5.5
Skills invoked:
/writing-tests,/writing-pr-descriptions,/stacking-prs.More context for AI agents
The queue name has no underscore because
cdp/types.tsrequires alphanumeric queue names andjob-queue-kafka.tsinterpolates them into topic and consumer group names.hogoverflowset the precedent.A rebase onto master picked up the new transactional send API in
cdp-api.ts, which enqueued straight onto theemailqueue. It now picks the lane through the same gate.cdpEmailQueuedTotalmoved intoemail-queues.tsand its label changed frompriority_classtoqueue; the queue-wait metric keeps the oldfast/bulklabel values so existing dashboards still resolve. An earlier revision dropped the priority classes; review found that regresses every team not yet in the rollout, so they stay.No open PR builds a transactional lane. Layer 3 of 4.