postvak_core/daemon/
client.rs

1//! A connection to the daemon: requests with replies, fire-and-forget sends, and the event subscription.
2use std::collections::{HashMap, HashSet};
3use std::fs::{self, File};
4use std::io::{self, BufRead, BufReader, Write};
5use std::net::Shutdown;
6use std::os::unix::fs::OpenOptionsExt;
7use std::os::unix::net::UnixStream;
8use std::os::unix::process::CommandExt;
9use std::path::{Path, PathBuf};
10use std::process::{Child, Stdio};
11use std::sync::atomic::{AtomicU64, Ordering};
12use std::sync::mpsc::{self, Receiver, RecvTimeoutError, Sender};
13use std::sync::{Arc, Mutex};
14use std::time::{Duration, Instant};
15
16use anyhow::{Context, anyhow, bail};
17
18use super::lock;
19use super::wire::{
20    self, ClientMessage, DaemonMessage, Outcome, PROTOCOL, Payload, Status, VERSION,
21};
22use crate::paths::{self, Paths};
23use crate::rules::Action;
24use crate::store::{Message, Store};
25use crate::sync::{Command, Event, EventResults, RequestId};
26
27pub const NO_REPLY: &str =
28    "no reply from the daemon within 120 s; the command may still run, see `postvak log`";
29const LOG_TAIL_LINES: usize = 20;
30const REPLY_TIMEOUT: Duration = Duration::from_secs(120);
31const RETRY: Duration = Duration::from_millis(100);
32const START_WAIT: Duration = Duration::from_secs(5);
33const STOP_WAIT: Duration = Duration::from_secs(5);
34
35/// The error of a request that lost its daemon; a caller holding the client should reconnect.
36#[derive(Debug, thiserror::Error)]
37#[error("{0}")]
38pub struct Stopped(String);
39
40/// A daemon of a newer postvak than this one: this program has to restart, and the daemon keeps running.
41#[derive(Debug, thiserror::Error)]
42#[error("the daemon is version {theirs}, newer than this postvak ({ours}); restart this program")]
43pub struct NewerDaemon {
44    pub theirs: String,
45    pub ours: String,
46}
47
48pub struct Client {
49    transport: Transport,
50    inbox: Arc<Mutex<Inbox>>,
51    next_id: AtomicU64,
52    log: PathBuf,
53}
54
55enum Transport {
56    Socket(Mutex<UnixStream>),
57    #[cfg(any(test, feature = "testing"))]
58    Memory {
59        accounts: Vec<String>,
60        commands: Sender<(String, Command)>,
61        events: Mutex<Option<Receiver<Event>>>,
62    },
63}
64
65impl Transport {
66    /// The daemon connection; None for a test's in-memory client.
67    fn socket(&self) -> Option<&Mutex<UnixStream>> {
68        match self {
69            Transport::Socket(stream) => Some(stream),
70            #[cfg(any(test, feature = "testing"))]
71            Transport::Memory { .. } => None,
72        }
73    }
74}
75
76/// What the reader thread delivers to; `closed` once the daemon hung up.
77#[derive(Default)]
78struct Inbox {
79    closed: bool,
80    /// The daemon's ids of our own `ActionDone`s already published, whose broadcast copies are dropped.
81    own: HashSet<RequestId>,
82    /// The account of each `send` still unanswered, so a refusal can name it.
83    sent: HashMap<u64, String>,
84    subscriber: Option<Sender<Event>>,
85    /// Called after each event the subscriber gets, and once the daemon hung up.
86    wake: Option<Box<dyn Fn() + Send>>,
87    waiters: HashMap<u64, Sender<Outcome>>,
88}
89
90impl Inbox {
91    /// A reply to one of our `send`s goes to the subscriber with request 0, which marks it as this client's own. The
92    /// daemon broadcasts the same outcome right after; `receive` drops that copy of an `ActionDone`, and a failure's copy
93    /// carries the daemon's id.
94    fn deliver(&mut self, id: u64, outcome: Outcome) {
95        let account = self.sent.remove(&id);
96        if let Some(waiter) = self.waiters.remove(&id) {
97            let _ = waiter.send(outcome);
98            return;
99        }
100        let Some(account) = account else { return };
101        match outcome {
102            Outcome::Error(message) => self.publish(Event::CommandFailed {
103                account,
104                request: 0,
105                message,
106            }),
107            Outcome::Ok(Payload::Event(Event::ActionDone {
108                account,
109                folder,
110                results,
111                request,
112            })) => {
113                self.own.insert(request);
114                self.publish(Event::ActionDone {
115                    account,
116                    folder,
117                    results,
118                    request: 0,
119                });
120            }
121            Outcome::Ok(_) => {}
122        }
123    }
124
125    /// A broadcast event, unless it is the copy of our own `ActionDone` that `deliver` already published.
126    fn receive(&mut self, event: Event) {
127        if let Event::ActionDone { request, .. } = &event
128            && self.own.remove(request)
129        {
130            return;
131        }
132        self.publish(event);
133    }
134
135    fn publish(&mut self, event: Event) {
136        let Some(subscriber) = &self.subscriber else {
137            return;
138        };
139        if subscriber.send(event).is_err() {
140            self.subscriber = None;
141            self.wake = None;
142        } else if let Some(wake) = &self.wake {
143            wake();
144        }
145    }
146
147    /// Dropping the senders wakes every waiter with "the daemon stopped" and ends the subscription.
148    fn close(&mut self) {
149        let wake = self.wake.take();
150        *self = Inbox {
151            closed: true,
152            ..Inbox::default()
153        };
154        if let Some(wake) = wake {
155            wake();
156        }
157    }
158}
159
160#[derive(Debug, thiserror::Error)]
161pub(super) enum ConnectError {
162    /// Nothing answered, or the connection closed before the daemon's hello.
163    #[error("no daemon answered: {0}")]
164    Down(io::Error),
165    /// The daemon speaks another protocol or is another version; `connection` is the one that saw its hello.
166    #[error("{reason}")]
167    Mismatch {
168        reason: String,
169        /// The daemon's version, when its hello was understood.
170        theirs: Option<String>,
171        connection: BufReader<UnixStream>,
172    },
173}
174
175/// The daemon connection a CLI command or the MCP server makes on first use, and again once its daemon hung up.
176pub struct LazyClient {
177    paths: Paths,
178    /// Shared, so a long request does not hold up other callers.
179    slot: Mutex<Option<Arc<Client>>>,
180}
181
182impl LazyClient {
183    pub fn new(paths: &Paths) -> LazyClient {
184        LazyClient {
185            paths: paths.clone(),
186            slot: Mutex::new(None),
187        }
188    }
189
190    /// Starts out with a connection the caller made.
191    pub fn with_client(paths: &Paths, client: Client) -> LazyClient {
192        LazyClient {
193            paths: paths.clone(),
194            slot: Mutex::new(Some(Arc::new(client))),
195        }
196    }
197
198    /// The connection, starting the daemon when none answers.
199    pub fn client(&self) -> anyhow::Result<Arc<Client>> {
200        let mut slot = lock(&self.slot);
201        if let Some(client) = slot.as_ref().filter(|client| !client.is_closed()) {
202            return Ok(client.clone());
203        }
204        let client = Arc::new(Client::connect_or_start(&self.paths)?);
205        *slot = Some(client.clone());
206        Ok(client)
207    }
208
209    /// The daemon's reply to `command`, as far as `pick` takes it; any other reply is an error.
210    pub fn request<T>(
211        &self,
212        account: &str,
213        command: Command,
214        pick: impl FnOnce(Event) -> Result<T, Event>,
215    ) -> anyhow::Result<T> {
216        let client = self.client()?;
217        let result = client.request(account, command);
218        self.forget_if_stopped(&client, &result);
219        pick(result?).map_err(|reply| anyhow!("unexpected reply from the daemon: {reply:?}"))
220    }
221
222    /// Runs a direct action through the daemon; the outcome per uid.
223    pub fn apply(
224        &self,
225        account: &str,
226        folder: &str,
227        uids: &[u32],
228        action: &Action,
229        by: &str,
230    ) -> anyhow::Result<EventResults> {
231        let command = Command::Apply {
232            folder: folder.into(),
233            uids: uids.to_vec(),
234            action: action.clone(),
235            by: by.into(),
236        };
237        self.request(account, command, |reply| match reply {
238            Event::ActionDone { results, .. } => Ok(results),
239            other => Err(other),
240        })
241    }
242
243    /// Runs rule `name` over the account's existing mail through the daemon: messages evaluated, actions taken, and
244    /// the errors that did not stop the run.
245    pub fn apply_rule(
246        &self,
247        account: &str,
248        name: &str,
249    ) -> anyhow::Result<(usize, usize, Vec<String>)> {
250        let command = Command::ApplyRule { name: name.into() };
251        self.request(account, command, |reply| match reply {
252            Event::RuleApplied {
253                evaluated,
254                actions,
255                errors,
256                ..
257            } => Ok((evaluated, actions, errors)),
258            other => Err(other),
259        })
260    }
261
262    /// Searches the account's server for every word of `query`, in `folder` or every folder; the stored hits.
263    pub fn search_server(
264        &self,
265        account: &str,
266        folder: Option<&str>,
267        query: &str,
268    ) -> anyhow::Result<Vec<(String, u32)>> {
269        let command = Command::SearchServer {
270            folder: folder.map(str::to_string),
271            query: query.into(),
272        };
273        self.request(account, command, |reply| match reply {
274            Event::ServerSearchDone { hits, .. } => Ok(hits),
275            other => Err(other),
276        })
277    }
278
279    /// Appends a trashed `.eml` back into its folder through the daemon; the folder it went to.
280    pub fn restore(&self, account: &str, file: PathBuf) -> anyhow::Result<String> {
281        self.request(account, Command::Restore { file }, |reply| match reply {
282            Event::Restored { folder, .. } => Ok(folder),
283            other => Err(other),
284        })
285    }
286
287    /// The message's full raw bytes: from the store, or fetched through the daemon once; a stored body needs no daemon.
288    pub fn raw_message(
289        &self,
290        account: &str,
291        store: &Store,
292        msg: &Message,
293    ) -> anyhow::Result<Vec<u8>> {
294        if let Some(raw) = store.raw(&msg.folder, msg.uid)? {
295            return Ok(raw);
296        }
297        let fetch = Command::FetchBody {
298            folder: msg.folder.clone(),
299            uid: msg.uid,
300        };
301        self.request(account, fetch, Ok)?;
302        store
303            .raw(&msg.folder, msg.uid)?
304            .ok_or_else(|| anyhow!("no body for {}/{} after fetching", msg.folder, msg.uid))
305    }
306
307    /// Drops `client` when its daemon is gone, so the next call reconnects.
308    fn forget_if_stopped<T>(&self, client: &Arc<Client>, result: &anyhow::Result<T>) {
309        let stopped = result
310            .as_ref()
311            .is_err_and(|e| e.chain().any(|cause| cause.is::<Stopped>()));
312        let mut slot = lock(&self.slot);
313        if stopped && slot.as_ref().is_some_and(|held| Arc::ptr_eq(held, client)) {
314            *slot = None;
315        }
316    }
317}
318
319impl Client {
320    /// Connects to a running daemon; Err when none answers or it is another version.
321    #[cfg(any(test, feature = "testing"))]
322    pub fn connect(paths: &Paths) -> anyhow::Result<Client> {
323        Ok(connect_as(paths, VERSION, false)?)
324    }
325
326    /// Connects to whatever daemon answers, of any version; None when nothing does.
327    pub fn connect_any_version(paths: &Paths) -> anyhow::Result<Option<Client>> {
328        match connect_as(paths, VERSION, true) {
329            Ok(client) => Ok(Some(client)),
330            Err(ConnectError::Down(e))
331                if matches!(
332                    e.kind(),
333                    io::ErrorKind::ConnectionRefused
334                        | io::ErrorKind::NotFound
335                        | io::ErrorKind::UnexpectedEof
336                ) =>
337            {
338                Ok(None)
339            }
340            Err(e) => Err(e.into()),
341        }
342    }
343
344    /// Connects, starting this binary as the daemon when none answers, and replacing a daemon of an older version or
345    /// another protocol; a newer daemon is an error, so an old window never kills its upgrade.
346    pub fn connect_or_start(paths: &Paths) -> anyhow::Result<Client> {
347        let exe = crate::paths::stable_exe().context("finding this postvak")?;
348        Client::connect_or_start_with(paths, &exe, VERSION)
349    }
350
351    pub fn connect_or_start_with(
352        paths: &Paths,
353        exe: &Path,
354        version: &str,
355    ) -> anyhow::Result<Client> {
356        match connect_as(paths, version, false) {
357            Ok(client) => return Ok(client),
358            Err(ConnectError::Mismatch {
359                theirs: Some(theirs),
360                ..
361            }) if is_newer(&theirs, version) => {
362                bail!(NewerDaemon {
363                    theirs,
364                    ours: version.into()
365                })
366            }
367            Err(ConnectError::Mismatch {
368                reason, connection, ..
369            }) => {
370                log::info!("{reason}; restarting it");
371                stop_daemon(paths, connection);
372            }
373            Err(ConnectError::Down(_)) => {}
374        }
375        start_daemon(paths, exe, version)
376    }
377
378    /// True once the daemon hung up; a holder should connect again.
379    pub fn is_closed(&self) -> bool {
380        lock(&self.inbox).closed
381    }
382
383    /// Sends and waits for the reply, up to 120 s unless `reply_timeout` says the command may take longer.
384    pub fn request(&self, account: &str, command: Command) -> anyhow::Result<Event> {
385        let timeout = reply_timeout(&command);
386        self.request_until(account, command, timeout)
387    }
388
389    pub(super) fn request_until(
390        &self,
391        account: &str,
392        command: Command,
393        timeout: Option<Duration>,
394    ) -> anyhow::Result<Event> {
395        let message = |id| ClientMessage::Command {
396            id,
397            account: account.into(),
398            command,
399        };
400        match self.call(message, timeout)? {
401            Payload::Event(event) => Ok(event),
402            other => bail!("unexpected reply from the daemon: {other:?}"),
403        }
404    }
405
406    /// Sends without waiting; a refusal arrives as `Event::CommandFailed` on the subscription.
407    pub fn send(&self, account: &str, command: Command) -> bool {
408        #[cfg(any(test, feature = "testing"))]
409        if let Transport::Memory { commands, .. } = &self.transport {
410            return commands.send((account.into(), command)).is_ok();
411        }
412        let Some(stream) = self.transport.socket() else {
413            return false;
414        };
415        let id = self.next_id.fetch_add(1, Ordering::Relaxed);
416        let message = ClientMessage::Command {
417            id,
418            account: account.into(),
419            command,
420        };
421        let line = match wire::line(&message) {
422            Ok(line) => line,
423            Err(e) => {
424                lock(&self.inbox).publish(Event::CommandFailed {
425                    account: account.into(),
426                    request: 0,
427                    message: encode_failure(&e),
428                });
429                return true;
430            }
431        };
432        {
433            let mut inbox = lock(&self.inbox);
434            if inbox.closed {
435                return false;
436            }
437            inbox.sent.insert(id, account.into());
438        }
439        let written = lock(stream).write_all(line.as_bytes()).is_ok();
440        if !written {
441            lock(&self.inbox).sent.remove(&id);
442        }
443        written
444    }
445
446    /// Every broadcast event from now on, plus refusals of `send`. Call once.
447    pub fn subscribe(&self) -> anyhow::Result<Receiver<Event>> {
448        self.subscribe_waking(|| {})
449    }
450
451    /// As `subscribe`, calling `wake` after each event and once the daemon hung up, so a window can repaint.
452    pub fn subscribe_waking(
453        &self,
454        wake: impl Fn() + Send + 'static,
455    ) -> anyhow::Result<Receiver<Event>> {
456        #[cfg(any(test, feature = "testing"))]
457        if let Transport::Memory { events, .. } = &self.transport {
458            return lock(events)
459                .take()
460                .ok_or_else(|| anyhow!("already subscribed"));
461        }
462        let (subscriber, events) = mpsc::channel();
463        {
464            let mut inbox = lock(&self.inbox);
465            if inbox.subscriber.is_some() {
466                bail!("already subscribed");
467            }
468            inbox.subscriber = Some(subscriber);
469            inbox.wake = Some(Box::new(wake));
470        }
471        if let Err(e) = self.call(|id| ClientMessage::Subscribe { id }, Some(REPLY_TIMEOUT)) {
472            let mut inbox = lock(&self.inbox);
473            inbox.subscriber = None;
474            inbox.wake = None;
475            return Err(e);
476        }
477        Ok(events)
478    }
479
480    pub fn status(&self) -> anyhow::Result<Status> {
481        #[cfg(any(test, feature = "testing"))]
482        if let Transport::Memory { accounts, .. } = &self.transport {
483            return Ok(memory_status(accounts));
484        }
485        match self.call(|id| ClientMessage::Status { id }, Some(REPLY_TIMEOUT))? {
486            Payload::Status(status) => Ok(status),
487            other => bail!("unexpected reply from the daemon: {other:?}"),
488        }
489    }
490
491    pub fn shutdown(&self) -> anyhow::Result<()> {
492        self.call(|id| ClientMessage::Shutdown { id }, Some(REPLY_TIMEOUT))?;
493        Ok(())
494    }
495
496    /// Stops the daemon and waits for its socket to go; returns the pid it had.
497    pub fn stop_and_wait(&self, paths: &Paths) -> anyhow::Result<u32> {
498        let pid = self.status()?.pid;
499        self.shutdown()?;
500        let deadline = Instant::now() + STOP_WAIT;
501        while paths.daemon_socket().exists() {
502            if Instant::now() >= deadline {
503                bail!("the daemon (pid {pid}) did not stop within 5 s");
504            }
505            std::thread::sleep(Duration::from_millis(50));
506        }
507        Ok(pid)
508    }
509
510    /// A client with no daemon behind it: commands appear on the returned receiver, events are injected by the test.
511    #[cfg(any(test, feature = "testing"))]
512    pub fn in_memory(accounts: &[&str]) -> (Client, Receiver<(String, Command)>, Sender<Event>) {
513        let (commands, sent) = mpsc::channel();
514        let (inject, events) = mpsc::channel();
515        let client = Client {
516            transport: Transport::Memory {
517                accounts: accounts.iter().map(|name| name.to_string()).collect(),
518                commands,
519                events: Mutex::new(Some(events)),
520            },
521            inbox: Arc::default(),
522            next_id: AtomicU64::new(1),
523            log: PathBuf::new(),
524        };
525        (client, sent, inject)
526    }
527
528    /// Sends the message `with_id` builds and waits for its reply; without a timeout only a lost daemon ends the wait.
529    fn call(
530        &self,
531        with_id: impl FnOnce(u64) -> ClientMessage,
532        timeout: Option<Duration>,
533    ) -> anyhow::Result<Payload> {
534        let Some(stream) = self.transport.socket() else {
535            bail!("in-memory client has no daemon");
536        };
537        let id = self.next_id.fetch_add(1, Ordering::Relaxed);
538        let line = wire::line(&with_id(id)).map_err(|e| anyhow!(encode_failure(&e)))?;
539        let (waiter, reply) = mpsc::channel();
540        {
541            let mut inbox = lock(&self.inbox);
542            if inbox.closed {
543                return Err(self.stopped());
544            }
545            inbox.waiters.insert(id, waiter);
546        }
547        if lock(stream).write_all(line.as_bytes()).is_err() {
548            lock(&self.inbox).waiters.remove(&id);
549            return Err(self.stopped());
550        }
551        let received = match timeout {
552            Some(timeout) => reply.recv_timeout(timeout),
553            None => reply.recv().map_err(|_| RecvTimeoutError::Disconnected),
554        };
555        match received {
556            Ok(Outcome::Ok(payload)) => Ok(payload),
557            Ok(Outcome::Error(message)) => Err(anyhow!(message)),
558            Err(RecvTimeoutError::Timeout) => {
559                lock(&self.inbox).waiters.remove(&id);
560                Err(anyhow!(NO_REPLY))
561            }
562            Err(RecvTimeoutError::Disconnected) => Err(self.stopped()),
563        }
564    }
565
566    fn stopped(&self) -> anyhow::Error {
567        anyhow::Error::new(Stopped(format!(
568            "the daemon stopped; see {}",
569            self.log.display()
570        )))
571    }
572}
573
574impl Drop for Client {
575    fn drop(&mut self) {
576        if let Some(stream) = self.transport.socket() {
577            let _ = lock(stream).shutdown(Shutdown::Both);
578        }
579    }
580}
581
582/// How long `request` waits: a sync pass, a body fetch over a folder or a rule over stored mail can outlast 120 s, and
583/// giving up early would drop the client and let an auto-started daemon idle out mid-run.
584pub(super) fn reply_timeout(command: &Command) -> Option<Duration> {
585    match command {
586        Command::ApplyRule { .. }
587        | Command::FetchBodies { .. }
588        | Command::SearchServer { .. }
589        | Command::SyncNow => None,
590        Command::Apply { .. } | Command::FetchBody { .. } | Command::Restore { .. } => {
591            Some(REPLY_TIMEOUT)
592        }
593    }
594}
595
596fn encode_failure(error: &io::Error) -> String {
597    format!("could not encode the request: {error}")
598}
599
600#[cfg(any(test, feature = "testing"))]
601fn memory_status(accounts: &[String]) -> Status {
602    use super::wire::AccountStatus;
603    Status {
604        pid: std::process::id(),
605        version: VERSION.into(),
606        uptime_secs: 0,
607        clients: 1,
608        accounts: accounts
609            .iter()
610            .map(|name| AccountStatus {
611                name: name.clone(),
612                activity: None,
613            })
614            .collect(),
615    }
616}
617
618/// Exchanges hello as `version`; with `any_version` only the protocol has to match.
619pub(super) fn connect_as(
620    paths: &Paths,
621    version: &str,
622    any_version: bool,
623) -> Result<Client, ConnectError> {
624    let stream = UnixStream::connect(paths.daemon_socket()).map_err(ConnectError::Down)?;
625    let mut reader = handshake(&stream, version, any_version)?;
626    let inbox = Arc::new(Mutex::new(Inbox::default()));
627    let delivered = inbox.clone();
628    std::thread::Builder::new()
629        .name("daemon-replies".into())
630        .spawn(move || read_messages(&mut reader, &delivered))
631        .map_err(ConnectError::Down)?;
632    Ok(Client {
633        transport: Transport::Socket(Mutex::new(stream)),
634        inbox,
635        next_id: AtomicU64::new(1),
636        log: paths.daemon_log(),
637    })
638}
639
640/// Sends our hello and checks the daemon's; returns the reader positioned after it.
641fn handshake(
642    stream: &UnixStream,
643    version: &str,
644    any_version: bool,
645) -> Result<BufReader<UnixStream>, ConnectError> {
646    let hello = ClientMessage::Hello {
647        protocol: PROTOCOL,
648        version: version.into(),
649    };
650    wire::write_line(&mut &*stream, &hello).map_err(ConnectError::Down)?;
651    stream
652        .set_read_timeout(Some(START_WAIT))
653        .map_err(ConnectError::Down)?;
654    let mut reader = BufReader::new(stream.try_clone().map_err(ConnectError::Down)?);
655    let mut line = String::new();
656    if reader.read_line(&mut line).map_err(ConnectError::Down)? == 0 {
657        return Err(ConnectError::Down(io::ErrorKind::UnexpectedEof.into()));
658    }
659    let (reason, theirs) = match serde_json::from_str(&line) {
660        Ok(DaemonMessage::Hello {
661            protocol,
662            version: theirs,
663            ..
664        }) if protocol == PROTOCOL && (any_version || theirs == version) => {
665            stream.set_read_timeout(None).map_err(ConnectError::Down)?;
666            return Ok(reader);
667        }
668        Ok(DaemonMessage::Hello {
669            protocol,
670            version: theirs,
671            ..
672        }) => (
673            format!("daemon is version {theirs} (protocol {protocol}); this postvak is {version}"),
674            Some(theirs),
675        ),
676        _ => (
677            format!("the daemon's hello was not understood; this postvak is {version}"),
678            None,
679        ),
680    };
681    Err(ConnectError::Mismatch {
682        reason,
683        theirs,
684        connection: reader,
685    })
686}
687
688/// Whether version `theirs` is newer than `ours`, comparing dotted numbers; a pre-release or garbage sorts as older.
689pub(super) fn is_newer(theirs: &str, ours: &str) -> bool {
690    fn parts(version: &str) -> Option<Vec<u64>> {
691        version.split('.').map(|part| part.parse().ok()).collect()
692    }
693    parts(theirs) > parts(ours)
694}
695
696fn read_messages(reader: &mut BufReader<UnixStream>, inbox: &Mutex<Inbox>) {
697    for line in reader.lines() {
698        let Ok(line) = line else { break };
699        match serde_json::from_str(&line) {
700            Ok(DaemonMessage::Reply { id, outcome }) => lock(inbox).deliver(id, outcome),
701            Ok(DaemonMessage::Event(event)) => lock(inbox).receive(event),
702            Ok(DaemonMessage::Hello { .. }) => {}
703            Err(e) => log::warn!("unreadable message from the daemon: {e}"),
704        }
705    }
706    lock(inbox).close();
707}
708
709/// Shuts down the daemon behind `connection`, waits for it to hang up, then up to 5 s for its socket to go.
710fn stop_daemon(paths: &Paths, mut connection: BufReader<UnixStream>) {
711    let shutdown = ClientMessage::Shutdown { id: 0 };
712    if wire::write_line(&mut connection.get_ref(), &shutdown).is_ok() {
713        // The read timeout from the handshake bounds this wait.
714        let _ = io::copy(&mut connection, &mut io::sink());
715    }
716    let deadline = Instant::now() + START_WAIT;
717    while paths.daemon_socket().exists() && Instant::now() < deadline {
718        std::thread::sleep(RETRY);
719    }
720}
721
722/// Spawns `exe` as the daemon and connects within 5 s, spawning again when it exited and no socket is left.
723fn start_daemon(paths: &Paths, exe: &Path, version: &str) -> anyhow::Result<Client> {
724    let deadline = Instant::now() + START_WAIT;
725    let mut child = spawn_daemon(paths, exe)?;
726    let connected = loop {
727        std::thread::sleep(RETRY);
728        // We started our own binary; whatever now serves is the daemon to use, so only the protocol must match.
729        match connect_as(paths, version, true) {
730            Ok(client) => break Ok(client),
731            Err(ConnectError::Mismatch { reason, .. }) => break Err(anyhow!(reason)),
732            Err(ConnectError::Down(_)) => {}
733        }
734        if Instant::now() >= deadline {
735            let log = paths.daemon_log();
736            break Err(anyhow!(
737                "could not start the daemon; last lines of {}:\n{}",
738                log.display(),
739                log_tail(&log)
740            ));
741        }
742        if matches!(child.try_wait(), Ok(Some(_))) && !paths.daemon_socket().exists() {
743            child = spawn_daemon(paths, exe)?;
744        }
745    };
746    reap(child);
747    connected
748}
749
750fn spawn_daemon(paths: &Paths, exe: &Path) -> anyhow::Result<Child> {
751    paths::create_private_dir(&paths.state_dir)?;
752    let log = File::options()
753        .create(true)
754        .append(true)
755        .mode(0o600)
756        .open(paths.daemon_log())
757        .with_context(|| format!("opening {}", paths.daemon_log().display()))?;
758    let mut command = std::process::Command::new(exe);
759    command
760        .args(["run", "--idle-exit", "60"])
761        .current_dir("/")
762        .stdin(Stdio::null())
763        .stdout(log.try_clone()?)
764        .stderr(log)
765        .process_group(0);
766    if let Some(home) = &paths.home {
767        command.env("POSTVAK_HOME", home);
768    }
769    command
770        .spawn()
771        .with_context(|| format!("starting {}", exe.display()))
772}
773
774/// Waits on the daemon in the background so it does not linger as a zombie.
775fn reap(mut child: Child) {
776    std::thread::spawn(move || child.wait());
777}
778
779fn log_tail(log: &Path) -> String {
780    let text = fs::read(log)
781        .map(|bytes| String::from_utf8_lossy(&bytes).into_owned())
782        .unwrap_or_default();
783    let lines: Vec<&str> = text.lines().collect();
784    lines[lines.len().saturating_sub(LOG_TAIL_LINES)..].join("\n")
785}
postvak_core/daemon/client.rs

postvak_core/daemon/
client.rs
