diff --git a/crates/nunciod/Cargo.toml b/crates/nunciod/Cargo.toml index d1ee17f..060404f 100644 --- a/crates/nunciod/Cargo.toml +++ b/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"] } diff --git a/crates/nunciod/src/main.rs b/crates/nunciod/src/main.rs index 29bb348..fcdc090 100644 --- a/crates/nunciod/src/main.rs +++ b/crates/nunciod/src/main.rs @@ -153,6 +153,7 @@ async fn main() -> Result<(), Box> { 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> { 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 { diff --git a/crates/nunciod/src/outbox.rs b/crates/nunciod/src/outbox.rs index 062c38a..0d5bb8b 100644 --- a/crates/nunciod/src/outbox.rs +++ b/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), diff --git a/crates/nunciod/tests/outbox_executor_e2e_test.rs b/crates/nunciod/tests/outbox_executor_e2e_test.rs index d88509a..d6fd97f 100644 --- a/crates/nunciod/tests/outbox_executor_e2e_test.rs +++ b/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, 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, 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 { + 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" + ); +}