From 71bdb653e0e7ee987fb471aeb7a9259873c61f5f Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 10 Jun 2026 09:52:47 +0000 Subject: [PATCH] perf(notify): bounded-concurrency fan-out for grant notification emails MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Creating a grant for a group dispatched the notification emails one at a time inside the HTTP request — 30 members × ~500 ms of SMTP ≈ 15 s holding the POST /api/grants response (the code carried a TODO acknowledging it). Dispatches are independent (coalescing and rate-limiting key on the (granter, recipient) pair, distinct per member), so run them through `buffered(6)`: ~6× less wall time for group fan-outs while capping parallel SMTP sessions, with outcome order still matching member order. https://claude.ai/code/session_01Dp3oWon5GBMVn4j3QXZdgx --- .../recipient_notification_service.rs | 28 ++++++++++++++----- 1 file changed, 21 insertions(+), 7 deletions(-) diff --git a/src/application/services/recipient_notification_service.rs b/src/application/services/recipient_notification_service.rs index fce62b27..ca832d2e 100644 --- a/src/application/services/recipient_notification_service.rs +++ b/src/application/services/recipient_notification_service.rs @@ -70,6 +70,11 @@ use crate::domain::services::authorization::{Resource, Subject}; use crate::infrastructure::repositories::pg::UserPgRepository; use crate::interfaces::middleware::rate_limit::RateLimiter; +/// Concurrent per-recipient dispatches in flight during a group fan-out. +/// High enough to collapse a 30-member group's serial SMTP latency, +/// low enough not to flood the relay (most reject >10 parallel sessions). +const NOTIFY_DISPATCH_CONCURRENCY: usize = 6; + /// What triggered the notification — purely an audit discriminator. /// `GrantCreated` → fired implicitly when a grant lands; `ManualResend` /// → granter explicitly clicked "Notify by email" in My Shares. @@ -268,13 +273,22 @@ impl RecipientNotificationService { ); } - let mut outcomes = Vec::with_capacity(members.len()); - for member in &members { - let outcome = self - .dispatch_to_one_user(granter, member, resource, trigger) - .await; - outcomes.push(outcome); - } + // SMTP dispatch dominates each iteration (hundreds of ms per + // recipient) and the iterations are independent — coalescing and + // rate-limiting key on (granter, recipient), which is distinct per + // member. Bounded concurrency keeps a 30-member group grant from + // holding the HTTP response for 15+ s of serial sends while still + // capping the pressure on the SMTP relay. `buffered` (not + // `buffer_unordered`) preserves the member order of the outcomes. + use futures::stream::{self, StreamExt}; + let outcomes: Vec = stream::iter(members) + .map(|member| async move { + self.dispatch_to_one_user(granter, &member, resource, trigger) + .await + }) + .buffered(NOTIFY_DISPATCH_CONCURRENCY) + .collect() + .await; Ok(NotifyOutcomeSet { outcomes }) }