forge
ExploreLog in

Tahinli/forge

public
0 watching
★ 0
10.3 KiB
1//! The mail worker. It leases due outbox rows and sends them. A transient failure is retried on
2//! the schedule in `forge_core::mail::retry_delay`; a permanent one, or the last failed try, gives
3//! up. It sleeps until the next due time, or until `AppState::mail_waker` is notified (a message
4//! was queued, or the settings changed). While mail is off it sends nothing and keeps the
5//! messages queued.
6
7use std::time::Duration;
8
9use forge_core::mail::{MailError, OutboxEntry, OutgoingMail, retry_delay};
10use forge_core::time::{now_stamp, stamp_after};
11use time::OffsetDateTime;
12use time::format_description::well_known::Rfc3339;
13
14use crate::state::AppState;
15
16/// The longest the worker sleeps when nothing wakes it.
17const POLL: Duration = Duration::from_secs(60);
18/// The shortest sleep, so a due time that has already passed does not spin the loop.
19const MIN_SLEEP: Duration = Duration::from_secs(1);
20/// Messages leased per pass.
21const BATCH: u32 = 20;
22/// How long a leased row stays with this pass before another may take it.
23const LEASE: Duration = Duration::from_secs(5 * 60);
24
25/// Starts the worker task. It runs until the process exits.
26pub fn spawn(state: AppState) {
27 tokio::spawn(run(state));
28}
29
30async fn run(state: AppState) {
31 loop {
32 let nap = match tick(&state).await {
33 Ok(nap) => nap,
34 Err(e) => {
35 tracing::error!(error = %e, "mail: could not run a pass");
36 POLL
37 }
38 };
39 tokio::select! {
40 _ = tokio::time::sleep(nap) => {}
41 _ = state.mail_waker.notified() => {}
42 }
43 }
44}
45
46/// One pass: sends what is due when mail is on. Returns how long to sleep before the next one.
47async fn tick(state: &AppState) -> forge_core::Result<Duration> {
48 let settings = state.store.mail_settings().await?;
49 if !settings.enabled {
50 return Ok(POLL);
51 }
52 let now = now_stamp();
53 let lease_until = stamp_after(LEASE);
54 for entry in state.store.lease_mail(&now, &lease_until, BATCH).await? {
55 deliver(state, &entry).await;
56 }
57 let next = state.store.next_mail_due().await?;
58 Ok(sleep_for(next.as_deref(), OffsetDateTime::now_utc()))
59}
60
61/// Sends one leased message and records the outcome.
62async fn deliver(state: &AppState, entry: &OutboxEntry) {
63 let mail = OutgoingMail {
64 to: entry.to_addr.clone(),
65 subject: entry.subject.clone(),
66 body: entry.body.clone(),
67 message_id: entry.message_id.clone(),
68 in_reply_to: entry.in_reply_to.clone(),
69 };
70 let recorded = match state.mailer.send(&mail).await {
71 Ok(()) => state.store.mark_mail_sent(&entry.id, &now_stamp()).await,
72 Err(MailError::Permanent(error)) => {
73 tracing::warn!(id = %entry.id, error = %error, "mail: permanent failure, giving up");
74 state.store.mark_mail_failed(&entry.id, &error, None).await
75 }
76 Err(MailError::Transient(error)) => {
77 // `attempts` counts the tries before this one; this try makes one more.
78 let next = retry_delay(entry.attempts + 1).map(stamp_after);
79 if next.is_none() {
80 tracing::warn!(id = %entry.id, error = %error, "mail: last try failed, giving up");
81 } else {
82 tracing::info!(id = %entry.id, error = %error, "mail: try failed, will retry");
83 }
84 state
85 .store
86 .mark_mail_failed(&entry.id, &error, next.as_deref())
87 .await
88 }
89 };
90 if let Err(e) = recorded {
91 tracing::error!(id = %entry.id, error = %e, "mail: could not record the outcome");
92 }
93}
94
95/// How long the worker sleeps: until `next_due` (an RFC 3339 stamp), clamped to
96/// `MIN_SLEEP..=POLL`. `None` or an unparsable stamp gives `POLL`.
97pub fn sleep_for(next_due: Option<&str>, now: OffsetDateTime) -> Duration {
98 let Some(due) = next_due.and_then(|s| OffsetDateTime::parse(s, &Rfc3339).ok()) else {
99 return POLL;
100 };
101 let until = Duration::try_from(due - now).unwrap_or(Duration::ZERO);
102 until.clamp(MIN_SLEEP, POLL)
103}
104
105#[cfg(test)]
106mod tests {
107 use std::sync::Arc;
108
109 use forge_core::mail::{MailSettingsUpdate, OutboxStatus, Security};
110 use forge_core::time::{now_stamp, stamp_after};
111
112 use super::*;
113 use crate::mail::{FakeMailer, enqueue};
114 use crate::test_support::test_state;
115
116 async fn enabled_state(dir: &std::path::Path) -> (AppState, Arc<FakeMailer>) {
117 let mut state = test_state(dir).await;
118 let fake = Arc::new(FakeMailer::default());
119 state.mailer = fake.clone();
120 state
121 .store
122 .update_mail_settings(&MailSettingsUpdate {
123 host: "smtp.example",
124 port: 587,
125 security: Security::StartTls,
126 username: "",
127 from_addr: "forge@example.com",
128 from_name: "forge",
129 enabled: true,
130 password: None,
131 })
132 .await
133 .expect("settings");
134 (state, fake)
135 }
136
137 #[test]
138 fn sleep_stops_at_the_next_due_time_and_is_clamped() {
139 use time::macros::datetime;
140
141 let now = datetime!(2026-10-15 12:00:00 UTC);
142 let stamp = |t: OffsetDateTime| t.format(&Rfc3339).expect("stamp");
143 assert_eq!(sleep_for(None, now), POLL);
144 let in_10s = stamp(now + time::Duration::seconds(10));
145 assert_eq!(sleep_for(Some(&in_10s), now), Duration::from_secs(10));
146 let in_1h = stamp(now + time::Duration::hours(1));
147 assert_eq!(sleep_for(Some(&in_1h), now), POLL);
148 let past = stamp(now - time::Duration::seconds(5));
149 assert_eq!(sleep_for(Some(&past), now), MIN_SLEEP);
150 assert_eq!(sleep_for(Some("soon"), now), POLL);
151 }
152
153 #[tokio::test]
154 async fn a_queued_message_is_sent_once() {
155 let dir = tempfile::tempdir().expect("tempdir");
156 let (state, fake) = enabled_state(dir.path()).await;
157 assert!(
158 enqueue(&state, "reset:1", "a@example.com", "Hi", "body")
159 .await
160 .expect("queued")
161 );
162 // A second enqueue with the same key is a no-op.
163 assert!(
164 !enqueue(&state, "reset:1", "a@example.com", "Hi", "body")
165 .await
166 .expect("again")
167 );
168
169 tick(&state).await.expect("pass");
170 tick(&state).await.expect("second pass");
171 let sent = fake.sent();
172 assert_eq!(sent.len(), 1);
173 assert_eq!(sent[0].to, "a@example.com");
174 let row = state.store.recent_mail(10).await.expect("rows");
175 assert_eq!(row.len(), 1);
176 assert_eq!(row[0].status(), OutboxStatus::Sent);
177 }
178
179 #[tokio::test]
180 async fn nothing_is_sent_while_mail_is_off() {
181 let dir = tempfile::tempdir().expect("tempdir");
182 let (state, fake) = enabled_state(dir.path()).await;
183 state
184 .store
185 .update_mail_settings(&MailSettingsUpdate {
186 host: "smtp.example",
187 port: 587,
188 security: Security::StartTls,
189 username: "",
190 from_addr: "forge@example.com",
191 from_name: "forge",
192 enabled: false,
193 password: None,
194 })
195 .await
196 .expect("off");
197 enqueue(&state, "k", "a@example.com", "Hi", "body")
198 .await
199 .expect("queued");
200 assert_eq!(tick(&state).await.expect("pass"), POLL);
201 assert!(fake.sent().is_empty());
202 let row = state.store.recent_mail(10).await.expect("rows");
203 assert_eq!(row[0].status(), OutboxStatus::Queued);
204 assert_eq!(row[0].attempts, 0, "waiting costs no attempts");
205 }
206
207 #[tokio::test]
208 async fn a_transient_failure_is_retried_later_and_a_permanent_one_gives_up() {
209 let dir = tempfile::tempdir().expect("tempdir");
210 let (state, fake) = enabled_state(dir.path()).await;
211 enqueue(&state, "retry", "a@example.com", "Hi", "body")
212 .await
213 .expect("queued");
214 enqueue(&state, "bounce", "b@example.com", "Hi", "body")
215 .await
216 .expect("queued");
217 fake.fail_next(MailError::Transient("timeout".into()));
218 fake.fail_next(MailError::Permanent("550 no such user".into()));
219
220 tick(&state).await.expect("pass");
221 let rows = state.store.recent_mail(10).await.expect("rows");
222 let by_key = |to: &str| rows.iter().find(|r| r.to_addr == to).expect("row").clone();
223
224 let retried = by_key("a@example.com");
225 assert_eq!(retried.status(), OutboxStatus::Queued);
226 assert_eq!(retried.attempts, 1);
227 assert_eq!(retried.last_error.as_deref(), Some("timeout"));
228 let due = retried.next_attempt_at.expect("scheduled");
229 assert!(due > now_stamp(), "the retry waits about a minute");
230 assert!(due <= stamp_after(Duration::from_secs(120)));
231
232 let bounced = by_key("b@example.com");
233 assert_eq!(bounced.status(), OutboxStatus::GaveUp);
234 assert_eq!(bounced.attempts, 1);
235 assert_eq!(bounced.last_error.as_deref(), Some("550 no such user"));
236 assert_eq!(fake.sent().len(), 0, "nothing reached the transport");
237 }
238
239 #[tokio::test]
240 async fn the_last_try_gives_up_and_an_earlier_one_reschedules() {
241 let dir = tempfile::tempdir().expect("tempdir");
242 let (state, fake) = enabled_state(dir.path()).await;
243 enqueue(&state, "stubborn", "a@example.com", "Hi", "body")
244 .await
245 .expect("queued");
246 let row = state.store.recent_mail(10).await.expect("rows").remove(0);
247 fake.fail_next(MailError::Transient("timeout".into()));
248 fake.fail_next(MailError::Transient("timeout".into()));
249
250 // The row has failed MAX_ATTEMPTS - 1 times already: this try is the last one.
251 let last = OutboxEntry {
252 attempts: forge_core::mail::MAX_ATTEMPTS - 1,
253 ..row.clone()
254 };
255 deliver(&state, &last).await;
256 let after = state.store.recent_mail(10).await.expect("rows").remove(0);
257 assert_eq!(after.status(), OutboxStatus::GaveUp);
258 assert_eq!(after.last_error.as_deref(), Some("timeout"));
259
260 // The same failure on an early try is rescheduled instead.
261 let early = OutboxEntry { attempts: 1, ..row };
262 deliver(&state, &early).await;
263 let after = state.store.recent_mail(10).await.expect("rows").remove(0);
264 assert_eq!(after.status(), OutboxStatus::Queued);
265 }
266}