postvak_core/daemon/
mod.rs

1//! The one process that owns the engine, serving it to the GUI, CLI and MCP over a private Unix socket.
2use std::collections::HashMap;
3use std::fs::{self, File, TryLockError};
4use std::io::{self, BufRead, BufReader, IsTerminal, Read, Write};
5use std::net::Shutdown;
6use std::os::unix::fs::PermissionsExt;
7use std::os::unix::net::{UnixListener, UnixStream};
8use std::path::Path;
9use std::sync::atomic::{AtomicBool, Ordering};
10use std::sync::mpsc::{self, Receiver, Sender};
11use std::sync::{Arc, Mutex, MutexGuard, PoisonError};
12use std::thread::JoinHandle;
13use std::time::{Duration, Instant};
14
15use anyhow::{Context, anyhow, bail};
16
17use crate::config::Config;
18use crate::engine::{self, Connector, Engine};
19use crate::message::clean;
20use crate::paths::{self, FileWatch, Paths};
21use crate::store::Store;
22use crate::sync::{self, Activity, Command, Event, RequestId};
23use wire::{AccountStatus, ClientMessage, DaemonMessage, Outcome, Payload, Status};
24
25pub mod client;
26pub mod service;
27#[cfg(any(test, feature = "testing"))]
28pub mod test_support;
29#[cfg(test)]
30mod tests;
31pub mod wire;
32
33pub use client::Client;
34
35const FILE_POLL: Duration = Duration::from_secs(2);
36/// How long exit waits for clients to take their last replies before closing their connections.
37const FLUSH_GRACE: Duration = Duration::from_secs(1);
38const LOG_LIMIT: u64 = 1_048_576;
39const POLL: Duration = Duration::from_millis(200);
40
41pub struct Options {
42    /// Exit after this long without a connected client; None never idles out.
43    pub idle_exit: Option<Duration>,
44    pub connect: Connector,
45    /// Sends desktop notifications; tests turn it off.
46    pub notify: bool,
47    /// Stops on SIGTERM and SIGINT as on `daemon stop`; tests, which run many daemons in one process, turn it off.
48    pub signals: bool,
49}
50
51impl Options {
52    pub fn foreground() -> Options {
53        Options {
54            idle_exit: None,
55            connect: engine::imap_connector(),
56            notify: true,
57            signals: true,
58        }
59    }
60
61    pub fn auto_started(idle: Duration) -> Options {
62        Options {
63            idle_exit: Some(idle),
64            ..Options::foreground()
65        }
66    }
67}
68
69/// Serves until `shutdown`, idle exit, or a fatal error; returns after the engine stopped and the socket file is gone.
70pub fn run(paths: &Paths, options: Options) -> anyhow::Result<()> {
71    // The private state directory guards the socket between `bind` and its chmod.
72    paths::create_private_dir(&paths.state_dir)?;
73    let _lock = take_daemon_lock(&paths.daemon_lock())?;
74    truncate_large_log(&paths.daemon_log())?;
75    let socket = paths.daemon_socket();
76    let listener = bind_private(&socket)?;
77    let signalled = Arc::new(AtomicBool::new(false));
78    let signals = if options.signals {
79        StopSignals::register(&signalled)?
80    } else {
81        StopSignals(Vec::new())
82    };
83    let served = serve(paths, options, listener, signalled);
84    drop(signals);
85    let _ = fs::remove_file(&socket);
86    served
87}
88
89/// The `[account] …` line for an event, as `postvak run` and `sync` print it.
90pub fn report(event: &Event) {
91    match event {
92        Event::NewMail {
93            account,
94            from,
95            subject,
96            ..
97        } => println!(
98            "[{account}] new mail from {}: {}",
99            clean(from, false),
100            clean(subject, false)
101        ),
102        Event::Synced {
103            account,
104            new_messages,
105            actions,
106            ..
107        } => println!("[{account}] synced: {new_messages} new, {actions} rule actions"),
108        Event::CommandFailed {
109            account, message, ..
110        }
111        | Event::Error { account, message } => report_error(account, message),
112        Event::Activity { account, activity } => log::debug!("[{account}] {activity:?}"),
113        Event::ActionDone { .. }
114        | Event::BodiesFetched { .. }
115        | Event::BodyReady { .. }
116        | Event::Restored { .. }
117        | Event::RuleApplied { .. }
118        | Event::ServerSearchDone { .. } => {}
119    }
120}
121
122/// The `[account] error: …` line `report` prints for a failure.
123pub fn report_error(account: &str, message: &str) {
124    eprintln!("[{account}] error: {}", clean(message, false));
125}
126
127fn truncate_large_log(log: &Path) -> io::Result<()> {
128    match fs::metadata(log) {
129        Ok(meta) if meta.len() > LOG_LIMIT => File::options().write(true).open(log)?.set_len(0),
130        _ => Ok(()),
131    }
132}
133
134/// Locks `path` and writes our pid there; held until the returned file drops.
135fn take_daemon_lock(path: &Path) -> anyhow::Result<File> {
136    let mut file = File::options()
137        .read(true)
138        .write(true)
139        .create(true)
140        .truncate(false)
141        .open(path)?;
142    match file.try_lock() {
143        Ok(()) => {
144            file.set_len(0)?;
145            write!(file, "{}", std::process::id())?;
146            Ok(file)
147        }
148        Err(TryLockError::WouldBlock) => {
149            let mut pid = String::new();
150            file.read_to_string(&mut pid)?;
151            let pid: Option<u32> = pid.trim().parse().ok();
152            bail!(
153                "already running (pid {})",
154                pid.map_or_else(|| "unknown".to_string(), |pid| pid.to_string())
155            )
156        }
157        Err(TryLockError::Error(e)) => Err(e.into()),
158    }
159}
160
161/// Replaces a leftover socket file and binds a fresh one only we can open.
162fn bind_private(socket: &Path) -> anyhow::Result<UnixListener> {
163    match fs::remove_file(socket) {
164        Err(e) if e.kind() != io::ErrorKind::NotFound => {
165            return Err(e).with_context(|| format!("removing {}", socket.display()));
166        }
167        _ => {}
168    }
169    let listener = UnixListener::bind(socket).map_err(|e| match e.kind() {
170        io::ErrorKind::InvalidInput => anyhow!("socket path too long: {}", socket.display()),
171        _ => anyhow::Error::new(e).context(format!("binding {}", socket.display())),
172    })?;
173    fs::set_permissions(socket, fs::Permissions::from_mode(0o600))?;
174    listener.set_nonblocking(true)?;
175    Ok(listener)
176}
177
178/// SIGTERM (`docker stop`, `systemctl stop`, launchd) and SIGINT (Ctrl-C) set the stop flag, so the daemon stops as on
179/// `daemon stop`; a second one exits at once. PID 1 in a container ignores a signal it has no handler for, so without
180/// this `docker stop` waits out its timeout and kills the daemon.
181struct StopSignals(Vec<signal_hook::SigId>);
182
183impl StopSignals {
184    fn register(signalled: &Arc<AtomicBool>) -> io::Result<StopSignals> {
185        use signal_hook::consts::{SIGINT, SIGTERM};
186        use signal_hook::flag;
187        let mut ids = Vec::new();
188        for signal in [SIGTERM, SIGINT] {
189            // The shutdown goes first, so it sees the flag only from the second signal on.
190            ids.push(flag::register_conditional_shutdown(
191                signal,
192                128 + signal,
193                signalled.clone(),
194            )?);
195            ids.push(flag::register(signal, signalled.clone())?);
196        }
197        Ok(StopSignals(ids))
198    }
199}
200
201impl Drop for StopSignals {
202    fn drop(&mut self) {
203        for id in self.0.drain(..) {
204            signal_hook::low_level::unregister(id);
205        }
206    }
207}
208
209type ClientId = u64;
210
211struct ClientConn {
212    out: Sender<String>,
213    subscribed: bool,
214    writer: JoinHandle<()>,
215}
216
217struct Hub {
218    activity: HashMap<String, Activity>,
219    clients: HashMap<ClientId, ClientConn>,
220    last_client_left: Instant,
221    next_request: RequestId,
222    /// The daemon's request id to the client that sent it and that client's own id.
223    pending: HashMap<RequestId, (ClientId, u64)>,
224    shutdown: bool,
225    started: Instant,
226}
227
228impl Hub {
229    fn new() -> Hub {
230        let started = Instant::now();
231        Hub {
232            activity: HashMap::new(),
233            clients: HashMap::new(),
234            last_client_left: started,
235            next_request: 0,
236            pending: HashMap::new(),
237            shutdown: false,
238            started,
239        }
240    }
241
242    fn send(&self, client: ClientId, message: &DaemonMessage) {
243        let Some(conn) = self.clients.get(&client) else {
244            return;
245        };
246        match wire::line(message) {
247            Ok(line) => {
248                let _ = conn.out.send(line);
249            }
250            Err(e) => log::error!("could not encode a message: {e}"),
251        }
252    }
253
254    fn reply(&self, client: ClientId, id: u64, outcome: Outcome) {
255        self.send(client, &DaemonMessage::Reply { id, outcome });
256    }
257
258    fn broadcast(&self, event: &Event) {
259        let line = match wire::line(&DaemonMessage::Event(event.clone())) {
260            Ok(line) => line,
261            Err(e) => return log::error!("could not encode an event: {e}"),
262        };
263        for conn in self.clients.values().filter(|conn| conn.subscribed) {
264            let _ = conn.out.send(line.clone());
265        }
266    }
267}
268
269/// Readers reach the engine through its own mutex, so a slow engine call never stalls the hub.
270struct Shared {
271    engine: Mutex<Option<Engine>>,
272    hub: Mutex<Hub>,
273    /// Set by a stop signal.
274    signalled: Arc<AtomicBool>,
275    watch: Mutex<Watch>,
276}
277
278/// A panic elsewhere must not take the daemon down with a poisoned lock.
279fn lock<T>(mutex: &Mutex<T>) -> MutexGuard<'_, T> {
280    mutex.lock().unwrap_or_else(PoisonError::into_inner)
281}
282
283fn serve(
284    paths: &Paths,
285    options: Options,
286    listener: UnixListener,
287    signalled: Arc<AtomicBool>,
288) -> anyhow::Result<()> {
289    let watch = Watch::new(paths);
290    let config = Config::load(&paths.config_file())?;
291    migrate_stores(&config, paths);
292    let (engine, events) = Engine::start(&config, paths, options.connect.clone());
293    let shared = Arc::new(Shared {
294        engine: Mutex::new(Some(engine)),
295        hub: Mutex::new(Hub::new()),
296        signalled,
297        watch: Mutex::new(watch),
298    });
299    let notify: fn(&str, &str) = if options.notify {
300        crate::notify::new_mail
301    } else {
302        |_, _| {}
303    };
304    let router = std::thread::Builder::new()
305        .name("daemon-events".into())
306        .spawn({
307            let shared = shared.clone();
308            move || route_events(&shared, events, io::stdout().is_terminal(), notify)
309        })?;
310    let connections = accept_until_done(&shared, &listener, options.idle_exit);
311    // Connects fail at once from here on instead of waiting out the hello timeout; the socket file and lock remain.
312    drop(listener);
313    stop(&shared, router, connections);
314    Ok(())
315}
316
317/// Clients open the stores as soon as the daemon answers, so every migration has to be done before it does; a store
318/// that fails to open fails again in its account's thread, which reports it.
319fn migrate_stores(config: &Config, paths: &Paths) {
320    for account in &config.accounts {
321        if let Err(e) = Store::open_account(paths, &account.name) {
322            log::warn!("[{}] could not open the store: {e}", account.name);
323        }
324    }
325}
326
327/// config.toml and rules.toml, looked at every 2 s and before each client request.
328struct Watch {
329    config: FileWatch,
330    rules: FileWatch,
331    checked: Instant,
332}
333
334impl Watch {
335    fn new(paths: &Paths) -> Watch {
336        Watch {
337            config: FileWatch::new(paths.config_file()),
338            rules: FileWatch::new(paths.rules_file()),
339            checked: Instant::now(),
340        }
341    }
342
343    /// The config, loaded again once it changed, and whether the rules file changed; `poll` skips the check until
344    /// FILE_POLL passed since the last one.
345    fn changed(&mut self, poll: bool) -> (Option<Config>, bool) {
346        if poll && self.checked.elapsed() < FILE_POLL {
347            return (None, false);
348        }
349        self.checked = Instant::now();
350        let config = self.config.changed().then(|| {
351            Config::load(&self.config.path)
352                .inspect_err(|e| log::error!("{e}; keeping the running accounts"))
353                .ok()
354        });
355        (config.flatten(), self.rules.changed())
356    }
357}
358
359struct Connection {
360    stream: UnixStream,
361    reader: JoinHandle<()>,
362}
363
364/// Applies a changed config.toml or rules.toml and starts the accounts whose old thread ended.
365fn apply_file_changes(shared: &Shared, poll: bool) {
366    // Held throughout, so a request never overtakes a change another thread is still applying.
367    let mut watch = lock(&shared.watch);
368    let (config, rules) = watch.changed(poll);
369    if let Some(engine) = lock(&shared.engine).as_mut() {
370        if let Some(config) = &config {
371            engine.apply_config(config);
372        }
373        engine.start_ready();
374        if rules {
375            // Rules act on mail synced after they change, so every running account syncs at once; request 0 has
376            // nobody waiting for its reply.
377            for account in engine.accounts() {
378                engine.send(&account, 0, Command::SyncNow);
379            }
380        }
381    }
382}
383
384fn accept_until_done(
385    shared: &Arc<Shared>,
386    listener: &UnixListener,
387    idle_exit: Option<Duration>,
388) -> Vec<Connection> {
389    let mut connections: Vec<Connection> = Vec::new();
390    let mut next_client: ClientId = 0;
391    while !done(shared, idle_exit) {
392        apply_file_changes(shared, true);
393        match listener.accept() {
394            Ok((stream, _)) => {
395                next_client += 1;
396                match open_connection(shared, stream, next_client) {
397                    Ok(connection) => connections.push(connection),
398                    Err(e) => log::warn!("could not serve a client: {e}"),
399                }
400            }
401            Err(e) if e.kind() == io::ErrorKind::WouldBlock => std::thread::sleep(POLL),
402            Err(e) => {
403                log::warn!("accept failed: {e}");
404                std::thread::sleep(POLL);
405            }
406        }
407        connections.retain(|connection| !connection.reader.is_finished());
408    }
409    connections
410}
411
412fn done(shared: &Shared, idle_exit: Option<Duration>) -> bool {
413    if shared.signalled.load(Ordering::Acquire) {
414        log::info!("stopping on a signal");
415        return true;
416    }
417    let hub = lock(&shared.hub);
418    hub.shutdown
419        || idle_exit
420            .is_some_and(|idle| hub.clients.is_empty() && hub.last_client_left.elapsed() >= idle)
421}
422
423/// Stops the engine, delivers the replies its threads still send, then closes every connection.
424fn stop(shared: &Shared, router: JoinHandle<()>, connections: Vec<Connection>) {
425    lock(&shared.hub).shutdown = true;
426    let engine = lock(&shared.engine).take();
427    drop(engine);
428    let _ = router.join();
429    let clients: Vec<ClientConn> = lock(&shared.hub)
430        .clients
431        .drain()
432        .map(|(_, conn)| conn)
433        .collect();
434    let writers: Vec<JoinHandle<()>> = clients.into_iter().map(|conn| conn.writer).collect();
435    let deadline = Instant::now() + FLUSH_GRACE;
436    while writers.iter().any(|writer| !writer.is_finished()) && Instant::now() < deadline {
437        std::thread::sleep(Duration::from_millis(10));
438    }
439    for connection in &connections {
440        let _ = connection.stream.shutdown(Shutdown::Both);
441    }
442    writers.into_iter().for_each(|writer| drop(writer.join()));
443    for connection in connections {
444        let _ = connection.reader.join();
445    }
446}
447
448/// Senders and subjects only go to a terminal, never into daemon.log; errors go to both.
449fn reported(event: &Event, terminal: bool) -> bool {
450    terminal || matches!(event, Event::CommandFailed { .. } | Event::Error { .. })
451}
452
453fn route_events(
454    shared: &Shared,
455    events: Receiver<Event>,
456    terminal: bool,
457    notify: impl Fn(&str, &str),
458) {
459    for event in events {
460        if reported(&event, terminal) {
461            report(&event);
462        }
463        // The sync sends NewMail only for mail that should notify, rules included.
464        if let Event::NewMail { from, subject, .. } = &event {
465            notify(from, subject);
466        }
467        let mut hub = lock(&shared.hub);
468        if let Event::Activity { account, activity } = &event {
469            hub.activity.insert(account.clone(), activity.clone());
470        }
471        for request in event.request_ids() {
472            if let Some((client, id)) = hub.pending.remove(&request) {
473                hub.reply(client, id, outcome_of(&event));
474            }
475        }
476        hub.broadcast(&event);
477    }
478}
479
480fn outcome_of(event: &Event) -> Outcome {
481    match event {
482        Event::CommandFailed { message, .. } => Outcome::Error(message.clone()),
483        _ => Outcome::Ok(Payload::Event(event.clone())),
484    }
485}
486
487fn open_connection(
488    shared: &Arc<Shared>,
489    stream: UnixStream,
490    client: ClientId,
491) -> io::Result<Connection> {
492    // macOS hands out accepted sockets with the listener's non-blocking flag.
493    stream.set_nonblocking(false)?;
494    let kept = stream.try_clone()?;
495    let shared = shared.clone();
496    let reader = std::thread::Builder::new()
497        .name(format!("daemon-client-{client}"))
498        .spawn(move || serve_client(&shared, client, stream))?;
499    Ok(Connection {
500        stream: kept,
501        reader,
502    })
503}
504
505fn serve_client(shared: &Shared, client: ClientId, stream: UnixStream) {
506    if let Err(e) = converse(shared, client, &stream) {
507        log::debug!("client {client}: {e}");
508    }
509    if let Some(conn) = unregister(shared, client) {
510        drop(conn.out);
511        let _ = conn.writer.join();
512    }
513    let _ = stream.shutdown(Shutdown::Both);
514}
515
516fn converse(shared: &Shared, client: ClientId, stream: &UnixStream) -> io::Result<()> {
517    let mut lines = BufReader::new(stream).lines();
518    let Some(first) = lines.next().transpose()? else {
519        return Ok(());
520    };
521    if !matches!(
522        serde_json::from_str(&first),
523        Ok(ClientMessage::Hello { .. })
524    ) {
525        return Ok(());
526    }
527    if !register(shared, client, stream.try_clone()?)? {
528        return Ok(());
529    }
530    for line in lines {
531        if !handle_line(shared, client, &line?) {
532            break;
533        }
534    }
535    Ok(())
536}
537
538/// Adds the client with its writer thread and greets it; false once the daemon is stopping.
539fn register(shared: &Shared, client: ClientId, stream: UnixStream) -> io::Result<bool> {
540    let (out, lines) = mpsc::channel();
541    let writer = std::thread::Builder::new()
542        .name(format!("daemon-writer-{client}"))
543        .spawn(move || write_lines(stream, lines))?;
544    let mut hub = lock(&shared.hub);
545    if hub.shutdown {
546        return Ok(false);
547    }
548    hub.clients.insert(
549        client,
550        ClientConn {
551            out,
552            subscribed: false,
553            writer,
554        },
555    );
556    let hello = DaemonMessage::Hello {
557        protocol: wire::PROTOCOL,
558        version: wire::VERSION.into(),
559        pid: std::process::id(),
560    };
561    hub.send(client, &hello);
562    Ok(true)
563}
564
565fn write_lines(mut stream: UnixStream, lines: Receiver<String>) {
566    for line in lines {
567        if stream.write_all(line.as_bytes()).is_err() {
568            return;
569        }
570    }
571}
572
573fn unregister(shared: &Shared, client: ClientId) -> Option<ClientConn> {
574    let mut hub = lock(&shared.hub);
575    let conn = hub.clients.remove(&client)?;
576    hub.pending.retain(|_, (owner, _)| *owner != client);
577    if hub.clients.is_empty() {
578        hub.last_client_left = Instant::now();
579    }
580    Some(conn)
581}
582
583/// Answers one line; false when the connection should close.
584fn handle_line(shared: &Shared, client: ClientId, line: &str) -> bool {
585    let message = match serde_json::from_str::<ClientMessage>(line) {
586        Ok(message) => message,
587        Err(e) => {
588            let Some(id) = id_in(line) else {
589                return false;
590            };
591            lock(&shared.hub).reply(client, id, Outcome::Error(format!("bad message: {e}")));
592            return true;
593        }
594    };
595    // A request right after a client saved config.toml sees the accounts it saved.
596    apply_file_changes(shared, false);
597    match message {
598        ClientMessage::Hello { .. } => {}
599        ClientMessage::Subscribe { id } => {
600            let mut hub = lock(&shared.hub);
601            if let Some(conn) = hub.clients.get_mut(&client) {
602                conn.subscribed = true;
603            }
604            hub.reply(client, id, Outcome::Ok(Payload::Done));
605        }
606        ClientMessage::Status { id } => {
607            let status = status(shared);
608            lock(&shared.hub).reply(client, id, Outcome::Ok(Payload::Status(status)));
609        }
610        ClientMessage::Shutdown { id } => {
611            let mut hub = lock(&shared.hub);
612            hub.reply(client, id, Outcome::Ok(Payload::Done));
613            hub.shutdown = true;
614        }
615        ClientMessage::Command {
616            id,
617            account,
618            command,
619        } => submit(shared, client, id, &account, command),
620    }
621    true
622}
623
624/// The `id` of a message that did not parse, at its top level or inside its one variant.
625fn id_in(line: &str) -> Option<u64> {
626    let value: serde_json::Value = serde_json::from_str(line).ok()?;
627    let variant = value.as_object()?.values().next();
628    value.get("id").or_else(|| variant?.get("id"))?.as_u64()
629}
630
631fn status(shared: &Shared) -> Status {
632    let accounts = lock(&shared.engine)
633        .as_ref()
634        .map(Engine::accounts)
635        .unwrap_or_default();
636    let hub = lock(&shared.hub);
637    Status {
638        pid: std::process::id(),
639        version: wire::VERSION.into(),
640        uptime_secs: hub.started.elapsed().as_secs(),
641        clients: hub.clients.len(),
642        accounts: accounts
643            .into_iter()
644            .map(|name| AccountStatus {
645                activity: hub.activity.get(&name).cloned(),
646                name,
647            })
648            .collect(),
649    }
650}
651
652/// Refuses the command at once when its account cannot run it, and otherwise hands it to the engine.
653fn submit(shared: &Shared, client: ClientId, id: u64, account: &str, command: Command) {
654    // None once `stop` took the engine.
655    let known = lock(&shared.engine).as_ref().map(|engine| {
656        let running = engine.accounts().iter().any(|name| name == account);
657        (running, engine.is_restarting(account))
658    });
659    let request = {
660        let mut hub = lock(&shared.hub);
661        let refused = match known {
662            None => Some("the daemon is stopping".to_string()),
663            Some((running, restarting)) => refusal(
664                account,
665                &command,
666                running,
667                restarting,
668                hub.activity.get(account),
669            ),
670        };
671        if let Some(refusal) = refused {
672            hub.reply(client, id, Outcome::Error(refusal));
673            return;
674        }
675        hub.next_request += 1;
676        let request = hub.next_request;
677        hub.pending.insert(request, (client, id));
678        request
679    };
680    let sent = lock(&shared.engine)
681        .as_ref()
682        .is_some_and(|engine| engine.send(account, request, command));
683    if !sent {
684        let mut hub = lock(&shared.hub);
685        if let Some((client, id)) = hub.pending.remove(&request) {
686            hub.reply(
687                client,
688                id,
689                Outcome::Error(format!("{account} is not running")),
690            );
691        }
692    }
693}
694
695/// Why a command for `account` cannot run now, if it cannot. SyncNow is never refused for an offline account: it
696/// wakes the thread from its back-off to connect again at once.
697fn refusal(
698    account: &str,
699    command: &Command,
700    running: bool,
701    restarting: bool,
702    activity: Option<&Activity>,
703) -> Option<String> {
704    match (running, activity) {
705        (false, _) if restarting => Some(format!("{account} is restarting")),
706        (false, _) => Some(format!("no account named '{account}'")),
707        (true, _) if matches!(command, Command::SyncNow) => None,
708        (true, Some(Activity::Offline { reason, retry_at })) => {
709            Some(sync::offline_message(account, reason, *retry_at))
710        }
711        (true, _) => None,
712    }
713}
postvak_core/daemon/mod.rs

postvak_core/daemon/
mod.rs
