Skip to main content

max / goingson

9.3 KB · 252 lines History Blame Raw
1 //! The outbox drainer: what actually sends a queued message.
2 //!
3 //! <!-- wiki: quasi-overview -->
4 //!
5 //! GoingsOn uses an outbox explicitly, and send-later is a feature of it
6 //! rather than a setting bolted beside it.
7 //!
8 //! # Why the app has one at all
9 //!
10 //! `send_email` is async. A described route handler is synchronous, by
11 //! `quasi_router`'s Decision 6, which exists so egui and a terminal need no
12 //! runtime. So a described compose screen cannot send; it can only write, and
13 //! queueing is a write.
14 //!
15 //! That is the constraint. What makes an outbox better than the thing it works
16 //! around is what it is once it exists: a message you can see before it goes
17 //! and stop, a message you can schedule, and a send that survives being
18 //! offline instead of failing at the instant somebody pressed the button. The
19 //! distinction is Send versus Queue.
20 //!
21 //! # Nothing here is described, and that is the design
22 //!
23 //! The description says "queue this". What drains the queue is not a screen and
24 //! has no address: it is a pass over the store, on the app's own clock. That
25 //! division is why the outbox answers the async problem rather than moving it —
26 //! the async lives out here, where there has always been a runtime.
27 //!
28 //! # The shape
29 //!
30 //! A tokio interval with a cancel token, the same shape as
31 //! `email_sync_scheduler` and the notification pass, so a reader who has read
32 //! either of those knows how this one starts, stops and survives a failing
33 //! tick.
34 //!
35 //! # What a failed send does
36 //!
37 //! Stamps the error, counts the attempt, and leaves the message queued. It does
38 //! not delete, and it does not stop trying: an SMTP server that is down at
39 //! 09:00 is usually up at 09:20, and a message silently dropped for that is
40 //! worse than one still sitting in the outbox with a reason on it.
41 //!
42 //! What it does do is back off, on the attempt count, so a message that can
43 //! never go (a bad address, a rejected credential) stops occupying every tick.
44 //! `send_attempts` is the whole of that state, and it resets when somebody
45 //! takes the message back out of the outbox to edit it.
46
47 use std::sync::Arc;
48
49 use tauri::Manager;
50 use tokio::time::{Duration, interval};
51 use tokio_util::sync::CancellationToken;
52 use tracing::{debug, error, info};
53
54 use crate::commands::SendEmailInput;
55 use crate::state::{AppState, DESKTOP_USER_ID};
56
57 /// How often the drainer wakes.
58 ///
59 /// A minute matches `email_sync_scheduler`. It is also what makes "queue"
60 /// acceptable as the only way to send: a message leaves within a minute of
61 /// being written, which is not immediate and is not a wait anybody watches.
62 const CHECK_INTERVAL_SECS: u64 = 60;
63
64 /// How many ticks a message waits after its nth failure.
65 ///
66 /// Doubling, capped. A message that has failed once is retried on the next
67 /// wake; one that has failed six times is retried every half hour or so. The
68 /// cap matters more than the curve: without it a message that failed twenty
69 /// times would effectively never be retried again, which is a silent drop
70 /// wearing a backoff's clothes.
71 const MAX_BACKOFF_TICKS: u32 = 32;
72
73 /// Whether this tick should try a message that has already failed `attempts`
74 /// times, given how many ticks have passed.
75 ///
76 /// Pure, so the curve is testable without a clock or a database.
77 #[must_use]
78 pub fn due_on_tick(attempts: i32, tick: u64) -> bool {
79 if attempts <= 0 {
80 return true;
81 }
82 let every = u64::from(
83 2u32.saturating_pow(u32::try_from(attempts).unwrap_or(u32::MAX).min(16))
84 .min(MAX_BACKOFF_TICKS),
85 );
86 tick.is_multiple_of(every)
87 }
88
89 /// Start the drainer.
90 ///
91 /// Runs until cancelled. A tick that fails logs and returns; the next one tries
92 /// again, which is the same contract the sync scheduler makes.
93 pub async fn start_outbox_drainer(app: tauri::AppHandle, cancel: CancellationToken) {
94 let mut check_interval = interval(Duration::from_secs(CHECK_INTERVAL_SECS));
95 let mut tick: u64 = 0;
96
97 info!("Outbox drainer started (checking every {CHECK_INTERVAL_SECS} seconds)");
98
99 loop {
100 tokio::select! {
101 () = cancel.cancelled() => {
102 info!("Outbox drainer shutting down");
103 break;
104 }
105 _ = check_interval.tick() => {}
106 }
107 tick = tick.wrapping_add(1);
108
109 // None during startup, before `AppState` is managed. The next tick
110 // picks it up, same as the sync scheduler.
111 let Some(state) = app.try_state::<Arc<AppState>>() else {
112 debug!("Outbox drainer: state not ready yet");
113 continue;
114 };
115 let state: Arc<AppState> = state.inner().clone();
116
117 drain_once(&state, tick).await;
118 }
119 }
120
121 /// One pass over the outbox.
122 ///
123 /// Split from the loop so a test can run a pass without a Tauri handle or a
124 /// minute of waiting.
125 pub async fn drain_once(state: &Arc<AppState>, tick: u64) {
126 let now = chrono::Utc::now();
127 let due = match state.emails.list_due(DESKTOP_USER_ID, now) {
128 Ok(due) => due,
129 Err(error) => {
130 error!("Outbox drainer: could not read the outbox: {error}");
131 return;
132 }
133 };
134
135 for email in due {
136 if !due_on_tick(email.send_attempts, tick) {
137 continue;
138 }
139
140 // A queued draft with no account cannot be sent and never will be: the
141 // account is what holds the SMTP credentials. Stamped rather than
142 // retried, so it sits in the outbox saying why instead of failing
143 // silently once a minute forever.
144 let Some(account_id) = email.draft_account_id else {
145 if let Err(error) = state.emails.record_send_failure(
146 email.id,
147 DESKTOP_USER_ID,
148 "No account chosen, so there is nothing to send it from.",
149 ) {
150 error!("Outbox drainer: could not record the failure: {error}");
151 }
152 continue;
153 };
154
155 // The blobs, not the paths the files were picked from. A queued
156 // message may go hours after it was written and that file can have been
157 // moved, renamed or deleted; a blob is content-addressed under
158 // `<data_dir>/blobs` and `blob_gc` keeps it while a row references it.
159 //
160 // The row's `filename` is what the recipient sees, because a blob is
161 // named by its hash.
162 let attachments = match state.attachments.list_for_email(email.id, DESKTOP_USER_ID) {
163 Ok(files) => files,
164 Err(error) => {
165 error!(
166 "Outbox drainer: could not read {}'s files: {error}",
167 email.id
168 );
169 continue;
170 }
171 };
172 let attachment_paths = attachments
173 .iter()
174 .map(|file| {
175 crate::commands::attachment::blob_path(&state.data_dir, &file.blob_hash)
176 .to_string_lossy()
177 .into_owned()
178 })
179 .collect();
180
181 let input = SendEmailInput {
182 account_id,
183 to_address: email.to.clone(),
184 cc_address: email.cc_address.clone(),
185 bcc_address: email.bcc_address.clone(),
186 subject: email.subject.clone(),
187 body: email.body.clone(),
188 project_id: email.project_id,
189 in_reply_to: email.in_reply_to.clone(),
190 references: None,
191 thread_id: email.thread_id.clone(),
192 attachment_paths,
193 };
194
195 match crate::commands::send_email_inner(state, input).await {
196 Ok(_) => {
197 // The queued draft has served its purpose and the send wrote
198 // its own copy, which is what `send_email_draft` does too.
199 if let Err(error) = state.emails.delete(email.id, DESKTOP_USER_ID) {
200 error!(
201 "Outbox drainer: sent {} but could not clear it: {error}",
202 email.id
203 );
204 } else {
205 info!("Outbox drainer: sent {}", email.id);
206 }
207 }
208 Err(error) => {
209 let said = error.to_string();
210 debug!("Outbox drainer: {} did not go: {said}", email.id);
211 if let Err(error) =
212 state
213 .emails
214 .record_send_failure(email.id, DESKTOP_USER_ID, &said)
215 {
216 error!("Outbox drainer: could not record the failure: {error}");
217 }
218 }
219 }
220 }
221 }
222
223 #[cfg(test)]
224 mod tests {
225 use super::*;
226
227 #[test]
228 fn a_message_that_has_never_failed_goes_on_the_next_tick() {
229 assert!(due_on_tick(0, 1));
230 assert!(due_on_tick(0, 7));
231 }
232
233 #[test]
234 fn a_failed_message_backs_off_and_the_backoff_is_capped() {
235 // Once failed: every other tick.
236 assert!(due_on_tick(1, 2));
237 assert!(!due_on_tick(1, 3));
238
239 // The cap is the point. Without it, twenty failures is a silent drop
240 // wearing a backoff's clothes.
241 let every_at_cap: Vec<u64> = (1..=64).filter(|t| due_on_tick(20, *t)).collect();
242 assert_eq!(every_at_cap, vec![32, 64]);
243 }
244
245 #[test]
246 fn a_wild_attempt_count_cannot_panic_the_drainer() {
247 // `send_attempts` is a database column and this is arithmetic on it.
248 assert!(due_on_tick(i32::MAX, 32));
249 assert!(due_on_tick(-1, 1), "negative reads as never failed");
250 }
251 }
252