From 1e9fff69906e5e5cef588c51773cafe2c46e2fb1 Mon Sep 17 00:00:00 2001 From: Claudia L Date: Mon, 13 Jul 2026 11:58:01 +0200 Subject: [PATCH 1/3] feat(email): add bounce/complaint suppression list MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Refuses to send to addresses that previously hard-bounced or filed a spam complaint, since repeatedly emailing one is exactly what gets a sending domain's reputation downgraded by receiving mail providers. - New suppressed_recipients table + SuppressionService (suppress/unsuppress/ suppressed_among), normalized (trim+lowercase) on both write and lookup. - Checked in EmailService::send after the email row is inserted but before any domain/provider work. - Checks to/cc/bcc together, not just to — a suppressed address left in cc or bcc would otherwise still receive mail. - Drops only the suppressed addresses rather than capturing the whole send — a suppressed address mixed into `to` alongside legitimate recipients must not deny delivery to everyone on the email. - If every `to` address ends up suppressed (or `to` was already empty), the email is captured (not sent) with a clear error_message. Co-Authored-By: Claude Sonnet 5 --- .../src/handlers/tracking_tests.rs | 6 +- crates/temps-email/src/plugin.rs | 11 +- .../temps-email/src/services/email_service.rs | 175 ++++++++++- crates/temps-email/src/services/mod.rs | 2 + .../src/services/suppression_service.rs | 287 ++++++++++++++++++ crates/temps-entities/src/lib.rs | 1 + .../src/suppressed_recipients.rs | 41 +++ ...711_000002_create_suppressed_recipients.rs | 49 +++ crates/temps-migrations/src/migration/mod.rs | 2 + 9 files changed, 563 insertions(+), 11 deletions(-) create mode 100644 crates/temps-email/src/services/suppression_service.rs create mode 100644 crates/temps-entities/src/suppressed_recipients.rs create mode 100644 crates/temps-migrations/src/migration/m20260711_000002_create_suppressed_recipients.rs diff --git a/crates/temps-email/src/handlers/tracking_tests.rs b/crates/temps-email/src/handlers/tracking_tests.rs index d035bdc57..26ec6c52c 100644 --- a/crates/temps-email/src/handlers/tracking_tests.rs +++ b/crates/temps-email/src/handlers/tracking_tests.rs @@ -24,8 +24,8 @@ mod tests { use crate::handlers::tracking::{public_routes, routes}; use crate::handlers::types::AppState; use crate::services::{ - DomainService, EmailService, ProviderService, TrackingService, ValidationConfig, - ValidationService, + DomainService, EmailService, ProviderService, SuppressionService, TrackingService, + ValidationConfig, ValidationService, }; // ============================================ @@ -120,11 +120,13 @@ mod tests { config_service, "http://localhost:3000".to_string(), )); + let suppression_service = Arc::new(SuppressionService::new(db.db.clone())); let email_service = Arc::new(EmailService::new( db.db.clone(), provider_service.clone(), domain_service.clone(), tracking_service.clone(), + suppression_service, )); let validation_service = Arc::new(ValidationService::new(ValidationConfig::default())); diff --git a/crates/temps-email/src/plugin.rs b/crates/temps-email/src/plugin.rs index 360ac3c85..b1bbba789 100644 --- a/crates/temps-email/src/plugin.rs +++ b/crates/temps-email/src/plugin.rs @@ -13,8 +13,8 @@ use utoipa::OpenApi as OpenApiTrait; use crate::handlers::{self, AppState, EmailApiDoc}; use crate::services::{ - DomainService, EmailService, ProviderService, TrackingService, ValidationConfig, - ValidationService, + DomainService, EmailService, ProviderService, SuppressionService, TrackingService, + ValidationConfig, ValidationService, }; use temps_dns::services::DnsProviderService; @@ -61,12 +61,17 @@ impl TempsPlugin for EmailPlugin { let tracking_service = Arc::new(TrackingService::new(db.clone(), config_service)); context.register_service(tracking_service.clone()); - // Create EmailService with tracking support + // Create SuppressionService — bounce/complaint do-not-send list + let suppression_service = Arc::new(SuppressionService::new(db.clone())); + context.register_service(suppression_service.clone()); + + // Create EmailService with tracking + suppression support let email_service = Arc::new(EmailService::new( db.clone(), provider_service.clone(), domain_service.clone(), tracking_service.clone(), + suppression_service.clone(), )); context.register_service(email_service.clone()); diff --git a/crates/temps-email/src/services/email_service.rs b/crates/temps-email/src/services/email_service.rs index 4831d0217..d094b0f9c 100644 --- a/crates/temps-email/src/services/email_service.rs +++ b/crates/temps-email/src/services/email_service.rs @@ -12,7 +12,7 @@ use uuid::Uuid; use crate::errors::EmailError; use crate::providers::SendEmailRequest as ProviderSendRequest; -use crate::services::{DomainService, ProviderService, TrackingService}; +use crate::services::{DomainService, ProviderService, SuppressionService, TrackingService}; /// Trait for rewriting HTML to inject tracking (pixel + click links). /// Implemented by `temps-email-tracking::HtmlTrackingRewriter`. @@ -27,6 +27,7 @@ pub struct EmailService { domain_service: Arc, tracking_rewriter: Option>, tracking_service: Arc, + suppression_service: Arc, } /// Request to send an email @@ -75,6 +76,7 @@ impl EmailService { provider_service: Arc, domain_service: Arc, tracking_service: Arc, + suppression_service: Arc, ) -> Self { Self { db, @@ -82,6 +84,7 @@ impl EmailService { domain_service, tracking_rewriter: None, tracking_service, + suppression_service, } } @@ -181,6 +184,62 @@ impl EmailService { } } + // Refuse to send to previously hard-bounced/complained addresses — + // repeatedly emailing one is exactly what gets a sending domain's + // reputation downgraded by receiving mail providers. Checked after + // the row is inserted (still visible for debugging) but before any + // domain/provider work, since it's independent of both. + // + // Checks to/cc/bcc together (a suppressed address left in cc/bcc + // would otherwise still receive mail), and drops only the + // suppressed addresses rather than capturing the whole send — a + // suppressed address mixed into `to` alongside legitimate + // recipients used to silently deny delivery to everyone on the + // email, not just the bad address. + let mut all_recipients: Vec = request.to.clone(); + all_recipients.extend(request.cc.iter().flatten().cloned()); + all_recipients.extend(request.bcc.iter().flatten().cloned()); + + let suppressed = self + .suppression_service + .suppressed_among(&all_recipients) + .await?; + + if !suppressed.is_empty() { + info!( + "Dropping suppressed recipient(s) from email {}: {:?}", + email_id, suppressed + ); + } + let (to, cc, bcc) = + filter_suppressed_recipients(request.to, request.cc, request.bcc, &suppressed); + + // Nothing left to send to (either every `to` address was + // suppressed, or `to` was already empty) — capture instead of + // sending an email with no primary recipient. + if to.is_empty() { + info!( + "Refusing to send email {} — all recipient(s) suppressed: {:?}", + email_id, suppressed + ); + + let mut active_model: emails::ActiveModel = email_model.into(); + active_model.status = Set("captured".to_string()); + active_model.error_message = Set(Some(format!( + "Recipient(s) suppressed (previous hard bounce or complaint): {}", + suppressed.join(", ") + ))); + active_model.sent_at = Set(Some(Utc::now())); + + active_model.update(self.db.as_ref()).await?; + + return Ok(SendEmailResponse { + id: email_id, + status: "captured".to_string(), + provider_message_id: None, + }); + } + // If no domain configured, capture email without sending (Mailhog-like behavior) let domain = match domain { Some(d) => d, @@ -198,7 +257,7 @@ impl EmailService { info!( "Email captured (no domain configured), id: {}, from: {}, to: {:?}", - email_id, request.from, request.to + email_id, request.from, to ); return Ok(SendEmailResponse { @@ -256,7 +315,7 @@ impl EmailService { info!( "Email captured (no provider), id: {}, from: {}, to: {:?}", - email_id, request.from, request.to + email_id, request.from, to ); return Ok(SendEmailResponse { @@ -299,9 +358,9 @@ impl EmailService { let provider_request = ProviderSendRequest { from: request.from, from_name: request.from_name, - to: request.to, - cc: request.cc, - bcc: request.bcc, + to, + cc, + bcc, reply_to: request.reply_to, subject: request.subject, html: tracked_html, @@ -447,6 +506,33 @@ pub struct EmailStats { pub captured: u64, } +/// Drop suppressed addresses from `to`/`cc`/`bcc`. `suppressed` is the +/// (already-normalized) output of `SuppressionService::suppressed_among`. +/// +/// Filters each list independently rather than rejecting the whole send — +/// a suppressed address mixed into `to` alongside legitimate recipients +/// must not deny delivery to everyone on the email, just to itself. +fn filter_suppressed_recipients( + to: Vec, + cc: Option>, + bcc: Option>, + suppressed: &[String], +) -> (Vec, Option>, Option>) { + if suppressed.is_empty() { + return (to, cc, bcc); + } + + let suppressed_set: std::collections::HashSet<&str> = + suppressed.iter().map(String::as_str).collect(); + let keep = |addr: &String| !suppressed_set.contains(SuppressionService::normalize(addr).as_str()); + + ( + to.into_iter().filter(&keep).collect(), + cc.map(|list| list.into_iter().filter(&keep).collect()), + bcc.map(|list| list.into_iter().filter(&keep).collect()), + ) +} + #[cfg(test)] mod tests { use super::*; @@ -501,11 +587,13 @@ mod tests { config_service, "http://localhost:3000".to_string(), )); + let suppression_service = Arc::new(SuppressionService::new(db.db.clone())); let email_service = EmailService::new( db.db.clone(), Arc::new(provider_service.clone()), Arc::new(domain_service.clone()), tracking_service, + suppression_service, ); (db, email_service, provider_service, domain_service) } @@ -662,6 +750,81 @@ mod tests { assert!(domain.is_none()); } + #[test] + fn filter_suppressed_recipients_passes_through_when_nothing_suppressed() { + let (to, cc, bcc) = filter_suppressed_recipients( + vec!["a@example.com".to_string()], + Some(vec!["b@example.com".to_string()]), + None, + &[], + ); + assert_eq!(to, vec!["a@example.com"]); + assert_eq!(cc, Some(vec!["b@example.com".to_string()])); + assert_eq!(bcc, None); + } + + #[test] + fn filter_suppressed_recipients_drops_only_the_suppressed_cc_address() { + // A suppressed address in `cc` used to be invisible to the check + // entirely — it must be dropped from the send, and it must not take + // the legitimate `to` recipient down with it. + let (to, cc, bcc) = filter_suppressed_recipients( + vec!["good@example.com".to_string()], + Some(vec![ + "bad@example.com".to_string(), + "also-good@example.com".to_string(), + ]), + None, + &["bad@example.com".to_string()], + ); + assert_eq!(to, vec!["good@example.com"]); + assert_eq!(cc, Some(vec!["also-good@example.com".to_string()])); + assert_eq!(bcc, None); + } + + #[test] + fn filter_suppressed_recipients_keeps_other_to_addresses() { + // One suppressed address mixed into `to` used to capture the whole + // send — the other `to` recipients must still get the email. + let (to, cc, bcc) = filter_suppressed_recipients( + vec![ + "bad@example.com".to_string(), + "good@example.com".to_string(), + ], + None, + None, + &["bad@example.com".to_string()], + ); + assert_eq!(to, vec!["good@example.com"]); + assert_eq!(cc, None); + assert_eq!(bcc, None); + } + + #[test] + fn filter_suppressed_recipients_matches_case_and_whitespace_insensitively() { + // `suppressed_among` returns normalized (trimmed/lowercased) forms + // from the DB — the filter must normalize candidates the same way, + // not compare raw strings. + let (to, _, _) = filter_suppressed_recipients( + vec![" Bad@Example.COM ".to_string(), "good@example.com".to_string()], + None, + None, + &["bad@example.com".to_string()], + ); + assert_eq!(to, vec!["good@example.com"]); + } + + #[test] + fn filter_suppressed_recipients_empties_to_when_all_suppressed() { + let (to, _, _) = filter_suppressed_recipients( + vec!["bad@example.com".to_string()], + None, + None, + &["bad@example.com".to_string()], + ); + assert!(to.is_empty()); + } + #[test] fn test_list_emails_options_builder() { // Test that list options can be constructed with various filters diff --git a/crates/temps-email/src/services/mod.rs b/crates/temps-email/src/services/mod.rs index 94be4b7ba..5686c453f 100644 --- a/crates/temps-email/src/services/mod.rs +++ b/crates/temps-email/src/services/mod.rs @@ -3,6 +3,7 @@ mod domain_service; mod email_service; mod provider_service; +mod suppression_service; mod tracking_service; #[cfg(test)] mod tracking_service_integration_tests; @@ -17,6 +18,7 @@ pub use provider_service::{ CreateProviderRequest, ProviderCredentials, ProviderService, TestEmailResult, UpdateProviderOutcome, UpdateProviderRequest, }; +pub use suppression_service::{SuppressionReason, SuppressionService}; pub use tracking_service::{ExtractedLink, TrackingEvent, TrackingService, TransformResult}; pub use validation::{ MiscResult, MxResult, ProxyConfig, ReachabilityStatus, SmtpResult, SyntaxResult, diff --git a/crates/temps-email/src/services/suppression_service.rs b/crates/temps-email/src/services/suppression_service.rs new file mode 100644 index 000000000..ecdb021b9 --- /dev/null +++ b/crates/temps-email/src/services/suppression_service.rs @@ -0,0 +1,287 @@ +//! Suppression list — recipients who must not receive further email due to +//! a hard bounce, a spam complaint, or a manual admin action. Checked by +//! `EmailService::send` before every send: without this, a permanently-bad +//! or complained address kept getting mailed on every subsequent send, +//! which is the exact pattern that gets a sending domain downgraded by +//! receiving mail providers. + +use sea_orm::{ + ActiveModelTrait, ActiveValue::Set, ColumnTrait, DatabaseConnection, EntityTrait, + PaginatorTrait, QueryFilter, QueryOrder, +}; +use std::sync::Arc; +use temps_entities::suppressed_recipients; + +use crate::errors::EmailError; + +/// Why an address was suppressed. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum SuppressionReason { + /// A hard/permanent bounce (mailbox doesn't exist, domain rejects mail, …). + Bounced, + /// The recipient marked a message as spam. + Complained, + /// An admin suppressed (or un-suppressed) the address by hand. + Manual, +} + +impl SuppressionReason { + fn as_str(&self) -> &'static str { + match self { + SuppressionReason::Bounced => "bounced", + SuppressionReason::Complained => "complained", + SuppressionReason::Manual => "manual", + } + } +} + +/// Service for managing the email suppression list. +pub struct SuppressionService { + db: Arc, +} + +impl SuppressionService { + pub fn new(db: Arc) -> Self { + Self { db } + } + + /// Trim + lowercase for storage/lookup comparison. `pub(crate)` so + /// callers filtering a recipient list against `suppressed_among`'s + /// results (which come back normalized, not in their original casing) + /// can match them correctly. + pub(crate) fn normalize(email: &str) -> String { + email.trim().to_lowercase() + } + + /// Add an address to the suppression list, or update its reason if it's + /// already there (e.g. a bounce followed later by a complaint). + pub async fn suppress( + &self, + email: &str, + reason: SuppressionReason, + domain_id: Option, + detail: Option, + ) -> Result<(), EmailError> { + let normalized = Self::normalize(email); + + let existing = suppressed_recipients::Entity::find() + .filter(suppressed_recipients::Column::Email.eq(normalized.clone())) + .one(self.db.as_ref()) + .await?; + + match existing { + Some(model) => { + let mut active: suppressed_recipients::ActiveModel = model.into(); + active.reason = Set(reason.as_str().to_string()); + active.domain_id = Set(domain_id); + active.detail = Set(detail); + active.update(self.db.as_ref()).await?; + } + None => { + let active = suppressed_recipients::ActiveModel { + email: Set(normalized), + reason: Set(reason.as_str().to_string()), + domain_id: Set(domain_id), + detail: Set(detail), + ..Default::default() + }; + active.insert(self.db.as_ref()).await?; + } + } + + Ok(()) + } + + /// Remove an address from the suppression list (manual admin override — + /// e.g. the mailbox was fixed, or the bounce/complaint was a mistake). + pub async fn unsuppress(&self, email: &str) -> Result<(), EmailError> { + let normalized = Self::normalize(email); + suppressed_recipients::Entity::delete_many() + .filter(suppressed_recipients::Column::Email.eq(normalized)) + .exec(self.db.as_ref()) + .await?; + Ok(()) + } + + /// Is this single address currently suppressed? + pub async fn is_suppressed(&self, email: &str) -> Result { + let normalized = Self::normalize(email); + let count = suppressed_recipients::Entity::find() + .filter(suppressed_recipients::Column::Email.eq(normalized)) + .count(self.db.as_ref()) + .await?; + Ok(count > 0) + } + + /// Which of these addresses are currently suppressed — one query instead + /// of N, for `EmailService::send` checking every `to` recipient at once. + pub async fn suppressed_among(&self, emails: &[String]) -> Result, EmailError> { + if emails.is_empty() { + return Ok(Vec::new()); + } + let normalized: Vec = emails.iter().map(|e| Self::normalize(e)).collect(); + let rows = suppressed_recipients::Entity::find() + .filter(suppressed_recipients::Column::Email.is_in(normalized)) + .all(self.db.as_ref()) + .await?; + Ok(rows.into_iter().map(|r| r.email).collect()) + } + + /// Paginated list of the whole suppression list, most recent first. + pub async fn list( + &self, + page: u64, + page_size: u64, + ) -> Result<(Vec, u64), EmailError> { + let page = page.max(1); + let page_size = std::cmp::min(page_size, 100).max(1); + + let paginator = suppressed_recipients::Entity::find() + .order_by_desc(suppressed_recipients::Column::CreatedAt) + .paginate(self.db.as_ref(), page_size); + + let total = paginator.num_items().await?; + let items = paginator.fetch_page(page - 1).await?; + Ok((items, total)) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use temps_database::test_utils::TestDatabase; + + async fn setup() -> (TestDatabase, SuppressionService) { + let db = TestDatabase::with_migrations().await.unwrap(); + let service = SuppressionService::new(db.db.clone()); + (db, service) + } + + #[test] + fn suppression_reason_as_str() { + assert_eq!(SuppressionReason::Bounced.as_str(), "bounced"); + assert_eq!(SuppressionReason::Complained.as_str(), "complained"); + assert_eq!(SuppressionReason::Manual.as_str(), "manual"); + } + + #[test] + fn normalize_trims_and_lowercases() { + assert_eq!( + SuppressionService::normalize(" Person@Example.COM "), + "person@example.com" + ); + } + + #[tokio::test] + async fn not_suppressed_by_default() { + let (_db, service) = setup().await; + assert!(!service.is_suppressed("nobody@example.com").await.unwrap()); + } + + #[tokio::test] + async fn suppress_then_is_suppressed() { + let (_db, service) = setup().await; + service + .suppress( + "Bounced@Example.com", + SuppressionReason::Bounced, + None, + Some("mailbox does not exist".to_string()), + ) + .await + .unwrap(); + + // Case/whitespace-insensitive lookup. + assert!(service + .is_suppressed(" bounced@example.com ") + .await + .unwrap()); + } + + #[tokio::test] + async fn suppress_is_idempotent_and_updates_reason() { + let (_db, service) = setup().await; + let email = "person@example.com"; + + service + .suppress(email, SuppressionReason::Bounced, None, None) + .await + .unwrap(); + service + .suppress(email, SuppressionReason::Complained, None, None) + .await + .unwrap(); + + let (rows, total) = service.list(1, 10).await.unwrap(); + assert_eq!(total, 1, "re-suppressing must not create a duplicate row"); + assert_eq!(rows[0].reason, "complained"); + } + + #[tokio::test] + async fn unsuppress_removes_the_address() { + let (_db, service) = setup().await; + let email = "person@example.com"; + service + .suppress(email, SuppressionReason::Manual, None, None) + .await + .unwrap(); + assert!(service.is_suppressed(email).await.unwrap()); + + service.unsuppress(email).await.unwrap(); + assert!(!service.is_suppressed(email).await.unwrap()); + } + + #[tokio::test] + async fn unsuppress_nonexistent_is_a_noop() { + let (_db, service) = setup().await; + assert!(service.unsuppress("nobody@example.com").await.is_ok()); + } + + #[tokio::test] + async fn suppressed_among_returns_only_matches() { + let (_db, service) = setup().await; + service + .suppress("bad@example.com", SuppressionReason::Bounced, None, None) + .await + .unwrap(); + + let result = service + .suppressed_among(&[ + "bad@example.com".to_string(), + "good@example.com".to_string(), + ]) + .await + .unwrap(); + + assert_eq!(result, vec!["bad@example.com".to_string()]); + } + + #[tokio::test] + async fn suppressed_among_empty_input_short_circuits() { + let (_db, service) = setup().await; + assert_eq!(service.suppressed_among(&[]).await.unwrap(), Vec::::new()); + } + + #[tokio::test] + async fn list_is_paginated_most_recent_first() { + let (_db, service) = setup().await; + for i in 0..3 { + service + .suppress( + &format!("person{i}@example.com"), + SuppressionReason::Manual, + None, + None, + ) + .await + .unwrap(); + } + + let (page1, total) = service.list(1, 2).await.unwrap(); + assert_eq!(total, 3); + assert_eq!(page1.len(), 2); + + let (page2, _) = service.list(2, 2).await.unwrap(); + assert_eq!(page2.len(), 1); + } +} diff --git a/crates/temps-entities/src/lib.rs b/crates/temps-entities/src/lib.rs index f49a1789c..c803c909f 100644 --- a/crates/temps-entities/src/lib.rs +++ b/crates/temps-entities/src/lib.rs @@ -94,6 +94,7 @@ pub mod sessions; pub mod source_type; pub mod static_asset_cache; pub mod static_bundles; +pub mod suppressed_recipients; pub mod tls_acme_certificates; pub mod types; pub mod upstream_config; diff --git a/crates/temps-entities/src/suppressed_recipients.rs b/crates/temps-entities/src/suppressed_recipients.rs new file mode 100644 index 000000000..22a73e7b5 --- /dev/null +++ b/crates/temps-entities/src/suppressed_recipients.rs @@ -0,0 +1,41 @@ +//! Suppressed recipients entity — addresses that must not receive further +//! email due to a hard bounce, a spam complaint, or a manual admin action. +//! Checked by `EmailService::send` before every send. + +use sea_orm::entity::prelude::*; +use serde::{Deserialize, Serialize}; +use temps_core::DBDateTime; + +#[derive(Clone, Debug, PartialEq, DeriveEntityModel, Eq, Serialize, Deserialize)] +#[sea_orm(table_name = "suppressed_recipients")] +pub struct Model { + #[sea_orm(primary_key)] + pub id: i32, + /// Always lowercased/trimmed before storage — see SuppressionService::normalize. + pub email: String, + /// "bounced" | "complained" | "manual" + pub reason: String, + /// Which domain's send triggered this, if known. Diagnostic only — + /// suppression is enforced globally, not scoped to one sending domain. + pub domain_id: Option, + pub detail: Option, + pub created_at: DBDateTime, +} + +#[derive(Copy, Clone, Debug, EnumIter, DeriveRelation)] +pub enum Relation { + #[sea_orm( + belongs_to = "super::email_domains::Entity", + from = "Column::DomainId", + to = "super::email_domains::Column::Id" + )] + EmailDomain, +} + +impl Related for Entity { + fn to() -> RelationDef { + Relation::EmailDomain.def() + } +} + +impl ActiveModelBehavior for ActiveModel {} diff --git a/crates/temps-migrations/src/migration/m20260711_000002_create_suppressed_recipients.rs b/crates/temps-migrations/src/migration/m20260711_000002_create_suppressed_recipients.rs new file mode 100644 index 000000000..99100c06c --- /dev/null +++ b/crates/temps-migrations/src/migration/m20260711_000002_create_suppressed_recipients.rs @@ -0,0 +1,49 @@ +//! Suppression list: recipients who must not receive further email due to a +//! hard bounce, a spam complaint, or a manual admin action. Without this, +//! nothing stopped a permanently-bad or complained address from being +//! emailed again on the next send, which is exactly the pattern that gets a +//! sending domain's reputation downgraded by receiving mail providers. +//! Enforced globally (not per sending domain) in `EmailService::send`. + +use sea_orm_migration::prelude::*; + +pub struct Migration; + +impl MigrationName for Migration { + fn name(&self) -> &str { + "m20260711_000002_create_suppressed_recipients" + } +} + +#[async_trait::async_trait] +impl MigrationTrait for Migration { + async fn up(&self, manager: &SchemaManager) -> Result<(), DbErr> { + manager + .get_connection() + .execute_unprepared( + r#" + CREATE TABLE IF NOT EXISTS suppressed_recipients ( + id SERIAL PRIMARY KEY, + email TEXT NOT NULL, + reason TEXT NOT NULL, + domain_id INTEGER REFERENCES email_domains(id) ON DELETE SET NULL, + detail TEXT, + created_at TIMESTAMPTZ NOT NULL DEFAULT now() + ); + + CREATE UNIQUE INDEX IF NOT EXISTS idx_suppressed_recipients_email + ON suppressed_recipients (email); + "#, + ) + .await?; + Ok(()) + } + + async fn down(&self, manager: &SchemaManager) -> Result<(), DbErr> { + manager + .get_connection() + .execute_unprepared("DROP TABLE IF EXISTS suppressed_recipients;") + .await?; + Ok(()) + } +} diff --git a/crates/temps-migrations/src/migration/mod.rs b/crates/temps-migrations/src/migration/mod.rs index ee77f0fa1..01cfc7537 100644 --- a/crates/temps-migrations/src/migration/mod.rs +++ b/crates/temps-migrations/src/migration/mod.rs @@ -144,6 +144,7 @@ mod m20260705_000001_add_visitor_unique_index; mod m20260707_000001_add_external_service_to_logs; mod m20260707_000002_add_external_services_container_name; mod m20260708_000001_add_node_id_to_monitoring_alert_rules; +mod m20260711_000002_create_suppressed_recipients; pub struct Migrator; @@ -293,6 +294,7 @@ impl MigratorTrait for Migrator { Box::new(m20260707_000001_add_external_service_to_logs::Migration), Box::new(m20260707_000002_add_external_services_container_name::Migration), Box::new(m20260708_000001_add_node_id_to_monitoring_alert_rules::Migration), + Box::new(m20260711_000002_create_suppressed_recipients::Migration), ] } } From 967157c041de64ae7797e1758d5d86f9547cf8a4 Mon Sep 17 00:00:00 2001 From: Claudia L Date: Mon, 13 Jul 2026 12:15:10 +0200 Subject: [PATCH 2/3] feat(email): add multi-provider failover with retry, circuit breaker, and rate limiting MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Previously a domain had exactly one provider, and any send failure went straight to "captured" — a transient SES throttle or a Scaleway blip permanently dropped the email instead of trying again or falling back to a backup provider. - Ordered failover chain per domain: primary provider, then configured fallbacks in priority order (new `email_domain_fallback_providers` table + CRUD endpoints). Inactive providers are dropped from the chain, so disabling one now actually takes it out of the send path. - Error classification (`EmailError::SendFailed { retryable, .. }`) per provider: HTTP status for Scaleway (5xx/429 retryable), SdkError kind for SES (throttling/timeout/network retryable, message-level rejection not), SMTP reply code via lettre's is_transient/is_timeout for SMTP. - Bounded retry (2 attempts) against a provider before moving to the next one in the chain — only for errors classified as transient. - Per-provider circuit breaker (opens after 5 consecutive failures, cooldown before retrying) and rate limiter (`rate_limit_per_minute`, operator-configurable per provider, sliding window), both in-memory and scoped to the process (control-plane code, not hot-path). - `emails.provider_id`/`retry_count` record which provider ultimately handled the send and how many attempts it took, across the whole chain. Co-Authored-By: Claude Sonnet 5 --- crates/temps-email/src/errors.rs | 25 +++ crates/temps-email/src/handlers/audit.rs | 32 +++ crates/temps-email/src/handlers/domains.rs | 175 +++++++++++++++- crates/temps-email/src/handlers/mod.rs | 5 + crates/temps-email/src/handlers/providers.rs | 5 + crates/temps-email/src/handlers/types.rs | 29 +++ crates/temps-email/src/providers/scaleway.rs | 37 +++- crates/temps-email/src/providers/ses.rs | 49 ++++- crates/temps-email/src/providers/smtp.rs | 12 +- .../temps-email/src/services/email_service.rs | 157 ++++++++------ crates/temps-email/src/services/mod.rs | 1 + .../src/services/provider_service.rs | 170 ++++++++++++++- crates/temps-email/src/services/resilience.rs | 195 ++++++++++++++++++ .../src/email_domain_fallback_providers.rs | 49 +++++ crates/temps-entities/src/email_providers.rs | 2 + crates/temps-entities/src/emails.rs | 7 + crates/temps-entities/src/lib.rs | 1 + ...712_000001_add_email_failover_and_retry.rs | 72 +++++++ crates/temps-migrations/src/migration/mod.rs | 2 + 19 files changed, 949 insertions(+), 76 deletions(-) create mode 100644 crates/temps-email/src/services/resilience.rs create mode 100644 crates/temps-entities/src/email_domain_fallback_providers.rs create mode 100644 crates/temps-migrations/src/migration/m20260712_000001_add_email_failover_and_retry.rs diff --git a/crates/temps-email/src/errors.rs b/crates/temps-email/src/errors.rs index 343209ec1..a43e22de6 100644 --- a/crates/temps-email/src/errors.rs +++ b/crates/temps-email/src/errors.rs @@ -2,6 +2,8 @@ use thiserror::Error; +use crate::providers::EmailProviderType; + #[derive(Error, Debug)] pub enum EmailError { #[error("Database error: {0}")] @@ -51,6 +53,19 @@ pub enum EmailError { #[error("Tracking rewrite failed for email {email_id}: {reason}")] TrackingRewrite { email_id: String, reason: String }, + + /// A provider's `send()` call failed. Unlike the flat per-provider + /// variants above (used for identity/domain management calls), + /// this carries a `retryable` classification derived from the + /// underlying transport error (HTTP status, SMTP reply code, or AWS SDK + /// error kind) so the send path knows whether retrying — same provider + /// or the next one in the failover chain — can plausibly succeed. + #[error("Failed to send email via {provider}: {reason}")] + SendFailed { + provider: EmailProviderType, + retryable: bool, + reason: String, + }, } impl From for EmailError { @@ -58,3 +73,13 @@ impl From for EmailError { EmailError::Serialization(err.to_string()) } } + +impl EmailError { + /// Whether this failure is transient and worth retrying (against the + /// same provider or the next one in a domain's failover chain). Every + /// variant besides `SendFailed { retryable: true, .. }` represents a + /// config, validation, or lookup problem that a retry cannot fix. + pub fn is_retryable(&self) -> bool { + matches!(self, EmailError::SendFailed { retryable: true, .. }) + } +} diff --git a/crates/temps-email/src/handlers/audit.rs b/crates/temps-email/src/handlers/audit.rs index 7035ffac9..0cf518d1f 100644 --- a/crates/temps-email/src/handlers/audit.rs +++ b/crates/temps-email/src/handlers/audit.rs @@ -226,6 +226,38 @@ impl AuditOperation for EmailDomainDeletedAudit { } } +#[derive(Debug, Clone, Serialize)] +pub struct EmailDomainFallbackProviderChangedAudit { + pub context: AuditContext, + pub domain_id: i32, + pub provider_id: i32, + /// "added" or "removed" + pub action: String, + pub priority: Option, +} + +impl AuditOperation for EmailDomainFallbackProviderChangedAudit { + fn operation_type(&self) -> String { + "EMAIL_DOMAIN_FALLBACK_PROVIDER_CHANGED".to_string() + } + + fn user_id(&self) -> i32 { + self.context.user_id + } + + fn ip_address(&self) -> Option { + self.context.ip_address.clone() + } + + fn user_agent(&self) -> &str { + &self.context.user_agent + } + + fn serialize(&self) -> anyhow::Result { + serde_json::to_string(self).map_err(|e| anyhow::anyhow!("Failed to serialize: {}", e)) + } +} + // ======================================== // Email Audit Types // ======================================== diff --git a/crates/temps-email/src/handlers/domains.rs b/crates/temps-email/src/handlers/domains.rs index caa1cebf4..ccc8e1f97 100644 --- a/crates/temps-email/src/handlers/domains.rs +++ b/crates/temps-email/src/handlers/domains.rs @@ -18,10 +18,14 @@ use temps_core::{ use temps_dns::providers::{DnsProvider, DnsRecordContent, DnsRecordRequest}; use tracing::{error, info, warn}; -use super::audit::{EmailDomainCreatedAudit, EmailDomainDeletedAudit, EmailDomainVerifiedAudit}; +use super::audit::{ + EmailDomainCreatedAudit, EmailDomainDeletedAudit, EmailDomainFallbackProviderChangedAudit, + EmailDomainVerifiedAudit, +}; use super::types::{ - AppState, CreateEmailDomainRequest, DnsRecordResponse, DnsRecordSetupResult, - EmailDomainResponse, EmailDomainWithDnsResponse, SetupDnsRequest, SetupDnsResponse, + AddFallbackProviderRequest, AppState, CreateEmailDomainRequest, DnsRecordResponse, + DnsRecordSetupResult, EmailDomainFallbackProviderResponse, EmailDomainResponse, + EmailDomainWithDnsResponse, SetupDnsRequest, SetupDnsResponse, }; use crate::errors::EmailError; use crate::services::CreateDomainRequest; @@ -65,6 +69,10 @@ impl From for Problem { .with_title("Internal Server Error") .with_detail(error.to_string()) } + + EmailError::SendFailed { .. } => problemdetails::new(StatusCode::BAD_GATEWAY) + .with_title("Provider Send Failed") + .with_detail(error.to_string()), } } } @@ -87,6 +95,14 @@ pub fn routes() -> Router> { ) .route("/email-domains/{id}/verify", post(verify_domain)) .route("/email-domains/{id}/setup-dns", post(setup_dns)) + .route( + "/email-domains/{id}/fallback-providers", + get(list_domain_fallback_providers).post(add_domain_fallback_provider), + ) + .route( + "/email-domains/{id}/fallback-providers/{provider_id}", + axum::routing::delete(remove_domain_fallback_provider), + ) } /// Create a new email domain @@ -520,6 +536,159 @@ pub async fn delete_email_domain( Ok(StatusCode::NO_CONTENT) } +/// List a domain's fallback providers (send failover chain, priority order) +#[utoipa::path( + tag = "Email Domains", + get, + path = "/email-domains/{id}/fallback-providers", + responses( + (status = 200, description = "Fallback providers in priority order", body = Vec), + (status = 401, description = "Unauthorized"), + (status = 403, description = "Insufficient permissions"), + (status = 404, description = "Domain not found"), + (status = 500, description = "Internal server error") + ), + params( + ("id" = i32, Path, description = "Domain ID") + ), + security(("bearer_auth" = [])) +)] +pub async fn list_domain_fallback_providers( + RequireAuth(auth): RequireAuth, + State(state): State>, + Path(id): Path, +) -> Result { + permission_guard!(auth, EmailDomainsRead); + + // 404 if the domain itself doesn't exist, rather than silently + // returning an empty list for a nonexistent id. + state.domain_service.get(id).await?; + + let links = state.provider_service.list_fallback_providers(id).await?; + + let response: Vec = links + .into_iter() + .map(|l| EmailDomainFallbackProviderResponse { + id: l.id, + domain_id: l.domain_id, + provider_id: l.provider_id, + priority: l.priority, + created_at: l.created_at.to_rfc3339(), + }) + .collect(); + + Ok(Json(response)) +} + +/// Add (or re-prioritize) a fallback provider for a domain's send failover chain +#[utoipa::path( + tag = "Email Domains", + post, + path = "/email-domains/{id}/fallback-providers", + request_body = AddFallbackProviderRequest, + responses( + (status = 200, description = "Fallback provider added", body = EmailDomainFallbackProviderResponse), + (status = 401, description = "Unauthorized"), + (status = 403, description = "Insufficient permissions"), + (status = 404, description = "Domain or provider not found"), + (status = 500, description = "Internal server error") + ), + params( + ("id" = i32, Path, description = "Domain ID") + ), + security(("bearer_auth" = [])) +)] +pub async fn add_domain_fallback_provider( + RequireAuth(auth): RequireAuth, + State(state): State>, + axum::Extension(metadata): axum::Extension, + Path(id): Path, + Json(request): Json, +) -> Result { + permission_guard!(auth, EmailDomainsWrite); + + state.domain_service.get(id).await?; + + let link = state + .provider_service + .add_fallback_provider(id, request.provider_id, request.priority) + .await?; + + let audit = EmailDomainFallbackProviderChangedAudit { + context: AuditContext { + user_id: auth.user_id(), + ip_address: Some(metadata.ip_address.clone()), + user_agent: metadata.user_agent.clone(), + }, + domain_id: id, + provider_id: request.provider_id, + action: "added".to_string(), + priority: Some(request.priority), + }; + if let Err(e) = state.audit_service.create_audit_log(&audit).await { + error!("Failed to create audit log: {}", e); + } + + Ok(Json(EmailDomainFallbackProviderResponse { + id: link.id, + domain_id: link.domain_id, + provider_id: link.provider_id, + priority: link.priority, + created_at: link.created_at.to_rfc3339(), + })) +} + +/// Remove a fallback provider from a domain's send failover chain +#[utoipa::path( + tag = "Email Domains", + delete, + path = "/email-domains/{id}/fallback-providers/{provider_id}", + responses( + (status = 204, description = "Fallback provider removed (or wasn't configured)"), + (status = 401, description = "Unauthorized"), + (status = 403, description = "Insufficient permissions"), + (status = 404, description = "Domain not found"), + (status = 500, description = "Internal server error") + ), + params( + ("id" = i32, Path, description = "Domain ID"), + ("provider_id" = i32, Path, description = "Provider ID") + ), + security(("bearer_auth" = [])) +)] +pub async fn remove_domain_fallback_provider( + RequireAuth(auth): RequireAuth, + State(state): State>, + axum::Extension(metadata): axum::Extension, + Path((id, provider_id)): Path<(i32, i32)>, +) -> Result { + permission_guard!(auth, EmailDomainsWrite); + + state.domain_service.get(id).await?; + + state + .provider_service + .remove_fallback_provider(id, provider_id) + .await?; + + let audit = EmailDomainFallbackProviderChangedAudit { + context: AuditContext { + user_id: auth.user_id(), + ip_address: Some(metadata.ip_address.clone()), + user_agent: metadata.user_agent.clone(), + }, + domain_id: id, + provider_id, + action: "removed".to_string(), + priority: None, + }; + if let Err(e) = state.audit_service.create_audit_log(&audit).await { + error!("Failed to create audit log: {}", e); + } + + Ok(StatusCode::NO_CONTENT) +} + /// Setup DNS records for an email domain using a configured DNS provider #[utoipa::path( tag = "Email Domains", diff --git a/crates/temps-email/src/handlers/mod.rs b/crates/temps-email/src/handlers/mod.rs index 66577d2d1..eb396a416 100644 --- a/crates/temps-email/src/handlers/mod.rs +++ b/crates/temps-email/src/handlers/mod.rs @@ -50,6 +50,9 @@ pub fn configure_public_routes() -> Router> { domains::verify_domain, domains::delete_email_domain, domains::setup_dns, + domains::list_domain_fallback_providers, + domains::add_domain_fallback_provider, + domains::remove_domain_fallback_provider, // Emails emails::send_email, emails::list_emails, @@ -86,6 +89,8 @@ pub fn configure_public_routes() -> Router> { types::SetupDnsRequest, types::SetupDnsResponse, types::DnsRecordSetupResult, + types::EmailDomainFallbackProviderResponse, + types::AddFallbackProviderRequest, // Email types types::SendEmailRequestBody, types::SendEmailResponseBody, diff --git a/crates/temps-email/src/handlers/providers.rs b/crates/temps-email/src/handlers/providers.rs index da6313233..d99075b0e 100644 --- a/crates/temps-email/src/handlers/providers.rs +++ b/crates/temps-email/src/handlers/providers.rs @@ -176,6 +176,7 @@ pub async fn create_email_provider( .unwrap_or(EmailProviderTypeRoute::Ses), region: provider.region, is_active: provider.is_active, + rate_limit_per_minute: provider.rate_limit_per_minute, credentials: masked_credentials, created_at: provider.created_at.to_rfc3339(), updated_at: provider.updated_at.to_rfc3339(), @@ -226,6 +227,7 @@ pub async fn list_email_providers( .unwrap_or(EmailProviderTypeRoute::Ses), region: p.region, is_active: p.is_active, + rate_limit_per_minute: p.rate_limit_per_minute, credentials: masked_credentials, created_at: p.created_at.to_rfc3339(), updated_at: p.updated_at.to_rfc3339(), @@ -278,6 +280,7 @@ pub async fn get_email_provider( .unwrap_or(EmailProviderTypeRoute::Ses), region: provider.region, is_active: provider.is_active, + rate_limit_per_minute: provider.rate_limit_per_minute, credentials: masked_credentials, created_at: provider.created_at.to_rfc3339(), updated_at: provider.updated_at.to_rfc3339(), @@ -407,6 +410,7 @@ pub async fn update_email_provider( name: request.name, region: request.region, is_active: request.is_active, + rate_limit_per_minute: request.rate_limit_per_minute.map(Some), credentials, }; @@ -453,6 +457,7 @@ pub async fn update_email_provider( .unwrap_or(EmailProviderTypeRoute::Ses), region: provider.region, is_active: provider.is_active, + rate_limit_per_minute: provider.rate_limit_per_minute, credentials: masked_credentials, created_at: provider.created_at.to_rfc3339(), updated_at: provider.updated_at.to_rfc3339(), diff --git a/crates/temps-email/src/handlers/types.rs b/crates/temps-email/src/handlers/types.rs index be77c6880..9d842fdaf 100644 --- a/crates/temps-email/src/handlers/types.rs +++ b/crates/temps-email/src/handlers/types.rs @@ -172,6 +172,11 @@ pub struct UpdateEmailProviderRequest { pub region: Option, #[serde(default, skip_serializing_if = "Option::is_none")] pub is_active: Option, + /// Send-path cap for this provider. Omit to leave unchanged; set to + /// clamp throughput (e.g. a rate-limited SMTP relay). + #[serde(default, skip_serializing_if = "Option::is_none")] + #[schema(example = 120)] + pub rate_limit_per_minute: Option, #[serde(default, skip_serializing_if = "Option::is_none")] pub ses_credentials: Option, #[serde(default, skip_serializing_if = "Option::is_none")] @@ -189,6 +194,9 @@ pub struct EmailProviderResponse { #[schema(example = "us-east-1")] pub region: String, pub is_active: bool, + /// Send-path cap for this provider, if configured. `null` = unlimited. + #[schema(example = 120)] + pub rate_limit_per_minute: Option, /// Masked credentials for display pub credentials: serde_json::Value, #[schema(example = "2025-12-03T10:30:00Z")] @@ -297,6 +305,27 @@ pub struct EmailDomainWithDnsResponse { pub dns_records: Vec, } +/// A backup provider configured for a domain's send failover chain. +#[derive(Debug, Serialize, ToSchema)] +pub struct EmailDomainFallbackProviderResponse { + pub id: i32, + pub domain_id: i32, + pub provider_id: i32, + /// Lower priority is tried first, after the domain's primary provider. + pub priority: i32, + #[schema(example = "2025-12-03T10:30:00Z")] + pub created_at: String, +} + +/// Request to add (or re-prioritize) a fallback provider for a domain. +#[derive(Debug, Deserialize, ToSchema)] +pub struct AddFallbackProviderRequest { + pub provider_id: i32, + /// Lower priority is tried first, after the domain's primary provider. + #[serde(default)] + pub priority: i32, +} + /// Request to setup DNS records using a configured DNS provider #[derive(Debug, Deserialize, ToSchema)] pub struct SetupDnsRequest { diff --git a/crates/temps-email/src/providers/scaleway.rs b/crates/temps-email/src/providers/scaleway.rs index c1f726ce3..4583fbecd 100644 --- a/crates/temps-email/src/providers/scaleway.rs +++ b/crates/temps-email/src/providers/scaleway.rs @@ -398,7 +398,14 @@ impl EmailProvider for ScalewayProvider { .json(&request) .send() .await - .map_err(|e| EmailError::Scaleway(format!("Failed to send email: {}", e)))?; + .map_err(|e| EmailError::SendFailed { + provider: EmailProviderType::Scaleway, + // A request that never reached Scaleway (DNS/connect/timeout) is + // worth retrying — it says nothing about whether the message itself + // is deliverable. + retryable: true, + reason: format!("Failed to send email: {}", e), + })?; if !response.status().is_success() { let status = response.status(); @@ -407,23 +414,33 @@ impl EmailProvider for ScalewayProvider { .await .unwrap_or_else(|_| "Unknown error".to_string()); error!("Failed to send email via Scaleway ({}): {}", status, body); - return Err(EmailError::Scaleway(format!( - "Failed to send email ({}): {}", - status, body - ))); + return Err(EmailError::SendFailed { + provider: EmailProviderType::Scaleway, + // 5xx and 429 are Scaleway-side transient conditions; any other + // 4xx (bad request, invalid recipient, auth failure) won't change + // on retry. + retryable: status.is_server_error() || status.as_u16() == 429, + reason: format!("Failed to send email ({}): {}", status, body), + }); } - let email_response: ScalewayEmailResponse = response - .json() - .await - .map_err(|e| EmailError::Scaleway(format!("Failed to parse email response: {}", e)))?; + let email_response: ScalewayEmailResponse = + response.json().await.map_err(|e| EmailError::SendFailed { + provider: EmailProviderType::Scaleway, + retryable: false, + reason: format!("Failed to parse email response: {}", e), + })?; let message_id = email_response .emails .first() .and_then(|e| e.message_id.clone()) .or_else(|| email_response.emails.first().map(|e| e.id.clone())) - .ok_or_else(|| EmailError::Scaleway("No message ID returned".to_string()))?; + .ok_or_else(|| EmailError::SendFailed { + provider: EmailProviderType::Scaleway, + retryable: false, + reason: "No message ID returned".to_string(), + })?; debug!("Email sent successfully, message_id: {}", message_id); diff --git a/crates/temps-email/src/providers/ses.rs b/crates/temps-email/src/providers/ses.rs index 221a9ccc8..02e98b50c 100644 --- a/crates/temps-email/src/providers/ses.rs +++ b/crates/temps-email/src/providers/ses.rs @@ -57,6 +57,36 @@ fn extract_ses_error_details( } } +/// Classify whether an SES send failure is worth retrying. Network/timeout +/// failures never reached AWS and are always worth another attempt; service +/// errors are retryable only when SES itself reports a throttling/capacity +/// condition (surfaced as an HTTP 429/5xx-equivalent error code) rather than +/// a message-level rejection (bad recipient, unverified sender, suspended +/// account, ...) that will fail identically on retry. +fn is_ses_error_retryable( + e: &aws_sdk_sesv2::error::SdkError, +) -> bool { + use aws_sdk_sesv2::error::SdkError; + + match e { + SdkError::TimeoutError(_) => true, + SdkError::DispatchFailure(dispatch_err) => { + dispatch_err.is_io() || dispatch_err.is_timeout() + } + SdkError::ResponseError(_) => true, + SdkError::ConstructionFailure(_) => false, + SdkError::ServiceError(service_err) => { + let message = format!("{}", service_err.err()).to_lowercase(); + message.contains("throttl") + || message.contains("too many requests") + || message.contains("limit exceeded") + || message.contains("service unavailable") + || message.contains("internal") + } + _ => false, + } +} + /// AWS SES credentials configuration #[derive(Debug, Clone, Serialize, Deserialize)] pub struct SesCredentials { @@ -468,13 +498,26 @@ impl EmailProvider for SesProvider { let result = send_request.send().await.map_err(|e| { // Extract detailed error information from AWS SDK error let error_message = extract_ses_error_details(&e); - error!("Failed to send email via SES: {}", error_message); - EmailError::AwsSes(error_message) + let retryable = is_ses_error_retryable(&e); + error!( + "Failed to send email via SES ({}): {}", + if retryable { "retryable" } else { "permanent" }, + error_message + ); + EmailError::SendFailed { + provider: EmailProviderType::Ses, + retryable, + reason: error_message, + } })?; let message_id = result .message_id() - .ok_or_else(|| EmailError::AwsSes("No message ID returned".to_string()))? + .ok_or_else(|| EmailError::SendFailed { + provider: EmailProviderType::Ses, + retryable: false, + reason: "No message ID returned".to_string(), + })? .to_string(); debug!("Email sent successfully, message_id: {}", message_id); diff --git a/crates/temps-email/src/providers/smtp.rs b/crates/temps-email/src/providers/smtp.rs index 61c59c30c..62cc4d68d 100644 --- a/crates/temps-email/src/providers/smtp.rs +++ b/crates/temps-email/src/providers/smtp.rs @@ -279,7 +279,17 @@ impl EmailProvider for SmtpProvider { self.transport.send(message).await.map_err(|e| { error!("Failed to send email via SMTP: {}", e); - EmailError::Smtp(format!("Failed to send email: {}", e)) + // 4xx SMTP replies (is_transient) and connection-level failures + // (is_client/is_timeout/is_transport_shutdown) are worth retrying; + // a 5xx permanent reply from the relay will reject the same + // message again. + let retryable = + e.is_transient() || e.is_timeout() || e.is_client() || e.is_transport_shutdown(); + EmailError::SendFailed { + provider: EmailProviderType::Smtp, + retryable, + reason: format!("Failed to send email: {}", e), + } })?; debug!("Email sent via SMTP, message_id: {}", message_id); diff --git a/crates/temps-email/src/services/email_service.rs b/crates/temps-email/src/services/email_service.rs index d094b0f9c..9e0b99a49 100644 --- a/crates/temps-email/src/services/email_service.rs +++ b/crates/temps-email/src/services/email_service.rs @@ -292,21 +292,14 @@ impl EmailService { }); } - // Try to get provider - if not configured, capture email - let provider = match self.provider_service.get(domain.provider_id).await { - Ok(p) => Some(p), - Err(e) => { - info!( - "No provider configured for domain '{}', capturing email without sending (Mailhog mode)", - domain.domain - ); - debug!("Provider lookup error: {}", e); - None - } - }; - - // If no provider, mark as captured and return success - if provider.is_none() { + // Build the domain's failover chain: primary provider first, then + // configured fallbacks in priority order. `get_send_chain` also + // drops inactive providers, so disabling a provider now actually + // takes it out of the send path instead of only hiding it from + // provider-selection UI. + let chain = self.provider_service.get_send_chain(&domain).await?; + + if chain.is_empty() { let mut active_model: emails::ActiveModel = email_model.into(); active_model.status = Set("captured".to_string()); active_model.sent_at = Set(Some(Utc::now())); @@ -325,36 +318,6 @@ impl EmailService { }); } - let provider = provider.unwrap(); - - let provider_instance = match self - .provider_service - .create_provider_instance(&provider) - .await - { - Ok(instance) => instance, - Err(e) => { - // Provider exists but failed to create instance - capture email instead of failing - info!( - "Failed to create provider instance, capturing email without sending: {}", - e - ); - let mut active_model: emails::ActiveModel = email_model.into(); - active_model.status = Set("captured".to_string()); - active_model.error_message = Set(Some(format!("Provider unavailable: {}", e))); - active_model.sent_at = Set(Some(Utc::now())); - active_model.update(self.db.as_ref()).await?; - - return Ok(SendEmailResponse { - id: email_id, - status: "captured".to_string(), - provider_message_id: None, - }); - } - }; - - // Use tracked HTML (with open/click tracking injected) if available - let provider_request = ProviderSendRequest { from: request.from, from_name: request.from_name, @@ -368,19 +331,97 @@ impl EmailService { headers: request.headers, }; - match provider_instance.send(&provider_request).await { - Ok(response) => { - // Update email with success status - let mut active_model: emails::ActiveModel = email_model.clone().into(); + // Try each provider in the chain in order. Within a provider, retry + // once more only if the failure was classified as transient + // (`EmailError::is_retryable`) — a permanent rejection (bad + // recipient, auth failure, unverified sender) fails identically on + // retry, so move straight to the next provider instead of wasting a + // second attempt against the same one. + const MAX_ATTEMPTS_PER_PROVIDER: u32 = 2; + const RETRY_DELAY: std::time::Duration = std::time::Duration::from_millis(500); + + let mut total_attempts: i32 = 0; + let mut last_provider_id: Option = None; + let mut last_error: Option = None; + let mut sent: Option = None; + + 'chain: for provider in &chain { + if !self.provider_service.circuit_allows(provider.id) { + debug!( + "Skipping provider {} ({}) for email {} — circuit breaker open", + provider.id, provider.name, email_id + ); + continue; + } + if !self.provider_service.try_acquire_rate_limit(provider) { + debug!( + "Skipping provider {} ({}) for email {} — rate limit exceeded", + provider.id, provider.name, email_id + ); + continue; + } + + let provider_instance = match self + .provider_service + .create_provider_instance(provider) + .await + { + Ok(instance) => instance, + Err(e) => { + warn!( + "Failed to create provider instance {} ({}) for email {}: {}", + provider.id, provider.name, email_id, e + ); + last_provider_id = Some(provider.id); + last_error = Some(format!("{}: provider unavailable ({})", provider.name, e)); + continue; + } + }; + + for attempt in 1..=MAX_ATTEMPTS_PER_PROVIDER { + total_attempts += 1; + last_provider_id = Some(provider.id); + + match provider_instance.send(&provider_request).await { + Ok(response) => { + self.provider_service.record_send_success(provider.id); + sent = Some(response); + break 'chain; + } + Err(e) => { + let retryable = e.is_retryable(); + warn!( + "Send attempt {}/{} via {} ({}) failed for email {}: {}", + attempt, MAX_ATTEMPTS_PER_PROVIDER, provider.name, provider.id, + email_id, e + ); + last_error = Some(format!("{}: {}", provider.name, e)); + + if !retryable || attempt == MAX_ATTEMPTS_PER_PROVIDER { + self.provider_service.record_send_failure(provider.id); + break; + } + tokio::time::sleep(RETRY_DELAY).await; + } + } + } + } + + let mut active_model: emails::ActiveModel = email_model.into(); + active_model.retry_count = Set(total_attempts); + active_model.provider_id = Set(last_provider_id); + + match sent { + Some(response) => { active_model.status = Set("sent".to_string()); active_model.provider_message_id = Set(Some(response.message_id.clone())); active_model.sent_at = Set(Some(Utc::now())); - let _email_model = active_model.update(self.db.as_ref()).await?; + active_model.update(self.db.as_ref()).await?; info!( - "Email sent successfully, id: {}, provider_message_id: {}", - email_id, response.message_id + "Email sent successfully, id: {}, provider_message_id: {}, attempts: {}", + email_id, response.message_id, total_attempts ); Ok(SendEmailResponse { @@ -389,16 +430,16 @@ impl EmailService { provider_message_id: Some(response.message_id), }) } - Err(e) => { - // Provider send failed - capture email instead of failing + None => { + let reason = last_error + .unwrap_or_else(|| "All providers unavailable (circuit open or rate limited)".to_string()); info!( - "Failed to send email via provider, capturing instead: {}", - e + "Failed to send email {} via {} provider(s) in failover chain, capturing instead: {}", + email_id, chain.len(), reason ); - let mut active_model: emails::ActiveModel = email_model.into(); active_model.status = Set("captured".to_string()); - active_model.error_message = Set(Some(format!("Send failed: {}", e))); + active_model.error_message = Set(Some(format!("Send failed: {}", reason))); active_model.sent_at = Set(Some(Utc::now())); active_model.update(self.db.as_ref()).await?; diff --git a/crates/temps-email/src/services/mod.rs b/crates/temps-email/src/services/mod.rs index 5686c453f..428883644 100644 --- a/crates/temps-email/src/services/mod.rs +++ b/crates/temps-email/src/services/mod.rs @@ -3,6 +3,7 @@ mod domain_service; mod email_service; mod provider_service; +mod resilience; mod suppression_service; mod tracking_service; #[cfg(test)] diff --git a/crates/temps-email/src/services/provider_service.rs b/crates/temps-email/src/services/provider_service.rs index 3f08864dd..8f2b88fa6 100644 --- a/crates/temps-email/src/services/provider_service.rs +++ b/crates/temps-email/src/services/provider_service.rs @@ -6,7 +6,7 @@ use sea_orm::{ }; use std::sync::Arc; use temps_core::EncryptionService; -use temps_entities::email_providers; +use temps_entities::{email_domain_fallback_providers, email_domains, email_providers}; use tracing::{debug, error}; use crate::errors::EmailError; @@ -14,12 +14,17 @@ use crate::providers::{ EmailProvider, EmailProviderType, ScalewayCredentials, ScalewayProvider, SesCredentials, SesProvider, SmtpCredentials, SmtpProvider, }; +use crate::services::resilience::{ProviderCircuitBreaker, ProviderRateLimiter}; /// Service for managing email providers #[derive(Clone)] pub struct ProviderService { db: Arc, encryption_service: Arc, + /// Per-provider send-path circuit breaker. Lives here (not per-request) + /// so failure/success history persists across sends. + circuit_breaker: Arc, + rate_limiter: Arc, } /// Request to create a new email provider @@ -62,6 +67,10 @@ pub struct UpdateProviderRequest { /// how operators rotate `name`/`region` without re-typing secrets. pub credentials: Option, pub is_active: Option, + /// Send-path rate cap. Outer `None` leaves the current value untouched; + /// `Some(None)` explicitly clears it back to unlimited; `Some(Some(n))` + /// sets the cap to `n` sends/minute. + pub rate_limit_per_minute: Option>, } /// Summary of what changed during an update. Used for audit logging. @@ -89,6 +98,8 @@ impl ProviderService { Self { db, encryption_service, + circuit_breaker: Arc::new(ProviderCircuitBreaker::new()), + rate_limiter: Arc::new(ProviderRateLimiter::new()), } } @@ -160,6 +171,154 @@ impl ProviderService { Ok(providers) } + /// The ordered list of providers to try for a domain's send: its primary + /// provider (`email_domains.provider_id`) first, then its configured + /// fallback providers in ascending `priority` order. Inactive providers + /// and duplicates (a provider set as both primary and a fallback) are + /// dropped — callers loop this and move to the next entry on failure. + pub async fn get_send_chain( + &self, + domain: &email_domains::Model, + ) -> Result, EmailError> { + let mut chain = Vec::new(); + let mut seen = std::collections::HashSet::new(); + + if let Ok(primary) = self.get(domain.provider_id).await { + if primary.is_active { + seen.insert(primary.id); + chain.push(primary); + } + } + + let fallback_links = email_domain_fallback_providers::Entity::find() + .filter(email_domain_fallback_providers::Column::DomainId.eq(domain.id)) + .order_by_asc(email_domain_fallback_providers::Column::Priority) + .all(self.db.as_ref()) + .await?; + + // Batch-fetch every fallback provider in one query instead of one + // `self.get(id)` per link, then re-apply the links' priority order — + // fallback chains are short today, but this scales with however many + // an operator configures instead of being O(n) queries per send. + let fallback_provider_ids: Vec = fallback_links + .iter() + .map(|link| link.provider_id) + .filter(|id| !seen.contains(id)) + .collect(); + + let fallback_providers: std::collections::HashMap = + if fallback_provider_ids.is_empty() { + std::collections::HashMap::new() + } else { + email_providers::Entity::find() + .filter(email_providers::Column::Id.is_in(fallback_provider_ids)) + .all(self.db.as_ref()) + .await? + .into_iter() + .map(|p| (p.id, p)) + .collect() + }; + + for link in fallback_links { + if seen.contains(&link.provider_id) { + continue; + } + if let Some(provider) = fallback_providers.get(&link.provider_id) { + if provider.is_active { + seen.insert(provider.id); + chain.push(provider.clone()); + } + } + } + + Ok(chain) + } + + /// Add (or re-prioritize, if it already exists) a fallback provider for + /// a domain. + pub async fn add_fallback_provider( + &self, + domain_id: i32, + provider_id: i32, + priority: i32, + ) -> Result { + // Ensure the provider actually exists — surfaces a clear 404 instead + // of a foreign-key violation. + self.get(provider_id).await?; + + let existing = email_domain_fallback_providers::Entity::find() + .filter(email_domain_fallback_providers::Column::DomainId.eq(domain_id)) + .filter(email_domain_fallback_providers::Column::ProviderId.eq(provider_id)) + .one(self.db.as_ref()) + .await?; + + if let Some(existing) = existing { + let mut active_model: email_domain_fallback_providers::ActiveModel = existing.into(); + active_model.priority = Set(priority); + return Ok(active_model.update(self.db.as_ref()).await?); + } + + let link = email_domain_fallback_providers::ActiveModel { + domain_id: Set(domain_id), + provider_id: Set(provider_id), + priority: Set(priority), + ..Default::default() + }; + Ok(link.insert(self.db.as_ref()).await?) + } + + /// List a domain's fallback providers in priority order. + pub async fn list_fallback_providers( + &self, + domain_id: i32, + ) -> Result, EmailError> { + Ok(email_domain_fallback_providers::Entity::find() + .filter(email_domain_fallback_providers::Column::DomainId.eq(domain_id)) + .order_by_asc(email_domain_fallback_providers::Column::Priority) + .all(self.db.as_ref()) + .await?) + } + + /// Remove a fallback provider link. A no-op (not an error) if it wasn't + /// configured — removing a fallback that's already gone is idempotent. + pub async fn remove_fallback_provider( + &self, + domain_id: i32, + provider_id: i32, + ) -> Result<(), EmailError> { + email_domain_fallback_providers::Entity::delete_many() + .filter(email_domain_fallback_providers::Column::DomainId.eq(domain_id)) + .filter(email_domain_fallback_providers::Column::ProviderId.eq(provider_id)) + .exec(self.db.as_ref()) + .await?; + Ok(()) + } + + /// Whether the send path should attempt this provider right now (its + /// circuit breaker isn't open from recent consecutive failures). + pub fn circuit_allows(&self, provider_id: i32) -> bool { + self.circuit_breaker.allow(provider_id) + } + + /// Whether this provider is under its configured per-minute send cap. + /// Consumes one unit of budget if it returns `true`. + pub fn try_acquire_rate_limit(&self, provider: &email_providers::Model) -> bool { + self.rate_limiter + .try_acquire(provider.id, provider.rate_limit_per_minute) + } + + /// Record a successful send against a provider — resets its circuit + /// breaker's consecutive-failure count. + pub fn record_send_success(&self, provider_id: i32) { + self.circuit_breaker.record_success(provider_id); + } + + /// Record a failed send attempt against a provider — counts toward + /// tripping its circuit breaker open. + pub fn record_send_failure(&self, provider_id: i32) { + self.circuit_breaker.record_failure(provider_id); + } + /// Delete a provider pub async fn delete(&self, id: i32) -> Result<(), EmailError> { let provider = self.get(id).await?; @@ -239,6 +398,13 @@ impl ProviderService { } } + if let Some(rate_limit_per_minute) = request.rate_limit_per_minute { + if rate_limit_per_minute != existing.rate_limit_per_minute { + active.rate_limit_per_minute = Set(rate_limit_per_minute); + changed_fields.push("rate_limit_per_minute".to_string()); + } + } + if let Some(new_credentials) = request.credentials { let new_type = new_credentials.provider_type(); if new_type != existing_type { @@ -822,6 +988,7 @@ mod tests { region: "us-east-1".to_string(), credentials: encrypted, is_active: true, + rate_limit_per_minute: None, created_at: chrono::Utc::now(), updated_at: chrono::Utc::now(), }; @@ -858,6 +1025,7 @@ mod tests { region: "fr-par".to_string(), credentials: encrypted, is_active: true, + rate_limit_per_minute: None, created_at: chrono::Utc::now(), updated_at: chrono::Utc::now(), }; diff --git a/crates/temps-email/src/services/resilience.rs b/crates/temps-email/src/services/resilience.rs new file mode 100644 index 000000000..aea483bff --- /dev/null +++ b/crates/temps-email/src/services/resilience.rs @@ -0,0 +1,195 @@ +//! In-memory per-provider circuit breaker and rate limiter for the email +//! send path. This is control-plane code — email sends are nowhere near +//! proxy/ingest hot-path volume — so a `Mutex>` keyed by the +//! small, bounded set of configured provider ids is fine here (see +//! CLAUDE.md's hot-path-vs-control-plane distinction). + +use std::collections::HashMap; +use std::sync::Mutex; +use std::time::{Duration, Instant}; + +const FAILURE_THRESHOLD: u32 = 5; +const OPEN_COOLDOWN: Duration = Duration::from_secs(60); + +#[derive(Clone, Copy, Default)] +struct CircuitState { + consecutive_failures: u32, + /// `Some` while the circuit is open (failing fast). Cleared once the + /// cooldown elapses to admit a half-open trial. + opened_at: Option, +} + +/// Consecutive-failure circuit breaker, one state machine per provider id. +/// Trips open after `FAILURE_THRESHOLD` consecutive failures and fails fast +/// for `OPEN_COOLDOWN` before admitting trial requests again. +/// +/// Simplification: once the cooldown elapses, *every* concurrent caller is +/// admitted (not just a single half-open trial) until the next recorded +/// failure re-opens the circuit. At email-send concurrency this is an +/// acceptable trade for staying lock-free across the actual send call; a +/// strict single-trial half-open state would need an in-flight marker. +pub struct ProviderCircuitBreaker { + states: Mutex>, +} + +impl Default for ProviderCircuitBreaker { + fn default() -> Self { + Self::new() + } +} + +impl ProviderCircuitBreaker { + pub fn new() -> Self { + Self { + states: Mutex::new(HashMap::new()), + } + } + + /// Whether a send attempt against this provider should be allowed right + /// now. `false` means "skip this provider, try the next one in the + /// failover chain" — not a hard error. + pub fn allow(&self, provider_id: i32) -> bool { + let mut states = self.states.lock().unwrap_or_else(|e| e.into_inner()); + let state = states.entry(provider_id).or_default(); + match state.opened_at { + None => true, + Some(opened_at) if opened_at.elapsed() >= OPEN_COOLDOWN => { + state.opened_at = None; + true + } + Some(_) => false, + } + } + + pub fn record_success(&self, provider_id: i32) { + let mut states = self.states.lock().unwrap_or_else(|e| e.into_inner()); + states.entry(provider_id).or_default().consecutive_failures = 0; + } + + pub fn record_failure(&self, provider_id: i32) { + let mut states = self.states.lock().unwrap_or_else(|e| e.into_inner()); + let state = states.entry(provider_id).or_default(); + state.consecutive_failures += 1; + if state.consecutive_failures >= FAILURE_THRESHOLD { + state.opened_at = Some(Instant::now()); + } + } +} + +/// Sliding-window per-provider send rate limiter, mirroring the pattern in +/// `temps_auth::rate_limit::AuthRateLimiter`. The limit itself +/// (`email_providers.rate_limit_per_minute`) is operator-configured per +/// provider rather than a global constant, so a slow SMTP relay and a +/// high-throughput SES account can coexist. +pub struct ProviderRateLimiter { + windows: Mutex>>, +} + +impl Default for ProviderRateLimiter { + fn default() -> Self { + Self::new() + } +} + +impl ProviderRateLimiter { + pub fn new() -> Self { + Self { + windows: Mutex::new(HashMap::new()), + } + } + + /// Returns `true` and records the attempt if the provider is under its + /// per-minute cap; returns `false` without recording anything if not, so + /// a denied attempt doesn't consume capacity it never used and the + /// caller is free to try the next provider in the chain. + /// `limit_per_minute: None` means unlimited (always allowed). + pub fn try_acquire(&self, provider_id: i32, limit_per_minute: Option) -> bool { + let Some(limit) = limit_per_minute else { + return true; + }; + if limit <= 0 { + return false; + } + + let now = Instant::now(); + let window_start = now - Duration::from_secs(60); + + let mut windows = self.windows.lock().unwrap_or_else(|e| e.into_inner()); + let timestamps = windows.entry(provider_id).or_default(); + timestamps.retain(|t| *t > window_start); + + if timestamps.len() >= limit as usize { + false + } else { + timestamps.push(now); + true + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn circuit_allows_until_threshold() { + let cb = ProviderCircuitBreaker::new(); + for _ in 0..FAILURE_THRESHOLD - 1 { + cb.record_failure(1); + assert!(cb.allow(1), "should still be closed below threshold"); + } + cb.record_failure(1); + assert!(!cb.allow(1), "should trip open at threshold"); + } + + #[test] + fn circuit_success_resets_failure_count() { + let cb = ProviderCircuitBreaker::new(); + for _ in 0..FAILURE_THRESHOLD - 1 { + cb.record_failure(1); + } + cb.record_success(1); + cb.record_failure(1); + assert!(cb.allow(1), "one failure after a reset shouldn't trip it"); + } + + #[test] + fn circuit_is_per_provider() { + let cb = ProviderCircuitBreaker::new(); + for _ in 0..FAILURE_THRESHOLD { + cb.record_failure(1); + } + assert!(!cb.allow(1)); + assert!(cb.allow(2), "a different provider's circuit is independent"); + } + + #[test] + fn rate_limiter_unlimited_when_none() { + let rl = ProviderRateLimiter::new(); + for _ in 0..1000 { + assert!(rl.try_acquire(1, None)); + } + } + + #[test] + fn rate_limiter_denies_over_cap() { + let rl = ProviderRateLimiter::new(); + assert!(rl.try_acquire(1, Some(2))); + assert!(rl.try_acquire(1, Some(2))); + assert!(!rl.try_acquire(1, Some(2)), "third attempt exceeds the cap of 2/min"); + } + + #[test] + fn rate_limiter_zero_cap_denies_everything() { + let rl = ProviderRateLimiter::new(); + assert!(!rl.try_acquire(1, Some(0))); + } + + #[test] + fn rate_limiter_is_per_provider() { + let rl = ProviderRateLimiter::new(); + assert!(rl.try_acquire(1, Some(1))); + assert!(!rl.try_acquire(1, Some(1))); + assert!(rl.try_acquire(2, Some(1)), "a different provider has its own budget"); + } +} diff --git a/crates/temps-entities/src/email_domain_fallback_providers.rs b/crates/temps-entities/src/email_domain_fallback_providers.rs new file mode 100644 index 000000000..060df170f --- /dev/null +++ b/crates/temps-entities/src/email_domain_fallback_providers.rs @@ -0,0 +1,49 @@ +//! Backup providers tried, in `priority` order, after a domain's primary +//! provider (`email_domains.provider_id`) is exhausted. Checked by +//! `ProviderService::get_send_chain` on every send. + +use sea_orm::entity::prelude::*; +use serde::{Deserialize, Serialize}; +use temps_core::DBDateTime; + +#[derive(Clone, Debug, PartialEq, DeriveEntityModel, Eq, Serialize, Deserialize)] +#[sea_orm(table_name = "email_domain_fallback_providers")] +pub struct Model { + #[sea_orm(primary_key)] + pub id: i32, + pub domain_id: i32, + pub provider_id: i32, + /// Lower priority is tried first, after the domain's primary provider. + pub priority: i32, + pub created_at: DBDateTime, +} + +#[derive(Copy, Clone, Debug, EnumIter, DeriveRelation)] +pub enum Relation { + #[sea_orm( + belongs_to = "super::email_domains::Entity", + from = "Column::DomainId", + to = "super::email_domains::Column::Id" + )] + EmailDomain, + #[sea_orm( + belongs_to = "super::email_providers::Entity", + from = "Column::ProviderId", + to = "super::email_providers::Column::Id" + )] + EmailProvider, +} + +impl Related for Entity { + fn to() -> RelationDef { + Relation::EmailDomain.def() + } +} + +impl Related for Entity { + fn to() -> RelationDef { + Relation::EmailProvider.def() + } +} + +impl ActiveModelBehavior for ActiveModel {} diff --git a/crates/temps-entities/src/email_providers.rs b/crates/temps-entities/src/email_providers.rs index e4ee6daf0..b53c84018 100644 --- a/crates/temps-entities/src/email_providers.rs +++ b/crates/temps-entities/src/email_providers.rs @@ -17,6 +17,8 @@ pub struct Model { /// Encrypted JSON with provider credentials pub credentials: String, pub is_active: bool, + /// Send-path cap enforced by `ProviderRateLimiter`. `None` = unlimited. + pub rate_limit_per_minute: Option, pub created_at: DBDateTime, pub updated_at: DBDateTime, } diff --git a/crates/temps-entities/src/emails.rs b/crates/temps-entities/src/emails.rs index 93288128a..baa618de6 100644 --- a/crates/temps-entities/src/emails.rs +++ b/crates/temps-entities/src/emails.rs @@ -39,6 +39,13 @@ pub struct Model { pub click_count: i32, pub first_opened_at: Option, pub first_clicked_at: Option, + /// Which provider last attempted (or ultimately completed) the send — + /// set on both success and exhausted-retry capture, so the UI can show + /// which link in the domain's failover chain was used. + pub provider_id: Option, + /// Total send attempts across every provider in the domain's failover + /// chain, incremented on each retryable failure. + pub retry_count: i32, } #[derive(Copy, Clone, Debug, EnumIter, DeriveRelation)] diff --git a/crates/temps-entities/src/lib.rs b/crates/temps-entities/src/lib.rs index c803c909f..b33f8fd04 100644 --- a/crates/temps-entities/src/lib.rs +++ b/crates/temps-entities/src/lib.rs @@ -41,6 +41,7 @@ pub mod deployments; pub mod dns_managed_domains; pub mod dns_providers; pub mod domains; +pub mod email_domain_fallback_providers; pub mod email_domains; pub mod email_events; pub mod email_links; diff --git a/crates/temps-migrations/src/migration/m20260712_000001_add_email_failover_and_retry.rs b/crates/temps-migrations/src/migration/m20260712_000001_add_email_failover_and_retry.rs new file mode 100644 index 000000000..92563f3e4 --- /dev/null +++ b/crates/temps-migrations/src/migration/m20260712_000001_add_email_failover_and_retry.rs @@ -0,0 +1,72 @@ +//! Multi-provider failover + retry/circuit-breaker support for the email +//! send path. Previously a domain had exactly one provider (`email_domains. +//! provider_id`) and any send failure went straight to "captured" — a +//! transient SES throttle or a Scaleway blip permanently dropped the email +//! instead of trying again or falling back to a backup provider. +//! +//! - `email_domain_fallback_providers`: ordered backup providers tried, in +//! `priority` order, after the domain's primary provider is exhausted. +//! - `emails.provider_id` / `emails.retry_count`: which provider actually +//! attempted the send and how many attempts were made across the chain, +//! so the UI can show why a send eventually succeeded or was captured. +//! - `email_providers.rate_limit_per_minute`: optional per-provider cap +//! enforced on the send path (NULL = unlimited), operator-configurable +//! per provider rather than a global env var. + +use sea_orm_migration::prelude::*; + +pub struct Migration; + +impl MigrationName for Migration { + fn name(&self) -> &str { + "m20260712_000001_add_email_failover_and_retry" + } +} + +#[async_trait::async_trait] +impl MigrationTrait for Migration { + async fn up(&self, manager: &SchemaManager) -> Result<(), DbErr> { + manager + .get_connection() + .execute_unprepared( + r#" + ALTER TABLE emails + ADD COLUMN IF NOT EXISTS provider_id INTEGER REFERENCES email_providers(id) ON DELETE SET NULL, + ADD COLUMN IF NOT EXISTS retry_count INTEGER NOT NULL DEFAULT 0; + + ALTER TABLE email_providers + ADD COLUMN IF NOT EXISTS rate_limit_per_minute INTEGER; + + CREATE TABLE IF NOT EXISTS email_domain_fallback_providers ( + id SERIAL PRIMARY KEY, + domain_id INTEGER NOT NULL REFERENCES email_domains(id) ON DELETE CASCADE, + provider_id INTEGER NOT NULL REFERENCES email_providers(id) ON DELETE CASCADE, + priority INTEGER NOT NULL DEFAULT 0, + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + UNIQUE (domain_id, provider_id) + ); + + CREATE INDEX IF NOT EXISTS idx_email_domain_fallback_providers_domain + ON email_domain_fallback_providers (domain_id, priority); + "#, + ) + .await?; + Ok(()) + } + + async fn down(&self, manager: &SchemaManager) -> Result<(), DbErr> { + manager + .get_connection() + .execute_unprepared( + r#" + DROP TABLE IF EXISTS email_domain_fallback_providers; + ALTER TABLE email_providers DROP COLUMN IF EXISTS rate_limit_per_minute; + ALTER TABLE emails + DROP COLUMN IF EXISTS retry_count, + DROP COLUMN IF EXISTS provider_id; + "#, + ) + .await?; + Ok(()) + } +} diff --git a/crates/temps-migrations/src/migration/mod.rs b/crates/temps-migrations/src/migration/mod.rs index 01cfc7537..68e2d983c 100644 --- a/crates/temps-migrations/src/migration/mod.rs +++ b/crates/temps-migrations/src/migration/mod.rs @@ -145,6 +145,7 @@ mod m20260707_000001_add_external_service_to_logs; mod m20260707_000002_add_external_services_container_name; mod m20260708_000001_add_node_id_to_monitoring_alert_rules; mod m20260711_000002_create_suppressed_recipients; +mod m20260712_000001_add_email_failover_and_retry; pub struct Migrator; @@ -295,6 +296,7 @@ impl MigratorTrait for Migrator { Box::new(m20260707_000002_add_external_services_container_name::Migration), Box::new(m20260708_000001_add_node_id_to_monitoring_alert_rules::Migration), Box::new(m20260711_000002_create_suppressed_recipients::Migration), + Box::new(m20260712_000001_add_email_failover_and_retry::Migration), ] } } From 7cec0f5764e852593b6134713d94beb667daef88 Mon Sep 17 00:00:00 2001 From: Claudia L Date: Mon, 13 Jul 2026 13:06:35 +0200 Subject: [PATCH 3/3] refactor(email): rescope to retry + error classification, drop failover MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Per review: failover is enterprise-shaped for OSS core (single/no-provider is the typical self-hosted setup), and the circuit breaker/rate limiter only make sense in service of a failover chain — without one, "skip" means capturing the email with zero send attempts, which is strictly worse than main for the common case (a non-retryable rejection could trip the circuit and drop healthy mail; a rate-limited burst silently fails instead of deferring). Dropped: - email_domain_fallback_providers table, its 3 CRUD endpoints, audit type, and ProviderService::get_send_chain - resilience.rs (circuit breaker + rate limiter) - email_providers.rate_limit_per_minute Kept, and fixed per review: - EmailError::SendFailed { retryable } + per-provider classification - Bounded 2-attempt retry with delay, only for transient errors - emails.provider_id / emails.attempt_count (renamed from retry_count — a clean first-try send now unambiguously records 1, not "0 retries") - The inactive-provider fix: a domain's provider is now checked for is_active before the send path uses it (previously only hid it from selection UI while sends kept going through it regardless) — covered by a new regression test - SES retryability now matches on SendEmailError's typed variants (TooManyRequestsException, LimitExceededException, ...) via ProvideErrorMetadata::code() instead of string-matching the error message, which is fragile against SDK wording changes - The whole attempt sequence (both tries) is now bounded by an overall 45s deadline via tokio::time::timeout, so a slow/hanging provider can't stall the request indefinitely Co-Authored-By: Claude Sonnet 5 --- crates/temps-email/src/handlers/audit.rs | 32 --- crates/temps-email/src/handlers/domains.rs | 171 +----------- crates/temps-email/src/handlers/mod.rs | 5 - crates/temps-email/src/handlers/providers.rs | 5 - crates/temps-email/src/handlers/types.rs | 29 --- crates/temps-email/src/providers/ses.rs | 50 ++-- .../temps-email/src/services/email_service.rs | 246 ++++++++++++------ crates/temps-email/src/services/mod.rs | 1 - .../src/services/provider_service.rs | 170 +----------- crates/temps-email/src/services/resilience.rs | 195 -------------- .../src/email_domain_fallback_providers.rs | 49 ---- crates/temps-entities/src/email_providers.rs | 2 - crates/temps-entities/src/emails.rs | 12 +- crates/temps-entities/src/lib.rs | 1 - ...712_000001_add_email_failover_and_retry.rs | 72 ----- ...0260712_000001_add_email_retry_tracking.rs | 48 ++++ crates/temps-migrations/src/migration/mod.rs | 4 +- 17 files changed, 253 insertions(+), 839 deletions(-) delete mode 100644 crates/temps-email/src/services/resilience.rs delete mode 100644 crates/temps-entities/src/email_domain_fallback_providers.rs delete mode 100644 crates/temps-migrations/src/migration/m20260712_000001_add_email_failover_and_retry.rs create mode 100644 crates/temps-migrations/src/migration/m20260712_000001_add_email_retry_tracking.rs diff --git a/crates/temps-email/src/handlers/audit.rs b/crates/temps-email/src/handlers/audit.rs index 0cf518d1f..7035ffac9 100644 --- a/crates/temps-email/src/handlers/audit.rs +++ b/crates/temps-email/src/handlers/audit.rs @@ -226,38 +226,6 @@ impl AuditOperation for EmailDomainDeletedAudit { } } -#[derive(Debug, Clone, Serialize)] -pub struct EmailDomainFallbackProviderChangedAudit { - pub context: AuditContext, - pub domain_id: i32, - pub provider_id: i32, - /// "added" or "removed" - pub action: String, - pub priority: Option, -} - -impl AuditOperation for EmailDomainFallbackProviderChangedAudit { - fn operation_type(&self) -> String { - "EMAIL_DOMAIN_FALLBACK_PROVIDER_CHANGED".to_string() - } - - fn user_id(&self) -> i32 { - self.context.user_id - } - - fn ip_address(&self) -> Option { - self.context.ip_address.clone() - } - - fn user_agent(&self) -> &str { - &self.context.user_agent - } - - fn serialize(&self) -> anyhow::Result { - serde_json::to_string(self).map_err(|e| anyhow::anyhow!("Failed to serialize: {}", e)) - } -} - // ======================================== // Email Audit Types // ======================================== diff --git a/crates/temps-email/src/handlers/domains.rs b/crates/temps-email/src/handlers/domains.rs index ccc8e1f97..c21b2e0ff 100644 --- a/crates/temps-email/src/handlers/domains.rs +++ b/crates/temps-email/src/handlers/domains.rs @@ -18,14 +18,10 @@ use temps_core::{ use temps_dns::providers::{DnsProvider, DnsRecordContent, DnsRecordRequest}; use tracing::{error, info, warn}; -use super::audit::{ - EmailDomainCreatedAudit, EmailDomainDeletedAudit, EmailDomainFallbackProviderChangedAudit, - EmailDomainVerifiedAudit, -}; +use super::audit::{EmailDomainCreatedAudit, EmailDomainDeletedAudit, EmailDomainVerifiedAudit}; use super::types::{ - AddFallbackProviderRequest, AppState, CreateEmailDomainRequest, DnsRecordResponse, - DnsRecordSetupResult, EmailDomainFallbackProviderResponse, EmailDomainResponse, - EmailDomainWithDnsResponse, SetupDnsRequest, SetupDnsResponse, + AppState, CreateEmailDomainRequest, DnsRecordResponse, DnsRecordSetupResult, + EmailDomainResponse, EmailDomainWithDnsResponse, SetupDnsRequest, SetupDnsResponse, }; use crate::errors::EmailError; use crate::services::CreateDomainRequest; @@ -95,14 +91,6 @@ pub fn routes() -> Router> { ) .route("/email-domains/{id}/verify", post(verify_domain)) .route("/email-domains/{id}/setup-dns", post(setup_dns)) - .route( - "/email-domains/{id}/fallback-providers", - get(list_domain_fallback_providers).post(add_domain_fallback_provider), - ) - .route( - "/email-domains/{id}/fallback-providers/{provider_id}", - axum::routing::delete(remove_domain_fallback_provider), - ) } /// Create a new email domain @@ -536,159 +524,6 @@ pub async fn delete_email_domain( Ok(StatusCode::NO_CONTENT) } -/// List a domain's fallback providers (send failover chain, priority order) -#[utoipa::path( - tag = "Email Domains", - get, - path = "/email-domains/{id}/fallback-providers", - responses( - (status = 200, description = "Fallback providers in priority order", body = Vec), - (status = 401, description = "Unauthorized"), - (status = 403, description = "Insufficient permissions"), - (status = 404, description = "Domain not found"), - (status = 500, description = "Internal server error") - ), - params( - ("id" = i32, Path, description = "Domain ID") - ), - security(("bearer_auth" = [])) -)] -pub async fn list_domain_fallback_providers( - RequireAuth(auth): RequireAuth, - State(state): State>, - Path(id): Path, -) -> Result { - permission_guard!(auth, EmailDomainsRead); - - // 404 if the domain itself doesn't exist, rather than silently - // returning an empty list for a nonexistent id. - state.domain_service.get(id).await?; - - let links = state.provider_service.list_fallback_providers(id).await?; - - let response: Vec = links - .into_iter() - .map(|l| EmailDomainFallbackProviderResponse { - id: l.id, - domain_id: l.domain_id, - provider_id: l.provider_id, - priority: l.priority, - created_at: l.created_at.to_rfc3339(), - }) - .collect(); - - Ok(Json(response)) -} - -/// Add (or re-prioritize) a fallback provider for a domain's send failover chain -#[utoipa::path( - tag = "Email Domains", - post, - path = "/email-domains/{id}/fallback-providers", - request_body = AddFallbackProviderRequest, - responses( - (status = 200, description = "Fallback provider added", body = EmailDomainFallbackProviderResponse), - (status = 401, description = "Unauthorized"), - (status = 403, description = "Insufficient permissions"), - (status = 404, description = "Domain or provider not found"), - (status = 500, description = "Internal server error") - ), - params( - ("id" = i32, Path, description = "Domain ID") - ), - security(("bearer_auth" = [])) -)] -pub async fn add_domain_fallback_provider( - RequireAuth(auth): RequireAuth, - State(state): State>, - axum::Extension(metadata): axum::Extension, - Path(id): Path, - Json(request): Json, -) -> Result { - permission_guard!(auth, EmailDomainsWrite); - - state.domain_service.get(id).await?; - - let link = state - .provider_service - .add_fallback_provider(id, request.provider_id, request.priority) - .await?; - - let audit = EmailDomainFallbackProviderChangedAudit { - context: AuditContext { - user_id: auth.user_id(), - ip_address: Some(metadata.ip_address.clone()), - user_agent: metadata.user_agent.clone(), - }, - domain_id: id, - provider_id: request.provider_id, - action: "added".to_string(), - priority: Some(request.priority), - }; - if let Err(e) = state.audit_service.create_audit_log(&audit).await { - error!("Failed to create audit log: {}", e); - } - - Ok(Json(EmailDomainFallbackProviderResponse { - id: link.id, - domain_id: link.domain_id, - provider_id: link.provider_id, - priority: link.priority, - created_at: link.created_at.to_rfc3339(), - })) -} - -/// Remove a fallback provider from a domain's send failover chain -#[utoipa::path( - tag = "Email Domains", - delete, - path = "/email-domains/{id}/fallback-providers/{provider_id}", - responses( - (status = 204, description = "Fallback provider removed (or wasn't configured)"), - (status = 401, description = "Unauthorized"), - (status = 403, description = "Insufficient permissions"), - (status = 404, description = "Domain not found"), - (status = 500, description = "Internal server error") - ), - params( - ("id" = i32, Path, description = "Domain ID"), - ("provider_id" = i32, Path, description = "Provider ID") - ), - security(("bearer_auth" = [])) -)] -pub async fn remove_domain_fallback_provider( - RequireAuth(auth): RequireAuth, - State(state): State>, - axum::Extension(metadata): axum::Extension, - Path((id, provider_id)): Path<(i32, i32)>, -) -> Result { - permission_guard!(auth, EmailDomainsWrite); - - state.domain_service.get(id).await?; - - state - .provider_service - .remove_fallback_provider(id, provider_id) - .await?; - - let audit = EmailDomainFallbackProviderChangedAudit { - context: AuditContext { - user_id: auth.user_id(), - ip_address: Some(metadata.ip_address.clone()), - user_agent: metadata.user_agent.clone(), - }, - domain_id: id, - provider_id, - action: "removed".to_string(), - priority: None, - }; - if let Err(e) = state.audit_service.create_audit_log(&audit).await { - error!("Failed to create audit log: {}", e); - } - - Ok(StatusCode::NO_CONTENT) -} - /// Setup DNS records for an email domain using a configured DNS provider #[utoipa::path( tag = "Email Domains", diff --git a/crates/temps-email/src/handlers/mod.rs b/crates/temps-email/src/handlers/mod.rs index eb396a416..66577d2d1 100644 --- a/crates/temps-email/src/handlers/mod.rs +++ b/crates/temps-email/src/handlers/mod.rs @@ -50,9 +50,6 @@ pub fn configure_public_routes() -> Router> { domains::verify_domain, domains::delete_email_domain, domains::setup_dns, - domains::list_domain_fallback_providers, - domains::add_domain_fallback_provider, - domains::remove_domain_fallback_provider, // Emails emails::send_email, emails::list_emails, @@ -89,8 +86,6 @@ pub fn configure_public_routes() -> Router> { types::SetupDnsRequest, types::SetupDnsResponse, types::DnsRecordSetupResult, - types::EmailDomainFallbackProviderResponse, - types::AddFallbackProviderRequest, // Email types types::SendEmailRequestBody, types::SendEmailResponseBody, diff --git a/crates/temps-email/src/handlers/providers.rs b/crates/temps-email/src/handlers/providers.rs index d99075b0e..da6313233 100644 --- a/crates/temps-email/src/handlers/providers.rs +++ b/crates/temps-email/src/handlers/providers.rs @@ -176,7 +176,6 @@ pub async fn create_email_provider( .unwrap_or(EmailProviderTypeRoute::Ses), region: provider.region, is_active: provider.is_active, - rate_limit_per_minute: provider.rate_limit_per_minute, credentials: masked_credentials, created_at: provider.created_at.to_rfc3339(), updated_at: provider.updated_at.to_rfc3339(), @@ -227,7 +226,6 @@ pub async fn list_email_providers( .unwrap_or(EmailProviderTypeRoute::Ses), region: p.region, is_active: p.is_active, - rate_limit_per_minute: p.rate_limit_per_minute, credentials: masked_credentials, created_at: p.created_at.to_rfc3339(), updated_at: p.updated_at.to_rfc3339(), @@ -280,7 +278,6 @@ pub async fn get_email_provider( .unwrap_or(EmailProviderTypeRoute::Ses), region: provider.region, is_active: provider.is_active, - rate_limit_per_minute: provider.rate_limit_per_minute, credentials: masked_credentials, created_at: provider.created_at.to_rfc3339(), updated_at: provider.updated_at.to_rfc3339(), @@ -410,7 +407,6 @@ pub async fn update_email_provider( name: request.name, region: request.region, is_active: request.is_active, - rate_limit_per_minute: request.rate_limit_per_minute.map(Some), credentials, }; @@ -457,7 +453,6 @@ pub async fn update_email_provider( .unwrap_or(EmailProviderTypeRoute::Ses), region: provider.region, is_active: provider.is_active, - rate_limit_per_minute: provider.rate_limit_per_minute, credentials: masked_credentials, created_at: provider.created_at.to_rfc3339(), updated_at: provider.updated_at.to_rfc3339(), diff --git a/crates/temps-email/src/handlers/types.rs b/crates/temps-email/src/handlers/types.rs index 9d842fdaf..be77c6880 100644 --- a/crates/temps-email/src/handlers/types.rs +++ b/crates/temps-email/src/handlers/types.rs @@ -172,11 +172,6 @@ pub struct UpdateEmailProviderRequest { pub region: Option, #[serde(default, skip_serializing_if = "Option::is_none")] pub is_active: Option, - /// Send-path cap for this provider. Omit to leave unchanged; set to - /// clamp throughput (e.g. a rate-limited SMTP relay). - #[serde(default, skip_serializing_if = "Option::is_none")] - #[schema(example = 120)] - pub rate_limit_per_minute: Option, #[serde(default, skip_serializing_if = "Option::is_none")] pub ses_credentials: Option, #[serde(default, skip_serializing_if = "Option::is_none")] @@ -194,9 +189,6 @@ pub struct EmailProviderResponse { #[schema(example = "us-east-1")] pub region: String, pub is_active: bool, - /// Send-path cap for this provider, if configured. `null` = unlimited. - #[schema(example = 120)] - pub rate_limit_per_minute: Option, /// Masked credentials for display pub credentials: serde_json::Value, #[schema(example = "2025-12-03T10:30:00Z")] @@ -305,27 +297,6 @@ pub struct EmailDomainWithDnsResponse { pub dns_records: Vec, } -/// A backup provider configured for a domain's send failover chain. -#[derive(Debug, Serialize, ToSchema)] -pub struct EmailDomainFallbackProviderResponse { - pub id: i32, - pub domain_id: i32, - pub provider_id: i32, - /// Lower priority is tried first, after the domain's primary provider. - pub priority: i32, - #[schema(example = "2025-12-03T10:30:00Z")] - pub created_at: String, -} - -/// Request to add (or re-prioritize) a fallback provider for a domain. -#[derive(Debug, Deserialize, ToSchema)] -pub struct AddFallbackProviderRequest { - pub provider_id: i32, - /// Lower priority is tried first, after the domain's primary provider. - #[serde(default)] - pub priority: i32, -} - /// Request to setup DNS records using a configured DNS provider #[derive(Debug, Deserialize, ToSchema)] pub struct SetupDnsRequest { diff --git a/crates/temps-email/src/providers/ses.rs b/crates/temps-email/src/providers/ses.rs index 02e98b50c..e54b70e3c 100644 --- a/crates/temps-email/src/providers/ses.rs +++ b/crates/temps-email/src/providers/ses.rs @@ -57,16 +57,20 @@ fn extract_ses_error_details( } } -/// Classify whether an SES send failure is worth retrying. Network/timeout -/// failures never reached AWS and are always worth another attempt; service -/// errors are retryable only when SES itself reports a throttling/capacity -/// condition (surfaced as an HTTP 429/5xx-equivalent error code) rather than -/// a message-level rejection (bad recipient, unverified sender, suspended -/// account, ...) that will fail identically on retry. -fn is_ses_error_retryable( - e: &aws_sdk_sesv2::error::SdkError, +/// Classify whether an SES `send_email` failure is worth retrying. +/// Network/timeout failures never reached AWS and are always worth another +/// attempt. Service errors are retryable only for AWS's own typed +/// throttling/capacity exceptions — a message-level rejection (bad +/// recipient, unverified sender, suspended/paused account) will fail +/// identically on retry. Matches on the concrete `SendEmailError` variants +/// rather than string-matching the error message, which is fragile against +/// SDK wording changes. +fn is_send_email_error_retryable( + e: &aws_sdk_sesv2::error::SdkError, ) -> bool { use aws_sdk_sesv2::error::SdkError; + use aws_sdk_sesv2::operation::send_email::SendEmailError; + use aws_sdk_sesv2::error::ProvideErrorMetadata; match e { SdkError::TimeoutError(_) => true, @@ -75,14 +79,26 @@ fn is_ses_error_retryable( } SdkError::ResponseError(_) => true, SdkError::ConstructionFailure(_) => false, - SdkError::ServiceError(service_err) => { - let message = format!("{}", service_err.err()).to_lowercase(); - message.contains("throttl") - || message.contains("too many requests") - || message.contains("limit exceeded") - || message.contains("service unavailable") - || message.contains("internal") - } + SdkError::ServiceError(service_err) => match service_err.err() { + SendEmailError::TooManyRequestsException(_) + | SendEmailError::LimitExceededException(_) => true, + SendEmailError::AccountSuspendedException(_) + | SendEmailError::BadRequestException(_) + | SendEmailError::MailFromDomainNotVerifiedException(_) + | SendEmailError::MessageRejected(_) + | SendEmailError::NotFoundException(_) + | SendEmailError::SendingPausedException(_) => false, + // Any future/unhandled variant: fall back to the error code for + // AWS-side throttling/capacity conditions rather than assuming + // permanent. + other => matches!( + other.code(), + Some("ThrottlingException") + | Some("ServiceUnavailable") + | Some("InternalFailure") + | Some("InternalServerError") + ), + }, _ => false, } } @@ -498,7 +514,7 @@ impl EmailProvider for SesProvider { let result = send_request.send().await.map_err(|e| { // Extract detailed error information from AWS SDK error let error_message = extract_ses_error_details(&e); - let retryable = is_ses_error_retryable(&e); + let retryable = is_send_email_error_retryable(&e); error!( "Failed to send email via SES ({}): {}", if retryable { "retryable" } else { "permanent" }, diff --git a/crates/temps-email/src/services/email_service.rs b/crates/temps-email/src/services/email_service.rs index 9e0b99a49..ec0a34442 100644 --- a/crates/temps-email/src/services/email_service.rs +++ b/crates/temps-email/src/services/email_service.rs @@ -292,31 +292,87 @@ impl EmailService { }); } - // Build the domain's failover chain: primary provider first, then - // configured fallbacks in priority order. `get_send_chain` also - // drops inactive providers, so disabling a provider now actually - // takes it out of the send path instead of only hiding it from - // provider-selection UI. - let chain = self.provider_service.get_send_chain(&domain).await?; - - if chain.is_empty() { - let mut active_model: emails::ActiveModel = email_model.into(); - active_model.status = Set("captured".to_string()); - active_model.sent_at = Set(Some(Utc::now())); + // Look up the domain's configured provider. Previously `is_active` + // only hid a provider from selection UI while the send path kept + // using it regardless — checked explicitly below so disabling a + // provider actually takes it out of the send path. + let provider = match self.provider_service.get(domain.provider_id).await { + Ok(p) => Some(p), + Err(e) => { + info!( + "No provider configured for domain '{}', capturing email without sending (Mailhog mode)", + domain.domain + ); + debug!("Provider lookup error: {}", e); + None + } + }; - active_model.update(self.db.as_ref()).await?; + let provider = match provider { + Some(p) if p.is_active => p, + Some(p) => { + info!( + "Provider '{}' for domain '{}' is inactive, capturing email without sending", + p.name, domain.domain + ); + let mut active_model: emails::ActiveModel = email_model.into(); + active_model.status = Set("captured".to_string()); + active_model.error_message = + Set(Some(format!("Provider '{}' is inactive", p.name))); + active_model.sent_at = Set(Some(Utc::now())); + active_model.update(self.db.as_ref()).await?; - info!( - "Email captured (no provider), id: {}, from: {}, to: {:?}", - email_id, request.from, to - ); + return Ok(SendEmailResponse { + id: email_id, + status: "captured".to_string(), + provider_message_id: None, + }); + } + None => { + let mut active_model: emails::ActiveModel = email_model.into(); + active_model.status = Set("captured".to_string()); + active_model.sent_at = Set(Some(Utc::now())); - return Ok(SendEmailResponse { - id: email_id, - status: "captured".to_string(), - provider_message_id: None, - }); - } + active_model.update(self.db.as_ref()).await?; + + info!( + "Email captured (no provider), id: {}, from: {}, to: {:?}", + email_id, request.from, to + ); + + return Ok(SendEmailResponse { + id: email_id, + status: "captured".to_string(), + provider_message_id: None, + }); + } + }; + + let provider_instance = match self + .provider_service + .create_provider_instance(&provider) + .await + { + Ok(instance) => instance, + Err(e) => { + // Provider exists but failed to create instance - capture email instead of failing + info!( + "Failed to create provider instance, capturing email without sending: {}", + e + ); + let mut active_model: emails::ActiveModel = email_model.into(); + active_model.status = Set("captured".to_string()); + active_model.error_message = Set(Some(format!("Provider unavailable: {}", e))); + active_model.sent_at = Set(Some(Utc::now())); + active_model.update(self.db.as_ref()).await?; + + return Ok(SendEmailResponse { + id: email_id, + status: "captured".to_string(), + provider_message_id: None, + }); + } + }; let provider_request = ProviderSendRequest { from: request.from, @@ -331,85 +387,61 @@ impl EmailService { headers: request.headers, }; - // Try each provider in the chain in order. Within a provider, retry - // once more only if the failure was classified as transient + // Retry once more only if the failure was classified as transient // (`EmailError::is_retryable`) — a permanent rejection (bad // recipient, auth failure, unverified sender) fails identically on - // retry, so move straight to the next provider instead of wasting a - // second attempt against the same one. - const MAX_ATTEMPTS_PER_PROVIDER: u32 = 2; + // retry, so a second attempt would just waste time. The whole + // attempt sequence is bounded by an overall deadline so a + // slow/hanging provider can't stall this request indefinitely even + // across multiple attempts. + const MAX_ATTEMPTS: u32 = 2; const RETRY_DELAY: std::time::Duration = std::time::Duration::from_millis(500); + const TOTAL_SEND_DEADLINE: std::time::Duration = std::time::Duration::from_secs(45); - let mut total_attempts: i32 = 0; - let mut last_provider_id: Option = None; - let mut last_error: Option = None; - let mut sent: Option = None; + let attempt_result = tokio::time::timeout(TOTAL_SEND_DEADLINE, async { + let mut attempt_count: i32 = 0; + let mut last_error: Option = None; - 'chain: for provider in &chain { - if !self.provider_service.circuit_allows(provider.id) { - debug!( - "Skipping provider {} ({}) for email {} — circuit breaker open", - provider.id, provider.name, email_id - ); - continue; - } - if !self.provider_service.try_acquire_rate_limit(provider) { - debug!( - "Skipping provider {} ({}) for email {} — rate limit exceeded", - provider.id, provider.name, email_id - ); - continue; - } - - let provider_instance = match self - .provider_service - .create_provider_instance(provider) - .await - { - Ok(instance) => instance, - Err(e) => { - warn!( - "Failed to create provider instance {} ({}) for email {}: {}", - provider.id, provider.name, email_id, e - ); - last_provider_id = Some(provider.id); - last_error = Some(format!("{}: provider unavailable ({})", provider.name, e)); - continue; - } - }; - - for attempt in 1..=MAX_ATTEMPTS_PER_PROVIDER { - total_attempts += 1; - last_provider_id = Some(provider.id); + for attempt in 1..=MAX_ATTEMPTS { + attempt_count += 1; match provider_instance.send(&provider_request).await { - Ok(response) => { - self.provider_service.record_send_success(provider.id); - sent = Some(response); - break 'chain; - } + Ok(response) => return (attempt_count, Some(response), None), Err(e) => { let retryable = e.is_retryable(); warn!( - "Send attempt {}/{} via {} ({}) failed for email {}: {}", - attempt, MAX_ATTEMPTS_PER_PROVIDER, provider.name, provider.id, - email_id, e + "Send attempt {}/{} via {} failed for email {}: {}", + attempt, MAX_ATTEMPTS, provider.name, email_id, e ); - last_error = Some(format!("{}: {}", provider.name, e)); + last_error = Some(e.to_string()); - if !retryable || attempt == MAX_ATTEMPTS_PER_PROVIDER { - self.provider_service.record_send_failure(provider.id); + if !retryable || attempt == MAX_ATTEMPTS { break; } tokio::time::sleep(RETRY_DELAY).await; } } } - } + + (attempt_count, None, last_error) + }) + .await; + + let (attempt_count, sent, last_error) = match attempt_result { + Ok(result) => result, + Err(_) => ( + MAX_ATTEMPTS as i32, + None, + Some(format!( + "Send timed out after {:?} across all attempts", + TOTAL_SEND_DEADLINE + )), + ), + }; let mut active_model: emails::ActiveModel = email_model.into(); - active_model.retry_count = Set(total_attempts); - active_model.provider_id = Set(last_provider_id); + active_model.attempt_count = Set(attempt_count); + active_model.provider_id = Set(Some(provider.id)); match sent { Some(response) => { @@ -421,7 +453,7 @@ impl EmailService { info!( "Email sent successfully, id: {}, provider_message_id: {}, attempts: {}", - email_id, response.message_id, total_attempts + email_id, response.message_id, attempt_count ); Ok(SendEmailResponse { @@ -431,11 +463,10 @@ impl EmailService { }) } None => { - let reason = last_error - .unwrap_or_else(|| "All providers unavailable (circuit open or rate limited)".to_string()); + let reason = last_error.unwrap_or_else(|| "unknown error".to_string()); info!( - "Failed to send email {} via {} provider(s) in failover chain, capturing instead: {}", - email_id, chain.len(), reason + "Failed to send email {} via provider {}, capturing instead: {}", + email_id, provider.name, reason ); active_model.status = Set("captured".to_string()); @@ -1040,6 +1071,49 @@ mod tests { assert_eq!(response.status, "captured"); } + #[tokio::test] + async fn test_send_email_inactive_provider_is_captured_not_sent() { + // Previously `is_active` only hid a provider from selection UI — + // the send path kept using it regardless. A verified domain + // pointing at a disabled provider must be captured, not attempt an + // actual send through credentials the operator turned off. + let (db, email_service, provider_service, _domain_service) = setup_test_env().await; + + let provider = create_test_provider(&provider_service).await; + provider_service.set_active(provider.id, false).await.unwrap(); + + let domain = create_test_domain(&db.db, provider.id, "test-inactive.example.com").await; + let mut active_model: temps_entities::email_domains::ActiveModel = domain.into(); + active_model.status = sea_orm::ActiveValue::Set("verified".to_string()); + active_model.update(db.db.as_ref()).await.unwrap(); + + let request = SendEmailRequest { + from: "sender@test-inactive.example.com".to_string(), + from_name: None, + to: vec!["recipient@test.com".to_string()], + cc: None, + bcc: None, + reply_to: None, + subject: "Test".to_string(), + html: Some("

Test

".to_string()), + text: None, + headers: None, + tags: None, + track_opens: false, + track_clicks: false, + }; + + let result = email_service.send(request).await.unwrap(); + + assert_eq!(result.status, "captured"); + let stored = email_service.get(result.id).await.unwrap(); + assert!(stored + .error_message + .as_deref() + .unwrap() + .contains("is inactive")); + } + #[tokio::test] async fn test_list_emails_with_filters() { let (_db, email_service, _provider_service, _domain_service) = setup_test_env().await; diff --git a/crates/temps-email/src/services/mod.rs b/crates/temps-email/src/services/mod.rs index 428883644..5686c453f 100644 --- a/crates/temps-email/src/services/mod.rs +++ b/crates/temps-email/src/services/mod.rs @@ -3,7 +3,6 @@ mod domain_service; mod email_service; mod provider_service; -mod resilience; mod suppression_service; mod tracking_service; #[cfg(test)] diff --git a/crates/temps-email/src/services/provider_service.rs b/crates/temps-email/src/services/provider_service.rs index 8f2b88fa6..3f08864dd 100644 --- a/crates/temps-email/src/services/provider_service.rs +++ b/crates/temps-email/src/services/provider_service.rs @@ -6,7 +6,7 @@ use sea_orm::{ }; use std::sync::Arc; use temps_core::EncryptionService; -use temps_entities::{email_domain_fallback_providers, email_domains, email_providers}; +use temps_entities::email_providers; use tracing::{debug, error}; use crate::errors::EmailError; @@ -14,17 +14,12 @@ use crate::providers::{ EmailProvider, EmailProviderType, ScalewayCredentials, ScalewayProvider, SesCredentials, SesProvider, SmtpCredentials, SmtpProvider, }; -use crate::services::resilience::{ProviderCircuitBreaker, ProviderRateLimiter}; /// Service for managing email providers #[derive(Clone)] pub struct ProviderService { db: Arc, encryption_service: Arc, - /// Per-provider send-path circuit breaker. Lives here (not per-request) - /// so failure/success history persists across sends. - circuit_breaker: Arc, - rate_limiter: Arc, } /// Request to create a new email provider @@ -67,10 +62,6 @@ pub struct UpdateProviderRequest { /// how operators rotate `name`/`region` without re-typing secrets. pub credentials: Option, pub is_active: Option, - /// Send-path rate cap. Outer `None` leaves the current value untouched; - /// `Some(None)` explicitly clears it back to unlimited; `Some(Some(n))` - /// sets the cap to `n` sends/minute. - pub rate_limit_per_minute: Option>, } /// Summary of what changed during an update. Used for audit logging. @@ -98,8 +89,6 @@ impl ProviderService { Self { db, encryption_service, - circuit_breaker: Arc::new(ProviderCircuitBreaker::new()), - rate_limiter: Arc::new(ProviderRateLimiter::new()), } } @@ -171,154 +160,6 @@ impl ProviderService { Ok(providers) } - /// The ordered list of providers to try for a domain's send: its primary - /// provider (`email_domains.provider_id`) first, then its configured - /// fallback providers in ascending `priority` order. Inactive providers - /// and duplicates (a provider set as both primary and a fallback) are - /// dropped — callers loop this and move to the next entry on failure. - pub async fn get_send_chain( - &self, - domain: &email_domains::Model, - ) -> Result, EmailError> { - let mut chain = Vec::new(); - let mut seen = std::collections::HashSet::new(); - - if let Ok(primary) = self.get(domain.provider_id).await { - if primary.is_active { - seen.insert(primary.id); - chain.push(primary); - } - } - - let fallback_links = email_domain_fallback_providers::Entity::find() - .filter(email_domain_fallback_providers::Column::DomainId.eq(domain.id)) - .order_by_asc(email_domain_fallback_providers::Column::Priority) - .all(self.db.as_ref()) - .await?; - - // Batch-fetch every fallback provider in one query instead of one - // `self.get(id)` per link, then re-apply the links' priority order — - // fallback chains are short today, but this scales with however many - // an operator configures instead of being O(n) queries per send. - let fallback_provider_ids: Vec = fallback_links - .iter() - .map(|link| link.provider_id) - .filter(|id| !seen.contains(id)) - .collect(); - - let fallback_providers: std::collections::HashMap = - if fallback_provider_ids.is_empty() { - std::collections::HashMap::new() - } else { - email_providers::Entity::find() - .filter(email_providers::Column::Id.is_in(fallback_provider_ids)) - .all(self.db.as_ref()) - .await? - .into_iter() - .map(|p| (p.id, p)) - .collect() - }; - - for link in fallback_links { - if seen.contains(&link.provider_id) { - continue; - } - if let Some(provider) = fallback_providers.get(&link.provider_id) { - if provider.is_active { - seen.insert(provider.id); - chain.push(provider.clone()); - } - } - } - - Ok(chain) - } - - /// Add (or re-prioritize, if it already exists) a fallback provider for - /// a domain. - pub async fn add_fallback_provider( - &self, - domain_id: i32, - provider_id: i32, - priority: i32, - ) -> Result { - // Ensure the provider actually exists — surfaces a clear 404 instead - // of a foreign-key violation. - self.get(provider_id).await?; - - let existing = email_domain_fallback_providers::Entity::find() - .filter(email_domain_fallback_providers::Column::DomainId.eq(domain_id)) - .filter(email_domain_fallback_providers::Column::ProviderId.eq(provider_id)) - .one(self.db.as_ref()) - .await?; - - if let Some(existing) = existing { - let mut active_model: email_domain_fallback_providers::ActiveModel = existing.into(); - active_model.priority = Set(priority); - return Ok(active_model.update(self.db.as_ref()).await?); - } - - let link = email_domain_fallback_providers::ActiveModel { - domain_id: Set(domain_id), - provider_id: Set(provider_id), - priority: Set(priority), - ..Default::default() - }; - Ok(link.insert(self.db.as_ref()).await?) - } - - /// List a domain's fallback providers in priority order. - pub async fn list_fallback_providers( - &self, - domain_id: i32, - ) -> Result, EmailError> { - Ok(email_domain_fallback_providers::Entity::find() - .filter(email_domain_fallback_providers::Column::DomainId.eq(domain_id)) - .order_by_asc(email_domain_fallback_providers::Column::Priority) - .all(self.db.as_ref()) - .await?) - } - - /// Remove a fallback provider link. A no-op (not an error) if it wasn't - /// configured — removing a fallback that's already gone is idempotent. - pub async fn remove_fallback_provider( - &self, - domain_id: i32, - provider_id: i32, - ) -> Result<(), EmailError> { - email_domain_fallback_providers::Entity::delete_many() - .filter(email_domain_fallback_providers::Column::DomainId.eq(domain_id)) - .filter(email_domain_fallback_providers::Column::ProviderId.eq(provider_id)) - .exec(self.db.as_ref()) - .await?; - Ok(()) - } - - /// Whether the send path should attempt this provider right now (its - /// circuit breaker isn't open from recent consecutive failures). - pub fn circuit_allows(&self, provider_id: i32) -> bool { - self.circuit_breaker.allow(provider_id) - } - - /// Whether this provider is under its configured per-minute send cap. - /// Consumes one unit of budget if it returns `true`. - pub fn try_acquire_rate_limit(&self, provider: &email_providers::Model) -> bool { - self.rate_limiter - .try_acquire(provider.id, provider.rate_limit_per_minute) - } - - /// Record a successful send against a provider — resets its circuit - /// breaker's consecutive-failure count. - pub fn record_send_success(&self, provider_id: i32) { - self.circuit_breaker.record_success(provider_id); - } - - /// Record a failed send attempt against a provider — counts toward - /// tripping its circuit breaker open. - pub fn record_send_failure(&self, provider_id: i32) { - self.circuit_breaker.record_failure(provider_id); - } - /// Delete a provider pub async fn delete(&self, id: i32) -> Result<(), EmailError> { let provider = self.get(id).await?; @@ -398,13 +239,6 @@ impl ProviderService { } } - if let Some(rate_limit_per_minute) = request.rate_limit_per_minute { - if rate_limit_per_minute != existing.rate_limit_per_minute { - active.rate_limit_per_minute = Set(rate_limit_per_minute); - changed_fields.push("rate_limit_per_minute".to_string()); - } - } - if let Some(new_credentials) = request.credentials { let new_type = new_credentials.provider_type(); if new_type != existing_type { @@ -988,7 +822,6 @@ mod tests { region: "us-east-1".to_string(), credentials: encrypted, is_active: true, - rate_limit_per_minute: None, created_at: chrono::Utc::now(), updated_at: chrono::Utc::now(), }; @@ -1025,7 +858,6 @@ mod tests { region: "fr-par".to_string(), credentials: encrypted, is_active: true, - rate_limit_per_minute: None, created_at: chrono::Utc::now(), updated_at: chrono::Utc::now(), }; diff --git a/crates/temps-email/src/services/resilience.rs b/crates/temps-email/src/services/resilience.rs deleted file mode 100644 index aea483bff..000000000 --- a/crates/temps-email/src/services/resilience.rs +++ /dev/null @@ -1,195 +0,0 @@ -//! In-memory per-provider circuit breaker and rate limiter for the email -//! send path. This is control-plane code — email sends are nowhere near -//! proxy/ingest hot-path volume — so a `Mutex>` keyed by the -//! small, bounded set of configured provider ids is fine here (see -//! CLAUDE.md's hot-path-vs-control-plane distinction). - -use std::collections::HashMap; -use std::sync::Mutex; -use std::time::{Duration, Instant}; - -const FAILURE_THRESHOLD: u32 = 5; -const OPEN_COOLDOWN: Duration = Duration::from_secs(60); - -#[derive(Clone, Copy, Default)] -struct CircuitState { - consecutive_failures: u32, - /// `Some` while the circuit is open (failing fast). Cleared once the - /// cooldown elapses to admit a half-open trial. - opened_at: Option, -} - -/// Consecutive-failure circuit breaker, one state machine per provider id. -/// Trips open after `FAILURE_THRESHOLD` consecutive failures and fails fast -/// for `OPEN_COOLDOWN` before admitting trial requests again. -/// -/// Simplification: once the cooldown elapses, *every* concurrent caller is -/// admitted (not just a single half-open trial) until the next recorded -/// failure re-opens the circuit. At email-send concurrency this is an -/// acceptable trade for staying lock-free across the actual send call; a -/// strict single-trial half-open state would need an in-flight marker. -pub struct ProviderCircuitBreaker { - states: Mutex>, -} - -impl Default for ProviderCircuitBreaker { - fn default() -> Self { - Self::new() - } -} - -impl ProviderCircuitBreaker { - pub fn new() -> Self { - Self { - states: Mutex::new(HashMap::new()), - } - } - - /// Whether a send attempt against this provider should be allowed right - /// now. `false` means "skip this provider, try the next one in the - /// failover chain" — not a hard error. - pub fn allow(&self, provider_id: i32) -> bool { - let mut states = self.states.lock().unwrap_or_else(|e| e.into_inner()); - let state = states.entry(provider_id).or_default(); - match state.opened_at { - None => true, - Some(opened_at) if opened_at.elapsed() >= OPEN_COOLDOWN => { - state.opened_at = None; - true - } - Some(_) => false, - } - } - - pub fn record_success(&self, provider_id: i32) { - let mut states = self.states.lock().unwrap_or_else(|e| e.into_inner()); - states.entry(provider_id).or_default().consecutive_failures = 0; - } - - pub fn record_failure(&self, provider_id: i32) { - let mut states = self.states.lock().unwrap_or_else(|e| e.into_inner()); - let state = states.entry(provider_id).or_default(); - state.consecutive_failures += 1; - if state.consecutive_failures >= FAILURE_THRESHOLD { - state.opened_at = Some(Instant::now()); - } - } -} - -/// Sliding-window per-provider send rate limiter, mirroring the pattern in -/// `temps_auth::rate_limit::AuthRateLimiter`. The limit itself -/// (`email_providers.rate_limit_per_minute`) is operator-configured per -/// provider rather than a global constant, so a slow SMTP relay and a -/// high-throughput SES account can coexist. -pub struct ProviderRateLimiter { - windows: Mutex>>, -} - -impl Default for ProviderRateLimiter { - fn default() -> Self { - Self::new() - } -} - -impl ProviderRateLimiter { - pub fn new() -> Self { - Self { - windows: Mutex::new(HashMap::new()), - } - } - - /// Returns `true` and records the attempt if the provider is under its - /// per-minute cap; returns `false` without recording anything if not, so - /// a denied attempt doesn't consume capacity it never used and the - /// caller is free to try the next provider in the chain. - /// `limit_per_minute: None` means unlimited (always allowed). - pub fn try_acquire(&self, provider_id: i32, limit_per_minute: Option) -> bool { - let Some(limit) = limit_per_minute else { - return true; - }; - if limit <= 0 { - return false; - } - - let now = Instant::now(); - let window_start = now - Duration::from_secs(60); - - let mut windows = self.windows.lock().unwrap_or_else(|e| e.into_inner()); - let timestamps = windows.entry(provider_id).or_default(); - timestamps.retain(|t| *t > window_start); - - if timestamps.len() >= limit as usize { - false - } else { - timestamps.push(now); - true - } - } -} - -#[cfg(test)] -mod tests { - use super::*; - - #[test] - fn circuit_allows_until_threshold() { - let cb = ProviderCircuitBreaker::new(); - for _ in 0..FAILURE_THRESHOLD - 1 { - cb.record_failure(1); - assert!(cb.allow(1), "should still be closed below threshold"); - } - cb.record_failure(1); - assert!(!cb.allow(1), "should trip open at threshold"); - } - - #[test] - fn circuit_success_resets_failure_count() { - let cb = ProviderCircuitBreaker::new(); - for _ in 0..FAILURE_THRESHOLD - 1 { - cb.record_failure(1); - } - cb.record_success(1); - cb.record_failure(1); - assert!(cb.allow(1), "one failure after a reset shouldn't trip it"); - } - - #[test] - fn circuit_is_per_provider() { - let cb = ProviderCircuitBreaker::new(); - for _ in 0..FAILURE_THRESHOLD { - cb.record_failure(1); - } - assert!(!cb.allow(1)); - assert!(cb.allow(2), "a different provider's circuit is independent"); - } - - #[test] - fn rate_limiter_unlimited_when_none() { - let rl = ProviderRateLimiter::new(); - for _ in 0..1000 { - assert!(rl.try_acquire(1, None)); - } - } - - #[test] - fn rate_limiter_denies_over_cap() { - let rl = ProviderRateLimiter::new(); - assert!(rl.try_acquire(1, Some(2))); - assert!(rl.try_acquire(1, Some(2))); - assert!(!rl.try_acquire(1, Some(2)), "third attempt exceeds the cap of 2/min"); - } - - #[test] - fn rate_limiter_zero_cap_denies_everything() { - let rl = ProviderRateLimiter::new(); - assert!(!rl.try_acquire(1, Some(0))); - } - - #[test] - fn rate_limiter_is_per_provider() { - let rl = ProviderRateLimiter::new(); - assert!(rl.try_acquire(1, Some(1))); - assert!(!rl.try_acquire(1, Some(1))); - assert!(rl.try_acquire(2, Some(1)), "a different provider has its own budget"); - } -} diff --git a/crates/temps-entities/src/email_domain_fallback_providers.rs b/crates/temps-entities/src/email_domain_fallback_providers.rs deleted file mode 100644 index 060df170f..000000000 --- a/crates/temps-entities/src/email_domain_fallback_providers.rs +++ /dev/null @@ -1,49 +0,0 @@ -//! Backup providers tried, in `priority` order, after a domain's primary -//! provider (`email_domains.provider_id`) is exhausted. Checked by -//! `ProviderService::get_send_chain` on every send. - -use sea_orm::entity::prelude::*; -use serde::{Deserialize, Serialize}; -use temps_core::DBDateTime; - -#[derive(Clone, Debug, PartialEq, DeriveEntityModel, Eq, Serialize, Deserialize)] -#[sea_orm(table_name = "email_domain_fallback_providers")] -pub struct Model { - #[sea_orm(primary_key)] - pub id: i32, - pub domain_id: i32, - pub provider_id: i32, - /// Lower priority is tried first, after the domain's primary provider. - pub priority: i32, - pub created_at: DBDateTime, -} - -#[derive(Copy, Clone, Debug, EnumIter, DeriveRelation)] -pub enum Relation { - #[sea_orm( - belongs_to = "super::email_domains::Entity", - from = "Column::DomainId", - to = "super::email_domains::Column::Id" - )] - EmailDomain, - #[sea_orm( - belongs_to = "super::email_providers::Entity", - from = "Column::ProviderId", - to = "super::email_providers::Column::Id" - )] - EmailProvider, -} - -impl Related for Entity { - fn to() -> RelationDef { - Relation::EmailDomain.def() - } -} - -impl Related for Entity { - fn to() -> RelationDef { - Relation::EmailProvider.def() - } -} - -impl ActiveModelBehavior for ActiveModel {} diff --git a/crates/temps-entities/src/email_providers.rs b/crates/temps-entities/src/email_providers.rs index b53c84018..e4ee6daf0 100644 --- a/crates/temps-entities/src/email_providers.rs +++ b/crates/temps-entities/src/email_providers.rs @@ -17,8 +17,6 @@ pub struct Model { /// Encrypted JSON with provider credentials pub credentials: String, pub is_active: bool, - /// Send-path cap enforced by `ProviderRateLimiter`. `None` = unlimited. - pub rate_limit_per_minute: Option, pub created_at: DBDateTime, pub updated_at: DBDateTime, } diff --git a/crates/temps-entities/src/emails.rs b/crates/temps-entities/src/emails.rs index baa618de6..190b14a54 100644 --- a/crates/temps-entities/src/emails.rs +++ b/crates/temps-entities/src/emails.rs @@ -39,13 +39,13 @@ pub struct Model { pub click_count: i32, pub first_opened_at: Option, pub first_clicked_at: Option, - /// Which provider last attempted (or ultimately completed) the send — - /// set on both success and exhausted-retry capture, so the UI can show - /// which link in the domain's failover chain was used. + /// Which provider attempted (or ultimately completed) the send — set on + /// both success and exhausted-retry capture. `None` if the domain has no + /// active provider configured (never attempted). pub provider_id: Option, - /// Total send attempts across every provider in the domain's failover - /// chain, incremented on each retryable failure. - pub retry_count: i32, + /// Total send attempts against `provider_id` (a clean first-try success + /// records 1), incremented on each retryable failure. + pub attempt_count: i32, } #[derive(Copy, Clone, Debug, EnumIter, DeriveRelation)] diff --git a/crates/temps-entities/src/lib.rs b/crates/temps-entities/src/lib.rs index b33f8fd04..c803c909f 100644 --- a/crates/temps-entities/src/lib.rs +++ b/crates/temps-entities/src/lib.rs @@ -41,7 +41,6 @@ pub mod deployments; pub mod dns_managed_domains; pub mod dns_providers; pub mod domains; -pub mod email_domain_fallback_providers; pub mod email_domains; pub mod email_events; pub mod email_links; diff --git a/crates/temps-migrations/src/migration/m20260712_000001_add_email_failover_and_retry.rs b/crates/temps-migrations/src/migration/m20260712_000001_add_email_failover_and_retry.rs deleted file mode 100644 index 92563f3e4..000000000 --- a/crates/temps-migrations/src/migration/m20260712_000001_add_email_failover_and_retry.rs +++ /dev/null @@ -1,72 +0,0 @@ -//! Multi-provider failover + retry/circuit-breaker support for the email -//! send path. Previously a domain had exactly one provider (`email_domains. -//! provider_id`) and any send failure went straight to "captured" — a -//! transient SES throttle or a Scaleway blip permanently dropped the email -//! instead of trying again or falling back to a backup provider. -//! -//! - `email_domain_fallback_providers`: ordered backup providers tried, in -//! `priority` order, after the domain's primary provider is exhausted. -//! - `emails.provider_id` / `emails.retry_count`: which provider actually -//! attempted the send and how many attempts were made across the chain, -//! so the UI can show why a send eventually succeeded or was captured. -//! - `email_providers.rate_limit_per_minute`: optional per-provider cap -//! enforced on the send path (NULL = unlimited), operator-configurable -//! per provider rather than a global env var. - -use sea_orm_migration::prelude::*; - -pub struct Migration; - -impl MigrationName for Migration { - fn name(&self) -> &str { - "m20260712_000001_add_email_failover_and_retry" - } -} - -#[async_trait::async_trait] -impl MigrationTrait for Migration { - async fn up(&self, manager: &SchemaManager) -> Result<(), DbErr> { - manager - .get_connection() - .execute_unprepared( - r#" - ALTER TABLE emails - ADD COLUMN IF NOT EXISTS provider_id INTEGER REFERENCES email_providers(id) ON DELETE SET NULL, - ADD COLUMN IF NOT EXISTS retry_count INTEGER NOT NULL DEFAULT 0; - - ALTER TABLE email_providers - ADD COLUMN IF NOT EXISTS rate_limit_per_minute INTEGER; - - CREATE TABLE IF NOT EXISTS email_domain_fallback_providers ( - id SERIAL PRIMARY KEY, - domain_id INTEGER NOT NULL REFERENCES email_domains(id) ON DELETE CASCADE, - provider_id INTEGER NOT NULL REFERENCES email_providers(id) ON DELETE CASCADE, - priority INTEGER NOT NULL DEFAULT 0, - created_at TIMESTAMPTZ NOT NULL DEFAULT now(), - UNIQUE (domain_id, provider_id) - ); - - CREATE INDEX IF NOT EXISTS idx_email_domain_fallback_providers_domain - ON email_domain_fallback_providers (domain_id, priority); - "#, - ) - .await?; - Ok(()) - } - - async fn down(&self, manager: &SchemaManager) -> Result<(), DbErr> { - manager - .get_connection() - .execute_unprepared( - r#" - DROP TABLE IF EXISTS email_domain_fallback_providers; - ALTER TABLE email_providers DROP COLUMN IF EXISTS rate_limit_per_minute; - ALTER TABLE emails - DROP COLUMN IF EXISTS retry_count, - DROP COLUMN IF EXISTS provider_id; - "#, - ) - .await?; - Ok(()) - } -} diff --git a/crates/temps-migrations/src/migration/m20260712_000001_add_email_retry_tracking.rs b/crates/temps-migrations/src/migration/m20260712_000001_add_email_retry_tracking.rs new file mode 100644 index 000000000..d4de0d819 --- /dev/null +++ b/crates/temps-migrations/src/migration/m20260712_000001_add_email_retry_tracking.rs @@ -0,0 +1,48 @@ +//! Retry/error-classification support for the email send path. Previously +//! any send failure went straight to "captured" — a transient SES throttle +//! or a Scaleway blip permanently dropped the email instead of retrying. +//! +//! `emails.provider_id` / `emails.attempt_count` record which provider +//! attempted the send and how many attempts it took, so the UI can show why +//! a send eventually succeeded or was captured. + +use sea_orm_migration::prelude::*; + +pub struct Migration; + +impl MigrationName for Migration { + fn name(&self) -> &str { + "m20260712_000001_add_email_retry_tracking" + } +} + +#[async_trait::async_trait] +impl MigrationTrait for Migration { + async fn up(&self, manager: &SchemaManager) -> Result<(), DbErr> { + manager + .get_connection() + .execute_unprepared( + r#" + ALTER TABLE emails + ADD COLUMN IF NOT EXISTS provider_id INTEGER REFERENCES email_providers(id) ON DELETE SET NULL, + ADD COLUMN IF NOT EXISTS attempt_count INTEGER NOT NULL DEFAULT 0; + "#, + ) + .await?; + Ok(()) + } + + async fn down(&self, manager: &SchemaManager) -> Result<(), DbErr> { + manager + .get_connection() + .execute_unprepared( + r#" + ALTER TABLE emails + DROP COLUMN IF EXISTS attempt_count, + DROP COLUMN IF EXISTS provider_id; + "#, + ) + .await?; + Ok(()) + } +} diff --git a/crates/temps-migrations/src/migration/mod.rs b/crates/temps-migrations/src/migration/mod.rs index 68e2d983c..62cf7d136 100644 --- a/crates/temps-migrations/src/migration/mod.rs +++ b/crates/temps-migrations/src/migration/mod.rs @@ -145,7 +145,7 @@ mod m20260707_000001_add_external_service_to_logs; mod m20260707_000002_add_external_services_container_name; mod m20260708_000001_add_node_id_to_monitoring_alert_rules; mod m20260711_000002_create_suppressed_recipients; -mod m20260712_000001_add_email_failover_and_retry; +mod m20260712_000001_add_email_retry_tracking; pub struct Migrator; @@ -296,7 +296,7 @@ impl MigratorTrait for Migrator { Box::new(m20260707_000002_add_external_services_container_name::Migration), Box::new(m20260708_000001_add_node_id_to_monitoring_alert_rules::Migration), Box::new(m20260711_000002_create_suppressed_recipients::Migration), - Box::new(m20260712_000001_add_email_failover_and_retry::Migration), + Box::new(m20260712_000001_add_email_retry_tracking::Migration), ] } }