cadmus_core/task/
mod.rs

1//! Long-running background task infrastructure.
2//!
3//! This module provides a trait-based system for defining and managing
4//! background tasks that run alongside the main application loop.
5//!
6//! # Architecture
7//!
8//! - [`BackgroundTask`] trait defines the interface for long-running tasks
9//! - [`TaskManager`] spawns and manages task lifecycles
10//! - [`ShutdownSignal`] provides graceful shutdown coordination
11//!
12//! # Example
13//!
14//! ```no_run
15//! use std::sync::mpsc::Sender;
16//! use std::time::Duration;
17//!
18//! use cadmus_core::task::{BackgroundTask, ShutdownSignal, TaskId};
19//! use cadmus_core::view::Event;
20//!
21//! struct MyTask;
22//!
23//! impl BackgroundTask for MyTask {
24//!     fn id(&self) -> TaskId {
25//!         TaskId::Placeholder
26//!     }
27//!
28//!     fn run(&mut self, hub: &Sender<Event>, shutdown: &ShutdownSignal) {
29//!         while !shutdown.should_stop() {
30//!             // Do work...
31//!             if shutdown.wait(Duration::from_secs(60)) {
32//!                 break;
33//!             }
34//!         }
35//!     }
36//! }
37//! ```
38
39#[cfg(any(all(feature = "test", feature = "kobo"), doc))]
40mod dbus_monitor;
41pub mod dictionary_index;
42#[cfg(any(feature = "test", doc))]
43mod hello_world;
44pub mod import;
45pub mod thumbnail;
46#[cfg(any(feature = "kobo", doc))]
47mod wifi_status_monitor;
48
49use std::collections::{HashMap, VecDeque};
50use std::sync::atomic::{AtomicBool, Ordering};
51use std::sync::mpsc::{self, Receiver, Sender};
52use std::thread::{self, JoinHandle};
53use std::time::Duration;
54
55use thiserror::Error;
56
57use crate::db::Database;
58use crate::settings::Settings;
59use crate::view::Event;
60
61/// Errors that can occur during task management operations.
62#[derive(Error, Debug)]
63pub enum TaskError {
64    /// A task with the given ID is already running.
65    #[error("task '{0}' is already running")]
66    AlreadyRunning(TaskId),
67
68    /// A task with the given ID is not running.
69    #[error("task '{0}' is not running")]
70    NotRunning(TaskId),
71}
72
73/// Unique identifier for a background task.
74#[derive(Debug, Clone, PartialEq, Eq, Hash)]
75pub enum TaskId {
76    /// A tmp placeholder until there is a Task always available.
77    Placeholder,
78    /// Library import task.
79    Import,
80    /// Thumbnail extraction background task.
81    ThumbnailExtraction,
82    /// Dictionary index background task.
83    DictionaryIndex,
84    /// The example task that prints periodically (test builds only).
85    #[cfg(any(feature = "test", doc))]
86    HelloWorld,
87    /// D-Bus system bus monitor (test + kobo builds only).
88    #[cfg(any(all(feature = "test", feature = "kobo"), doc))]
89    DbusMonitor,
90    /// WiFi status monitor using dhcpcd-dbus (kobo builds only).
91    #[cfg(any(feature = "kobo", doc))]
92    WifiStatusMonitor,
93    /// Test-only task for unit tests.
94    #[cfg(test)]
95    TestTask,
96    /// Second test-only task for unit tests.
97    #[cfg(test)]
98    TestTask2,
99}
100
101impl std::fmt::Display for TaskId {
102    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
103        match self {
104            TaskId::Placeholder => write!(f, "placeholder"),
105            TaskId::Import => write!(f, "import"),
106            TaskId::ThumbnailExtraction => write!(f, "thumbnail_extraction"),
107            TaskId::DictionaryIndex => write!(f, "dictionary_index"),
108            #[cfg(feature = "test")]
109            TaskId::HelloWorld => write!(f, "hello_world"),
110            #[cfg(all(feature = "test", feature = "kobo"))]
111            TaskId::DbusMonitor => write!(f, "dbus_monitor"),
112            #[cfg(feature = "kobo")]
113            TaskId::WifiStatusMonitor => write!(f, "wifi_status_monitor"),
114            #[cfg(test)]
115            TaskId::TestTask => write!(f, "test_task"),
116            #[cfg(test)]
117            TaskId::TestTask2 => write!(f, "test_task_2"),
118        }
119    }
120}
121
122/// Signal for coordinating graceful shutdown of background tasks.
123///
124/// Tasks should periodically check [`should_stop`](Self::should_stop) or use
125/// [`wait`](Self::wait) to interrupt sleep when shutdown is requested.
126pub struct ShutdownSignal {
127    receiver: Receiver<()>,
128    /// Keeps the sender alive when no external owner exists, preventing
129    /// spurious `Disconnected` errors in `wait()`.
130    _sender_anchor: Option<Sender<()>>,
131    stopped: AtomicBool,
132}
133
134impl ShutdownSignal {
135    fn new(receiver: Receiver<()>) -> Self {
136        Self {
137            receiver,
138            _sender_anchor: None,
139            stopped: AtomicBool::new(false),
140        }
141    }
142
143    /// Creates a shutdown signal that never fires.
144    ///
145    /// Intended for use in tests and one-shot contexts where graceful shutdown
146    /// is not needed.
147    pub fn never() -> Self {
148        let (tx, rx) = mpsc::channel();
149        Self {
150            receiver: rx,
151            _sender_anchor: Some(tx),
152            stopped: AtomicBool::new(false),
153        }
154    }
155
156    /// Creates a shutdown signal from a raw receiver, for use in tests.
157    ///
158    /// Prefer [`never`](Self::never) when no shutdown is needed. Use this
159    /// when the test needs to trigger shutdown explicitly by sending `()` on
160    /// the corresponding `Sender`.
161    #[cfg(test)]
162    pub fn new_for_test(receiver: Receiver<()>) -> Self {
163        Self::new(receiver)
164    }
165
166    /// Returns `true` if shutdown has been requested.
167    ///
168    /// Once `true` is returned, all subsequent calls also return `true`
169    /// (the shutdown state is latched). This is non-blocking and suitable
170    /// for polling in tight loops.
171    pub fn should_stop(&self) -> bool {
172        if self.stopped.load(Ordering::Acquire) {
173            return true;
174        }
175        if self.receiver.try_recv().is_ok() {
176            self.stopped.store(true, Ordering::Release);
177            return true;
178        }
179        false
180    }
181
182    /// Waits for the given duration or until shutdown is requested.
183    ///
184    /// Returns `true` if shutdown was requested, `false` if the duration elapsed.
185    ///
186    /// This is the preferred method for tasks that sleep between work cycles.
187    pub fn wait(&self, duration: Duration) -> bool {
188        if self.stopped.load(Ordering::Acquire) {
189            return true;
190        }
191        match self.receiver.recv_timeout(duration) {
192            Ok(()) | Err(std::sync::mpsc::RecvTimeoutError::Disconnected) => {
193                self.stopped.store(true, Ordering::Release);
194                true
195            }
196            Err(std::sync::mpsc::RecvTimeoutError::Timeout) => false,
197        }
198    }
199}
200
201/// A long-running background task.
202///
203/// Implement this trait to define tasks that run in dedicated threads
204/// alongside the main application loop. Tasks receive the event hub
205/// to dispatch events and a shutdown signal for graceful termination.
206pub trait BackgroundTask: Send {
207    /// Returns the unique identifier for this task.
208    fn id(&self) -> TaskId;
209
210    /// Runs the task until shutdown is requested.
211    ///
212    /// This method is called in a dedicated thread. Use `hub` to send
213    /// events to the main loop and `shutdown` to check for termination.
214    fn run(&mut self, hub: &Sender<Event>, shutdown: &ShutdownSignal);
215
216    /// Called when the task is being stopped.
217    ///
218    /// Override this to perform cleanup. The default implementation does nothing.
219    fn stop(&mut self) {}
220
221    /// Returns a "finished" event to send after the task thread exits.
222    ///
223    /// The [`TaskManager`] sends this event after
224    /// observing the task's thread as finished. The default returns `None`.
225    fn finished_event(&self) -> Option<Event> {
226        None
227    }
228}
229
230struct RunningTask {
231    handle: JoinHandle<()>,
232    shutdown: Sender<()>,
233    /// Event to emit when the task is observed as naturally finished.
234    finished_event: Option<Event>,
235}
236
237/// Manages the lifecycle of background tasks.
238///
239/// The task manager spawns tasks in dedicated threads and provides
240/// methods to stop individual tasks or all tasks at once.
241pub struct TaskManager {
242    tasks: HashMap<TaskId, RunningTask>,
243    /// Library indices awaiting import while one is already running. The bool is the `force` flag.
244    pending_import_indices: VecDeque<(Option<usize>, bool)>,
245    /// Library indices awaiting thumbnail extraction while a run is in progress.
246    pending_thumbnail_indices: VecDeque<Option<usize>>,
247    /// Events from naturally finished tasks, waiting to be sent.
248    buffered_events: Vec<Event>,
249}
250
251impl TaskManager {
252    /// Creates a new empty task manager.
253    pub fn new() -> Self {
254        Self {
255            tasks: HashMap::new(),
256            pending_import_indices: VecDeque::new(),
257            pending_thumbnail_indices: VecDeque::new(),
258            buffered_events: Vec::new(),
259        }
260    }
261
262    /// Starts a background task in a new thread.
263    ///
264    /// The task receives a clone of `hub` for sending events and a
265    /// [`ShutdownSignal`] for graceful termination.
266    ///
267    /// Returns an error if a task with the same ID is already running.
268    #[cfg_attr(feature = "tracing", tracing::instrument(skip(self, task, hub), fields(task_id = tracing::field::Empty), ret))]
269    pub fn start(
270        &mut self,
271        task: Box<dyn BackgroundTask>,
272        hub: Sender<Event>,
273    ) -> Result<TaskId, TaskError> {
274        let id = task.id();
275
276        #[cfg(feature = "tracing")]
277        tracing::Span::current().record("task_id", tracing::field::display(&id));
278
279        if self.is_running(&id) {
280            return Err(TaskError::AlreadyRunning(id));
281        }
282
283        let (shutdown_tx, shutdown_rx) = mpsc::channel();
284        let shutdown_signal = ShutdownSignal::new(shutdown_rx);
285
286        let finished_event = task.finished_event();
287
288        let handle = thread::spawn(move || {
289            let mut task = task;
290            tracing::info!("task started");
291            task.run(&hub, &shutdown_signal);
292            task.stop();
293            tracing::info!("task stopped");
294        });
295
296        self.tasks.insert(
297            id.clone(),
298            RunningTask {
299                handle,
300                shutdown: shutdown_tx,
301                finished_event,
302            },
303        );
304
305        tracing::info!("task registered");
306        Ok(id)
307    }
308
309    /// Stops a running task by ID.
310    ///
311    /// Sends the shutdown signal and waits for the task thread to finish.
312    /// Returns an error if the task is not running.
313    #[cfg_attr(feature = "tracing", tracing::instrument(skip(self), fields(task_id = %id), ret))]
314    pub fn stop(&mut self, id: &TaskId) -> Result<(), TaskError> {
315        self.cleanup_finished();
316        if let Some(task) = self.tasks.remove(id) {
317            tracing::info!("sending shutdown signal");
318            if let Err(e) = task.shutdown.send(()) {
319                tracing::error!(error = %e, "failed to send shutdown signal");
320            }
321            if task.handle.join().is_err() {
322                tracing::error!("task thread panicked");
323            }
324            Ok(())
325        } else {
326            Err(TaskError::NotRunning(id.clone()))
327        }
328    }
329
330    /// Stops all running tasks.
331    ///
332    /// Sends shutdown signals to all tasks and waits for them to finish.
333    #[cfg_attr(feature = "tracing", tracing::instrument(skip(self), fields(task_count = tracing::field::Empty)))]
334    pub fn stop_all(&mut self) {
335        let tasks: Vec<_> = self.tasks.drain().collect();
336
337        #[cfg(feature = "tracing")]
338        tracing::Span::current().record("task_count", tasks.len());
339
340        if !tasks.is_empty() {
341            tracing::info!("stopping all tasks");
342        }
343        for (_, task) in &tasks {
344            if let Err(e) = task.shutdown.send(()) {
345                tracing::error!(error = %e, "failed to send shutdown signal");
346            }
347        }
348        for (_, task) in tasks {
349            if task.handle.join().is_err() {
350                tracing::error!("task thread panicked");
351            }
352        }
353    }
354
355    /// Removes entries for tasks whose threads have finished, buffering
356    /// their completion events only if the thread exited successfully.
357    fn cleanup_finished(&mut self) {
358        let finished: Vec<TaskId> = self
359            .tasks
360            .iter()
361            .filter(|(_, task)| task.handle.is_finished())
362            .map(|(id, _)| id.clone())
363            .collect();
364
365        for id in finished {
366            if let Some(task) = self.tasks.remove(&id) {
367                if task.handle.join().is_ok() {
368                    if let Some(evt) = task.finished_event {
369                        self.buffered_events.push(evt);
370                    }
371                } else {
372                    tracing::error!(task_id = %id, "task thread panicked");
373                }
374            }
375        }
376    }
377
378    /// Sends any buffered completion events from naturally finished tasks.
379    fn flush_buffered_events(&mut self, hub: &Sender<Event>) {
380        for evt in self.buffered_events.drain(..) {
381            hub.send(evt).ok();
382        }
383    }
384
385    /// Observes an event without consuming it.
386    ///
387    /// Must be called for every event before passing it to the view tree.
388    /// Always returns `false` — it never consumes events.
389    #[cfg_attr(
390        feature = "tracing",
391        tracing::instrument(skip(self, hub, database, settings))
392    )]
393    pub fn handle_event(
394        &mut self,
395        evt: &Event,
396        hub: &Sender<Event>,
397        database: &Database,
398        settings: &Settings,
399    ) -> bool {
400        self.cleanup_finished();
401        self.flush_buffered_events(hub);
402
403        match evt {
404            Event::ImportLibrary {
405                library_index,
406                force,
407            } => {
408                self.schedule_import(*library_index, *force, hub, database, settings);
409            }
410            Event::ImportFinished { library_index } => {
411                self.drain_pending_imports(hub, database, settings);
412                self.schedule_thumbnail_extraction(*library_index, hub, database, settings);
413            }
414            Event::ThumbnailExtractionFinished { .. } => {
415                self.drain_pending_thumbnails(hub, database, settings);
416            }
417            Event::ReindexDictionaries => {
418                self.schedule_dictionary_index(hub, database);
419            }
420            _ => {}
421        }
422        false
423    }
424
425    /// Schedules an import task, queuing the index if one is already running.
426    #[cfg_attr(feature = "tracing", tracing::instrument(skip_all))]
427    fn schedule_import(
428        &mut self,
429        library_index: Option<usize>,
430        force: bool,
431        hub: &Sender<Event>,
432        database: &Database,
433        settings: &Settings,
434    ) {
435        if self.is_running(&TaskId::Import) {
436            tracing::info!(library_index = ?library_index, force, "import already running, queueing");
437            self.pending_import_indices
438                .push_back((library_index, force));
439            return;
440        }
441
442        self.flush_buffered_events(hub);
443
444        let task = Box::new(import::ImportTask::new(
445            database.clone(),
446            settings.clone(),
447            library_index,
448            force,
449        ));
450
451        if let Err(e) = self.start(task, hub.clone()) {
452            tracing::warn!(error = %e, "failed to start import task");
453        }
454    }
455
456    /// Starts the next pending import when the current one finishes.
457    #[cfg_attr(feature = "tracing", tracing::instrument(skip_all))]
458    fn drain_pending_imports(
459        &mut self,
460        hub: &Sender<Event>,
461        database: &Database,
462        settings: &Settings,
463    ) {
464        if self.is_running(&TaskId::Import) || self.pending_import_indices.is_empty() {
465            return;
466        }
467
468        let Some((next, force)) = self.pending_import_indices.pop_front() else {
469            return;
470        };
471        self.schedule_import(next, force, hub, database, settings);
472    }
473
474    /// Schedules a dictionary index scan, stopping any running instance first.
475    #[cfg_attr(feature = "tracing", tracing::instrument(skip_all))]
476    fn schedule_dictionary_index(&mut self, hub: &Sender<Event>, database: &Database) {
477        if self.is_running(&TaskId::DictionaryIndex) {
478            tracing::debug!("stopping running dictionary index task for restart");
479            if let Err(e) = self.stop(&TaskId::DictionaryIndex) {
480                tracing::warn!(error = %e, "failed to stop dictionary_index task for restart");
481            }
482        }
483
484        self.flush_buffered_events(hub);
485
486        let task = Box::new(dictionary_index::DictionaryIndexTask::new(database.clone()));
487
488        if let Err(e) = self.start(task, hub.clone()) {
489            tracing::warn!(error = %e, "failed to start dictionary_index task");
490        }
491    }
492
493    /// Schedules a thumbnail extraction task, queuing the index if one is already running.
494    #[cfg_attr(feature = "tracing", tracing::instrument(skip_all))]
495    pub fn schedule_thumbnail_extraction(
496        &mut self,
497        library_index: Option<usize>,
498        hub: &Sender<Event>,
499        database: &Database,
500        settings: &Settings,
501    ) {
502        if self.is_running(&TaskId::ThumbnailExtraction) {
503            tracing::info!(library_index = ?library_index, "thumbnail extraction already running, queueing");
504            self.pending_thumbnail_indices.push_back(library_index);
505            return;
506        }
507
508        self.flush_buffered_events(hub);
509
510        let task = Box::new(thumbnail::ThumbnailExtractionTask::new(
511            database.clone(),
512            settings.clone(),
513            library_index,
514        ));
515
516        if let Err(e) = self.start(task, hub.clone()) {
517            tracing::warn!(error = %e, "failed to start thumbnail extraction task");
518        }
519    }
520
521    /// Starts the next pending thumbnail extraction when the current one finishes.
522    #[cfg_attr(feature = "tracing", tracing::instrument(skip_all))]
523    fn drain_pending_thumbnails(
524        &mut self,
525        hub: &Sender<Event>,
526        database: &Database,
527        settings: &Settings,
528    ) {
529        if self.is_running(&TaskId::ThumbnailExtraction)
530            || self.pending_thumbnail_indices.is_empty()
531        {
532            return;
533        }
534
535        let Some(next) = self.pending_thumbnail_indices.pop_front() else {
536            return;
537        };
538        self.schedule_thumbnail_extraction(next, hub, database, settings);
539    }
540
541    /// Returns `true` if a task with the given ID is running.
542    pub fn is_running(&mut self, id: &TaskId) -> bool {
543        self.cleanup_finished();
544        self.tasks.contains_key(id)
545    }
546
547    /// Returns the IDs of all running tasks.
548    pub fn running_tasks(&mut self) -> Vec<TaskId> {
549        self.cleanup_finished();
550        self.tasks.keys().cloned().collect()
551    }
552}
553
554impl Default for TaskManager {
555    fn default() -> Self {
556        Self::new()
557    }
558}
559
560impl Drop for TaskManager {
561    fn drop(&mut self) {
562        self.stop_all();
563    }
564}
565
566/// Registers background tasks that run at startup.
567///
568/// Call this during startup to add background tasks.
569/// Currently registers:
570/// - [`wifi_status_monitor::WifiStatusMonitorTask`] - monitors WiFi status via dhcpcd-dbus (kobo only)
571/// - [`hello_world::HelloWorldTask`] - prints "Hello world!" every minute (test only)
572/// - [`dbus_monitor::DbusMonitorTask`] - monitors D-Bus signals (test + kobo only, when `settings.logging.enable_dbus_log` is true)
573/// - [`import::ImportTask`] - runs an incremental import of all libraries on startup
574/// - [`dictionary_index::DictionaryIndexTask`] - indexes `.index` dictionary files into SQLite
575pub fn register_startup_tasks(
576    manager: &mut TaskManager,
577    hub: Sender<Event>,
578    settings: &Settings,
579    database: &Database,
580) {
581    #[cfg(feature = "kobo")]
582    {
583        let task = Box::new(wifi_status_monitor::WifiStatusMonitorTask);
584        if let Err(e) = manager.start(task, hub.clone()) {
585            tracing::warn!(error = %e, "failed to start wifi_status_monitor task");
586        }
587    }
588
589    #[cfg(feature = "test")]
590    {
591        let task = Box::new(hello_world::HelloWorldTask);
592        if let Err(e) = manager.start(task, hub.clone()) {
593            tracing::warn!(error = %e, "failed to start hello_world task");
594        }
595
596        #[cfg(feature = "kobo")]
597        if settings.logging.enable_dbus_log {
598            let task = Box::new(dbus_monitor::DbusMonitorTask);
599            if let Err(e) = manager.start(task, hub.clone()) {
600                tracing::warn!(error = %e, "failed to start dbus_monitor task");
601            }
602        }
603    }
604
605    manager.schedule_import(None, false, &hub, database, settings);
606
607    let task = Box::new(dictionary_index::DictionaryIndexTask::new(database.clone()));
608    if let Err(e) = manager.start(task, hub.clone()) {
609        tracing::warn!(error = %e, "failed to start dictionary_index task");
610    }
611}
612
613#[cfg(test)]
614mod tests {
615    use super::*;
616    use std::sync::mpsc;
617    use std::time::{Duration, Instant};
618
619    fn wait_until_not_running(manager: &mut TaskManager, id: &TaskId) {
620        let deadline = Instant::now() + Duration::from_secs(5);
621        while Instant::now() < deadline {
622            if !manager.is_running(id) {
623                return;
624            }
625            std::thread::sleep(Duration::from_millis(1));
626        }
627        panic!("task '{id}' did not finish within timeout");
628    }
629
630    struct InstantTask;
631
632    impl BackgroundTask for InstantTask {
633        fn id(&self) -> TaskId {
634            TaskId::TestTask2
635        }
636
637        fn run(&mut self, _hub: &Sender<Event>, _shutdown: &ShutdownSignal) {}
638    }
639
640    struct WaitingTask;
641
642    impl BackgroundTask for WaitingTask {
643        fn id(&self) -> TaskId {
644            TaskId::TestTask
645        }
646
647        fn run(&mut self, _hub: &Sender<Event>, shutdown: &ShutdownSignal) {
648            shutdown.wait(Duration::from_secs(60));
649        }
650    }
651
652    #[test]
653    fn start_and_stop() {
654        let mut manager = TaskManager::new();
655        let (hub, _rx) = mpsc::channel();
656
657        let id = manager.start(Box::new(WaitingTask), hub).unwrap();
658        assert!(manager.is_running(&id));
659
660        manager.stop(&id).unwrap();
661        assert!(!manager.is_running(&id));
662    }
663
664    #[test]
665    fn duplicate_start_returns_error() {
666        let mut manager = TaskManager::new();
667        let (hub, _rx) = mpsc::channel();
668
669        manager.start(Box::new(WaitingTask), hub.clone()).unwrap();
670        let err = manager.start(Box::new(WaitingTask), hub).unwrap_err();
671
672        assert!(matches!(err, TaskError::AlreadyRunning(TaskId::TestTask)));
673    }
674
675    #[test]
676    fn finished_task_is_cleaned_up() {
677        let mut manager = TaskManager::new();
678        let (hub, _rx) = mpsc::channel();
679
680        let id = manager.start(Box::new(InstantTask), hub).unwrap();
681
682        wait_until_not_running(&mut manager, &id);
683        assert!(!manager.is_running(&id));
684    }
685
686    #[test]
687    fn stop_finished_task_returns_not_running() {
688        let mut manager = TaskManager::new();
689        let (hub, _rx) = mpsc::channel();
690
691        let id = manager.start(Box::new(InstantTask), hub).unwrap();
692
693        wait_until_not_running(&mut manager, &id);
694        let err = manager.stop(&id).unwrap_err();
695
696        assert!(matches!(err, TaskError::NotRunning(TaskId::TestTask2)));
697    }
698
699    #[test]
700    fn running_tasks_excludes_finished() {
701        let mut manager = TaskManager::new();
702        let (hub, _rx) = mpsc::channel();
703
704        manager.start(Box::new(WaitingTask), hub.clone()).unwrap();
705        let instant_id = manager.start(Box::new(InstantTask), hub).unwrap();
706
707        wait_until_not_running(&mut manager, &instant_id);
708        let running = manager.running_tasks();
709
710        assert_eq!(running.len(), 1);
711        assert_eq!(running[0], TaskId::TestTask);
712
713        manager.stop_all();
714    }
715
716    #[test]
717    fn stop_all_stops_everything() {
718        let mut manager = TaskManager::new();
719        let (hub, _rx) = mpsc::channel();
720
721        manager.start(Box::new(WaitingTask), hub).unwrap();
722        manager.stop_all();
723
724        assert!(!manager.is_running(&TaskId::TestTask));
725    }
726
727    #[test]
728    fn test_thumbnail_extraction_task_lifecycle() {
729        let mut manager = TaskManager::new();
730        let (hub, _rx) = mpsc::channel();
731        let database = Database::new(":memory:").unwrap();
732        database.migrate().unwrap();
733        let settings = Settings::default();
734
735        manager.schedule_thumbnail_extraction(None, &hub, &database, &settings);
736
737        // Task exits quickly on an unseeded database, so wait for
738        // completion rather than asserting the transient running state.
739        wait_until_not_running(&mut manager, &TaskId::ThumbnailExtraction);
740        assert!(!manager.is_running(&TaskId::ThumbnailExtraction));
741
742        let err = manager.stop(&TaskId::ThumbnailExtraction).unwrap_err();
743        assert!(matches!(
744            err,
745            TaskError::NotRunning(TaskId::ThumbnailExtraction)
746        ));
747    }
748
749    #[test]
750    fn thumbnail_extraction_queues_when_running() {
751        let mut manager = TaskManager::new();
752        let (hub, _rx) = mpsc::channel();
753
754        // Simulate a running ThumbnailExtraction task with a blocking thread.
755        let (shutdown_tx, shutdown_rx) = mpsc::channel();
756        let blocking_handle = thread::spawn(move || {
757            let _ = shutdown_rx.recv();
758        });
759        manager.tasks.insert(
760            TaskId::ThumbnailExtraction,
761            RunningTask {
762                handle: blocking_handle,
763                shutdown: shutdown_tx,
764                finished_event: None,
765            },
766        );
767
768        let database = Database::new(":memory:").unwrap();
769        database.migrate().unwrap();
770        let settings = Settings::default();
771
772        manager.schedule_thumbnail_extraction(Some(0), &hub, &database, &settings);
773        manager.schedule_thumbnail_extraction(Some(1), &hub, &database, &settings);
774
775        assert_eq!(manager.pending_thumbnail_indices.len(), 2);
776
777        manager.stop(&TaskId::ThumbnailExtraction).unwrap();
778
779        manager.drain_pending_thumbnails(&hub, &database, &settings);
780        assert_eq!(manager.pending_thumbnail_indices.len(), 1);
781
782        wait_until_not_running(&mut manager, &TaskId::ThumbnailExtraction);
783
784        manager.drain_pending_thumbnails(&hub, &database, &settings);
785        assert!(manager.pending_thumbnail_indices.is_empty());
786
787        wait_until_not_running(&mut manager, &TaskId::ThumbnailExtraction);
788    }
789
790    #[test]
791    fn import_queue_preserves_force_flag() {
792        let mut manager = TaskManager::new();
793        let (hub, _rx) = mpsc::channel();
794        let database = Database::new(":memory:").unwrap();
795        database.migrate().unwrap();
796        let settings = Settings::default();
797
798        // Simulate a running import task with a blocking thread.
799        let (shutdown_tx, shutdown_rx) = mpsc::channel();
800        let blocking_handle = thread::spawn(move || {
801            let _ = shutdown_rx.recv();
802        });
803        manager.tasks.insert(
804            TaskId::Import,
805            RunningTask {
806                handle: blocking_handle,
807                shutdown: shutdown_tx,
808                finished_event: None,
809            },
810        );
811
812        manager.handle_event(
813            &Event::ImportLibrary {
814                library_index: Some(0),
815                force: true,
816            },
817            &hub,
818            &database,
819            &settings,
820        );
821
822        assert_eq!(
823            manager.pending_import_indices.front(),
824            Some(&(Some(0), true))
825        );
826
827        manager.stop(&TaskId::Import).unwrap();
828    }
829
830    #[test]
831    fn import_queue_preserves_force_false_flag() {
832        let mut manager = TaskManager::new();
833        let (hub, _rx) = mpsc::channel();
834        let database = Database::new(":memory:").unwrap();
835        database.migrate().unwrap();
836        let settings = Settings::default();
837
838        let (shutdown_tx, shutdown_rx) = mpsc::channel();
839        let blocking_handle = thread::spawn(move || {
840            let _ = shutdown_rx.recv();
841        });
842        manager.tasks.insert(
843            TaskId::Import,
844            RunningTask {
845                handle: blocking_handle,
846                shutdown: shutdown_tx,
847                finished_event: None,
848            },
849        );
850
851        manager.handle_event(
852            &Event::ImportLibrary {
853                library_index: None,
854                force: false,
855            },
856            &hub,
857            &database,
858            &settings,
859        );
860
861        assert_eq!(manager.pending_import_indices.front(), Some(&(None, false)));
862
863        manager.stop(&TaskId::Import).unwrap();
864    }
865}
cadmus_core/task/mod.rs

cadmus_core/task/
mod.rs
