smolboid/daemon/src/mailloop.rs

1186 lines
46 KiB
Rust

//! The mail loop: fetch from the daemon's mailbox, admit by ownership, apply
//! each mail to its feed, and send rate-limited error replies. The apply and
//! retention logic is pure and unit-tested; network glue stays thin.
use fumi_core::account::Account;
use fumi_core::address::Address;
use fumi_core::client;
use fumi_core::store::{now, Store};
use fumi_core::error::Error as FumiError;
use smolboid_core::feed::{Feed, MailLine, Record};
use smolboid_core::envelope;
use crate::config::Config;
use crate::dispatch::{self, Action, Command, Incoming, Request};
use crate::feedstore::FeedStore;
use crate::reason::Reason;
use crate::register;
use crate::state::{
global_allowed, payload_key, reply_allowed, FeedState, Global, Mode, PruneMode, State,
};
use std::sync::{Mutex, MutexGuard};
/// How many mails per cycle may make an outbound connection.
///
/// `poll` holds the state lock across each mail and a dial can block for
/// `timeout` seconds, so an unbounded batch of registrations would stall mail
/// processing for minutes. Anything over the budget is left in the store for
/// the next cycle rather than dropped.
const DIALS_PER_POLL: usize = 4;
/// What became of one mail, and the line the operator reads about it.
///
/// Every branch carries its own phrase so that `poll` can log all of them in
/// one place: a new branch cannot be added without saying what it did.
///
/// The daemon stays silent toward *senders* deliberately — a reply would
/// confirm whether an address exists, and that silence is a security
/// property. The operator's journal is a different audience and reaches
/// nobody else. Do not "fix" a quiet drop by answering the sender.
#[derive(Debug, PartialEq, Eq)]
enum Outcome {
/// The feed or state changed.
Applied(String),
/// A reply was queued, but nothing changed here.
Answered(String),
/// Nothing happened and the sender was told nothing.
Dropped(String),
/// Needed a dial the cycle could not afford. Left in the store.
Deferred,
}
impl Outcome {
fn line(&self) -> String {
match self {
Outcome::Applied(what) => format!("applied: {what}"),
Outcome::Answered(what) => format!("answered: {what}"),
Outcome::Dropped(why) => format!("dropped: {why}"),
Outcome::Deferred => "deferred: no dials left this cycle".to_string(),
}
}
}
/// A byte count as something a reader can picture. Deliberately coarse: the
/// exact figure helps nobody deciding whether to prune their feed.
fn human_size(bytes: usize) -> String {
match bytes {
b if b >= 1 << 20 => format!("{} MB", b / (1 << 20)),
b if b >= 1 << 10 => format!("{} KB", b / (1 << 10)),
b => format!("{b} bytes"),
}
}
/// Whether a request is one only an admin may run. Register and Help are not:
/// an admin sending either means it literally, and routing them through the
/// admin branch would answer with silence.
fn is_admin_command(request: &Request) -> bool {
matches!(
request,
Request::Admit(_) | Request::Block(_) | Request::Registration(_)
)
}
/// Everything needed to publish one record, before retention gets a say.
pub struct PendingRecord {
pub timestamp: String,
pub text: String,
pub repost: bool,
/// The Reply-To the mail carried, when the ordinary-send form was used.
pub reply_to: Option<String>,
pub mail: MailLine,
}
/// Failures that draw an error reply (PLAN.md, Management mail).
#[derive(Debug)]
pub enum ApplyError {
TooLong,
OverQuota(bool),
Collision(String),
FeedFull,
/// The author's Reply-To changed after the feed bound one: published
/// records would stop reconstructing (PLAN.md, smol:reply_to).
ReplyToChanged,
}
/// Applies a published record to the feed: length, quota, uniqueness, retention
/// (pruning in Fifo mode, rejecting in Off mode), then the append.
/// Returns the rendered size after appending, for the 90% warning.
pub fn apply_record(
feed: &mut Feed,
fs: &mut FeedState,
cfg: &Config,
record: PendingRecord,
now_unix: i64,
) -> Result<(), ApplyError> {
if record.text.chars().count() > cfg.max_post_chars {
return Err(ApplyError::TooLong);
}
State::roll_counters(fs, now_unix);
let (counter, limit) = if record.repost {
(&mut fs.reposts, cfg.reposts_per_day)
} else {
(&mut fs.posts, cfg.posts_per_day)
};
if *counter >= limit {
return Err(ApplyError::OverQuota(record.repost));
}
if feed.has_timestamp(&record.timestamp) {
// The signature pins the time, so a bump is impossible; the author
// resends a second later (post rule 2).
return Err(ApplyError::Collision(record.timestamp));
}
// The Reply-To must be the author's own address (key = sender), and the
// verifier derives it from `smol:author` plus the record's sender key. The
// first publishing mail adopts the author's own spelling of the address;
// a later change cannot be adopted without invalidating old records.
if let Some(reply_to) = &record.reply_to {
let Ok(addr) = Address::parse(reply_to) else {
return Err(ApplyError::ReplyToChanged);
};
if addr.identity != Some(record.mail.sender) || addr.short() != feed.author().unwrap_or_default() {
if feed.entries().is_empty() {
feed.set_meta("smol:author", &addr.short());
} else {
return Err(ApplyError::ReplyToChanged);
}
}
}
let entry = if record.repost {
Record::Repost {
timestamp: record.timestamp.clone(),
args: record.text.clone(),
}
} else {
Record::Post {
timestamp: record.timestamp.clone(),
text: record.text.clone(),
}
};
let rendered = entry.render();
let addition = rendered.len() + 1 + record.mail.render().len() + 1;
let mut size = feed.render().len() + addition;
if size > cfg.cap_bytes {
if fs.prune == PruneMode::Off {
return Err(ApplyError::FeedFull);
}
// FIFO: drop oldest records until the new one fits. If even an empty
// feed cannot hold it, the record simply exceeds the cap.
while size > cfg.cap_bytes {
let Some(oldest) = feed.entries().first().map(|e| e.record.timestamp().to_owned())
else {
return Err(ApplyError::FeedFull);
};
feed.delete(&oldest);
size = feed.render().len() + addition;
}
}
feed.append(entry, Some(record.mail));
*counter += 1;
// The 90% warning fires once per growth episode (PLAN.md, Retention).
let size = feed.render().len();
if size * 10 >= cfg.cap_bytes * 9 {
if !fs.warned {
fs.warned = true;
}
} else if fs.warned {
fs.warned = false;
}
Ok(())
}
/// Applies command fields to feed and state. Deletion failure is reported so
/// the caller can reply.
pub fn apply_commands(
feed: &mut Feed,
fs: &mut FeedState,
commands: &[Command],
) -> Result<usize, Reason> {
for command in commands {
match command {
Command::Nick(nick) => {
fs.nick = nick.clone();
feed.set_meta("nick", nick);
}
Command::Delete(timestamp) => {
if !feed.delete(timestamp) {
return Err(Reason::NoRecordAt { timestamp: timestamp.clone() });
}
}
Command::Prune(mode) => fs.prune = *mode,
}
}
Ok(commands.len())
}
pub struct Daemon {
pub cfg: Config,
pub store: Store,
pub account: Account,
pub state: Mutex<State>,
pub feeds: FeedStore,
}
impl Daemon {
/// One poll cycle. Returns a human summary line.
pub fn poll(&self) -> Result<String, String> {
client::fetch(&self.store, &self.account, false, false, self.cfg.timeout)
.map_err(|e| format!("fetch: {e}"))?;
// 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 answered = 0;
let mut dropped = 0;
let mut replies: Vec<(String, String, String)> = Vec::new(); // (address, subject, body)
let mut dials = 0;
let mut deferred = 0;
for stored in &mail {
let outcome = self.process(stored.id, &stored.envelope, &mut replies, &mut dials);
// One line per mail, so a silent drop is still visible to whoever
// runs this. Without it every outcome looked the same in the log.
if let Ok(outcome) = &outcome {
println!("{}", outcome.line());
}
// A deferred mail keeps its row, so the next cycle sees it again.
if outcome.as_ref().is_ok_and(|o| *o == Outcome::Deferred) {
deferred += 1;
continue;
}
let _ = self.store.delete_local("inbox", &stored.id);
match outcome {
Ok(Outcome::Applied(_)) => applied += 1,
Ok(Outcome::Answered(_)) => answered += 1,
Ok(_) => dropped += 1,
Err(e) => eprintln!("mail: {e}"),
}
}
let failed = self.send_replies(replies);
Ok(format!(
"{applied} applied, {answered} answered, {dropped} dropped{}{}{}",
match drained {
Some(n) => format!(", {n} backlog discarded"),
None => String::new(),
},
if deferred > 0 { format!(", {deferred} deferred") } else { 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.
///
/// Ordering matters twice over. The owner path runs first, so a mail from
/// a rotated owner is never mistaken for a registration. And on the
/// no-feed path the blocked check runs before anything touches the
/// network, so Twt-Block cannot be worked around by re-registering.
fn process(
&self,
id: [u8; 32],
envelope: &[u8],
replies: &mut Vec<(String, String, String)>,
dials: &mut usize,
) -> Result<Outcome, String> {
let opened = envelope::open(self.account.keys(), envelope, now())
.map_err(|e| format!("open {}: {e}", &fumi_core::crypto::b32(&id)[..8]))?;
let incoming = Incoming {
sender: opened.sender,
epk: opened.epk,
sig: opened.sig,
time: opened.time,
body: opened.body,
};
let mut state = self.state.lock().unwrap();
let owner = match state.feed_of(&opened.sender) {
Some(index) => Some(index),
None => self.follow_rotation(&incoming, &mut state)?,
};
match owner {
Some(index) => self.owner_mail(&incoming, index, &mut state, replies, dials),
None => self.stranger_mail(&incoming, &mut state, replies, dials),
}
}
/// A mail from someone who already has a feed.
fn owner_mail(
&self,
incoming: &Incoming,
index: usize,
state: &mut MutexGuard<'_, State>,
replies: &mut Vec<(String, String, String)>,
dials: &mut usize,
) -> Result<Outcome, String> {
let address = state.feeds[index].address.clone();
// Blocked: their token is withdrawn, so nothing they send applies and
// they get no reply either.
if !self.accepted(&address)? {
return Ok(Outcome::Dropped(format!("{address} is blocked")));
}
match dispatch::request(incoming) {
// An owner's admin command needs no resolve: the feed's address
// was verified when it was created and only a signed rotation
// chain can move it, so the key on file is enough.
Ok(Some(ref request)) if self.is_admin(&address) && is_admin_command(request) => {
self.run_admin(request, &address, state, replies, dials)
}
// Help, or a request only an admin may send — either way the
// reference is the useful answer, and it quietly omits the admin
// section for a non-admin.
Ok(Some(_)) => {
self.queue_help(state, incoming, &address, index, replies);
Ok(Outcome::Answered(format!("help for {address}")))
}
Ok(None) => self.apply(incoming, index, state, replies),
Err(reason) => {
let line = format!("{address}: {}", reason.text());
self.queue_reply(state, incoming, &address, &reason, replies);
Ok(Outcome::Answered(format!("refused {line}")))
}
}
}
/// A mail from someone with no feed. Everything here is gated on proving
/// they hold the address they claim.
fn stranger_mail(
&self,
incoming: &Incoming,
state: &mut MutexGuard<'_, State>,
replies: &mut Vec<(String, String, String)>,
dials: &mut usize,
) -> Result<Outcome, String> {
// No signed Reply-To, no address to answer or to register. Dropped in
// silence, as before: without one there is nothing to verify and
// replying would make this daemon a mail reflector for anyone holding
// a freshly minted keypair.
let Some(claim) = dispatch::claimed(incoming) else {
return Ok(Outcome::Dropped(
"no signed Reply-To, so there is no address to verify".into(),
));
};
let address = claim.short();
// Before any network call, so a blocked author cannot spend our
// budget by re-registering.
if self.store.accepted_state(&address).unwrap_or(None) == Some(false) {
return Ok(Outcome::Dropped(format!("{address} is blocked")));
}
let request = dispatch::request(incoming);
let is_admin_address = self.is_admin(&address);
let mode = state.registration(self.cfg.registration);
// Silent is the pre-self-service behaviour, and it must stay silent
// even for a well-formed request — except for an admin, who would
// otherwise have no way to turn registration back on.
if mode == Mode::Silent && !is_admin_address {
return Ok(Outcome::Dropped(format!(
"{address} wrote in, and registration is silent"
)));
}
if !self.host_allowed(&claim) {
return Ok(Outcome::Dropped(format!(
"{} is not in register_hosts, so {address} cannot register",
claim.host
)));
}
if *dials >= DIALS_PER_POLL {
return Ok(Outcome::Deferred);
}
match request {
// An admin with no feed still has to prove they hold the address,
// because a signed Reply-To naming an admin proves nothing: anyone
// can mint a key and sign a URI carrying someone else's address.
Ok(Some(ref request)) if is_admin_address && is_admin_command(request) => {
*dials += 1;
if !self.holds_address(&claim, &incoming.sender)? {
return Ok(Outcome::Dropped(format!(
"{address} does not resolve to the key that signed an admin command"
)));
}
self.run_admin(request, &address, state, replies, dials)
}
Ok(Some(Request::Register { nick })) => {
if mode == Mode::Closed {
self.queue_closed(state, incoming, &address, replies);
return Ok(Outcome::Answered(format!(
"told {address} registration is closed"
)));
}
// Counted before the dial, not after: a registration that
// fails its key check costs the same outbound connection as
// one that succeeds.
if !global_allowed(
state,
Global::Attempted,
self.cfg.registrations_per_hour,
now(),
) {
return Ok(Outcome::Dropped(format!(
"{address} tried to register, and this hour's limit of {} is spent",
self.cfg.registrations_per_hour
)));
}
*dials += 1;
self.register(&claim, nick.as_deref(), incoming, state, replies)
}
// Help, an admin command from a non-admin, no request at all, or a
// malformed one: all the same answer, which is how to get a feed.
// A non-admin is never told that admin commands exist.
_ => {
if mode == Mode::Closed {
self.queue_closed(state, incoming, &address, replies);
return Ok(Outcome::Answered(format!(
"told {address} registration is closed"
)));
}
if !global_allowed(state, Global::Informed, self.cfg.info_replies_per_hour, now()) {
return Ok(Outcome::Dropped(format!(
"{address} wrote in, and this hour's limit of {} info replies is spent",
self.cfg.info_replies_per_hour
)));
}
*dials += 1;
// Resolving first is what makes the reply safe. Sending to the
// claimed smol:// URI would pin the sender's key as verified
// under that address — a permanent denial of service against
// whoever really owns it — and sending to the short form
// without this check would mail the real owner on a stranger's
// behalf.
if !self.holds_address(&claim, &incoming.sender)? {
return Ok(Outcome::Dropped(format!(
"{address} does not resolve to the key that signed this mail"
)));
}
self.queue_info(state, incoming, &address, replies);
Ok(Outcome::Answered(format!("told {address} how to register")))
}
}
}
/// Whether `claim` resolves, on its own server, to the key that signed the
/// mail. This is the check that turns "someone claims this address" into
/// "the address holder wrote to us".
///
/// A resolve failure is not an error worth surfacing: the address may
/// simply not exist, which is the answer.
fn holds_address(&self, claim: &Address, sender: &[u8; 32]) -> Result<bool, String> {
let mut session = match client::connect(&self.store, claim, false, self.cfg.timeout) {
Ok(session) => session,
Err(e) => {
eprintln!("claim {}: {e}", claim.short());
return Ok(false);
}
};
let resolved = client::trust_key(&self.store, session.transport(), claim);
session.close();
match resolved {
Ok((key, _)) => Ok(&key == sender),
Err(e) => {
eprintln!("claim {}: {e}", claim.short());
Ok(false)
}
}
}
fn is_admin(&self, address: &str) -> bool {
self.cfg.admins.iter().any(|a| a == address)
}
/// An empty allowlist means any host, and a non-empty one is the only way
/// to stop the daemon dialling a host a stranger chose.
fn host_allowed(&self, claim: &Address) -> bool {
self.cfg.register_hosts.is_empty()
|| self.cfg.register_hosts.iter().any(|h| h == &claim.host)
}
/// Creates the feed and hands over the token. `expect` is the signing key,
/// so admission refuses an address that resolves to anyone else.
fn register(
&self,
claim: &Address,
nick: Option<&str>,
incoming: &Incoming,
state: &mut MutexGuard<'_, State>,
replies: &mut Vec<(String, String, String)>,
) -> Result<Outcome, String> {
let admitted = match register::admit(
register::Context {
cfg: &self.cfg,
store: &self.store,
account: &self.account,
feeds: &self.feeds,
},
state,
claim,
nick,
Some(&incoming.sender),
) {
Ok(admitted) => admitted,
Err(e) => {
// Includes the key-mismatch refusal, which must stay quiet:
// the sender is not the address holder, so there is nobody
// here to answer.
// Includes the key-mismatch refusal. Quiet toward the
// sender, loud here: this is the line an operator needs when
// a registration does not appear.
return Ok(Outcome::Dropped(format!(
"could not register {}: {e}",
claim.short()
)));
}
};
let url = register::feed_url(&self.cfg, &admitted.slug);
replies.push((
claim.short(),
crate::i18n::t("welcome-subject"),
crate::i18n::t1("welcome-body", "url", url.clone()),
));
if !admitted.synced {
eprintln!(
"register {}: the accept token did not reach the server",
claim.short()
);
}
if self.cfg.admin_notify {
self.notify_admins(&admitted, claim, &url, replies);
}
Ok(Outcome::Applied(format!(
"registered {} as {url}",
claim.short()
)))
}
fn notify_admins(
&self,
admitted: &register::Admitted,
claim: &Address,
url: &str,
replies: &mut Vec<(String, String, String)>,
) {
let mut args = fluent_bundle::FluentArgs::new();
args.set("address", claim.short());
args.set("url", url.to_owned());
args.set("fingerprint", fumi_core::crypto::fingerprint(&admitted.key));
let body = crate::i18n::t_args("admin-notify-body", Some(&args));
let subject = crate::i18n::t1("admin-notify-subject", "slug", admitted.slug.clone());
for admin in &self.cfg.admins {
replies.push((admin.clone(), subject.clone(), body.clone()));
}
}
/// Runs an admin command. The caller has already established that the
/// sender holds an admin address.
fn run_admin(
&self,
request: &Request,
address: &str,
state: &mut MutexGuard<'_, State>,
replies: &mut Vec<(String, String, String)>,
dials: &mut usize,
) -> Result<Outcome, String> {
let what = match request {
Request::Admit(target) => {
if *dials >= DIALS_PER_POLL {
return Ok(Outcome::Deferred);
}
*dials += 1;
let target_addr =
Address::parse(target).map_err(|e| format!("admit target: {e}"))?;
// expect = None: the target has not written in, so there is no
// signing key to match. The admin vouches, as on the CLI.
match register::admit(
register::Context {
cfg: &self.cfg,
store: &self.store,
account: &self.account,
feeds: &self.feeds,
},
state,
&target_addr,
None,
None,
) {
Ok(admitted) => {
let url = register::feed_url(&self.cfg, &admitted.slug);
if admitted.created {
replies.push((
target_addr.short(),
crate::i18n::t("welcome-subject"),
crate::i18n::t1("welcome-body", "url", url.clone()),
));
}
let mut args = fluent_bundle::FluentArgs::new();
args.set("address", target.clone());
args.set("url", url);
crate::i18n::t_args("admin-admitted", Some(&args))
}
Err(e) => {
let mut args = fluent_bundle::FluentArgs::new();
args.set("address", target.clone());
args.set("detail", e);
crate::i18n::t_args("admin-admit-failed", Some(&args))
}
}
}
Request::Block(target) => {
let said = |id: &str, detail: Option<String>| {
let mut args = fluent_bundle::FluentArgs::new();
args.set("address", target.clone());
if let Some(detail) = detail {
args.set("detail", detail);
}
crate::i18n::t_args(id, Some(&args))
};
match self.store.block(target) {
Ok(true) => {
match client::push_tokens(&self.store, &self.account, None, self.cfg.timeout)
{
Ok(_) => said("admin-blocked", None),
Err(e) => said("admin-block-half-done", Some(e.to_string())),
}
}
Ok(false) => said("admin-block-unknown", None),
Err(e) => said("admin-block-failed", Some(e.to_string())),
}
}
Request::Registration(mode) => {
state.set_registration(*mode);
state
.save(&self.cfg.data.join("state.json"))
.map_err(|e| format!("state: {e}"))?;
crate::i18n::t(match mode {
Mode::Open => "admin-registration-open",
Mode::Closed => "admin-registration-closed",
Mode::Silent => "admin-registration-silent",
})
}
// Not an admin command; the caller routes these elsewhere.
Request::Register { .. } | Request::Help => {
return Ok(Outcome::Dropped(
"an admin command that is not one; the caller routes these".into(),
))
}
};
replies.push((
address.to_owned(),
crate::i18n::t1("admin-ack-subject", "what", what.clone()),
crate::i18n::t1("admin-ack-body", "what", what.clone()),
));
Ok(Outcome::Applied(format!("admin {address}: {what}")))
}
/// The command reference, with the admin section only for an admin.
fn queue_help(
&self,
state: &mut MutexGuard<'_, State>,
incoming: &Incoming,
address: &str,
index: usize,
replies: &mut Vec<(String, String, String)>,
) {
let key = payload_key(&incoming.sender, incoming.time, &incoming.body);
if !reply_allowed(state, &incoming.sender, &key, now()) {
return;
}
let slug = state.feeds[index].username.clone();
let mut body = crate::i18n::t1("help-body", "url", register::feed_url(&self.cfg, &slug));
if self.is_admin(address) {
body.push_str(&crate::i18n::t("help-admin-section"));
}
replies.push((address.to_owned(), crate::i18n::t("help-subject"), body));
}
fn queue_info(
&self,
state: &mut MutexGuard<'_, State>,
incoming: &Incoming,
address: &str,
replies: &mut Vec<(String, String, String)>,
) {
let key = payload_key(&incoming.sender, incoming.time, &incoming.body);
if !reply_allowed(state, &incoming.sender, &key, now()) {
return;
}
let mailbox = format!("{}@{}", self.cfg.mailbox, self.cfg.smolmail_host);
replies.push((
address.to_owned(),
crate::i18n::t("info-subject"),
crate::i18n::t1("info-body", "mailbox", mailbox),
));
}
fn queue_closed(
&self,
state: &mut MutexGuard<'_, State>,
incoming: &Incoming,
address: &str,
replies: &mut Vec<(String, String, String)>,
) {
let key = payload_key(&incoming.sender, incoming.time, &incoming.body);
if !reply_allowed(state, &incoming.sender, &key, now()) {
return;
}
replies.push((
address.to_owned(),
crate::i18n::t("info-closed-subject"),
crate::i18n::t("info-closed-body"),
));
}
fn accepted(&self, address: &str) -> Result<bool, String> {
match self.store.accepted_state(address) {
Ok(Some(true)) => Ok(true),
Ok(_) => Ok(false),
Err(e) => Err(format!("accepted_state: {e}")),
}
}
/// An unknown sender may be a rotated owner: a signed Reply-To names the
/// address, and a verified chain from the key we hold to this one
/// (SPEC.md §7) moves the feed to the new key, after which the very mail
/// that announced the rotation applies as the rotated owner. Anything
/// else is dropped.
fn follow_rotation(
&self,
incoming: &Incoming,
state: &mut MutexGuard<'_, State>,
) -> Result<Option<usize>, String> {
let Some(addr) = dispatch::claimed(incoming) else {
return Ok(None);
};
// The claim alone proves only that the sender signed this URI. What
// makes it safe here is the next two lines: the address must already
// own a feed, and the chain below must lead from the key on file.
let Some(index) = state.feeds.iter().position(|f| f.address == addr.short()) else {
return Ok(None);
};
let Some(old) = state.feeds[index].latest_owner() else {
return Ok(None);
};
let mut session = client::connect(&self.store, &addr, false, self.cfg.timeout)
.map_err(|e| format!("rotation connect: {e}"))?;
let resolved = client::resolve(session.transport(), &addr.user)
.map_err(|e| format!("rotation resolve: {e}"))?;
session.close();
let (current, chain) = resolved;
if !client::walk_chain(&addr.user, &old, &current, &chain) {
return Ok(None);
}
if state.feeds[index].owner_keys.last().map(String::as_str)
!= Some(fumi_core::crypto::b32(&current).as_str())
{
state.feeds[index]
.owner_keys
.push(fumi_core::crypto::b32(&current));
}
Ok(Some(index))
}
fn apply(
&self,
incoming: &Incoming,
index: usize,
state: &mut MutexGuard<'_, State>,
replies: &mut Vec<(String, String, String)>,
) -> Result<Outcome, String> {
let username = state.feeds[index].username.clone();
let address = state.feeds[index].address.clone();
let mut feed = self.feeds.load(&username)?;
let before = feed.render();
let outcome = match dispatch::dispatch(incoming) {
Ok(Action::Post { timestamp, text, reply_to, mail }) => {
self.publish(
state, index, &mut feed, &address, incoming, replies,
PendingRecord { timestamp, text, reply_to, repost: false, mail },
)?
}
Ok(Action::Repost { timestamp, args, reply_to, mail }) => {
self.publish(
state, index, &mut feed, &address, incoming, replies,
PendingRecord { timestamp, text: args, reply_to, repost: true, mail },
)?
}
Ok(Action::Commands(commands)) => {
match apply_commands(&mut feed, &mut state.feeds[index], &commands) {
Ok(n) => Outcome::Applied(format!("{n} command(s) from {address}")),
Err(reason) => {
let line = reason.text();
self.queue_reply(state, incoming, &address, &reason, replies);
Outcome::Answered(format!("refused {address}: {line}"))
}
}
}
Ok(Action::Ignore) => {
Outcome::Dropped(format!("nothing to do in {address}'s mail"))
}
Err(reason) => {
// Someone watching their first post be refused needs to hear
// that their feed is fine, not just what went wrong.
let reason = match reason {
Reason::AcceptFieldOnPost if feed.entries().is_empty() => {
Reason::AcceptFieldOnFirstPost
}
other => other,
};
let line = reason.text();
self.queue_reply(state, incoming, &address, &reason, replies);
Outcome::Answered(format!("refused {address}: {line}"))
}
};
let after = feed.render();
if after != before {
self.feeds.save(&username, &feed)?;
}
// Counters, ledgers and prune modes change even when no record did.
state
.save(&self.cfg.data.join("state.json"))
.map_err(|e| e.to_string())?;
Ok(outcome)
}
#[allow(clippy::too_many_arguments)]
fn publish(
&self,
state: &mut MutexGuard<'_, State>,
index: usize,
feed: &mut Feed,
address: &str,
incoming: &Incoming,
replies: &mut Vec<(String, String, String)>,
record: PendingRecord,
) -> Result<Outcome, String> {
let was_warned = state.feeds[index].warned;
match apply_record(feed, &mut state.feeds[index], &self.cfg, record, incoming.time) {
Ok(()) => {
// The transition false -> true sends the cap warning once per
// growth episode (PLAN.md, Retention).
if state.feeds[index].warned && !was_warned {
replies.push((
address.to_owned(),
crate::i18n::t("cap-warning-subject"),
crate::i18n::t1("cap-warning-body", "cap", human_size(self.cfg.cap_bytes)),
));
}
Ok(Outcome::Applied(format!("post by {address}")))
}
Err(e) => {
let reason = match e {
ApplyError::TooLong => Reason::TooLong { max: self.cfg.max_post_chars },
ApplyError::OverQuota(repost) => Reason::OverQuota { repost },
ApplyError::Collision(timestamp) => Reason::Collision { timestamp },
ApplyError::FeedFull => Reason::FeedFull,
ApplyError::ReplyToChanged => Reason::ReplyToChanged {
author: feed.author().unwrap_or_default().to_owned(),
},
};
let line = reason.text();
self.queue_reply(state, incoming, address, &reason, replies);
Ok(Outcome::Answered(format!("refused {address}: {line}")))
}
}
}
/// One error reply per rejected payload, within the hourly budget.
fn queue_reply(
&self,
state: &mut MutexGuard<'_, State>,
incoming: &Incoming,
address: &str,
reason: &Reason,
replies: &mut Vec<(String, String, String)>,
) {
let key = payload_key(&incoming.sender, incoming.time, &incoming.body);
if !reply_allowed(state, &incoming.sender, &key, now()) {
return;
}
let text = std::str::from_utf8(&incoming.body)
.map(|t| t.chars().take(200).collect::<String>())
.unwrap_or_default();
replies.push((
address.to_owned(),
crate::i18n::t1("reply-subject", "reason", reason.text()),
reason.reply_body(&text),
));
}
/// 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 = 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 {
address: &addr,
text: body,
subject: Some(&subject),
reply_to: None,
headers: &[],
anonymous: false,
no_pad: true,
},
self.cfg.timeout,
) {
// 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;
}
}
failed
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::config::Config;
use crate::state::State;
use smolboid_core::verify;
fn feed_state() -> FeedState {
FeedState {
username: "tobias".into(),
address: "tobias@smol.place".into(),
owner_keys: vec![fumi_core::crypto::b32(&[1; 32])],
nick: "tobias".into(),
prune: PruneMode::Off,
warned: false,
day: State::day(0),
posts: 0,
reposts: 0,
}
}
/// A record renders at roughly 260 bytes: the post line plus a
/// `# smol:mail` line of three base32 values.
fn config(cap: usize) -> Config {
Config {
cap_bytes: cap,
..Config::default()
}
}
fn record(ts: &str, text: &str) -> PendingRecord {
PendingRecord {
timestamp: ts.into(),
text: text.into(),
repost: false,
reply_to: None,
mail: MailLine {
sender: [1; 32],
epk: [2; 32],
sig: [3; 64],
},
}
}
#[test]
fn sizes_read_as_sizes() {
assert_eq!(human_size(1_048_576), "1 MB");
assert_eq!(human_size(10 * 1_048_576), "10 MB");
assert_eq!(human_size(2048), "2 KB");
assert_eq!(human_size(1023), "1023 bytes");
assert_eq!(human_size(0), "0 bytes");
}
#[test]
fn posts_publish_and_verify() {
let mut feed = smolboid_core::Feed::new();
let mut fs = feed_state();
apply_record(&mut feed, &mut fs, &config(1 << 20), record("2026-10-01T12:47:00Z", "hello"), 0).unwrap();
assert_eq!(fs.posts, 1);
let report = verify(&feed);
assert_eq!(report.states, vec![smolboid_core::State::Unsigned]);
assert_eq!(feed.entries().len(), 1);
}
#[test]
fn length_cap_is_configurable() {
let mut feed = smolboid_core::Feed::new();
let mut fs = feed_state();
let cfg = Config { max_post_chars: 5, ..config(1 << 20) };
let five = record("2026-10-01T12:47:00Z", "🙂🙂🙂🙂🙂");
apply_record(&mut feed, &mut fs, &cfg, five, 0).unwrap();
let six = record("2026-10-01T12:47:01Z", "abcdef");
assert!(matches!(apply_record(&mut feed, &mut fs, &cfg, six, 0), Err(ApplyError::TooLong)));
assert_eq!(fs.posts, 1, "a refused post spends no quota");
}
#[test]
fn reposts_publish_as_repost_records() {
let mut feed = smolboid_core::Feed::new();
let mut fs = feed_state();
let record = PendingRecord {
timestamp: "2026-10-01T12:47:00Z".into(),
text: "@<dana https://x/d.txt> #abc1234".into(),
repost: true,
reply_to: None,
mail: MailLine { sender: [1; 32], epk: [2; 32], sig: [3; 64] },
};
apply_record(&mut feed, &mut fs, &config(1 << 20), record, 0).unwrap();
let entry = &feed.entries()[0];
assert!(matches!(&entry.record, Record::Repost { args, .. } if args == "@<dana https://x/d.txt> #abc1234"));
assert_eq!(fs.reposts, 1);
assert_eq!(fs.posts, 0);
assert!(feed.render().starts_with("# smol:repost 2026-10-01T12:47:00Z @<dana https://x/d.txt> #abc1234\n"));
}
#[test]
fn collisions_and_quotas_reject_without_appending() {
let mut feed = smolboid_core::Feed::new();
let mut fs = feed_state();
let cfg = config(1 << 20);
apply_record(&mut feed, &mut fs, &cfg, record("2026-10-01T12:47:00Z", "one"), 0).unwrap();
assert!(matches!(
apply_record(&mut feed, &mut fs, &cfg, record("2026-10-01T12:47:00Z", "two"), 0),
Err(ApplyError::Collision(_))
));
assert_eq!(feed.entries().len(), 1);
let mut fs = feed_state();
fs.posts = 50;
assert!(matches!(
apply_record(&mut feed, &mut fs, &config(1 << 20), record("2026-10-01T12:48:00Z", "x"), 0),
Err(ApplyError::OverQuota(_))
));
// A new UTC day rolls the counters.
let mut fs = feed_state();
fs.posts = 50;
apply_record(&mut feed, &mut fs, &config(1 << 20), record("2026-10-01T12:48:00Z", "x"), 1_760_000_000).unwrap();
}
#[test]
fn hard_limit_rejects_and_fifo_prunes_oldest() {
let mut feed = smolboid_core::Feed::new();
let mut fs = feed_state();
let cfg = config(300);
apply_record(&mut feed, &mut fs, &cfg, record("2026-10-01T12:47:00Z", "first post here"), 0).unwrap();
assert!(matches!(
apply_record(&mut feed, &mut fs, &cfg, record("2026-10-01T12:48:00Z", "second post"), 0),
Err(ApplyError::FeedFull)
));
assert_eq!(feed.entries().len(), 1, "off mode removes nothing");
fs.prune = PruneMode::Fifo;
apply_record(&mut feed, &mut fs, &cfg, record("2026-10-01T12:48:00Z", "second post"), 0).unwrap();
let entries = feed.entries();
assert_eq!(entries.len(), 1);
assert_eq!(entries[0].record.timestamp(), "2026-10-01T12:48:00Z");
// The quota counts published posts; pruning does not refund it.
assert_eq!(fs.posts, 2);
}
#[test]
fn warning_flag_flips_at_ninety_percent() {
let mut feed = smolboid_core::Feed::new();
let mut fs = feed_state();
let cfg = Config { max_post_chars: 700, ..config(1300) };
apply_record(&mut feed, &mut fs, &cfg, record("2026-10-01T12:47:00Z", "x"), 0).unwrap();
assert!(!fs.warned);
apply_record(&mut feed, &mut fs, &cfg, record("2026-10-01T12:47:01Z", &"y".repeat(700)), 0).unwrap();
assert!(fs.warned, "crossing 90% sets the flag once");
feed.delete("2026-10-01T12:47:01Z");
apply_record(&mut feed, &mut fs, &cfg, record("2026-10-01T12:47:02Z", "z"), 0).unwrap();
assert!(!fs.warned, "shrinking below the threshold resets it");
}
#[test]
fn commands_touch_feed_and_state() {
let mut feed = smolboid_core::Feed::new();
let mut fs = feed_state();
apply_record(&mut feed, &mut fs, &config(1 << 20), record("2026-10-01T12:47:00Z", "gone soon"), 0).unwrap();
apply_commands(
&mut feed,
&mut fs,
&[Command::Nick("toby".into()), Command::Prune(PruneMode::Fifo)],
)
.unwrap();
assert_eq!(fs.nick, "toby");
assert_eq!(feed.meta("nick"), Some("toby"));
assert_eq!(fs.prune, PruneMode::Fifo);
apply_commands(&mut feed, &mut fs, &[Command::Delete("2026-10-01T12:47:00Z".into())]).unwrap();
assert_eq!(feed.entries().len(), 0);
assert!(apply_commands(&mut feed, &mut fs, &[Command::Delete("2026-10-01T12:47:00Z".into())]).is_err());
}
}