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<SecretManager>,
     filter_engine: Arc<FilterEngine>,
     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<SyncDispatcher>,
 }
 
 #[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<SyncRequest>) -> Result<Response<SyncResponse>, 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<DatabaseEngine>,
+    secrets: Arc<SecretManager>,
+    event_bus: Arc<EventBus>,
+    filter_engine: Arc<FilterEngine>,
+    shutdown: ShutdownSignal,
+) -> Arc<SyncDispatcher> {
+    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<EventBus>,
@@ -2555,6 +2583,7 @@ pub async fn serve_with_shutdown(
     filter_engine: Arc<FilterEngine>,
     secrets: Arc<SecretManager>,
     token: impl Into<Arc<str>>,
+    sync_dispatcher: Arc<SyncDispatcher>,
     shutdown: impl std::future::Future<Output = ()> + 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<Arc<str>>,
     shutdown: impl std::future::Future<Output = ()> + 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<SyncDispatcher>,
     shutdown: impl std::future::Future<Output = ()> + Send + 'static,
 ) -> Result<(), GrpcServeError> {
     let token: Arc<str> = 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(

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`

crates/nunciod/src/lifecycle.rs

@@ -20,9 +20,26 @@ use tokio::sync::watch;
 #[derive(Clone)]
 pub struct ShutdownSignal {
     rx: watch::Receiver<bool>,
+    // 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<Arc<watch::Sender<bool>>>,
 }
 
 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<EventBus>) -> (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

crates/nunciod/src/main.rs

@@ -70,6 +70,27 @@ async fn main() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
     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<dyn std::error::Error + Send + Sync>> {
     // 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<dyn std::error::Error + Send + Sync>> {
                 };
                 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<dyn std::error::Error + Send + Sync>> {
             filter_engine,
             grpc_secrets,
             grpc_token,
+            sync_dispatcher,
             async move { grpc_shutdown_future.wait().await },
         ) => result,
         _ = async move {

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()

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<usize, SyncDispatchError>;
+
+/// 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<SyncError>),
+}
+
+/// 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<usize, SyncError>;
+}
+
+/// 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<DatabaseEngine>,
+    secrets: Arc<SecretManager>,
+    event_bus: Arc<EventBus>,
+    filter_engine: Arc<FilterEngine>,
+}
+
+impl ProductionAccountSyncer {
+    /// Build the production syncer from the daemon's shared subsystems.
+    pub fn new(
+        db: Arc<DatabaseEngine>,
+        secrets: Arc<SecretManager>,
+        event_bus: Arc<EventBus>,
+        filter_engine: Arc<FilterEngine>,
+    ) -> Self {
+        Self {
+            db,
+            secrets,
+            event_bus,
+            filter_engine,
+        }
+    }
+}
+
+#[async_trait]
+impl AccountSyncer for ProductionAccountSyncer {
+    async fn sync_account(&self, account_id: &str) -> Result<usize, SyncError> {
+        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::<usize>().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::<u64>().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<dyn AccountSyncer>,
+    semaphore: Arc<Semaphore>,
+    /// 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<HashMap<String, watch::Receiver<Option<SyncOutcome>>>>,
+    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<Inner>,
+}
+
+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<dyn AccountSyncer>,
+        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<dyn AccountSyncer>, 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::<Option<SyncOutcome>>(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<String>) -> 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::<Vec<_>>(),
+        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<Vec<String>>,
+        concurrent: AtomicUsize,
+        peak: AtomicUsize,
+        blocked: HashSet<String>,
+        gate: Arc<tokio::sync::Notify>,
+        per_call_delay: Option<Duration>,
+        fail: HashSet<String>,
+    }
+
+    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<String> {
+            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<usize, SyncError> {
+            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<MockSyncer>,
+        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<String> = 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 }
+        );
+    }
+}