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

cadmus_core/task/
mod.rs
