Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
24 changes: 17 additions & 7 deletions nodejs/src/cdp/services/messaging/email.service.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -549,29 +549,37 @@ describe('EmailService', () => {
describe('queue-to-send latency', () => {
// The job was created a minute ago and rescheduled after a throttle, so it became due again 5 seconds ago.
// Running the email step clones the invocation, which resets queueScheduledAt before the send.
const transactionalSend = (): void => {
invocation.state.globals.event.event = '$transactional_email'
invocation.hogFunction.metadata = { message_category_type: 'transactional' }
}

it.each([
['workflow', 'bulk', 'its rescheduled time', (): void => {}, 5_000, 60_000],
['workflow', 'bulk', 'false', 'its rescheduled time', (): void => {}, 5_000, 60_000],
['transactional_api', 'fast', 'false', 'its creation', transactionalSend, 60_000, Infinity],
// Alerts leave out a wait that the team sending cap causes on purpose.
[
'transactional_api',
'fast',
'its creation',
'true',
'its creation after a team cap delay',
(): void => {
invocation.state.globals.event.event = '$transactional_email'
invocation.hogFunction.metadata = { message_category_type: 'transactional' }
transactionalSend()
invocation.queueMetadata = { teamCapped: true }
},
60_000,
Infinity,
],
])(
'observes the wait of a retried %s send on the %s lane from %s',
async (source, priorityClass, _from, setUp, minWaitMs, maxWaitMs) => {
'observes the wait of a retried %s send on the %s lane with team_capped=%s from %s',
async (source, priorityClass, teamCapped, _from, setUp, minWaitMs, maxWaitMs) => {
setUp()
invocation.queue = 'email'
invocation.queueCreatedAt = DateTime.utc().minus({ seconds: 60 })
invocation.queueDueAt = DateTime.utc().minus({ seconds: 5 })
invocation.queueScheduledAt = undefined
const name = 'cdp_email_queue_to_send_ms'
const labels = { source, priority_class: priorityClass }
const labels = { source, priority_class: priorityClass, team_capped: teamCapped }
const countBefore = await metricValue(name, labels, `${name}_count`)
const sumBefore = await metricValue(name, labels, `${name}_sum`)

Expand Down Expand Up @@ -888,6 +896,8 @@ describe('EmailService', () => {
expect(result.finished).toBe(false)
// 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 })
const scheduledMs = result.invocation.queueScheduledAt!.toMillis()
expect(scheduledMs).toBeGreaterThanOrEqual(before + minMs)
expect(scheduledMs).toBeLessThan(before + maxMs + 5000)
Expand Down
10 changes: 6 additions & 4 deletions nodejs/src/cdp/services/messaging/email.service.ts
Original file line number Diff line number Diff line change
Expand Up @@ -118,12 +118,12 @@ const sesSendDurationMs = new Histogram({

const emailQueueToSendMs = new Histogram({
name: 'cdp_email_queue_to_send_ms',
help: 'Milliseconds an email waits on the email queue before its handoff to SES. For source="transactional_api" the wait starts when cdp-api created the job, so it is the time from the API\'s 202 to SES and includes every throttled or rate-limited retry. For source="workflow" the wait starts at the job\'s scheduled time: when the job joined the email queue, or when a delayed or rescheduled send became due, so a retried send measures only its last wait. priority_class is the email queue lane.',
labelNames: ['source', 'priority_class'],
help: 'Milliseconds an email waits on the email queue before its handoff to SES. For source="transactional_api" the wait starts when cdp-api created the job, so it is the time from the API\'s 202 to SES and includes every throttled or rate-limited retry. For source="workflow" the wait starts at the job\'s scheduled time: when the job joined the email queue, or when a delayed or rescheduled send became due, so a retried send measures only its last wait. priority_class is the email queue lane. team_capped is true when the team sending cap delayed the send, so that alerts can leave out a wait that the cap causes on purpose.',
labelNames: ['source', 'priority_class', 'team_capped'],
buckets: [100, 250, 500, 1_000, 2_500, 5_000, 10_000, 30_000, 60_000, 120_000, 300_000],
})

type EmailQueueWait = { source: 'transactional_api' | 'workflow'; startedAt: DateTime }
type EmailQueueWait = { source: 'transactional_api' | 'workflow'; startedAt: DateTime; teamCapped: boolean }

// A transactional API job is created on the email queue, so its creation time is when the caller got its 202. A
// workflow job can join the email queue long after it was created, so its wait starts at the time it became due.
Expand All @@ -132,7 +132,7 @@ function emailQueueWait(invocation: CyclotronJobInvocationHogFunction): EmailQue
const source =
invocation.state.globals.event?.event === TRANSACTIONAL_EMAIL_EVENT ? 'transactional_api' : 'workflow'
const startedAt = source === 'transactional_api' ? invocation.queueCreatedAt : invocation.queueDueAt
return startedAt ? { source, startedAt } : undefined
return startedAt ? { source, startedAt, teamCapped: invocation.queueMetadata?.teamCapped === true } : undefined
}

/**
Expand Down Expand Up @@ -693,6 +693,7 @@ export class EmailService {
// per-workflow limit above: createInvocationResult cleared queueParameters, and
// without them the rescheduled dequeue resumes the Hog VM and drops the send.
result.invocation.queueParameters = params
result.invocation.queueMetadata = { ...invocation.queueMetadata, teamCapped: true }
result.invocation.queueScheduledAt = DateTime.utc().plus({ milliseconds: capDelay.retryDelayMs })
const capRetrySeconds = Math.round(capDelay.retryDelayMs / 1000)
addLog(
Expand Down Expand Up @@ -1284,6 +1285,7 @@ export class EmailService {
.labels({
source: queueWait.source,
priority_class: getEmailQueuePriorityClass(result.invocation.hogFunction.metadata),
team_capped: queueWait.teamCapped ? 'true' : 'false',
})
.observe(Math.max(0, Date.now() - queueWait.startedAt.toMillis()))
}
Expand Down
Loading