fix: SendGrid webhook batching, idempotency-key stability, TTS circuit breaker, and cross-tenant job isolation

services/api/src/email/queue.rs

@@ -564,11 +564,15 @@ impl EmailQueue {
             .await?;
 
         // Derive a stable idempotency key for this job so retries never
-        // produce duplicate sends within the configured TTL window.
+        // produce duplicate sends within the configured TTL window. The key
+        // is bucketed off the job's original creation time (not "now") so
+        // that a retry delayed past an hour boundary by exponential backoff
+        // still computes the same key as earlier attempts of the same job.
         let idem = idempotency_key(
             &job.recipient_email,
             &job.template_name,
             &service.idempotency_secret,
+            job.created_at,
         );
 
         // Send email (deduplication handled inside send_email_idempotent)

services/api/src/email/service.rs

@@ -1,9 +1,10 @@
 use anyhow::{Context, Result};
+use chrono::{DateTime, Utc};
 use hmac::{Hmac, Mac};
 use redis::AsyncCommands as _;
 use serde_json::Value;
 use sha2::Sha256;
-use std::time::{Duration, SystemTime, UNIX_EPOCH};
+use std::time::Duration;
 use validator::ValidateEmail;
 
 use crate::cache::RedisCache;
@@ -34,16 +35,24 @@ impl Default for IdempotencyConfig {
 /// - A server-side secret prevents pre-computation by external attackers.
 /// - The hour bucket bounds the validity window to ~1 hour per key rotation.
 ///
+/// `bucket_time` is the timestamp the hour bucket is derived from. Callers
+/// that retry the same logical job (e.g. the email queue worker) must pass
+/// the job's original creation time rather than the current time — otherwise
+/// a retry whose exponential backoff crosses an hour boundary would compute a
+/// different key and defeat deduplication for a job that already succeeded.
+///
 /// The key format is `email:idem:<hex(HMAC)>`.
-pub fn idempotency_key(recipient: &str, template_name: &str, secret: &str) -> String {
+pub fn idempotency_key(
+    recipient: &str,
+    template_name: &str,
+    secret: &str,
+    bucket_time: DateTime<Utc>,
+) -> String {
     type HmacSha256 = Hmac<Sha256>;
 
-    // Hour bucket: seconds-since-epoch / 3600, so keys rotate each hour.
-    let hour_bucket = SystemTime::now()
-        .duration_since(UNIX_EPOCH)
-        .unwrap_or_default()
-        .as_secs()
-        / 3600;
+    // Hour bucket: seconds-since-epoch / 3600, so keys rotate each hour
+    // relative to `bucket_time` rather than wall-clock "now".
+    let hour_bucket = bucket_time.timestamp() / 3600;
 
     let mut mac = HmacSha256::new_from_slice(secret.as_bytes())
         .expect("HMAC accepts any key length");
@@ -183,7 +192,7 @@ impl EmailService {
         template_name: &str,
         template_data: &Value,
     ) -> Result<String> {
-        let idem = idempotency_key(recipient, template_name, &self.idempotency_secret);
+        let idem = idempotency_key(recipient, template_name, &self.idempotency_secret, Utc::now());
         self.send_email_idempotent(recipient, template_name, template_data, Some(&idem))
             .await
     }
@@ -441,36 +450,40 @@ mod tests {
 
     #[test]
     fn same_inputs_produce_same_key() {
-        let k1 = idempotency_key("user@example.com", "welcome_email", "secret");
-        let k2 = idempotency_key("user@example.com", "welcome_email", "secret");
+        let now = Utc::now();
+        let k1 = idempotency_key("user@example.com", "welcome_email", "secret", now);
+        let k2 = idempotency_key("user@example.com", "welcome_email", "secret", now);
         assert_eq!(k1, k2);
     }
 
     #[test]
     fn different_recipient_produces_different_key() {
-        let k1 = idempotency_key("alice@example.com", "welcome_email", "secret");
-        let k2 = idempotency_key("bob@example.com", "welcome_email", "secret");
+        let now = Utc::now();
+        let k1 = idempotency_key("alice@example.com", "welcome_email", "secret", now);
+        let k2 = idempotency_key("bob@example.com", "welcome_email", "secret", now);
         assert_ne!(k1, k2);
     }
 
     #[test]
     fn different_template_produces_different_key() {
-        let k1 = idempotency_key("user@example.com", "welcome_email", "secret");
-        let k2 = idempotency_key("user@example.com", "newsletter_confirmation", "secret");
+        let now = Utc::now();
+        let k1 = idempotency_key("user@example.com", "welcome_email", "secret", now);
+        let k2 = idempotency_key("user@example.com", "newsletter_confirmation", "secret", now);
         assert_ne!(k1, k2);
     }
 
     /// #932: key changes when the secret changes.
     #[test]
     fn different_secret_produces_different_key() {
-        let k1 = idempotency_key("user@example.com", "welcome_email", "secret-a");
-        let k2 = idempotency_key("user@example.com", "welcome_email", "secret-b");
+        let now = Utc::now();
+        let k1 = idempotency_key("user@example.com", "welcome_email", "secret-a", now);
+        let k2 = idempotency_key("user@example.com", "welcome_email", "secret-b", now);
         assert_ne!(k1, k2, "key must change when the HMAC secret changes");
     }
 
     #[test]
     fn key_has_expected_prefix() {
-        let key = idempotency_key("user@example.com", "t", "secret");
+        let key = idempotency_key("user@example.com", "t", "secret", Utc::now());
         assert!(key.starts_with("email:idem:"), "key should start with email:idem: prefix");
     }
 
@@ -489,14 +502,52 @@ mod tests {
     /// Retry produces the same key (within the same hour bucket).
     #[test]
     fn retry_produces_same_idempotency_key() {
-        let key_attempt_1 = idempotency_key("user@example.com", "newsletter_confirmation", "secret");
-        let key_attempt_2 = idempotency_key("user@example.com", "newsletter_confirmation", "secret");
+        let created_at = Utc::now();
+        let key_attempt_1 = idempotency_key("user@example.com", "newsletter_confirmation", "secret", created_at);
+        let key_attempt_2 = idempotency_key("user@example.com", "newsletter_confirmation", "secret", created_at);
         assert_eq!(
             key_attempt_1, key_attempt_2,
             "retry must produce the same idempotency key"
         );
     }
 
+    /// #1129: a retry delayed past the 1-hour bucket boundary (e.g. by the
+    /// 6th exponential-backoff attempt, ~64 minutes out) must still compute
+    /// the same idempotency key as the original attempt, since both are
+    /// derived from the job's original creation time rather than "now".
+    #[test]
+    fn retry_key_is_stable_across_hour_boundary_when_derived_from_job_creation_time() {
+        let created_at = Utc::now();
+        let key_at_creation = idempotency_key("user@example.com", "newsletter_confirmation", "secret", created_at);
+
+        // Simulate a retry scheduled more than an hour after the original
+        // attempt (e.g. the ~64-minute delay of the 6th backoff attempt).
+        let delayed_retry_time = created_at + chrono::Duration::minutes(90);
+        let key_at_delayed_retry = idempotency_key(
+            "user@example.com",
+            "newsletter_confirmation",
+            "secret",
+            created_at, // bucket is still derived from the original creation time
+        );
+        assert_eq!(
+            key_at_creation, key_at_delayed_retry,
+            "idempotency key must remain stable across the job's full retry lifetime"
+        );
+
+        // Sanity check: had the key been derived from wall-clock "now" at the
+        // delayed retry time instead of the job's creation time, it would differ.
+        let key_from_wall_clock_now = idempotency_key(
+            "user@example.com",
+            "newsletter_confirmation",
+            "secret",
+            delayed_retry_time,
+        );
+        assert_ne!(
+            key_at_creation, key_from_wall_clock_now,
+            "test precondition: 90 minutes must cross an hour bucket boundary"
+        );
+    }
+
     /// Two 429s followed by a 202: the service should succeed on the third attempt.
     #[tokio::test]
     async fn retry_succeeds_after_two_429s() {

services/api/src/email/webhook.rs

@@ -1,5 +1,6 @@
 use anyhow::{anyhow, Result};
 use axum::{extract::State, http::{HeaderMap, StatusCode}, response::IntoResponse, Json};
+use futures::stream::{self, StreamExt};
 use serde::{Deserialize, Serialize};
 use std::{sync::Arc, time::Duration};
 
@@ -17,6 +18,11 @@ const MAX_TEXT_FIELD_LEN: usize = 1024;
 /// Maximum length for structured identifier fields (email, event type, message_id).
 const MAX_ID_FIELD_LEN: usize = 254;
 
+/// Maximum number of webhook events processed concurrently within a single batch.
+/// Bounds DB/Redis connection usage while avoiding fully sequential processing
+/// of large SendGrid batches (which can contain hundreds of events per POST).
+const WEBHOOK_BATCH_CONCURRENCY: usize = 16;
+
 /// Raw deserialization target — accepts any valid JSON so we can validate
 /// before committing anything to the database.
 #[derive(Debug, Clone, Deserialize)]
@@ -185,19 +191,33 @@ impl WebhookHandler {
     }
 
     /// Process a list of already-sanitized SendGrid webhook events.
+    ///
+    /// Events are processed concurrently (bounded by `WEBHOOK_BATCH_CONCURRENCY`)
+    /// rather than one at a time, since a single POST can contain hundreds of
+    /// events and fully sequential processing risks holding the request open
+    /// past the SendGrid webhook timeout budget.
     pub async fn handle_sendgrid_webhook(
         &self,
         events: Vec<SendGridEvent>,
     ) -> Result<WebhookResponse> {
+        let results: Vec<(String, Result<()>)> = stream::iter(events)
+            .map(|event| async move {
+                let event_desc = event.event.clone();
+                (event_desc, self.process_event(event).await)
+            })
+            .buffer_unordered(WEBHOOK_BATCH_CONCURRENCY)
+            .collect()
+            .await;
+
         let mut processed = 0;
         let mut errors = Vec::new();
 
-        for event in events {
-            match self.process_event(event.clone()).await {
+        for (event_desc, result) in results {
+            match result {
                 Ok(_) => processed += 1,
                 Err(e) => {
                     tracing::error!("Error processing webhook event: {}", e);
-                    errors.push(format!("Event {}: {}", event.event, e));
+                    errors.push(format!("Event {}: {}", event_desc, e));
                 }
             }
         }

services/tts/src/HealthCheck.ts

@@ -264,7 +264,7 @@ export class HealthChecker {
   private checkJobStore(): HealthCheckStatus {
     try {
       // Verify job store is functional
-      const jobs = this.service.listJobs();
+      const jobs = this.service.listAllJobsUnscoped();
 
       return {
         status: "ok",
@@ -284,8 +284,8 @@ export class HealthChecker {
    */
   private checkJobQueueDepth(): HealthCheckStatus {
     try {
-      const pending = this.service.listJobs("pending").length;
-      const processing = this.service.listJobs("processing").length;
+      const pending = this.service.listAllJobsUnscoped("pending").length;
+      const processing = this.service.listAllJobsUnscoped("processing").length;
       const depth = pending + processing;
 
       if (depth >= MAX_QUEUE_DEPTH) {