crates/nunciod/Cargo.toml

@@ -40,3 +40,4 @@ fs4.workspace = true
 nuncio-cli = { path = "../nuncio-cli" }
 sqlx.workspace = true
 wiremock.workspace = true
+tokio = { workspace = true, features = ["test-util"] }

crates/nunciod/src/main.rs

@@ -153,6 +153,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
         account_secrets.clone(),
     ));
     let mut outbox_shutdown = shutdown_signal.clone();
+    let mut outbox_drain_shutdown = shutdown_signal.clone();
     let outbox_task = tokio::spawn(async move {
         let mut interval = tokio::time::interval(std::time::Duration::from_secs(5));
         loop {
@@ -162,13 +163,17 @@ async fn main() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
                     break;
                 }
                 _ = interval.tick() => {
-                    // Runs to completion once selected: a drain pass already
-                    // in progress is never aborted mid-write by a shutdown
-                    // signal that arrives after this branch was chosen.
+                    // A drain pass already in progress is never aborted
+                    // mid-write by a shutdown signal that arrives after this
+                    // branch was chosen -- but `execute_pending_mutations`
+                    // itself races each item's execution against
+                    // `outbox_drain_shutdown`, so a hung remote op cannot
+                    // block this pass (and therefore shutdown) indefinitely.
                     let summary = nunciod::outbox::execute_pending_mutations(
                         &db_outbox,
                         outbox_env.as_ref(),
                         50,
+                        &mut outbox_drain_shutdown,
                     )
                     .await;
                     if summary.completed > 0 || summary.failed > 0 {

crates/nunciod/src/outbox.rs

@@ -16,7 +16,9 @@
 //! exceeded, at which point it flips to `failed`. A permanent failure (the
 //! message no longer exists, a malformed payload, an unknown action) flips to
 //! `failed` immediately -- there is nothing a retry could fix. Success is
-//! never fabricated.
+//! never fabricated. A mutation whose execution exceeds
+//! [`PER_ITEM_EXECUTION_TIMEOUT`] is dispositioned the same way as any other
+//! transient failure -- it is RETRYABLE, never `completed`.
 //!
 //! # Testability
 //!
@@ -26,6 +28,7 @@
 //! without any live network. Production wires [`ProductionExecutionEnv`],
 //! which resolves real per-account engines from the store and the OS keyring.
 
+use crate::lifecycle::ShutdownSignal;
 use async_trait::async_trait;
 use nuncio_core::AccountConfig;
 use nuncio_filter::{
@@ -39,8 +42,18 @@ use nuncio_mail::{
 use nuncio_store::db::{DatabaseEngine, DatabaseError};
 use nuncio_store::vault::{SecretManager, VaultError, WEBHOOK_SIGNING_KEY_ACCOUNT};
 use std::sync::Arc;
+use std::time::Duration;
 use thiserror::Error;
 
+/// Bound on how long a single outbox mutation may run before it is treated
+/// as a retryable timeout. A healthy IMAP/JMAP/SMTP round trip or webhook
+/// POST comfortably finishes well inside this window; a value this generous
+/// still keeps one hung remote op from stalling every OTHER pending
+/// mutation -- across every account -- behind it in the same drain pass, and
+/// from blocking graceful shutdown for longer than the shutdown grace
+/// period tolerates.
+const PER_ITEM_EXECUTION_TIMEOUT: Duration = Duration::from_secs(30);
+
 /// Errors that can occur while resolving the resources needed to execute a
 /// single outbox mutation. These are surfaced from a [`RemoteExecutionEnv`]
 /// and treated as transient (retryable) by the executor unless it has already
@@ -119,11 +132,18 @@ pub struct ExecutionSummary {
 
 /// Drain up to `limit` pending mutations, executing each against `env` and
 /// persisting the honest outcome. Never marks a mutation `completed` unless the
-/// real operation succeeded. Returns a tally of the pass.
+/// real operation succeeded. Each item's execution is bounded by
+/// [`PER_ITEM_EXECUTION_TIMEOUT`] so one hung remote op cannot freeze delivery
+/// for every other item behind it; `shutdown` is raced against that same
+/// timeout so a hung item cannot delay graceful shutdown either -- on
+/// shutdown the drain pass stops immediately, leaving the in-flight item (and
+/// anything still queued behind it) untouched and `pending` for the next
+/// start. Returns a tally of the pass.
 pub async fn execute_pending_mutations(
     db: &DatabaseEngine,
     env: &dyn RemoteExecutionEnv,
     limit: usize,
+    shutdown: &mut ShutdownSignal,
 ) -> ExecutionSummary {
     let pending = match db.list_pending_mutations(limit).await {
         Ok(items) => items,
@@ -135,7 +155,23 @@ pub async fn execute_pending_mutations(
 
     let mut summary = ExecutionSummary::default();
     for item in pending {
-        let disposition = execute_one(db, env, &item).await;
+        let outcome = tokio::select! {
+            biased;
+            _ = shutdown.wait() => {
+                tracing::info!(
+                    mutation_id = %item.id,
+                    "outbox: drain pass interrupted by shutdown signal; item left pending"
+                );
+                break;
+            }
+            result = tokio::time::timeout(PER_ITEM_EXECUTION_TIMEOUT, execute_one(db, env, &item)) => result,
+        };
+        let disposition = match outcome {
+            Ok(disposition) => disposition,
+            Err(_elapsed) => Disposition::Retry(format!(
+                "mutation execution exceeded the {PER_ITEM_EXECUTION_TIMEOUT:?} timeout"
+            )),
+        };
         let next_retry = item.retry_count + 1;
         let (status, retry, tally): (&str, i32, fn(&mut ExecutionSummary)) = match disposition {
             Disposition::Completed => ("completed", item.retry_count, |s| s.completed += 1),

crates/nunciod/tests/outbox_executor_e2e_test.rs

@@ -13,16 +13,28 @@
 
 use async_trait::async_trait;
 use nuncio_core::model::Email;
+use nuncio_core::EventBus;
 use nuncio_filter::{FilterEngine, NsqlParser, ValidationOptions, WebhookDispatcher, WebhookError};
 use nuncio_mail::{
     MailBackend, MessageSender, MockMailBackend, MockMessageSender, RemoteMutationKind,
 };
 use nuncio_store::db::DatabaseEngine;
+use nunciod::lifecycle::{ShutdownController, ShutdownSignal};
 use nunciod::outbox::{execute_pending_mutations, RemoteExecutionEnv};
 use nunciod::sync::apply_filter_actions;
+use std::sync::atomic::{AtomicUsize, Ordering};
+use std::sync::Arc;
 use wiremock::matchers::{method, path};
 use wiremock::{Mock, MockServer, ResponseTemplate};
 
+/// A shutdown handle for tests that never trigger it. Holding the returned
+/// `ShutdownController` alive for the duration of the test matters: dropping
+/// it closes the underlying `watch` channel, which would make `wait()`
+/// resolve immediately and be mistaken for a real shutdown request.
+fn no_shutdown() -> (ShutdownController, ShutdownSignal) {
+    ShutdownController::new(Arc::new(EventBus::new()))
+}
+
 const ACCOUNT_ID: &str = "acct-outbox-e2e";
 const FOLDER_ID: &str = "INBOX";
 const CHECKPOINT: &str = "1:105";
@@ -145,6 +157,54 @@ async fn seed_and_enqueue(db: &DatabaseEngine, rule_nsql: &str, subject: &str) {
     );
 }
 
+/// Like [`seed_and_enqueue`], but for a message with its own surrogate id so
+/// several calls in the same test enqueue mutations against DISTINCT
+/// messages instead of colliding on the shared fixed id `sample_email`
+/// otherwise defaults to.
+async fn seed_and_enqueue_distinct(
+    db: &DatabaseEngine,
+    message_id: &str,
+    rule_nsql: &str,
+    subject: &str,
+) {
+    db.save_account(&sample_account())
+        .await
+        .expect("save account");
+    db.save_folder_sync_state(ACCOUNT_ID, FOLDER_ID, CHECKPOINT)
+        .await
+        .expect("save folder checkpoint");
+
+    // `remote_id`/`uid_validity` (not just `id`) are part of the store's
+    // `messages` identity UNIQUE index -- leaving them at `sample_email`'s
+    // shared default would make this INSERT OR REPLACE the OTHER seeded
+    // message sharing that same remote identity out from under it.
+    let mut email = sample_email(subject);
+    email.id = message_id.to_string();
+    email.remote_id = message_id.to_string();
+    db.save_email(&email).await.expect("save email");
+
+    let before = db
+        .list_pending_mutations(100)
+        .await
+        .expect("list pending")
+        .len();
+
+    let rule = NsqlParser::parse_rule("Outbox Rule", 1, rule_nsql).expect("parse rule");
+    let engine = FilterEngine::new(vec![rule]).expect("compile rule");
+    apply_filter_actions(db, &engine, &email).await;
+
+    let after = db
+        .list_pending_mutations(100)
+        .await
+        .expect("list pending")
+        .len();
+    assert_eq!(
+        after,
+        before + 1,
+        "the rule must enqueue exactly one new mutation"
+    );
+}
+
 #[tokio::test]
 async fn move_rule_executes_against_backend_and_completes_with_uidvalidity_checkpoint() {
     let (db, _dir) = DatabaseEngine::connect_ephemeral().await.expect("db");
@@ -156,7 +216,8 @@ async fn move_rule_executes_against_backend_and_completes_with_uidvalidity_check
     .await;
 
     let env = MockEnv::new();
-    let summary = execute_pending_mutations(&db, &env, 50).await;
+    let (_shutdown_ctrl, mut shutdown) = no_shutdown();
+    let summary = execute_pending_mutations(&db, &env, 50, &mut shutdown).await;
 
     assert_eq!(summary.completed, 1);
     assert_eq!(summary.failed, 0);
@@ -192,7 +253,8 @@ async fn flag_rule_executes_as_set_flagged() {
     .await;
 
     let env = MockEnv::new();
-    let summary = execute_pending_mutations(&db, &env, 50).await;
+    let (_shutdown_ctrl, mut shutdown) = no_shutdown();
+    let summary = execute_pending_mutations(&db, &env, 50, &mut shutdown).await;
 
     assert_eq!(summary.completed, 1);
     let applied = env.backend.applied_mutations();
@@ -214,7 +276,8 @@ async fn delete_rule_executes_as_delete() {
     .await;
 
     let env = MockEnv::new();
-    let summary = execute_pending_mutations(&db, &env, 50).await;
+    let (_shutdown_ctrl, mut shutdown) = no_shutdown();
+    let summary = execute_pending_mutations(&db, &env, 50, &mut shutdown).await;
 
     assert_eq!(summary.completed, 1);
     let applied = env.backend.applied_mutations();
@@ -234,7 +297,8 @@ async fn a_failing_backend_op_is_never_marked_completed() {
 
     let env = MockEnv::new();
     env.backend.set_should_fail(true);
-    let summary = execute_pending_mutations(&db, &env, 50).await;
+    let (_shutdown_ctrl, mut shutdown) = no_shutdown();
+    let summary = execute_pending_mutations(&db, &env, 50, &mut shutdown).await;
 
     // The op failed: nothing recorded, nothing completed, and the item is kept
     // for retry (still pending, with a bumped retry count) -- never faked.
@@ -258,7 +322,8 @@ async fn forward_rule_sends_via_the_message_sender() {
     .await;
 
     let env = MockEnv::new();
-    let summary = execute_pending_mutations(&db, &env, 50).await;
+    let (_shutdown_ctrl, mut shutdown) = no_shutdown();
+    let summary = execute_pending_mutations(&db, &env, 50, &mut shutdown).await;
 
     assert_eq!(summary.completed, 1);
     let sent = env.sender.sent_messages();
@@ -293,7 +358,8 @@ async fn webhook_rule_dispatches_a_signed_post_to_the_receiver() {
     .await;
 
     let env = MockEnv::new();
-    let summary = execute_pending_mutations(&db, &env, 50).await;
+    let (_shutdown_ctrl, mut shutdown) = no_shutdown();
+    let summary = execute_pending_mutations(&db, &env, 50, &mut shutdown).await;
 
     // Completion is genuine: wiremock's `.expect(1)` verifies the POST actually
     // arrived when the server drops.
@@ -320,7 +386,8 @@ async fn a_mutation_for_a_vanished_message_fails_permanently() {
         .expect("save mutation");
 
     let env = MockEnv::new();
-    let summary = execute_pending_mutations(&db, &env, 50).await;
+    let (_shutdown_ctrl, mut shutdown) = no_shutdown();
+    let summary = execute_pending_mutations(&db, &env, 50, &mut shutdown).await;
 
     // Cannot act on a message that no longer exists -- fail honestly, never
     // fabricate success, and never touch the backend.
@@ -332,3 +399,219 @@ async fn a_mutation_for_a_vanished_message_fails_permanently() {
         "a permanently-failed mutation is no longer pending"
     );
 }
+
+/// A [`RemoteExecutionEnv`] whose `mail_backend` call hangs (a `tokio::time`
+/// sleep far longer than any timeout under test, standing in for a stuck
+/// IMAP/JMAP op) for the first `hang_calls` invocations, then delegates to a
+/// normal [`MockEnv`]. Used to prove the per-item execution timeout fires.
+struct HangEnv {
+    inner: MockEnv,
+    calls: AtomicUsize,
+    hang_calls: usize,
+}
+
+impl HangEnv {
+    fn new(hang_calls: usize) -> Self {
+        Self {
+            inner: MockEnv::new(),
+            calls: AtomicUsize::new(0),
+            hang_calls,
+        }
+    }
+}
+
+#[async_trait]
+impl RemoteExecutionEnv for HangEnv {
+    async fn mail_backend(
+        &self,
+        account_id: &str,
+    ) -> Result<Box<dyn MailBackend>, nunciod::outbox::OutboxExecuteError> {
+        let call = self.calls.fetch_add(1, Ordering::SeqCst);
+        if call < self.hang_calls {
+            // A `tokio::time`-tracked sleep far longer than any per-item
+            // timeout under test -- NOT `std::future::pending`, which never
+            // registers with the time driver and so behaves unpredictably
+            // once `tokio::time::pause`'s auto-advance-on-idle interacts with
+            // real (non-timer) I/O elsewhere in the same test, like the
+            // SQLite pool's own connection bookkeeping. A real (virtual)
+            // sleep is what the timeout under test is meant to race against.
+            tokio::time::sleep(std::time::Duration::from_secs(3600)).await;
+        }
+        self.inner.mail_backend(account_id).await
+    }
+
+    async fn message_sender(
+        &self,
+        account_id: &str,
+    ) -> Result<Box<dyn MessageSender>, nunciod::outbox::OutboxExecuteError> {
+        self.inner.message_sender(account_id).await
+    }
+
+    async fn dispatch_webhook(
+        &self,
+        url: &str,
+        rule_id: &str,
+        message_id: &str,
+        subject: &str,
+        sender: &str,
+    ) -> Result<u16, WebhookError> {
+        self.inner
+            .dispatch_webhook(url, rule_id, message_id, subject, sender)
+            .await
+    }
+}
+
+/// Seed two mutations so the second proves the drain pass actually moves on
+/// after the first times out, rather than the whole pass having stalled.
+async fn seed_two_move_mutations(db: &DatabaseEngine) {
+    seed_and_enqueue_distinct(
+        db,
+        "surrogate-hang-1",
+        "WHERE subject CONTAINS 'Archive' ACTION MOVE TO 'Archive'",
+        "Please Archive this one",
+    )
+    .await;
+    seed_and_enqueue_distinct(
+        db,
+        "surrogate-hang-2",
+        "WHERE subject CONTAINS 'Trash' ACTION DELETE",
+        "Please Trash this one",
+    )
+    .await;
+}
+
+#[tokio::test]
+async fn a_hung_mutation_times_out_and_is_retried_never_completed() {
+    let (db, _dir) = DatabaseEngine::connect_ephemeral().await.expect("db");
+    seed_and_enqueue(
+        &db,
+        "WHERE subject CONTAINS 'Archive' ACTION MOVE TO 'Archive'",
+        "Please Archive",
+    )
+    .await;
+
+    // Paused only AFTER setup, and this pass touches the database exactly
+    // ONCE (a single `get_message`) before the hang -- mixing
+    // `tokio::time::pause`'s auto-advance-on-idle with real (non-timer)
+    // background I/O is only safe when there is a single, already-in-flight
+    // real operation for the auto-advance to race past; see the sibling
+    // two-pass test below for why a SECOND real DB operation in the same
+    // paused pass is deliberately avoided.
+    tokio::time::pause();
+    let env = HangEnv::new(usize::MAX);
+    let (_shutdown_ctrl, mut shutdown) = no_shutdown();
+    let summary = execute_pending_mutations(&db, &env, 10, &mut shutdown).await;
+    // Resume real time before any further DB access: `tokio::time::pause`'s
+    // auto-advance-on-idle only plays safely with the ONE real DB operation
+    // this pass performs (the `get_message` before the hang); a second real
+    // query made while still paused is what made the sibling two-pass test
+    // flaky before it was split across a resume point.
+    tokio::time::resume();
+
+    // The hung item is NEVER marked completed -- it is a retryable timeout,
+    // exactly like any other transient remote failure.
+    assert_eq!(summary.completed, 0);
+    assert_eq!(
+        summary.retried, 1,
+        "the hung item must be retried, not lost"
+    );
+    assert_eq!(summary.failed, 0);
+
+    let pending = db.list_pending_mutations(10).await.unwrap();
+    assert_eq!(pending.len(), 1, "the hung item stays pending for a retry");
+    assert_eq!(
+        pending[0].retry_count, 1,
+        "the timeout bumps the retry count exactly like any transient failure"
+    );
+    assert!(
+        env.inner.backend.applied_mutations().is_empty(),
+        "the hung item never genuinely reached the backend"
+    );
+}
+
+#[tokio::test]
+async fn a_timed_out_item_does_not_block_a_later_item_in_a_subsequent_pass() {
+    let (db, _dir) = DatabaseEngine::connect_ephemeral().await.expect("db");
+    seed_two_move_mutations(&db).await;
+
+    // First pass: bound to the FIRST (created-earliest) mutation only, so
+    // this pass's real DB work is a single `get_message` call before the
+    // hang, same as the sibling single-item timeout test. Real threaded
+    // SQLite I/O racing `tokio::time::pause`'s auto-advance is only
+    // exercised once per paused pass here -- see that test's comment for
+    // why a second real DB call in the SAME paused pass is unreliable.
+    tokio::time::pause();
+    let env = HangEnv::new(usize::MAX);
+    let (_shutdown_ctrl, mut shutdown) = no_shutdown();
+    let first_pass = execute_pending_mutations(&db, &env, 1, &mut shutdown).await;
+    assert_eq!(first_pass.completed, 0);
+    assert_eq!(first_pass.retried, 1, "the hung item must be retried");
+
+    // Second pass runs under ordinary real time against a normal
+    // (non-hanging) env, standing in for the outbox worker's next 5-second
+    // tick: the item that timed out in the first pass did not jam the queue
+    // -- both it (now against a healthy backend) and the item queued behind
+    // it are processed and genuinely complete.
+    tokio::time::resume();
+    let normal_env = MockEnv::new();
+    let (_shutdown_ctrl2, mut shutdown2) = no_shutdown();
+    let second_pass = execute_pending_mutations(&db, &normal_env, 10, &mut shutdown2).await;
+    assert_eq!(
+        second_pass.completed, 2,
+        "both the previously-timed-out item and the one behind it complete"
+    );
+    assert_eq!(second_pass.failed, 0);
+
+    let applied = normal_env.backend.applied_mutations();
+    assert_eq!(
+        applied.len(),
+        2,
+        "both mutations genuinely reached the backend"
+    );
+
+    // Neither pass ever fabricated a completion: the queue is now empty
+    // because both items genuinely succeeded, not because either was
+    // dropped or silently marked done while still hung.
+    assert!(db.list_pending_mutations(10).await.unwrap().is_empty());
+}
+
+#[tokio::test]
+async fn shutdown_signal_interrupts_a_hung_item_instead_of_waiting_out_the_timeout() {
+    let (db, _dir) = DatabaseEngine::connect_ephemeral().await.expect("db");
+    seed_and_enqueue(
+        &db,
+        "WHERE subject CONTAINS 'Archive' ACTION MOVE TO 'Archive'",
+        "Please Archive",
+    )
+    .await;
+
+    // Paused only after setup -- see the comment on the sibling timeout test
+    // for why pausing from the start would spuriously fail the DB pool.
+    tokio::time::pause();
+
+    // Hangs on every call -- the shutdown signal, not the timeout, must be
+    // what unblocks this drain pass.
+    let env = HangEnv::new(usize::MAX);
+    let (controller, mut shutdown) = ShutdownController::new(Arc::new(EventBus::new()));
+    tokio::spawn(async move {
+        tokio::time::sleep(std::time::Duration::from_millis(50)).await;
+        controller.trigger();
+    });
+
+    let summary = execute_pending_mutations(&db, &env, 10, &mut shutdown).await;
+    tokio::time::resume();
+
+    // The drain pass stopped on shutdown, well before the per-item timeout
+    // would have elapsed -- the hung item is untouched: not completed, not
+    // even retried, since its true disposition is still unknown.
+    assert_eq!(summary.completed, 0);
+    assert_eq!(summary.retried, 0);
+    assert_eq!(summary.failed, 0);
+
+    let pending = db.list_pending_mutations(10).await.unwrap();
+    assert_eq!(pending.len(), 1);
+    assert_eq!(
+        pending[0].retry_count, 0,
+        "shutdown must leave the interrupted item's disposition untouched"
+    );
+}