diff --git a/daemon/src/mailloop.rs b/daemon/src/mailloop.rs index fde3b63..90e5d44 100644 --- a/daemon/src/mailloop.rs +++ b/daemon/src/mailloop.rs @@ -161,20 +161,43 @@ pub struct Daemon { impl Daemon { /// One poll cycle. Returns a human summary line. pub fn poll(&self) -> Result { - let summary = client::fetch(&self.store, &self.account, false, false, self.cfg.timeout) + client::fetch(&self.store, &self.account, false, false, self.cfg.timeout) .map_err(|e| format!("fetch: {e}"))?; - if summary.stored == 0 { - return Ok("no new mail".into()); + + // Both tiers. A requests-tier row is where an unknown sender's mail + // lands, and the daemon never read that folder before, so those rows + // accumulated unseen; they are now handled and deleted like any other. + // + // Deliberately not keyed off `summary.stored`: a row left behind by an + // error path on an earlier cycle would then never be revisited. + let mut mail = self + .store + .mail("inbox") + .map_err(|e| format!("inbox: {e}"))?; + let requests = self + .store + .mail("requests") + .map_err(|e| format!("requests: {e}"))?; + + // Draining discards only the requests backlog; inbox mail fetched in + // the same cycle is still processed, so no author waits a cycle for a + // one-time cleanup that has nothing to do with them. + let drained = self.drain_backlog(&requests)?; + if drained.is_none() { + mail.extend(requests); } + if mail.is_empty() { + return Ok(match drained { + Some(n) => format!("discarded {n} requests-tier mails from before self-service"), + None => "no new mail".into(), + }); + } + let mut applied = 0; let mut dropped = 0; let mut replies: Vec<(String, String, String)> = Vec::new(); // (address, subject, body) - let inbox = self - .store - .mail("inbox") - .map_err(|e| format!("inbox: {e}"))?; - for stored in &inbox { + for stored in &mail { let outcome = self.process(stored.id, &stored.envelope, &mut replies); let _ = self.store.delete_local("inbox", &stored.id); match outcome { @@ -183,13 +206,48 @@ impl Daemon { Err(e) => eprintln!("mail: {e}"), } } - self.send_replies(replies)?; + let failed = self.send_replies(replies); Ok(format!( - "{} applied, {} dropped", - applied, dropped + "{applied} applied, {dropped} dropped{}{}", + match drained { + Some(n) => format!(", {n} backlog discarded"), + None => String::new(), + }, + if failed > 0 { format!(", {failed} replies undeliverable") } else { String::new() } )) } + /// Discards the requests-tier backlog once, the first time a version that + /// reads that folder runs. + /// + /// A live mail.db holds every requests-tier mail it ever received, because + /// nothing deleted them. Without this the first poll after an upgrade + /// would answer every stranger in that history at once, which is both a + /// surprise to them and a burst of outbound mail from us. + /// + /// A received_at watermark would be the wrong tool: that timestamp is the + /// server's, and any clock skew would silently discard real mail. + fn drain_backlog( + &self, + requests: &[fumi_core::store::Stored], + ) -> Result, String> { + { + let state = self.state.lock().unwrap(); + if state.drained { + return Ok(None); + } + } + for stored in requests { + let _ = self.store.delete_local("inbox", &stored.id); + } + let mut state = self.state.lock().unwrap(); + state.drained = true; + state + .save(&self.cfg.data.join("state.json")) + .map_err(|e| format!("state: {e}"))?; + Ok(Some(requests.len())) + } + /// Processes one stored envelope end to end. Ok(true) when the feed or /// state changed. fn process( @@ -400,10 +458,19 @@ impl Daemon { )); } - fn send_replies(&self, replies: Vec<(String, String, String)>) -> Result<(), String> { + /// Sends each queued reply, returning how many could not be delivered. + fn send_replies(&self, replies: Vec<(String, String, String)>) -> usize { + let mut failed = 0; for (address, subject, body) in replies { - let addr = Address::parse(&address).map_err(|e| format!("reply address: {e}"))?; - client::send( + let addr = match Address::parse(&address) { + Ok(addr) => addr, + Err(e) => { + eprintln!("reply address {address}: {e}"); + failed += 1; + continue; + } + }; + if let Err(e) = client::send( &self.store, &self.account, &client::SendDraft { @@ -416,10 +483,16 @@ impl Daemon { no_pad: true, }, self.cfg.timeout, - ) - .map_err(|e: FumiError| format!("reply to {address}: {e}"))?; + ) { + // One unreachable recipient must not swallow the rest of the + // batch: replies now go to arbitrary servers, and an author's + // accept token could be queued behind a host that is down. + let e: FumiError = e; + eprintln!("reply to {address}: {e}"); + failed += 1; + } } - Ok(()) + failed } }