postvak_core/
engine.rs

1//! Runs one sync thread per account and routes commands to them; the daemon runs it.
2use std::sync::Arc;
3use std::sync::atomic::{AtomicBool, Ordering};
4use std::sync::mpsc::{self, Receiver, Sender};
5use std::thread::JoinHandle;
6
7use crate::config::{AccountConfig, Config};
8use crate::mail_ops::MailOps;
9use crate::paths::Paths;
10use crate::sync::{self, Activity, Command, Event, Job, RequestId, SyncError};
11
12pub type Connector =
13    Arc<dyn Fn(&AccountConfig) -> Result<Box<dyn MailOps>, SyncError> + Send + Sync>;
14
15/// Connects to the account's real IMAP server.
16pub fn imap_connector() -> Connector {
17    Arc::new(|account| Ok(Box::new(sync::connect(account)?)))
18}
19
20struct AccountThread {
21    config: AccountConfig,
22    commands: Option<Sender<Job>>,
23    wake: Arc<AtomicBool>,
24    shutdown: Arc<AtomicBool>,
25    handle: Option<JoinHandle<()>>,
26}
27
28impl AccountThread {
29    fn not_started(config: AccountConfig) -> AccountThread {
30        AccountThread {
31            config,
32            commands: None,
33            wake: Arc::new(AtomicBool::new(false)),
34            shutdown: Arc::new(AtomicBool::new(false)),
35            handle: None,
36        }
37    }
38
39    fn signal_stop(&self) {
40        self.shutdown.store(true, Ordering::Release);
41        self.wake.store(true, Ordering::Release);
42    }
43
44    fn join(&mut self) {
45        if let Some(handle) = self.handle.take() {
46            let _ = handle.join();
47        }
48    }
49}
50
51struct Spawner {
52    paths: Paths,
53    events: Sender<Event>,
54    connect: Connector,
55}
56
57impl Spawner {
58    /// Spawns the account's sync thread; one that cannot spawn is reported once and stays without a thread.
59    fn spawn(&self, config: &AccountConfig) -> AccountThread {
60        let name = &config.name;
61        let (commands_tx, commands) = mpsc::channel();
62        let mut thread = AccountThread::not_started(config.clone());
63        let spawned = std::thread::Builder::new()
64            .name(format!("sync-{name}"))
65            .spawn({
66                let (account, paths, events) =
67                    (config.clone(), self.paths.clone(), self.events.clone());
68                let (shutdown, wake) = (thread.shutdown.clone(), thread.wake.clone());
69                let connect = self.connect.clone();
70                move || {
71                    let target = account.clone();
72                    sync::run_loop(
73                        account,
74                        paths,
75                        events,
76                        shutdown,
77                        commands,
78                        wake,
79                        move || connect(&target),
80                    )
81                }
82            });
83        match spawned {
84            Ok(handle) => {
85                thread.commands = Some(commands_tx);
86                thread.handle = Some(handle);
87            }
88            Err(e) => {
89                let reason = format!("could not start its sync thread: {e}");
90                let _ = self.events.send(Event::Error {
91                    account: name.clone(),
92                    message: reason.clone(),
93                });
94                // The daemon keeps the last activity, so status and windows that connect later still see why.
95                let _ = self.events.send(Event::Activity {
96                    account: name.clone(),
97                    activity: Activity::NotRunning { reason },
98                });
99            }
100        }
101        thread
102    }
103}
104
105pub struct Engine {
106    spawner: Spawner,
107    threads: Vec<AccountThread>,
108    /// The configured accounts in order; one whose old thread is still in `stopping` starts once that thread ended.
109    wanted: Vec<AccountConfig>,
110    /// Signalled threads of removed or changed accounts, by account name, until they end.
111    stopping: Vec<(String, JoinHandle<()>)>,
112}
113
114impl Engine {
115    /// Spawns a sync thread per account.
116    pub fn start(config: &Config, paths: &Paths, connect: Connector) -> (Engine, Receiver<Event>) {
117        let (events, received) = mpsc::channel();
118        let spawner = Spawner {
119            paths: paths.clone(),
120            events,
121            connect,
122        };
123        let threads = config
124            .accounts
125            .iter()
126            .map(|account| spawner.spawn(account))
127            .collect();
128        let engine = Engine {
129            spawner,
130            threads,
131            wanted: config.accounts.clone(),
132            stopping: Vec::new(),
133        };
134        (engine, received)
135    }
136
137    /// Signals the threads of removed and changed accounts, starts new ones, and leaves the rest; never waits. A changed
138    /// account starts again from `start_ready` once its old thread ended.
139    pub fn apply_config(&mut self, config: &Config) {
140        let (kept, stale): (Vec<_>, Vec<_>) = std::mem::take(&mut self.threads)
141            .into_iter()
142            .partition(|thread| config.accounts.contains(&thread.config));
143        for mut thread in stale {
144            thread.signal_stop();
145            if let Some(handle) = thread.handle.take() {
146                self.stopping.push((thread.config.name, handle));
147            }
148        }
149        self.threads = kept;
150        self.wanted = config.accounts.clone();
151        self.start_ready();
152    }
153
154    /// Starts each configured account that has no thread and no old one still ending; true once every one runs.
155    pub fn start_ready(&mut self) -> bool {
156        if self.threads.len() == self.wanted.len() {
157            return true;
158        }
159        self.stopping.retain(|(_, handle)| !handle.is_finished());
160        let mut idle = std::mem::take(&mut self.threads);
161        let mut threads = Vec::new();
162        for account in &self.wanted {
163            if let Some(index) = idle.iter().position(|thread| thread.config == *account) {
164                threads.push(idle.swap_remove(index));
165            } else if !self.stopping.iter().any(|(name, _)| *name == account.name) {
166                threads.push(self.spawner.spawn(account));
167            }
168        }
169        self.threads = threads;
170        self.threads.len() == self.wanted.len()
171    }
172
173    pub fn accounts(&self) -> Vec<String> {
174        self.threads
175            .iter()
176            .map(|thread| thread.config.name.clone())
177            .collect()
178    }
179
180    /// Whether the account is configured but has no thread yet, as while its old thread ends after a config change.
181    pub fn is_restarting(&self, account: &str) -> bool {
182        self.wanted.iter().any(|wanted| wanted.name == account)
183            && !self
184                .threads
185                .iter()
186                .any(|thread| thread.config.name == account)
187    }
188
189    /// Queues the command and wakes the account's IDLE; false when the account is not running.
190    pub fn send(&self, account: &str, request: RequestId, command: Command) -> bool {
191        let job = Job { request, command };
192        self.threads
193            .iter()
194            .find(|thread| thread.config.name == account)
195            .and_then(|thread| thread.commands.as_ref().map(|commands| (thread, commands)))
196            .is_some_and(|(thread, commands)| {
197                let sent = commands.send(job).is_ok();
198                thread.wake.store(true, Ordering::Release);
199                sent
200            })
201    }
202}
203
204impl Drop for Engine {
205    /// Signals every thread first so they stop together, then joins them.
206    fn drop(&mut self) {
207        self.threads.iter().for_each(AccountThread::signal_stop);
208        self.threads.iter_mut().for_each(AccountThread::join);
209        for (_, handle) in self.stopping.drain(..) {
210            let _ = handle.join();
211        }
212    }
213}
214
215#[cfg(test)]
216mod tests {
217    use super::*;
218    use crate::config::{AccountConfig, PasswordSource};
219
220    fn offline_account(name: &str) -> AccountConfig {
221        let password = PasswordSource::Command {
222            command: "printf x".into(),
223        };
224        AccountConfig {
225            port: 1,
226            notify: false,
227            ..AccountConfig::new(name, "127.0.0.1", "me@example.com", password)
228        }
229    }
230
231    #[test]
232    fn every_configured_account_starts() {
233        let dir = tempfile::tempdir().unwrap();
234        let config = config_of(&["home", "work"]);
235        let (engine, _events) = Engine::start(&config, &Paths::under(dir.path()), imap_connector());
236        assert_eq!(engine.accounts(), ["home", "work"]);
237    }
238
239    #[test]
240    fn stop_returns_promptly_while_an_account_is_offline() {
241        let dir = tempfile::tempdir().unwrap();
242        let config = Config {
243            accounts: vec![offline_account("work")],
244            ..Default::default()
245        };
246        let (engine, events) = Engine::start(&config, &Paths::under(dir.path()), imap_connector());
247        assert_eq!(engine.accounts(), ["work"]);
248        let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5);
249        loop {
250            let left = deadline.saturating_duration_since(std::time::Instant::now());
251            if let Event::Activity {
252                activity: Activity::Offline { .. },
253                ..
254            } = events.recv_timeout(left).unwrap()
255            {
256                break;
257            }
258        }
259        assert!(engine.send("work", 1, Command::SyncNow));
260        assert!(!engine.send("nope", 2, Command::SyncNow));
261        let started = std::time::Instant::now();
262        drop(engine);
263        assert!(started.elapsed() < std::time::Duration::from_secs(2));
264    }
265
266    #[test]
267    fn dropping_the_engine_stops_its_threads() {
268        let dir = tempfile::tempdir().unwrap();
269        let paths = Paths::under(dir.path());
270        let config = Config {
271            accounts: vec![offline_account("work")],
272            ..Default::default()
273        };
274        let (engine, events) = Engine::start(&config, &paths, imap_connector());
275        let started = std::time::Instant::now();
276        drop(engine);
277        assert!(started.elapsed() < std::time::Duration::from_secs(2));
278        events.try_iter().for_each(drop);
279        assert_eq!(
280            events.try_recv(),
281            Err(mpsc::TryRecvError::Disconnected),
282            "every sync thread has ended"
283        );
284    }
285
286    #[test]
287    fn a_command_for_an_offline_account_fails_with_its_request_id() {
288        let dir = tempfile::tempdir().unwrap();
289        let config = Config {
290            accounts: vec![offline_account("work")],
291            ..Default::default()
292        };
293        let (engine, events) = Engine::start(&config, &Paths::under(dir.path()), imap_connector());
294        let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5);
295        let mut failed = None;
296        while failed.is_none() {
297            let left = deadline.saturating_duration_since(std::time::Instant::now());
298            match events.recv_timeout(left).unwrap() {
299                Event::Activity {
300                    activity: Activity::Offline { .. },
301                    ..
302                } => assert!(engine.send("work", 4, Command::SyncNow)),
303                Event::CommandFailed { request, .. } => failed = Some(request),
304                _ => {}
305            }
306        }
307        assert_eq!(failed, Some(4));
308    }
309
310    use crate::mail_ops::RecordingOps;
311    use std::sync::Mutex;
312
313    const WAIT: std::time::Duration = std::time::Duration::from_secs(5);
314
315    /// A connector whose fake server idles until woken; every connection attempt is announced by account name.
316    fn recording_connector() -> (Connector, Receiver<String>) {
317        let (connected, announced) = mpsc::channel();
318        let connected = Mutex::new(connected);
319        let connector: Connector = Arc::new(move |account: &AccountConfig| {
320            let _ = connected.lock().unwrap().send(account.name.clone());
321            Ok(Box::new(RecordingOps::default().with_folder("INBOX", None)) as Box<dyn MailOps>)
322        });
323        (connector, announced)
324    }
325
326    fn config_of(names: &[&str]) -> Config {
327        Config {
328            accounts: names.iter().map(|name| offline_account(name)).collect(),
329            ..Default::default()
330        }
331    }
332
333    /// Starts what `apply_config` left waiting, for up to 5 s.
334    fn settle(engine: &mut Engine) {
335        let deadline = std::time::Instant::now() + WAIT;
336        while !engine.start_ready() {
337            assert!(
338                std::time::Instant::now() < deadline,
339                "an old thread never ended"
340            );
341            std::thread::sleep(std::time::Duration::from_millis(10));
342        }
343    }
344
345    #[test]
346    fn apply_config_never_waits_and_restarts_an_account_after_its_old_thread_ended() {
347        let dir = tempfile::tempdir().unwrap();
348        let (said, log) = mpsc::channel();
349        let said = Mutex::new(said);
350        let connector: Connector = Arc::new(move |account: &AccountConfig| {
351            let say = |what: &str| {
352                let line = format!("{what} {}", account.sync_interval_secs);
353                let _ = said.lock().unwrap().send(line);
354            };
355            say("connecting");
356            std::thread::sleep(std::time::Duration::from_secs(1));
357            say("connected");
358            Ok(Box::new(RecordingOps::default().with_folder("INBOX", None)) as Box<dyn MailOps>)
359        });
360        let mut config = config_of(&["a"]);
361        let (mut engine, _events) = Engine::start(&config, &Paths::under(dir.path()), connector);
362        assert_eq!(log.recv_timeout(WAIT).unwrap(), "connecting 120");
363        config.accounts[0].sync_interval_secs = 300;
364        let started = std::time::Instant::now();
365        engine.apply_config(&config);
366        assert!(started.elapsed() < std::time::Duration::from_millis(200));
367        assert!(
368            !engine.send("a", 1, Command::SyncNow),
369            "a is between threads"
370        );
371        settle(&mut engine);
372        assert_eq!(log.recv_timeout(WAIT).unwrap(), "connected 120");
373        assert_eq!(log.recv_timeout(WAIT).unwrap(), "connecting 300");
374    }
375
376    #[test]
377    fn the_engine_can_move_to_another_thread() {
378        fn assert_send<T: Send>() {}
379        assert_send::<Engine>();
380    }
381
382    #[test]
383    fn apply_config_restarts_only_changed_accounts() {
384        let dir = tempfile::tempdir().unwrap();
385        let (connector, connected) = recording_connector();
386        let mut config = config_of(&["a", "b"]);
387        let (mut engine, _events) = Engine::start(&config, &Paths::under(dir.path()), connector);
388        let mut first_connections = [
389            connected.recv_timeout(WAIT).unwrap(),
390            connected.recv_timeout(WAIT).unwrap(),
391        ];
392        first_connections.sort();
393        assert_eq!(first_connections, ["a", "b"]);
394
395        config.accounts[1].sync_interval_secs = 300;
396        engine.apply_config(&config);
397        settle(&mut engine);
398        assert_eq!(connected.recv_timeout(WAIT).unwrap(), "b");
399        assert_eq!(engine.accounts(), ["a", "b"]);
400
401        engine.apply_config(&config_of(&["a"]));
402        assert_eq!(engine.accounts(), ["a"]);
403        assert!(!engine.send("b", 1, Command::SyncNow));
404
405        engine.apply_config(&config_of(&["a", "c"]));
406        assert_eq!(connected.recv_timeout(WAIT).unwrap(), "c");
407        assert_eq!(engine.accounts(), ["a", "c"]);
408        drop(engine);
409        assert_eq!(
410            connected.try_iter().collect::<Vec<_>>(),
411            Vec::<String>::new(),
412            "a was never restarted, and no thread connects after the engine drops"
413        );
414    }
415
416    #[test]
417    fn a_command_reaches_the_account_thread() {
418        let dir = tempfile::tempdir().unwrap();
419        let (connector, _connected) = recording_connector();
420        let (engine, events) =
421            Engine::start(&config_of(&["a"]), &Paths::under(dir.path()), connector);
422        assert!(engine.send("a", 11, Command::SyncNow));
423        let deadline = std::time::Instant::now() + WAIT;
424        loop {
425            let left = deadline.saturating_duration_since(std::time::Instant::now());
426            if let Event::Synced { requests, .. } = events.recv_timeout(left).unwrap()
427                && requests == vec![11]
428            {
429                break;
430            }
431        }
432    }
433}
postvak_core/engine.rs

postvak_core/
engine.rs
