forge
ExploreLog in

Tahinli/forge

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