randogoth/smolboid
1
0
Fork 0
Code Issues Pull requests Projects Releases 4 Packages Wiki Activity Actions

feat: read the requests tier, drain its backlog once, and survive a bad reply

Browse source 
The daemon fetched requests-tier mail and then never looked at that folder, so
an unknown sender's mail sat in the local store unseen and undeleted. poll now
reads both tiers and deletes from both; delete_local removes by id from the
inbox table whatever the tier, so nothing can loop.

Dropped the `summary.stored == 0` early return. It meant a row left behind by
an error path on one cycle was only revisited if something new arrived later.
Emptiness is now decided from the rows actually in hand.

A live mail.db holds every requests-tier mail it ever received, so the first
poll of a version that reads them would answer every stranger in that history
at once. drain_backlog discards that backlog once and records it in state.
Not a received_at watermark: that timestamp is the server's, and clock skew
would silently discard real mail. It drains only the requests rows, so inbox
mail in the same cycle is still processed.

send_replies no longer aborts the batch on the first failure. It returned Err
on one unreachable recipient, dropping every later reply — tolerable when
replies only went to admitted authors, not once they go to arbitrary servers
and an author's accept token can be queued behind a host that is down. It now
counts failures and reports them in the poll summary.

The per-poll network cap is deliberately not here: the dials it bounds live in
the handlers, so it lands with them.
This commit is contained in:
randogoth 2026-10-14 20:11:01 +03:00
parent 2d2b42dd6f
commit 77d822cebe
1 changed files with 90 additions and 17 deletions
Download patch file Download diff file Expand all files Collapse all files

107
daemon/src/mailloop.rs
View file

@ -161,20 +161,43 @@ pub struct Daemon {
impl Daemon {
/// One poll cycle. Returns a human summary line.
pub fn poll(&self) -> Result<String, String> {
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<Option<usize>, 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
}
}