From ce18ccd62abf2e7e9537827df4cff00d4a92d8d5 Mon Sep 17 00:00:00 2001 From: James Maes Date: Fri, 31 Jul 2026 17:55:31 -0500 Subject: [PATCH] feat(nunciod): bounded-concurrency sync dispatcher with per-account timeout Route every production account sync through one SyncDispatcher that caps simultaneous syncs (NUNCIO_SYNC_CONCURRENCY, default 4), coalesces a duplicate in-flight sync for the same account, applies a per-account timeout (NUNCIO_SYNC_TIMEOUT, default 300s), and races in-flight syncs against the shared shutdown signal. The client Mail.Sync RPC and the daemon's CoreCommand sync consumer now share a single dispatcher instance so one slow account can no longer head-of-line-block the others. Refs #292 --- crates/nunciod/src/grpc.rs | 118 +++-- crates/nunciod/src/lib.rs | 1 + crates/nunciod/src/lifecycle.rs | 25 +- crates/nunciod/src/main.rs | 40 +- crates/nunciod/src/sync.rs | 178 +------ crates/nunciod/src/sync_dispatcher.rs | 688 ++++++++++++++++++++++++++ 6 files changed, 835 insertions(+), 215 deletions(-) create mode 100644 crates/nunciod/src/sync_dispatcher.rs diff --git a/crates/nunciod/src/grpc.rs b/crates/nunciod/src/grpc.rs index 7084a0c..913e3bc 100644 --- a/crates/nunciod/src/grpc.rs +++ b/crates/nunciod/src/grpc.rs @@ -5,7 +5,9 @@ //! //! This is the daemon's sole client-facing transport. +use crate::lifecycle::ShutdownSignal; use crate::pagination::{self, CursorField}; +use crate::sync_dispatcher::{sync_all_configured, ProductionAccountSyncer, SyncDispatcher}; use nuncio_cal::CalendarBackend; use nuncio_contacts::ContactsBackend; use nuncio_core::{CoreCommand, CoreEvent, EventBus}; @@ -1414,11 +1416,11 @@ impl Contacts for ContactsGrpcService { /// /// Production (`nunciod::main`, and every existing caller of [`serve`] / /// [`serve_on_listener`]) never constructs a non-default instance: -/// [`Default`] yields every field `None`, which routes `Mail/Sync` and -/// `Mail/SendMessage` through the exact same real production entry points -/// (`nunciod::sync::run_all_accounts_sync` / `run_account_sync`, -/// `nunciod::send::send_message_for_account`) as before this override -/// existed. Production code depends only on the `nuncio_mail::{MailBackend, +/// [`Default`] yields every field `None`, which routes `Mail/Sync` through +/// the shared `nunciod::sync_dispatcher::SyncDispatcher` (single account or +/// `sync_all_configured` fan-out) and `Mail/SendMessage` through +/// `nunciod::send::send_message_for_account`, exactly as production does. +/// Production code depends only on the `nuncio_mail::{MailBackend, /// MessageSender}` trait objects here -- never on `nuncio_mail`'s mock /// types -- so this seam never makes production depend on a test double. /// @@ -1464,6 +1466,11 @@ struct MailGrpcService { secrets: Arc, filter_engine: Arc, overrides: MailEngineOverrides, + /// Single bounded-concurrency dispatcher every production `Sync` funnels + /// through, so a client `Sync` and the daemon's scheduled syncs share one + /// concurrency cap, one in-flight coalescing map, and one per-account + /// timeout. The injected test backend path deliberately bypasses it. + sync_dispatcher: Arc, } #[tonic::async_trait] @@ -1683,12 +1690,20 @@ impl Mail for MailGrpcService { /// scopes the sync to one account, `None` syncs every configured /// account. /// + /// Production syncs funnel through the shared [`SyncDispatcher`]: a + /// single-account request goes through its bounded/coalesced/timed + /// per-account path, and an all-accounts request fans every configured + /// mail account out through that same path via + /// [`sync_all_configured`], so one slow account can no longer + /// head-of-line-block the others. + /// /// When [`MailEngineOverrides::mail_backend`] is injected, fetches from - /// it via [`crate::sync::sync_with_backend`] instead of resolving a - /// real per-account engine from keyring credentials -- this is what - /// lets a full-daemon E2E test drive a real inbound sync entirely over - /// this authenticated gRPC API with no live network, exactly as a real - /// client would trigger it (no un-intercepted back door). + /// it via [`crate::sync::sync_with_backend`] instead -- this is what lets + /// a full-daemon E2E test drive a real inbound sync entirely over this + /// authenticated gRPC API with no live network, exactly as a real client + /// would trigger it (no un-intercepted back door). The injected backend + /// is a single in-memory double, so it deliberately bypasses the + /// dispatcher's per-account bounding. async fn sync(&self, request: Request) -> Result, Status> { let account_id = request.into_inner().account_id; @@ -1703,23 +1718,12 @@ impl Mail for MailGrpcService { .await .map_err(|e| Status::internal(format!("sync failed: {e}")))? } else if let Some(account_id) = account_id { - crate::sync::run_account_sync( - &self.db, - &self.secrets, - &self.event_bus, - &self.filter_engine, - &account_id, - ) - .await - .map_err(|e| Status::internal(format!("sync failed: {e}")))? + self.sync_dispatcher + .sync_account(&account_id) + .await + .map_err(|e| Status::internal(format!("sync failed: {e}")))? } else { - crate::sync::run_all_accounts_sync( - &self.db, - &self.secrets, - &self.event_bus, - &self.filter_engine, - ) - .await + sync_all_configured(&self.sync_dispatcher, &self.db, &self.event_bus).await }; Ok(Response::new(SyncResponse { @@ -2460,6 +2464,29 @@ impl tonic::service::Interceptor for BearerAuthInterceptor { } } +/// Builds the production [`SyncDispatcher`] for serve entry points that are +/// not handed one by their caller (every path except production's +/// [`serve_with_shutdown`], which threads through the daemon-wide dispatcher +/// so the RPC and the scheduled syncs share one instance). Wraps a +/// [`ProductionAccountSyncer`] with concurrency/timeout bounds from the +/// environment; `shutdown` is the signal the dispatcher races in-flight syncs +/// against -- non-graceful entry points pass [`ShutdownSignal::never`]. +fn default_sync_dispatcher( + db: Arc, + secrets: Arc, + event_bus: Arc, + filter_engine: Arc, + shutdown: ShutdownSignal, +) -> Arc { + let syncer = Arc::new(ProductionAccountSyncer::new( + db, + secrets, + event_bus, + filter_engine, + )); + Arc::new(SyncDispatcher::from_env(syncer, shutdown)) +} + /// Binds a loopback TCP listener at `addr` and serves the `nuncio.v1.System`, /// `nuncio.v1.Accounts`, `nuncio.v1.Mail`, `nuncio.v1.Filters`, /// `nuncio.v1.Export`, `nuncio.v1.Audit`, and `nuncio.v1.Calendar` gRPC @@ -2548,6 +2575,7 @@ pub async fn serve_on_listener( /// as `shutdown` resolves, instead of running until the transport errors. /// /// `addr` MUST be a loopback address; see [`serve`]. +#[allow(clippy::too_many_arguments)] pub async fn serve_with_shutdown( addr: &str, event_bus: Arc, @@ -2555,6 +2583,7 @@ pub async fn serve_with_shutdown( filter_engine: Arc, secrets: Arc, token: impl Into>, + sync_dispatcher: Arc, shutdown: impl std::future::Future + Send + 'static, ) -> Result<(), GrpcServeError> { let socket_addr = addr @@ -2570,13 +2599,22 @@ pub async fn serve_with_shutdown( addr: addr.to_string(), source, })?; - serve_on_listener_with_shutdown( + // Production threads its own daemon-wide dispatcher (built over the shared + // shutdown signal) all the way to the Mail service, rather than letting a + // serve wrapper mint a fresh one, so the client `Sync` RPC and the + // scheduled syncs share a single concurrency cap and in-flight map. + serve_on_listener_with_overrides_and_shutdown( listener, event_bus, db, filter_engine, secrets, token, + MailEngineOverrides::default(), + CalendarEngineOverrides::default(), + ContactsEngineOverrides::default(), + AccountsEngineOverrides::default(), + sync_dispatcher, shutdown, ) .await @@ -2594,6 +2632,17 @@ pub async fn serve_on_listener_with_shutdown( token: impl Into>, shutdown: impl std::future::Future + Send + 'static, ) -> Result<(), GrpcServeError> { + // This entry point receives an opaque shutdown future rather than a + // `ShutdownSignal`, so it mints a fresh production dispatcher over a + // never-firing signal; production instead uses `serve_with_shutdown`, + // which threads the real shared dispatcher through. + let sync_dispatcher = default_sync_dispatcher( + db.clone(), + secrets.clone(), + event_bus.clone(), + filter_engine.clone(), + ShutdownSignal::never(), + ); serve_on_listener_with_overrides_and_shutdown( listener, event_bus, @@ -2605,6 +2654,7 @@ pub async fn serve_on_listener_with_shutdown( CalendarEngineOverrides::default(), ContactsEngineOverrides::default(), AccountsEngineOverrides::default(), + sync_dispatcher, shutdown, ) .await @@ -2636,7 +2686,16 @@ pub async fn serve_on_listener_with_overrides( ) -> Result<(), GrpcServeError> { // Delegates to the shutdown-aware variant with a shutdown future that // never resolves, preserving this function's exact prior behavior (run - // until the transport itself errors) for every existing caller. + // until the transport itself errors) for every existing caller. The + // dispatcher is minted here over a never-firing signal; only production's + // `serve_with_shutdown` threads a real shared dispatcher through. + let sync_dispatcher = default_sync_dispatcher( + db.clone(), + secrets.clone(), + event_bus.clone(), + filter_engine.clone(), + ShutdownSignal::never(), + ); serve_on_listener_with_overrides_and_shutdown( listener, event_bus, @@ -2648,6 +2707,7 @@ pub async fn serve_on_listener_with_overrides( calendar_overrides, contacts_overrides, accounts_overrides, + sync_dispatcher, std::future::pending(), ) .await @@ -2671,6 +2731,7 @@ pub async fn serve_on_listener_with_overrides_and_shutdown( calendar_overrides: CalendarEngineOverrides, contacts_overrides: ContactsEngineOverrides, accounts_overrides: AccountsEngineOverrides, + sync_dispatcher: Arc, shutdown: impl std::future::Future + Send + 'static, ) -> Result<(), GrpcServeError> { let token: Arc = token.into(); @@ -2708,6 +2769,7 @@ pub async fn serve_on_listener_with_overrides_and_shutdown( secrets: secrets.clone(), filter_engine: filter_engine.clone(), overrides, + sync_dispatcher, }; let mail_interceptor = BearerAuthInterceptor::new(token.clone()); let mail_svc = InterceptedService::new( diff --git a/crates/nunciod/src/lib.rs b/crates/nunciod/src/lib.rs index 93fcc13..8c24ff7 100644 --- a/crates/nunciod/src/lib.rs +++ b/crates/nunciod/src/lib.rs @@ -12,6 +12,7 @@ pub mod outbox; pub mod pagination; pub mod send; pub mod sync; +pub mod sync_dispatcher; pub use orchestrator::SelfHealingSyncOrchestrator; /// Default PERSISTENT database path for `nunciod`: `~/.nuncio/nuncio.db` diff --git a/crates/nunciod/src/lifecycle.rs b/crates/nunciod/src/lifecycle.rs index dcef665..107fd09 100644 --- a/crates/nunciod/src/lifecycle.rs +++ b/crates/nunciod/src/lifecycle.rs @@ -20,9 +20,26 @@ use tokio::sync::watch; #[derive(Clone)] pub struct ShutdownSignal { rx: watch::Receiver, + // Keeps the watch channel open for signals minted by [`ShutdownSignal::never`], + // which have no live [`ShutdownController`] owning the sender. A closed + // channel makes `wait` resolve (readers interpret that as "shutdown + // requested"), so a `never` signal must hold its sender alive; a real + // signal leaves this `None` because its controller owns the sender. + _keepalive: Option>>, } impl ShutdownSignal { + /// A signal that never fires, for call paths with no shutdown wiring of + /// their own (e.g. the non-graceful `serve_on_listener` entry point used + /// by tests). Holds its own sender alive so the channel never closes. + pub fn never() -> Self { + let (tx, rx) = watch::channel(false); + ShutdownSignal { + rx, + _keepalive: Some(Arc::new(tx)), + } + } + /// Resolves once shutdown has been requested. Safe to call from /// `tokio::select!` on every loop iteration: a task that observes /// shutdown mid-iteration finishes its current unit of work (the other @@ -59,7 +76,13 @@ impl ShutdownController { /// instead of that path staying reachable only from tests. pub fn new(event_bus: Arc) -> (Self, ShutdownSignal) { let (tx, rx) = watch::channel(false); - (Self { tx, event_bus }, ShutdownSignal { rx }) + ( + Self { tx, event_bus }, + ShutdownSignal { + rx, + _keepalive: None, + }, + ) } /// Initiates shutdown. Idempotent: calling it more than once (e.g. a diff --git a/crates/nunciod/src/main.rs b/crates/nunciod/src/main.rs index fcdc090..f483522 100644 --- a/crates/nunciod/src/main.rs +++ b/crates/nunciod/src/main.rs @@ -70,6 +70,27 @@ async fn main() -> Result<(), Box> { let initial_rules = nunciod::load_initial_filter_rules(&db).await?; let filter_engine = Arc::new(FilterEngine::new(initial_rules)?); + // Single bounded-concurrency sync dispatcher shared by BOTH the + // client-facing `Mail/Sync` RPC (via the gRPC server below) and this + // process's `CoreCommand` sync consumer, so account syncs from either + // origin queue against one concurrency cap, coalesce per account, and + // observe one per-account timeout -- instead of each entry point launching + // syncs independently and head-of-line-blocking one another. It races + // in-flight syncs against the SAME shutdown signal every other subsystem + // shares. + let sync_dispatcher = { + let syncer = Arc::new(nunciod::sync_dispatcher::ProductionAccountSyncer::new( + db.clone(), + account_secrets.clone(), + event_bus.clone(), + filter_engine.clone(), + )); + Arc::new(nunciod::sync_dispatcher::SyncDispatcher::from_env( + syncer, + shutdown_signal.clone(), + )) + }; + // Real inbound-sync `CoreCommand` consumer. This is the ONE place that // claims `EventBus`'s command receiver, so it is what finally makes // `CoreCommand::SyncAll` / @@ -82,9 +103,8 @@ async fn main() -> Result<(), Box> { // trigger was inert. let sync_command_task = if let Some(mut command_rx) = command_rx { let db_sync = db.clone(); - let secrets_sync = account_secrets.clone(); let event_bus_sync = event_bus.clone(); - let filter_engine_sync = filter_engine.clone(); + let dispatcher_sync = sync_dispatcher.clone(); let mut shutdown = shutdown_signal.clone(); Some(tokio::spawn(async move { loop { @@ -100,25 +120,16 @@ async fn main() -> Result<(), Box> { }; match cmd { CoreCommand::SyncAll => { - let synced = nunciod::sync::run_all_accounts_sync( + let synced = nunciod::sync_dispatcher::sync_all_configured( + &dispatcher_sync, &db_sync, - &secrets_sync, &event_bus_sync, - &filter_engine_sync, ) .await; tracing::info!("SyncAll completed: {} message(s) synced", synced); } CoreCommand::SyncAccount { account_id } => { - match nunciod::sync::run_account_sync( - &db_sync, - &secrets_sync, - &event_bus_sync, - &filter_engine_sync, - &account_id, - ) - .await - { + match dispatcher_sync.sync_account(&account_id).await { Ok(count) => tracing::info!( "SyncAccount({}) completed: {} message(s) synced", account_id, @@ -274,6 +285,7 @@ async fn main() -> Result<(), Box> { filter_engine, grpc_secrets, grpc_token, + sync_dispatcher, async move { grpc_shutdown_future.wait().await }, ) => result, _ = async move { diff --git a/crates/nunciod/src/sync.rs b/crates/nunciod/src/sync.rs index e2983ba..df2cf81 100644 --- a/crates/nunciod/src/sync.rs +++ b/crates/nunciod/src/sync.rs @@ -14,10 +14,12 @@ //! [`sync_with_backend`] takes `&dyn MailBackend` rather than a concrete //! engine, so it is exercised directly in tests with [`MockMailBackend`] -- //! no real network or OS keyring access ever happens inside it. Production -//! entry points ([`run_account_sync`], [`run_all_accounts_sync`]) are the -//! only places that resolve a real account's credentials and construct a -//! real backend; they then delegate the actual fetch/persist/event-emission -//! work to [`sync_with_backend`]. +//! entry point [`run_account_sync`] is the only place that resolves a real +//! account's credentials and constructs a real backend; it then delegates the +//! actual fetch/persist/event-emission work to [`sync_with_backend`]. An +//! all-accounts sync fans these per-account runs out through +//! [`crate::sync_dispatcher::sync_all_configured`], so concurrency and the +//! per-account timeout are bounded in one place rather than serialized here. use nuncio_core::{AccountConfig, AccountProtocol, CoreCommand, CoreEvent, EventBus}; use nuncio_filter::{FilterEngine, OutboxManager, RuleAction}; use nuncio_mail::{ImapEngine, JmapEngine, MailBackend, MailError}; @@ -408,87 +410,6 @@ pub async fn run_account_sync( } } -/// Production entry point for `CoreCommand::SyncAll`: syncs every persisted -/// account in turn, wrapping the whole batch in a single -/// `SyncStarted{None}`/`SyncCompleted{None}` pair (rather than one pair per -/// account). A single account failing to sync (missing credential, backend -/// error, etc.) publishes a `CoreEvent::Error` and does not stop the -/// remaining accounts from being attempted. Returns the total number of -/// messages synced across all accounts. -pub async fn run_all_accounts_sync( - db: &nuncio_store::db::DatabaseEngine, - secrets: &SecretManager, - event_bus: &EventBus, - filter_engine: &FilterEngine, -) -> usize { - event_bus.process_command(CoreCommand::SyncAll); - - let mut total_synced = 0usize; - match db.list_accounts().await { - Ok(accounts) => { - for config in accounts { - // DAV-protocol accounts (e.g. CalDAV) are not mail accounts; - // they sync through their own domain service, so skip them - // here rather than attempting to build a mail backend. - if config.protocol.is_dav() { - continue; - } - match secrets.get_secret(&config.keyring_secret_key) { - Ok(password) => { - let backend = match build_mail_backend(&config, &password) { - Ok(backend) => backend, - Err(e) => { - event_bus.process_command(CoreCommand::ReportError { - message: format!( - "sync skipped for account '{}': {e}", - config.id - ), - }); - continue; - } - }; - match fetch_and_persist( - db, - event_bus, - backend.as_ref(), - filter_engine, - Some(&config.id), - ) - .await - { - Ok(count) => total_synced += count, - Err(e) => { - event_bus.process_command(CoreCommand::ReportError { - message: format!( - "sync failed for account '{}': {e}", - config.id - ), - }); - } - } - } - Err(e) => { - event_bus.process_command(CoreCommand::ReportError { - message: format!( - "cannot read credential for account '{}': {e}", - config.id - ), - }); - } - } - } - } - Err(e) => { - event_bus.process_command(CoreCommand::ReportError { - message: format!("failed to list accounts for sync: {e}"), - }); - } - } - - event_bus.complete_sync(None); - total_synced -} - #[cfg(test)] mod tests { use super::*; @@ -956,93 +877,6 @@ mod tests { assert_eq!(event_bus.current_state().status, EngineStatus::Idle); } - #[tokio::test] - async fn run_all_accounts_sync_with_no_accounts_returns_zero_and_emits_events() { - let (db, _dir) = DatabaseEngine::connect_ephemeral() - .await - .expect("ephemeral db"); - let event_bus = EventBus::new(); - let mut events = event_bus.subscribe_events(); - let secrets = SecretManager::mock(); - - let filter_engine = empty_filter_engine(); - let total = run_all_accounts_sync(&db, &secrets, &event_bus, &filter_engine).await; - assert_eq!(total, 0); - - assert_eq!( - events.recv().await.expect("start event"), - CoreEvent::SyncStarted { account_id: None } - ); - assert_eq!( - events.recv().await.expect("complete event"), - CoreEvent::SyncCompleted { account_id: None } - ); - } - - #[tokio::test] - async fn run_all_accounts_sync_persists_across_multiple_accounts() { - let mock_server_a = MockServer::start().await; - let mock_server_b = MockServer::start().await; - mount_single_message_jmap_stubs(&mock_server_a, "acct-all-a").await; - mount_single_message_jmap_stubs(&mock_server_b, "acct-all-b").await; - - let (db, _dir) = DatabaseEngine::connect_ephemeral() - .await - .expect("ephemeral db"); - let event_bus = EventBus::new(); - let secrets = SecretManager::mock(); - - let config_a = sample_jmap_account_at("acct-all-a", &mock_server_a.uri()); - let config_b = sample_jmap_account_at("acct-all-b", &mock_server_b.uri()); - db.save_account(&config_a).await.expect("save account a"); - db.save_account(&config_b).await.expect("save account b"); - secrets - .set_secret(&config_a.keyring_secret_key, "token-a") - .expect("store credential a"); - secrets - .set_secret(&config_b.keyring_secret_key, "token-b") - .expect("store credential b"); - - let filter_engine = empty_filter_engine(); - let total = run_all_accounts_sync(&db, &secrets, &event_bus, &filter_engine).await; - // 2 accounts * 1 folder * 1 message each = 2 processed messages. - assert_eq!(total, 2); - assert_eq!(event_bus.current_state().status, EngineStatus::Idle); - } - - #[tokio::test] - async fn run_all_accounts_sync_reports_error_for_account_missing_credential() { - let (db, _dir) = DatabaseEngine::connect_ephemeral() - .await - .expect("ephemeral db"); - let event_bus = EventBus::new(); - let mut events = event_bus.subscribe_events(); - let secrets = SecretManager::mock(); - - let config = sample_jmap_account("acct-all-nocred"); - db.save_account(&config).await.expect("save account"); - // No credential stored in the vault for this account. - - let filter_engine = empty_filter_engine(); - let total = run_all_accounts_sync(&db, &secrets, &event_bus, &filter_engine).await; - assert_eq!(total, 0); - - assert_eq!( - events.recv().await.expect("start event"), - CoreEvent::SyncStarted { account_id: None } - ); - match events.recv().await.expect("error event") { - CoreEvent::Error { message } => { - assert!(message.contains("acct-all-nocred")); - } - other => panic!("expected CoreEvent::Error, got {other:?}"), - } - assert_eq!( - events.recv().await.expect("complete event"), - CoreEvent::SyncCompleted { account_id: None } - ); - } - #[tokio::test] async fn sync_with_backend_fires_immediate_mark_read_action_on_match() { let (db, _dir) = DatabaseEngine::connect_ephemeral() diff --git a/crates/nunciod/src/sync_dispatcher.rs b/crates/nunciod/src/sync_dispatcher.rs new file mode 100644 index 0000000..29718f5 --- /dev/null +++ b/crates/nunciod/src/sync_dispatcher.rs @@ -0,0 +1,688 @@ +//! Bounded-concurrency sync dispatcher. +//! +//! Account syncs used to be launched directly and independently from two +//! places (the client-facing `Mail/Sync` RPC handler and the daemon's +//! `CoreCommand` consumer), each of which ran an account's sync to completion +//! before the next one started. One slow or hung account could therefore +//! head-of-line-block every account behind it, and nothing bounded how many +//! syncs ran at once or how long a single account was allowed to take. +//! +//! [`SyncDispatcher`] is the single place all account syncs now funnel +//! through. It: +//! +//! * caps how many account syncs run at once with a [`Semaphore`] +//! (env-tunable via [`CONCURRENCY_ENV_VAR`], default +//! [`DEFAULT_CONCURRENCY`]); excess requests queue on the permit rather +//! than run unbounded; +//! * coalesces a second request for an account that is already syncing onto +//! the in-flight one instead of double-fetching -- a client `Sync` issued +//! while a scheduled sync of the same account is running joins it and +//! observes its result; +//! * bounds each account's sync with a per-account timeout (env-tunable via +//! [`TIMEOUT_ENV_VAR`], default [`DEFAULT_TIMEOUT`]); a timed-out sync is +//! cancelled cleanly and surfaces [`SyncDispatchError::Timeout`] without +//! poisoning the dispatcher for other accounts; and +//! * respects the shared shutdown signal: once shutdown is requested it stops +//! admitting new work and races each in-flight sync against the signal +//! (mirroring the outbox executor) so a hung account cannot delay graceful +//! shutdown beyond the daemon's grace period. +//! +//! # Testability +//! +//! The actual per-account sync is injected behind the [`AccountSyncer`] +//! trait, so the dispatcher's concurrency, coalescing, timeout, and shutdown +//! behavior is exercised deterministically with an in-memory mock syncer -- +//! no real network, keyring, or database. Production wires +//! [`ProductionAccountSyncer`], which delegates to +//! [`crate::sync::run_account_sync`]. + +use crate::lifecycle::ShutdownSignal; +use crate::sync::SyncError; +use async_trait::async_trait; +use nuncio_core::{CoreCommand, EventBus}; +use nuncio_filter::FilterEngine; +use nuncio_store::db::DatabaseEngine; +use nuncio_store::vault::SecretManager; +use std::collections::HashMap; +use std::sync::{Arc, Mutex, PoisonError}; +use std::time::Duration; +use thiserror::Error; +use tokio::sync::{watch, Semaphore}; + +/// Environment variable bounding how many account syncs may run at once. +pub const CONCURRENCY_ENV_VAR: &str = "NUNCIO_SYNC_CONCURRENCY"; + +/// Environment variable bounding a single account's sync, in whole seconds. +pub const TIMEOUT_ENV_VAR: &str = "NUNCIO_SYNC_TIMEOUT"; + +/// Default simultaneous-sync cap when [`CONCURRENCY_ENV_VAR`] is unset or +/// unparseable. Chosen to overlap a handful of accounts without letting an +/// unbounded fan-out open a socket per account at once. +pub const DEFAULT_CONCURRENCY: usize = 4; + +/// Default per-account sync timeout when [`TIMEOUT_ENV_VAR`] is unset or +/// unparseable. Generous enough for a healthy full mailbox fetch, tight +/// enough that a hung account cannot stall others (or shutdown) indefinitely. +pub const DEFAULT_TIMEOUT: Duration = Duration::from_secs(300); + +/// The outcome of a single dispatched account sync. Cloneable so a coalesced +/// result can fan out to every joined waiter. +type SyncOutcome = Result; + +/// Errors surfaced by [`SyncDispatcher::sync_account`]. +#[derive(Debug, Clone, Error)] +pub enum SyncDispatchError { + /// The account's sync exceeded the configured per-account timeout and was + /// cancelled. Retryable: nothing about the dispatcher is left broken. + #[error("account '{account_id}' sync exceeded the {timeout:?} per-account timeout")] + Timeout { + /// The account whose sync timed out. + account_id: String, + /// The timeout that was exceeded. + timeout: Duration, + }, + /// Shutdown was requested before or during the sync; no new work is + /// admitted and in-flight work is abandoned for the next start. + #[error("sync dispatcher is shutting down")] + ShuttingDown, + /// The in-flight sync ended without producing a result (e.g. its task was + /// dropped). Retryable. + #[error("account sync was cancelled before it produced a result")] + Cancelled, + /// The underlying account sync itself failed. Shared behind an [`Arc`] so + /// a coalesced failure can be reported to every joined waiter. + #[error(transparent)] + Sync(Arc), +} + +/// Injectable seam for the actual work of syncing one account. Production +/// uses [`ProductionAccountSyncer`]; tests inject an in-memory mock. +#[async_trait] +pub trait AccountSyncer: Send + Sync { + /// Sync exactly one account to completion, returning the number of + /// messages processed. + async fn sync_account(&self, account_id: &str) -> Result; +} + +/// Production [`AccountSyncer`]: delegates to [`crate::sync::run_account_sync`], +/// which resolves the account's credentials, builds a real mail backend, and +/// runs a full inbound sync with the standard engine-status lifecycle. +pub struct ProductionAccountSyncer { + db: Arc, + secrets: Arc, + event_bus: Arc, + filter_engine: Arc, +} + +impl ProductionAccountSyncer { + /// Build the production syncer from the daemon's shared subsystems. + pub fn new( + db: Arc, + secrets: Arc, + event_bus: Arc, + filter_engine: Arc, + ) -> Self { + Self { + db, + secrets, + event_bus, + filter_engine, + } + } +} + +#[async_trait] +impl AccountSyncer for ProductionAccountSyncer { + async fn sync_account(&self, account_id: &str) -> Result { + crate::sync::run_account_sync( + &self.db, + &self.secrets, + &self.event_bus, + &self.filter_engine, + account_id, + ) + .await + } +} + +/// Reads [`CONCURRENCY_ENV_VAR`], falling back to [`DEFAULT_CONCURRENCY`]. +/// A value of zero (or unparseable) is treated as the default; the effective +/// cap is always at least one. +fn concurrency_from_env() -> usize { + std::env::var(CONCURRENCY_ENV_VAR) + .ok() + .and_then(|v| v.trim().parse::().ok()) + .filter(|n| *n > 0) + .unwrap_or(DEFAULT_CONCURRENCY) +} + +/// Reads [`TIMEOUT_ENV_VAR`] as whole seconds, falling back to +/// [`DEFAULT_TIMEOUT`]. A value of zero (or unparseable) is treated as the +/// default rather than "no timeout", so a misconfiguration can never disable +/// the safety bound. +fn timeout_from_env() -> Duration { + std::env::var(TIMEOUT_ENV_VAR) + .ok() + .and_then(|v| v.trim().parse::().ok()) + .filter(|secs| *secs > 0) + .map(Duration::from_secs) + .unwrap_or(DEFAULT_TIMEOUT) +} + +/// Shared inner state, held behind an [`Arc`] so the dispatcher is cheap to +/// clone and can be moved into the per-account tasks it spawns. +struct Inner { + syncer: Arc, + semaphore: Arc, + /// Accounts with a sync currently in flight, mapped to a watch receiver + /// that publishes the sync's outcome once. A second request for the same + /// account clones the receiver and joins the in-flight sync. + inflight: Mutex>>>, + per_account_timeout: Duration, + shutdown: ShutdownSignal, +} + +impl Inner { + /// Run one account sync under the semaphore permit and per-account + /// timeout, racing both the permit wait and the sync itself against the + /// shutdown signal (mirroring the outbox executor). The permit is + /// released as soon as the sync resolves. + async fn run_guarded(&self, account_id: &str) -> SyncOutcome { + let mut shutdown = self.shutdown.clone(); + + let permit = tokio::select! { + biased; + _ = shutdown.wait() => return Err(SyncDispatchError::ShuttingDown), + acquired = self.semaphore.clone().acquire_owned() => match acquired { + Ok(permit) => permit, + // The semaphore is never closed in practice; treat a closed + // semaphore as a cancellation rather than fabricating success. + Err(_) => return Err(SyncDispatchError::Cancelled), + }, + }; + + let outcome = tokio::select! { + biased; + _ = shutdown.wait() => Err(SyncDispatchError::ShuttingDown), + result = tokio::time::timeout( + self.per_account_timeout, + self.syncer.sync_account(account_id), + ) => match result { + Ok(Ok(count)) => Ok(count), + Ok(Err(e)) => Err(SyncDispatchError::Sync(Arc::new(e))), + Err(_elapsed) => Err(SyncDispatchError::Timeout { + account_id: account_id.to_string(), + timeout: self.per_account_timeout, + }), + }, + }; + + drop(permit); + outcome + } +} + +/// The single entry point every account sync funnels through. Cheap to clone +/// (an [`Arc`] handle); every clone shares the same concurrency cap, +/// in-flight map, timeout, and shutdown signal. +#[derive(Clone)] +pub struct SyncDispatcher { + inner: Arc, +} + +impl SyncDispatcher { + /// Build a dispatcher with explicit bounds. `concurrency` is clamped up to + /// at least one so the dispatcher can always make progress. + pub fn new( + syncer: Arc, + shutdown: ShutdownSignal, + concurrency: usize, + per_account_timeout: Duration, + ) -> Self { + let concurrency = concurrency.max(1); + Self { + inner: Arc::new(Inner { + syncer, + semaphore: Arc::new(Semaphore::new(concurrency)), + inflight: Mutex::new(HashMap::new()), + per_account_timeout, + shutdown, + }), + } + } + + /// Build a dispatcher whose bounds come from the environment + /// ([`CONCURRENCY_ENV_VAR`] / [`TIMEOUT_ENV_VAR`]), falling back to + /// [`DEFAULT_CONCURRENCY`] / [`DEFAULT_TIMEOUT`]. + pub fn from_env(syncer: Arc, shutdown: ShutdownSignal) -> Self { + Self::new(syncer, shutdown, concurrency_from_env(), timeout_from_env()) + } + + /// Dispatch a sync for a single account, applying bounded concurrency, + /// in-flight coalescing, the per-account timeout, and shutdown handling. + /// + /// If a sync for `account_id` is already in flight, this joins it and + /// returns the SAME outcome rather than starting a duplicate fetch. + pub async fn sync_account(&self, account_id: &str) -> SyncOutcome { + if self.inner.shutdown.is_requested() { + return Err(SyncDispatchError::ShuttingDown); + } + + // Either join the in-flight sync for this account or become its owner + // by inserting a fresh result channel and spawning the guarded run. + // The lock is released before awaiting, so it never spans `.await`. + let mut rx = { + let mut inflight = self + .inner + .inflight + .lock() + .unwrap_or_else(PoisonError::into_inner); + if let Some(existing) = inflight.get(account_id) { + existing.clone() + } else { + let (tx, rx) = watch::channel::>(None); + inflight.insert(account_id.to_string(), rx.clone()); + + let inner = self.inner.clone(); + let id = account_id.to_string(); + tokio::spawn(async move { + let outcome = inner.run_guarded(&id).await; + // Remove the in-flight entry before publishing so a request + // arriving after completion starts a fresh sync rather than + // joining an already-finished one. + inner + .inflight + .lock() + .unwrap_or_else(PoisonError::into_inner) + .remove(&id); + let _ = tx.send(Some(outcome)); + }); + rx + } + }; + + // Clone the published outcome out of the borrowed watch guard before + // the guard (and `rx`) drop at the end of this scope. + let published = match rx.wait_for(Option::is_some).await { + Ok(guard) => guard.clone(), + // The sender was dropped without publishing (the owning task was + // itself dropped). Honestly report a cancellation, never success. + Err(_) => None, + }; + published.unwrap_or(Err(SyncDispatchError::Cancelled)) + } + + /// Dispatch syncs for several accounts concurrently, each through the same + /// bounded/coalesced/timed per-account path, and return the total number + /// of messages synced across the accounts that succeeded. Per-account + /// failures are logged and do not abort the batch (the underlying + /// production syncer additionally reports them on the event bus). + pub async fn sync_all(&self, account_ids: Vec) -> usize { + let mut set = tokio::task::JoinSet::new(); + for account_id in account_ids { + let dispatcher = self.clone(); + set.spawn(async move { + let result = dispatcher.sync_account(&account_id).await; + (account_id, result) + }); + } + + let mut total = 0usize; + while let Some(joined) = set.join_next().await { + match joined { + Ok((_, Ok(count))) => total += count, + Ok((account_id, Err(e))) => { + tracing::warn!("dispatched sync for account '{account_id}' failed: {e}"); + } + Err(join_err) => { + tracing::warn!("dispatched sync task failed to join: {join_err}"); + } + } + } + total + } +} + +/// Sync every configured mail account through `dispatcher`, wrapped in the +/// standard batch `SyncStarted{None}` / `SyncCompleted{None}` envelope. +/// +/// DAV-protocol accounts have no inbound mail backend and are skipped (they +/// sync through their own domain service). This is the single production +/// entry point for an all-accounts sync, shared by the `Mail/Sync` RPC +/// (`account_id == None`) and the daemon's `CoreCommand::SyncAll` consumer, so +/// the two never launch account syncs through separate, differently-bounded +/// paths. Returns the total number of messages synced across all accounts. +pub async fn sync_all_configured( + dispatcher: &SyncDispatcher, + db: &DatabaseEngine, + event_bus: &EventBus, +) -> usize { + event_bus.process_command(CoreCommand::SyncAll); + + let account_ids = match db.list_accounts().await { + Ok(accounts) => accounts + .into_iter() + .filter(|a| !a.protocol.is_dav()) + .map(|a| a.id) + .collect::>(), + Err(e) => { + event_bus.process_command(CoreCommand::ReportError { + message: format!("failed to list accounts for sync: {e}"), + }); + Vec::new() + } + }; + + let total = dispatcher.sync_all(account_ids).await; + event_bus.complete_sync(None); + total +} + +#[cfg(test)] +mod tests { + use super::*; + use nuncio_core::{AccountConfig, AccountProtocol, CoreEvent, TlsMode}; + use std::collections::HashSet; + use std::sync::atomic::{AtomicUsize, Ordering}; + use tokio::time::{sleep, Duration}; + + /// In-memory [`AccountSyncer`] that records invocations, tracks peak + /// concurrency, and can be told to block indefinitely for named accounts + /// (to model a hung account) or to add a fixed per-call delay (to make + /// concurrent calls genuinely overlap). + struct MockSyncer { + calls: Mutex>, + concurrent: AtomicUsize, + peak: AtomicUsize, + blocked: HashSet, + gate: Arc, + per_call_delay: Option, + fail: HashSet, + } + + impl MockSyncer { + fn new() -> Self { + Self { + calls: Mutex::new(Vec::new()), + concurrent: AtomicUsize::new(0), + peak: AtomicUsize::new(0), + blocked: HashSet::new(), + gate: Arc::new(tokio::sync::Notify::new()), + per_call_delay: None, + fail: HashSet::new(), + } + } + + fn with_delay(mut self, delay: Duration) -> Self { + self.per_call_delay = Some(delay); + self + } + + fn blocking(mut self, account_id: &str) -> Self { + self.blocked.insert(account_id.to_string()); + self + } + + fn failing(mut self, account_id: &str) -> Self { + self.fail.insert(account_id.to_string()); + self + } + + fn call_ids(&self) -> Vec { + self.calls + .lock() + .unwrap_or_else(PoisonError::into_inner) + .clone() + } + + fn call_count(&self) -> usize { + self.calls + .lock() + .unwrap_or_else(PoisonError::into_inner) + .len() + } + } + + #[async_trait] + impl AccountSyncer for MockSyncer { + async fn sync_account(&self, account_id: &str) -> Result { + self.calls + .lock() + .unwrap_or_else(PoisonError::into_inner) + .push(account_id.to_string()); + let current = self.concurrent.fetch_add(1, Ordering::SeqCst) + 1; + self.peak.fetch_max(current, Ordering::SeqCst); + + if self.blocked.contains(account_id) { + self.gate.notified().await; + } else if let Some(delay) = self.per_call_delay { + sleep(delay).await; + } + + self.concurrent.fetch_sub(1, Ordering::SeqCst); + + if self.fail.contains(account_id) { + return Err(SyncError::AccountNotFound(account_id.to_string())); + } + Ok(1) + } + } + + fn dispatcher_with( + syncer: Arc, + concurrency: usize, + timeout: Duration, + ) -> SyncDispatcher { + SyncDispatcher::new(syncer, ShutdownSignal::never(), concurrency, timeout) + } + + #[tokio::test] + async fn bounds_simultaneous_syncs_to_the_configured_cap() { + let syncer = Arc::new(MockSyncer::new().with_delay(Duration::from_millis(80))); + let dispatcher = dispatcher_with(syncer.clone(), 2, Duration::from_secs(30)); + + let mut set = tokio::task::JoinSet::new(); + for i in 0..6 { + let d = dispatcher.clone(); + set.spawn(async move { d.sync_account(&format!("acct-{i}")).await }); + } + while let Some(joined) = set.join_next().await { + assert!(matches!(joined, Ok(Ok(1))), "each distinct sync succeeds"); + } + + assert_eq!( + syncer.peak.load(Ordering::SeqCst), + 2, + "no more than the configured cap of 2 syncs may run at once" + ); + assert_eq!(syncer.call_count(), 6, "every distinct account is synced"); + } + + #[tokio::test] + async fn coalesces_a_duplicate_in_flight_sync_for_the_same_account() { + let syncer = Arc::new(MockSyncer::new().with_delay(Duration::from_millis(80))); + let dispatcher = dispatcher_with(syncer.clone(), 4, Duration::from_secs(30)); + + // Three concurrent requests for the SAME account, launched close + // enough together to all observe the first as still in flight. + let mut set = tokio::task::JoinSet::new(); + for _ in 0..3 { + let d = dispatcher.clone(); + set.spawn(async move { d.sync_account("acct-dup").await }); + } + while let Some(joined) = set.join_next().await { + assert!( + matches!(joined, Ok(Ok(1))), + "every joined waiter sees the result" + ); + } + + assert_eq!( + syncer.call_ids(), + vec!["acct-dup".to_string()], + "the underlying sync must run exactly once, not once per request" + ); + } + + #[tokio::test] + async fn a_hung_account_does_not_block_other_accounts() { + // Account A blocks forever; account B must still complete. With the + // cap at 2 both can hold a permit at once, proving the fix for the + // head-of-line blocking regression. + let syncer = Arc::new(MockSyncer::new().blocking("acct-A")); + let dispatcher = dispatcher_with(syncer.clone(), 2, Duration::from_secs(30)); + + let d_a = dispatcher.clone(); + let a = tokio::spawn(async move { d_a.sync_account("acct-A").await }); + + let b = dispatcher.sync_account("acct-B").await; + assert!( + matches!(b, Ok(1)), + "account B completes even while account A is hung, got {b:?}" + ); + + // A is still blocked; it has not resolved. + assert!(!a.is_finished(), "account A's sync is still in flight"); + a.abort(); + } + + #[tokio::test] + async fn a_hung_account_times_out_without_poisoning_the_dispatcher() { + let syncer = Arc::new(MockSyncer::new().blocking("acct-hung")); + let dispatcher = dispatcher_with(syncer.clone(), 2, Duration::from_millis(60)); + + let hung = dispatcher.sync_account("acct-hung").await; + assert!( + matches!(hung, Err(SyncDispatchError::Timeout { .. })), + "a hung account surfaces a typed Timeout, got {hung:?}" + ); + + // The dispatcher is not poisoned: a subsequent sync of a healthy + // account still works. + let ok = dispatcher.sync_account("acct-healthy").await; + assert!( + matches!(ok, Ok(1)), + "a later sync still works after a timeout, got {ok:?}" + ); + } + + #[tokio::test] + async fn surfaces_the_underlying_sync_error_typed() { + let syncer = Arc::new(MockSyncer::new().failing("acct-bad")); + let dispatcher = dispatcher_with(syncer, 2, Duration::from_secs(30)); + + let err = dispatcher.sync_account("acct-bad").await; + assert!( + matches!(err, Err(SyncDispatchError::Sync(_))), + "an underlying sync failure surfaces as SyncDispatchError::Sync, got {err:?}" + ); + } + + #[tokio::test] + async fn rejects_new_work_once_shutdown_is_requested() { + let event_bus = Arc::new(EventBus::new()); + let (controller, shutdown) = crate::lifecycle::ShutdownController::new(event_bus); + let syncer = Arc::new(MockSyncer::new()); + let dispatcher = SyncDispatcher::new(syncer.clone(), shutdown, 2, Duration::from_secs(30)); + + controller.trigger(); + + let rejected = dispatcher.sync_account("acct-late").await; + assert!( + matches!(rejected, Err(SyncDispatchError::ShuttingDown)), + "no new sync is admitted after shutdown, got {rejected:?}" + ); + assert_eq!( + syncer.call_count(), + 0, + "the underlying syncer must not even be invoked after shutdown" + ); + } + + #[tokio::test] + async fn shutdown_unblocks_an_in_flight_sync_promptly() { + let event_bus = Arc::new(EventBus::new()); + let (controller, shutdown) = crate::lifecycle::ShutdownController::new(event_bus); + let syncer = Arc::new(MockSyncer::new().blocking("acct-drain")); + let dispatcher = SyncDispatcher::new(syncer, shutdown, 2, Duration::from_secs(300)); + + let d = dispatcher.clone(); + let handle = tokio::spawn(async move { d.sync_account("acct-drain").await }); + + // Let the sync reach its blocking point, then request shutdown. + sleep(Duration::from_millis(30)).await; + controller.trigger(); + + let outcome = tokio::time::timeout(Duration::from_secs(5), handle) + .await + .expect("in-flight sync must unblock on shutdown, not hang") + .expect("task joins"); + assert!( + matches!(outcome, Err(SyncDispatchError::ShuttingDown)), + "a sync racing shutdown resolves as ShuttingDown, got {outcome:?}" + ); + } + + fn jmap_account(id: &str) -> AccountConfig { + AccountConfig { + id: id.to_string(), + name: "Test".to_string(), + email_address: format!("{id}@nuncio.mx"), + protocol: AccountProtocol::Jmap, + server_host: "jmap.nuncio.mx".to_string(), + server_port: 443, + smtp_host: "smtp.nuncio.mx".to_string(), + smtp_port: 465, + imap_tls_mode: TlsMode::ImplicitTls, + smtp_tls_mode: TlsMode::ImplicitTls, + keyring_secret_key: format!("nuncio/{id}"), + sync_interval_secs: 60, + collection_url: String::new(), + } + } + + fn caldav_account(id: &str) -> AccountConfig { + AccountConfig { + protocol: AccountProtocol::CalDav, + ..jmap_account(id) + } + } + + #[tokio::test] + async fn sync_all_configured_dispatches_every_mail_account_and_skips_dav() { + let (db, _dir) = DatabaseEngine::connect_ephemeral() + .await + .expect("ephemeral db"); + db.save_account(&jmap_account("acct-mail-1")) + .await + .expect("save mail account 1"); + db.save_account(&jmap_account("acct-mail-2")) + .await + .expect("save mail account 2"); + db.save_account(&caldav_account("acct-dav")) + .await + .expect("save dav account"); + + let event_bus = EventBus::new(); + let mut events = event_bus.subscribe_events(); + let syncer = Arc::new(MockSyncer::new()); + let dispatcher = dispatcher_with(syncer.clone(), 4, Duration::from_secs(30)); + + let total = sync_all_configured(&dispatcher, &db, &event_bus).await; + assert_eq!(total, 2, "both mail accounts synced one message each"); + + let mut synced: Vec = syncer.call_ids(); + synced.sort(); + assert_eq!( + synced, + vec!["acct-mail-1".to_string(), "acct-mail-2".to_string()], + "only the mail accounts are dispatched; the CalDAV account is skipped" + ); + + // The batch is wrapped in a single all-accounts start/complete pair. + assert_eq!( + events.recv().await.expect("start event"), + CoreEvent::SyncStarted { account_id: None } + ); + } +} -- 2.54.0