//! The outbox drainer: what actually sends a queued message. //! //! //! //! GoingsOn uses an outbox explicitly, and send-later is a feature of it //! rather than a setting bolted beside it. //! //! # Why the app has one at all //! //! `send_email` is async. A described route handler is synchronous, by //! `quasi_router`'s Decision 6, which exists so egui and a terminal need no //! runtime. So a described compose screen cannot send; it can only write, and //! queueing is a write. //! //! That is the constraint. What makes an outbox better than the thing it works //! around is what it is once it exists: a message you can see before it goes //! and stop, a message you can schedule, and a send that survives being //! offline instead of failing at the instant somebody pressed the button. The //! distinction is Send versus Queue. //! //! # Nothing here is described, and that is the design //! //! The description says "queue this". What drains the queue is not a screen and //! has no address: it is a pass over the store, on the app's own clock. That //! division is why the outbox answers the async problem rather than moving it — //! the async lives out here, where there has always been a runtime. //! //! # The shape //! //! A tokio interval with a cancel token, the same shape as //! `email_sync_scheduler` and the notification pass, so a reader who has read //! either of those knows how this one starts, stops and survives a failing //! tick. //! //! # What a failed send does //! //! Stamps the error, counts the attempt, and leaves the message queued. It does //! not delete, and it does not stop trying: an SMTP server that is down at //! 09:00 is usually up at 09:20, and a message silently dropped for that is //! worse than one still sitting in the outbox with a reason on it. //! //! What it does do is back off, on the attempt count, so a message that can //! never go (a bad address, a rejected credential) stops occupying every tick. //! `send_attempts` is the whole of that state, and it resets when somebody //! takes the message back out of the outbox to edit it. use std::sync::Arc; use tauri::Manager; use tokio::time::{Duration, interval}; use tokio_util::sync::CancellationToken; use tracing::{debug, error, info}; use crate::commands::SendEmailInput; use crate::state::{AppState, DESKTOP_USER_ID}; /// How often the drainer wakes. /// /// A minute matches `email_sync_scheduler`. It is also what makes "queue" /// acceptable as the only way to send: a message leaves within a minute of /// being written, which is not immediate and is not a wait anybody watches. const CHECK_INTERVAL_SECS: u64 = 60; /// How many ticks a message waits after its nth failure. /// /// Doubling, capped. A message that has failed once is retried on the next /// wake; one that has failed six times is retried every half hour or so. The /// cap matters more than the curve: without it a message that failed twenty /// times would effectively never be retried again, which is a silent drop /// wearing a backoff's clothes. const MAX_BACKOFF_TICKS: u32 = 32; /// Whether this tick should try a message that has already failed `attempts` /// times, given how many ticks have passed. /// /// Pure, so the curve is testable without a clock or a database. #[must_use] pub fn due_on_tick(attempts: i32, tick: u64) -> bool { if attempts <= 0 { return true; } let every = u64::from( 2u32.saturating_pow(u32::try_from(attempts).unwrap_or(u32::MAX).min(16)) .min(MAX_BACKOFF_TICKS), ); tick.is_multiple_of(every) } /// Start the drainer. /// /// Runs until cancelled. A tick that fails logs and returns; the next one tries /// again, which is the same contract the sync scheduler makes. pub async fn start_outbox_drainer(app: tauri::AppHandle, cancel: CancellationToken) { let mut check_interval = interval(Duration::from_secs(CHECK_INTERVAL_SECS)); let mut tick: u64 = 0; info!("Outbox drainer started (checking every {CHECK_INTERVAL_SECS} seconds)"); loop { tokio::select! { () = cancel.cancelled() => { info!("Outbox drainer shutting down"); break; } _ = check_interval.tick() => {} } tick = tick.wrapping_add(1); // None during startup, before `AppState` is managed. The next tick // picks it up, same as the sync scheduler. let Some(state) = app.try_state::>() else { debug!("Outbox drainer: state not ready yet"); continue; }; let state: Arc = state.inner().clone(); drain_once(&state, tick).await; } } /// One pass over the outbox. /// /// Split from the loop so a test can run a pass without a Tauri handle or a /// minute of waiting. pub async fn drain_once(state: &Arc, tick: u64) { let now = chrono::Utc::now(); let due = match state.emails.list_due(DESKTOP_USER_ID, now) { Ok(due) => due, Err(error) => { error!("Outbox drainer: could not read the outbox: {error}"); return; } }; for email in due { if !due_on_tick(email.send_attempts, tick) { continue; } // A queued draft with no account cannot be sent and never will be: the // account is what holds the SMTP credentials. Stamped rather than // retried, so it sits in the outbox saying why instead of failing // silently once a minute forever. let Some(account_id) = email.draft_account_id else { if let Err(error) = state.emails.record_send_failure( email.id, DESKTOP_USER_ID, "No account chosen, so there is nothing to send it from.", ) { error!("Outbox drainer: could not record the failure: {error}"); } continue; }; // The blobs, not the paths the files were picked from. A queued // message may go hours after it was written and that file can have been // moved, renamed or deleted; a blob is content-addressed under // `/blobs` and `blob_gc` keeps it while a row references it. // // The row's `filename` is what the recipient sees, because a blob is // named by its hash. let attachments = match state.attachments.list_for_email(email.id, DESKTOP_USER_ID) { Ok(files) => files, Err(error) => { error!( "Outbox drainer: could not read {}'s files: {error}", email.id ); continue; } }; let attachment_paths = attachments .iter() .map(|file| { crate::commands::attachment::blob_path(&state.data_dir, &file.blob_hash) .to_string_lossy() .into_owned() }) .collect(); let input = SendEmailInput { account_id, to_address: email.to.clone(), cc_address: email.cc_address.clone(), bcc_address: email.bcc_address.clone(), subject: email.subject.clone(), body: email.body.clone(), project_id: email.project_id, in_reply_to: email.in_reply_to.clone(), references: None, thread_id: email.thread_id.clone(), attachment_paths, }; match crate::commands::send_email_inner(state, input).await { Ok(_) => { // The queued draft has served its purpose and the send wrote // its own copy, which is what `send_email_draft` does too. if let Err(error) = state.emails.delete(email.id, DESKTOP_USER_ID) { error!( "Outbox drainer: sent {} but could not clear it: {error}", email.id ); } else { info!("Outbox drainer: sent {}", email.id); } } Err(error) => { let said = error.to_string(); debug!("Outbox drainer: {} did not go: {said}", email.id); if let Err(error) = state .emails .record_send_failure(email.id, DESKTOP_USER_ID, &said) { error!("Outbox drainer: could not record the failure: {error}"); } } } } } #[cfg(test)] mod tests { use super::*; #[test] fn a_message_that_has_never_failed_goes_on_the_next_tick() { assert!(due_on_tick(0, 1)); assert!(due_on_tick(0, 7)); } #[test] fn a_failed_message_backs_off_and_the_backoff_is_capped() { // Once failed: every other tick. assert!(due_on_tick(1, 2)); assert!(!due_on_tick(1, 3)); // The cap is the point. Without it, twenty failures is a silent drop // wearing a backoff's clothes. let every_at_cap: Vec = (1..=64).filter(|t| due_on_tick(20, *t)).collect(); assert_eq!(every_at_cap, vec![32, 64]); } #[test] fn a_wild_attempt_count_cannot_panic_the_drainer() { // `send_attempts` is a database column and this is arithmetic on it. assert!(due_on_tick(i32::MAX, 32)); assert!(due_on_tick(-1, 1), "negative reads as never failed"); } }