celers_core/
task.rs

1use crate::state::TaskState;
2use chrono::{DateTime, Utc};
3use serde::{Deserialize, Serialize};
4use std::collections::HashSet;
5use std::fmt;
6use uuid::Uuid;
7
8/// Unique identifier for a task
9pub type TaskId = Uuid;
10
11/// Batch utility functions for working with multiple tasks
12pub mod batch {
13    use super::{SerializedTask, TaskState, Uuid};
14
15    /// Validate a collection of tasks, returning all errors
16    ///
17    /// # Example
18    /// ```
19    /// use celers_core::{SerializedTask, task::batch};
20    ///
21    /// let tasks = vec![
22    ///     SerializedTask::new("task1".to_string(), vec![1, 2, 3]),
23    ///     SerializedTask::new("task2".to_string(), vec![4, 5, 6]),
24    /// ];
25    ///
26    /// let errors = batch::validate_all(&tasks);
27    /// assert!(errors.is_empty());
28    /// ```
29    #[must_use]
30    pub fn validate_all(tasks: &[SerializedTask]) -> Vec<(usize, String)> {
31        tasks
32            .iter()
33            .enumerate()
34            .filter_map(|(idx, task)| task.validate().err().map(|e| (idx, e)))
35            .collect()
36    }
37
38    /// Filter tasks by state
39    ///
40    /// # Example
41    /// ```
42    /// use celers_core::{SerializedTask, TaskState, task::batch};
43    ///
44    /// let mut tasks = vec![
45    ///     SerializedTask::new("task1".to_string(), vec![1, 2, 3]),
46    ///     SerializedTask::new("task2".to_string(), vec![4, 5, 6]),
47    /// ];
48    /// tasks[0].metadata.state = TaskState::Running;
49    ///
50    /// let running = batch::filter_by_state(&tasks, |s| matches!(s, TaskState::Running));
51    /// assert_eq!(running.len(), 1);
52    /// ```
53    #[must_use]
54    pub fn filter_by_state<F>(tasks: &[SerializedTask], predicate: F) -> Vec<&SerializedTask>
55    where
56        F: Fn(&TaskState) -> bool,
57    {
58        tasks
59            .iter()
60            .filter(|task| predicate(&task.metadata.state))
61            .collect()
62    }
63
64    /// Filter tasks by priority
65    ///
66    /// # Example
67    /// ```
68    /// use celers_core::{SerializedTask, task::batch};
69    ///
70    /// let tasks = vec![
71    ///     SerializedTask::new("task1".to_string(), vec![1]).with_priority(10),
72    ///     SerializedTask::new("task2".to_string(), vec![2]).with_priority(5),
73    ///     SerializedTask::new("task3".to_string(), vec![3]),
74    /// ];
75    ///
76    /// let high_priority = batch::filter_high_priority(&tasks);
77    /// assert_eq!(high_priority.len(), 2);
78    /// ```
79    #[must_use]
80    pub fn filter_high_priority(tasks: &[SerializedTask]) -> Vec<&SerializedTask> {
81        tasks
82            .iter()
83            .filter(|task| task.metadata.is_high_priority())
84            .collect()
85    }
86
87    /// Sort tasks by priority (highest first)
88    ///
89    /// # Example
90    /// ```
91    /// use celers_core::{SerializedTask, task::batch};
92    ///
93    /// let mut tasks = vec![
94    ///     SerializedTask::new("task1".to_string(), vec![1]).with_priority(5),
95    ///     SerializedTask::new("task2".to_string(), vec![2]).with_priority(10),
96    ///     SerializedTask::new("task3".to_string(), vec![3]).with_priority(1),
97    /// ];
98    ///
99    /// batch::sort_by_priority(&mut tasks);
100    /// assert_eq!(tasks[0].metadata.priority, 10);
101    /// assert_eq!(tasks[1].metadata.priority, 5);
102    /// assert_eq!(tasks[2].metadata.priority, 1);
103    /// ```
104    pub fn sort_by_priority(tasks: &mut [SerializedTask]) {
105        tasks.sort_by_key(|b| std::cmp::Reverse(b.metadata.priority));
106    }
107
108    /// Count tasks by state
109    ///
110    /// # Example
111    /// ```
112    /// use celers_core::{SerializedTask, TaskState, task::batch};
113    ///
114    /// let mut tasks = vec![
115    ///     SerializedTask::new("task1".to_string(), vec![1]),
116    ///     SerializedTask::new("task2".to_string(), vec![2]),
117    /// ];
118    /// tasks[0].metadata.state = TaskState::Running;
119    ///
120    /// let counts = batch::count_by_state(&tasks);
121    /// assert_eq!(counts.get("RUNNING"), Some(&1));
122    /// assert_eq!(counts.get("PENDING"), Some(&1));
123    /// ```
124    #[must_use]
125    pub fn count_by_state(tasks: &[SerializedTask]) -> std::collections::HashMap<String, usize> {
126        let mut counts = std::collections::HashMap::new();
127        for task in tasks {
128            *counts
129                .entry(task.metadata.state.name().to_string())
130                .or_insert(0) += 1;
131        }
132        counts
133    }
134
135    /// Check if any tasks have expired
136    ///
137    /// # Example
138    /// ```
139    /// use celers_core::{SerializedTask, task::batch};
140    ///
141    /// let tasks = vec![
142    ///     SerializedTask::new("task1".to_string(), vec![1]).with_timeout(60),
143    ///     SerializedTask::new("task2".to_string(), vec![2]),
144    /// ];
145    ///
146    /// // Fresh tasks shouldn't be expired
147    /// assert!(!batch::has_expired_tasks(&tasks));
148    /// ```
149    #[inline]
150    #[must_use]
151    pub fn has_expired_tasks(tasks: &[SerializedTask]) -> bool {
152        tasks.iter().any(super::SerializedTask::is_expired)
153    }
154
155    /// Get tasks that have expired
156    ///
157    /// # Example
158    /// ```
159    /// use celers_core::{SerializedTask, task::batch};
160    ///
161    /// let tasks = vec![
162    ///     SerializedTask::new("task1".to_string(), vec![1]).with_timeout(60),
163    ///     SerializedTask::new("task2".to_string(), vec![2]),
164    /// ];
165    ///
166    /// let expired = batch::get_expired_tasks(&tasks);
167    /// // Fresh tasks shouldn't be expired
168    /// assert_eq!(expired.len(), 0);
169    /// ```
170    #[inline]
171    #[must_use]
172    pub fn get_expired_tasks(tasks: &[SerializedTask]) -> Vec<&SerializedTask> {
173        tasks.iter().filter(|task| task.is_expired()).collect()
174    }
175
176    /// Calculate total payload size for a collection of tasks
177    ///
178    /// # Example
179    /// ```
180    /// use celers_core::{SerializedTask, task::batch};
181    ///
182    /// let tasks = vec![
183    ///     SerializedTask::new("task1".to_string(), vec![1, 2, 3]),
184    ///     SerializedTask::new("task2".to_string(), vec![4, 5]),
185    /// ];
186    ///
187    /// let total_size = batch::total_payload_size(&tasks);
188    /// assert_eq!(total_size, 5);
189    /// ```
190    #[must_use]
191    pub fn total_payload_size(tasks: &[SerializedTask]) -> usize {
192        tasks.iter().map(super::SerializedTask::payload_size).sum()
193    }
194
195    /// Find tasks with dependencies
196    ///
197    /// # Example
198    /// ```
199    /// use celers_core::{SerializedTask, task::batch};
200    /// use uuid::Uuid;
201    ///
202    /// let parent_id = Uuid::new_v4();
203    /// let tasks = vec![
204    ///     SerializedTask::new("task1".to_string(), vec![1]),
205    ///     SerializedTask::new("task2".to_string(), vec![2]).with_dependency(parent_id),
206    /// ];
207    ///
208    /// let with_deps = batch::filter_with_dependencies(&tasks);
209    /// assert_eq!(with_deps.len(), 1);
210    /// ```
211    #[must_use]
212    pub fn filter_with_dependencies(tasks: &[SerializedTask]) -> Vec<&SerializedTask> {
213        tasks
214            .iter()
215            .filter(|task| task.metadata.has_dependencies())
216            .collect()
217    }
218
219    /// Find tasks that can be retried
220    ///
221    /// # Example
222    /// ```
223    /// use celers_core::{SerializedTask, TaskState, task::batch};
224    ///
225    /// let mut tasks = vec![
226    ///     SerializedTask::new("task1".to_string(), vec![1]).with_max_retries(3),
227    ///     SerializedTask::new("task2".to_string(), vec![2]).with_max_retries(3),
228    /// ];
229    /// tasks[0].metadata.state = TaskState::Failed("error".to_string());
230    /// tasks[1].metadata.state = TaskState::Succeeded(vec![]);
231    ///
232    /// let can_retry = batch::filter_retryable(&tasks);
233    /// assert_eq!(can_retry.len(), 1);
234    /// ```
235    #[must_use]
236    pub fn filter_retryable(tasks: &[SerializedTask]) -> Vec<&SerializedTask> {
237        tasks.iter().filter(|task| task.can_retry()).collect()
238    }
239
240    /// Find tasks by name pattern (contains)
241    ///
242    /// # Example
243    /// ```
244    /// use celers_core::{SerializedTask, task::batch};
245    ///
246    /// let tasks = vec![
247    ///     SerializedTask::new("process_data".to_string(), vec![1]),
248    ///     SerializedTask::new("process_image".to_string(), vec![2]),
249    ///     SerializedTask::new("send_email".to_string(), vec![3]),
250    /// ];
251    ///
252    /// let process_tasks = batch::filter_by_name_pattern(&tasks, "process");
253    /// assert_eq!(process_tasks.len(), 2);
254    /// ```
255    #[must_use]
256    pub fn filter_by_name_pattern<'a>(
257        tasks: &'a [SerializedTask],
258        pattern: &str,
259    ) -> Vec<&'a SerializedTask> {
260        tasks
261            .iter()
262            .filter(|task| task.metadata.name.contains(pattern))
263            .collect()
264    }
265
266    /// Group tasks by their workflow group ID
267    ///
268    /// # Example
269    /// ```
270    /// use celers_core::{SerializedTask, task::batch};
271    /// use uuid::Uuid;
272    ///
273    /// let group1 = Uuid::new_v4();
274    /// let group2 = Uuid::new_v4();
275    ///
276    /// let tasks = vec![
277    ///     SerializedTask::new("task1".to_string(), vec![1]).with_group_id(group1),
278    ///     SerializedTask::new("task2".to_string(), vec![2]).with_group_id(group1),
279    ///     SerializedTask::new("task3".to_string(), vec![3]).with_group_id(group2),
280    ///     SerializedTask::new("task4".to_string(), vec![4]),
281    /// ];
282    ///
283    /// let groups = batch::group_by_workflow_id(&tasks);
284    /// assert_eq!(groups.len(), 2);
285    /// ```
286    #[must_use]
287    pub fn group_by_workflow_id(
288        tasks: &[SerializedTask],
289    ) -> std::collections::HashMap<Uuid, Vec<&SerializedTask>> {
290        let mut groups = std::collections::HashMap::new();
291        for task in tasks {
292            if let Some(group_id) = task.metadata.group_id {
293                groups.entry(group_id).or_insert_with(Vec::new).push(task);
294            }
295        }
296        groups
297    }
298
299    /// Find terminal tasks (succeeded or failed)
300    ///
301    /// # Example
302    /// ```
303    /// use celers_core::{SerializedTask, TaskState, task::batch};
304    ///
305    /// let mut tasks = vec![
306    ///     SerializedTask::new("task1".to_string(), vec![1]),
307    ///     SerializedTask::new("task2".to_string(), vec![2]),
308    /// ];
309    /// tasks[0].metadata.state = TaskState::Succeeded(vec![1, 2, 3]);
310    ///
311    /// let terminal = batch::filter_terminal(&tasks);
312    /// assert_eq!(terminal.len(), 1);
313    /// ```
314    #[must_use]
315    pub fn filter_terminal(tasks: &[SerializedTask]) -> Vec<&SerializedTask> {
316        tasks.iter().filter(|task| task.is_terminal()).collect()
317    }
318
319    /// Find active tasks (pending, running, or retrying)
320    ///
321    /// # Example
322    /// ```
323    /// use celers_core::{SerializedTask, TaskState, task::batch};
324    ///
325    /// let mut tasks = vec![
326    ///     SerializedTask::new("task1".to_string(), vec![1]),
327    ///     SerializedTask::new("task2".to_string(), vec![2]),
328    /// ];
329    /// tasks[1].metadata.state = TaskState::Succeeded(vec![]);
330    ///
331    /// let active = batch::filter_active(&tasks);
332    /// assert_eq!(active.len(), 1);
333    /// ```
334    #[must_use]
335    pub fn filter_active(tasks: &[SerializedTask]) -> Vec<&SerializedTask> {
336        tasks.iter().filter(|task| task.is_active()).collect()
337    }
338
339    /// Calculate average payload size
340    ///
341    /// # Example
342    /// ```
343    /// use celers_core::{SerializedTask, task::batch};
344    ///
345    /// let tasks = vec![
346    ///     SerializedTask::new("task1".to_string(), vec![1, 2, 3]),
347    ///     SerializedTask::new("task2".to_string(), vec![4, 5]),
348    /// ];
349    ///
350    /// let avg = batch::average_payload_size(&tasks);
351    /// assert_eq!(avg, 2); // (3 + 2) / 2 = 2
352    /// ```
353    #[must_use]
354    pub fn average_payload_size(tasks: &[SerializedTask]) -> usize {
355        if tasks.is_empty() {
356            0
357        } else {
358            total_payload_size(tasks) / tasks.len()
359        }
360    }
361
362    /// Find the oldest task by creation time
363    ///
364    /// # Example
365    /// ```
366    /// use celers_core::{SerializedTask, task::batch};
367    ///
368    /// let tasks = vec![
369    ///     SerializedTask::new("task1".to_string(), vec![1]),
370    ///     SerializedTask::new("task2".to_string(), vec![2]),
371    /// ];
372    ///
373    /// let oldest = batch::find_oldest(&tasks);
374    /// assert!(oldest.is_some());
375    /// ```
376    #[must_use]
377    pub fn find_oldest(tasks: &[SerializedTask]) -> Option<&SerializedTask> {
378        tasks.iter().min_by_key(|task| task.metadata.created_at)
379    }
380
381    /// Find the newest task by creation time
382    ///
383    /// # Example
384    /// ```
385    /// use celers_core::{SerializedTask, task::batch};
386    ///
387    /// let tasks = vec![
388    ///     SerializedTask::new("task1".to_string(), vec![1]),
389    ///     SerializedTask::new("task2".to_string(), vec![2]),
390    /// ];
391    ///
392    /// let newest = batch::find_newest(&tasks);
393    /// assert!(newest.is_some());
394    /// ```
395    #[must_use]
396    pub fn find_newest(tasks: &[SerializedTask]) -> Option<&SerializedTask> {
397        tasks.iter().max_by_key(|task| task.metadata.created_at)
398    }
399}
400
401/// Task metadata including execution information
402#[derive(Debug, Clone, Serialize, Deserialize)]
403pub struct TaskMetadata {
404    /// Unique task identifier
405    pub id: TaskId,
406
407    /// Task name/type identifier
408    pub name: String,
409
410    /// Current state of the task
411    pub state: TaskState,
412
413    /// When the task was created
414    pub created_at: DateTime<Utc>,
415
416    /// When the task was last updated
417    pub updated_at: DateTime<Utc>,
418
419    /// Maximum number of retry attempts
420    pub max_retries: u32,
421
422    /// Task timeout in seconds
423    pub timeout_secs: Option<u64>,
424
425    /// Task priority (higher = more important)
426    pub priority: i32,
427
428    /// Group ID (for workflow grouping)
429    #[serde(skip_serializing_if = "Option::is_none")]
430    pub group_id: Option<Uuid>,
431
432    /// Chord ID (for barrier synchronization)
433    #[serde(skip_serializing_if = "Option::is_none")]
434    pub chord_id: Option<Uuid>,
435
436    /// On-success chain link: name of task to enqueue with this task's result as first arg
437    #[serde(skip_serializing_if = "Option::is_none")]
438    pub on_success_link: Option<String>,
439
440    /// Task dependencies (tasks that must complete before this task can execute)
441    #[serde(skip_serializing_if = "HashSet::is_empty", default)]
442    pub dependencies: HashSet<TaskId>,
443}
444
445impl TaskMetadata {
446    #[inline]
447    #[must_use]
448    pub fn new(name: String) -> Self {
449        let now = Utc::now();
450        Self {
451            id: Uuid::new_v4(),
452            name,
453            state: TaskState::Pending,
454            created_at: now,
455            updated_at: now,
456            max_retries: 3,
457            timeout_secs: None,
458            priority: 0,
459            group_id: None,
460            chord_id: None,
461            on_success_link: None,
462            dependencies: HashSet::new(),
463        }
464    }
465
466    #[inline]
467    #[must_use]
468    pub fn with_max_retries(mut self, max_retries: u32) -> Self {
469        self.max_retries = max_retries;
470        self
471    }
472
473    #[inline]
474    #[must_use]
475    pub fn with_timeout(mut self, timeout_secs: u64) -> Self {
476        self.timeout_secs = Some(timeout_secs);
477        self
478    }
479
480    #[inline]
481    #[must_use]
482    pub fn with_priority(mut self, priority: i32) -> Self {
483        self.priority = priority;
484        self
485    }
486
487    /// Set the group ID for workflow grouping
488    #[inline]
489    #[must_use]
490    pub fn with_group_id(mut self, group_id: Uuid) -> Self {
491        self.group_id = Some(group_id);
492        self
493    }
494
495    /// Set the chord ID for barrier synchronization
496    #[inline]
497    #[must_use]
498    pub fn with_chord_id(mut self, chord_id: Uuid) -> Self {
499        self.chord_id = Some(chord_id);
500        self
501    }
502
503    /// Set the on-success chain link task name
504    #[inline]
505    #[must_use]
506    pub fn with_on_success_link(mut self, task_name: String) -> Self {
507        self.on_success_link = Some(task_name);
508        self
509    }
510
511    /// Get the age of the task (time since creation)
512    #[inline]
513    #[must_use]
514    pub fn age(&self) -> chrono::Duration {
515        Utc::now() - self.created_at
516    }
517
518    /// Check if the task has expired based on its timeout
519    #[inline]
520    #[must_use]
521    #[allow(clippy::cast_possible_wrap)]
522    pub fn is_expired(&self) -> bool {
523        if let Some(timeout_secs) = self.timeout_secs {
524            let elapsed = (Utc::now() - self.created_at).num_seconds();
525            elapsed > timeout_secs as i64
526        } else {
527            false
528        }
529    }
530
531    /// Check if the task is in a terminal state (Succeeded or Failed)
532    #[inline]
533    #[must_use]
534    pub fn is_terminal(&self) -> bool {
535        self.state.is_terminal()
536    }
537
538    /// Check if the task is in a running or active state
539    #[inline]
540    #[must_use]
541    pub fn is_active(&self) -> bool {
542        matches!(
543            self.state,
544            TaskState::Pending | TaskState::Reserved | TaskState::Running | TaskState::Retrying(_)
545        )
546    }
547
548    /// Validate the task metadata
549    ///
550    /// Returns an error if any of the metadata fields are invalid:
551    /// - Name must not be empty
552    /// - Max retries must be reasonable (< 1000)
553    /// - Timeout must be at least 1 second if set
554    /// - Priority must be in valid range (-2147483648 to 2147483647)
555    ///
556    /// # Errors
557    ///
558    /// Returns an error if validation fails.
559    pub fn validate(&self) -> Result<(), String> {
560        if self.name.is_empty() {
561            return Err("Task name cannot be empty".to_string());
562        }
563
564        if self.max_retries > 1000 {
565            return Err("Max retries cannot exceed 1000".to_string());
566        }
567
568        if let Some(timeout) = self.timeout_secs {
569            if timeout == 0 {
570                return Err("Timeout must be at least 1 second".to_string());
571            }
572            if timeout > 86400 {
573                return Err("Timeout cannot exceed 24 hours (86400 seconds)".to_string());
574            }
575        }
576
577        Ok(())
578    }
579
580    /// Check if task has a timeout configured
581    #[inline]
582    #[must_use]
583    pub fn has_timeout(&self) -> bool {
584        self.timeout_secs.is_some()
585    }
586
587    /// Check if task is part of a group
588    #[inline]
589    #[must_use]
590    pub fn has_group_id(&self) -> bool {
591        self.group_id.is_some()
592    }
593
594    /// Check if task is part of a chord
595    #[inline]
596    #[must_use]
597    pub fn has_chord_id(&self) -> bool {
598        self.chord_id.is_some()
599    }
600
601    /// Check if task has custom priority (non-zero)
602    #[inline]
603    #[must_use]
604    pub const fn has_priority(&self) -> bool {
605        self.priority != 0
606    }
607
608    /// Check if task has high priority (priority > 0)
609    #[inline]
610    #[must_use]
611    pub const fn is_high_priority(&self) -> bool {
612        self.priority > 0
613    }
614
615    /// Check if task has low priority (priority < 0)
616    #[inline]
617    #[must_use]
618    pub const fn is_low_priority(&self) -> bool {
619        self.priority < 0
620    }
621
622    /// Add a task dependency
623    #[inline]
624    #[must_use]
625    pub fn with_dependency(mut self, dependency: TaskId) -> Self {
626        self.dependencies.insert(dependency);
627        self
628    }
629
630    /// Add multiple task dependencies
631    #[inline]
632    #[must_use]
633    pub fn with_dependencies(mut self, dependencies: impl IntoIterator<Item = TaskId>) -> Self {
634        self.dependencies.extend(dependencies);
635        self
636    }
637
638    /// Check if task has any dependencies
639    #[inline]
640    #[must_use]
641    pub fn has_dependencies(&self) -> bool {
642        !self.dependencies.is_empty()
643    }
644
645    /// Get the number of dependencies
646    #[inline]
647    #[must_use]
648    pub fn dependency_count(&self) -> usize {
649        self.dependencies.len()
650    }
651
652    /// Check if a specific task is a dependency
653    #[inline]
654    #[must_use]
655    pub fn depends_on(&self, task_id: &TaskId) -> bool {
656        self.dependencies.contains(task_id)
657    }
658
659    /// Remove a dependency
660    #[inline]
661    pub fn remove_dependency(&mut self, task_id: &TaskId) -> bool {
662        self.dependencies.remove(task_id)
663    }
664
665    /// Clear all dependencies
666    #[inline]
667    pub fn clear_dependencies(&mut self) {
668        self.dependencies.clear();
669    }
670
671    // ===== Convenience State Checks =====
672
673    /// Check if task is in Pending state
674    #[inline]
675    #[must_use]
676    pub fn is_pending(&self) -> bool {
677        matches!(self.state, TaskState::Pending)
678    }
679
680    /// Check if task is in Running state
681    #[inline]
682    #[must_use]
683    pub fn is_running(&self) -> bool {
684        matches!(self.state, TaskState::Running)
685    }
686
687    /// Check if task is in Succeeded state
688    #[inline]
689    #[must_use]
690    pub fn is_succeeded(&self) -> bool {
691        matches!(self.state, TaskState::Succeeded(_))
692    }
693
694    /// Check if task is in Failed state
695    #[inline]
696    #[must_use]
697    pub fn is_failed(&self) -> bool {
698        matches!(self.state, TaskState::Failed(_))
699    }
700
701    /// Check if task is in Retrying state
702    #[inline]
703    #[must_use]
704    pub fn is_retrying(&self) -> bool {
705        matches!(self.state, TaskState::Retrying(_))
706    }
707
708    /// Check if task is in Reserved state
709    #[inline]
710    #[must_use]
711    pub fn is_reserved(&self) -> bool {
712        matches!(self.state, TaskState::Reserved)
713    }
714
715    // ===== Time-related Helpers =====
716
717    /// Get remaining time before timeout (None if no timeout or already expired)
718    ///
719    /// # Example
720    /// ```
721    /// use celers_core::TaskMetadata;
722    ///
723    /// let task = TaskMetadata::new("test".to_string()).with_timeout(60);
724    /// if let Some(remaining) = task.time_remaining() {
725    ///     println!("Task has {} seconds remaining", remaining.num_seconds());
726    /// }
727    /// ```
728    #[inline]
729    #[must_use]
730    #[allow(clippy::cast_possible_wrap)]
731    pub fn time_remaining(&self) -> Option<chrono::Duration> {
732        self.timeout_secs.and_then(|timeout| {
733            let elapsed = Utc::now() - self.created_at;
734            let timeout_duration = chrono::Duration::seconds(timeout as i64);
735            let remaining = timeout_duration - elapsed;
736            if remaining.num_seconds() > 0 {
737                Some(remaining)
738            } else {
739                None
740            }
741        })
742    }
743
744    /// Get the time elapsed since task creation
745    ///
746    /// # Example
747    /// ```
748    /// use celers_core::TaskMetadata;
749    ///
750    /// let task = TaskMetadata::new("test".to_string());
751    /// let elapsed = task.time_elapsed();
752    /// assert!(elapsed.num_seconds() >= 0);
753    /// ```
754    #[inline]
755    #[must_use]
756    pub fn time_elapsed(&self) -> chrono::Duration {
757        Utc::now() - self.created_at
758    }
759
760    // ===== Retry Helpers =====
761
762    /// Check if task can be retried based on current retry count
763    ///
764    /// # Example
765    /// ```
766    /// use celers_core::{TaskMetadata, TaskState};
767    ///
768    /// let mut task = TaskMetadata::new("test".to_string()).with_max_retries(3);
769    /// task.state = TaskState::Failed("error".to_string());
770    /// assert!(task.can_retry());
771    ///
772    /// task.state = TaskState::Retrying(3);
773    /// assert!(!task.can_retry());
774    /// ```
775    #[inline]
776    #[must_use]
777    pub fn can_retry(&self) -> bool {
778        self.state.can_retry(self.max_retries)
779    }
780
781    /// Get current retry count
782    ///
783    /// # Example
784    /// ```
785    /// use celers_core::{TaskMetadata, TaskState};
786    ///
787    /// let mut task = TaskMetadata::new("test".to_string());
788    /// task.state = TaskState::Retrying(2);
789    /// assert_eq!(task.retry_count(), 2);
790    /// ```
791    #[inline]
792    #[must_use]
793    pub const fn retry_count(&self) -> u32 {
794        self.state.retry_count()
795    }
796
797    /// Get remaining retry attempts
798    ///
799    /// # Example
800    /// ```
801    /// use celers_core::{TaskMetadata, TaskState};
802    ///
803    /// let mut task = TaskMetadata::new("test".to_string()).with_max_retries(5);
804    /// task.state = TaskState::Retrying(2);
805    /// assert_eq!(task.retries_remaining(), 3);
806    /// ```
807    #[inline]
808    #[must_use]
809    pub const fn retries_remaining(&self) -> u32 {
810        let current = self.retry_count();
811        self.max_retries.saturating_sub(current)
812    }
813
814    // ===== Workflow Helpers =====
815
816    /// Check if task is part of any workflow (group or chord)
817    ///
818    /// # Example
819    /// ```
820    /// use celers_core::TaskMetadata;
821    /// use uuid::Uuid;
822    ///
823    /// let task = TaskMetadata::new("test".to_string())
824    ///     .with_group_id(Uuid::new_v4());
825    /// assert!(task.is_part_of_workflow());
826    /// ```
827    #[inline]
828    #[must_use]
829    pub fn is_part_of_workflow(&self) -> bool {
830        self.group_id.is_some() || self.chord_id.is_some()
831    }
832
833    /// Get group ID if task is part of a group
834    #[inline]
835    #[must_use]
836    pub fn get_group_id(&self) -> Option<&Uuid> {
837        self.group_id.as_ref()
838    }
839
840    /// Get chord ID if task is part of a chord
841    #[inline]
842    #[must_use]
843    pub fn get_chord_id(&self) -> Option<&Uuid> {
844        self.chord_id.as_ref()
845    }
846
847    // ===== State Transition Helpers =====
848
849    /// Update task state to Running
850    ///
851    /// # Example
852    /// ```
853    /// use celers_core::{TaskMetadata, TaskState};
854    ///
855    /// let mut task = TaskMetadata::new("test".to_string());
856    /// task.mark_as_running();
857    /// assert!(task.is_running());
858    /// ```
859    #[inline]
860    pub fn mark_as_running(&mut self) {
861        self.state = TaskState::Running;
862        self.updated_at = Utc::now();
863    }
864
865    /// Update task state to Succeeded with result
866    ///
867    /// # Example
868    /// ```
869    /// use celers_core::{TaskMetadata, TaskState};
870    ///
871    /// let mut task = TaskMetadata::new("test".to_string());
872    /// task.mark_as_succeeded(vec![1, 2, 3]);
873    /// assert!(task.is_succeeded());
874    /// ```
875    #[inline]
876    pub fn mark_as_succeeded(&mut self, result: Vec<u8>) {
877        self.state = TaskState::Succeeded(result);
878        self.updated_at = Utc::now();
879    }
880
881    /// Update task state to Failed with error message
882    ///
883    /// # Example
884    /// ```
885    /// use celers_core::{TaskMetadata, TaskState};
886    ///
887    /// let mut task = TaskMetadata::new("test".to_string());
888    /// task.mark_as_failed("Connection timeout");
889    /// assert!(task.is_failed());
890    /// ```
891    #[inline]
892    pub fn mark_as_failed(&mut self, error: impl Into<String>) {
893        self.state = TaskState::Failed(error.into());
894        self.updated_at = Utc::now();
895    }
896
897    /// Clone task with a new ID (useful for task retry/duplication)
898    ///
899    /// # Example
900    /// ```
901    /// use celers_core::TaskMetadata;
902    ///
903    /// let task = TaskMetadata::new("test".to_string()).with_priority(5);
904    /// let cloned = task.with_new_id();
905    /// assert_ne!(task.id, cloned.id);
906    /// assert_eq!(task.name, cloned.name);
907    /// assert_eq!(task.priority, cloned.priority);
908    /// ```
909    #[inline]
910    #[must_use]
911    pub fn with_new_id(&self) -> Self {
912        let now = Utc::now();
913        Self {
914            id: Uuid::new_v4(),
915            name: self.name.clone(),
916            state: TaskState::Pending,
917            created_at: now,
918            updated_at: now,
919            max_retries: self.max_retries,
920            timeout_secs: self.timeout_secs,
921            priority: self.priority,
922            group_id: self.group_id,
923            chord_id: self.chord_id,
924            on_success_link: self.on_success_link.clone(),
925            dependencies: self.dependencies.clone(),
926        }
927    }
928}
929
930impl fmt::Display for TaskMetadata {
931    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
932        write!(
933            f,
934            "Task[{}] name={} state={} priority={} retries={}/{}",
935            &self.id.to_string()[..8],
936            self.name,
937            self.state,
938            self.priority,
939            self.state.retry_count(),
940            self.max_retries
941        )?;
942
943        if let Some(timeout) = self.timeout_secs {
944            write!(f, " timeout={timeout}s")?;
945        }
946
947        if let Some(chord_id) = self.chord_id {
948            write!(f, " chord={}", &chord_id.to_string()[..8])?;
949        }
950
951        Ok(())
952    }
953}
954
955/// Trait for tasks that can be executed
956#[async_trait::async_trait]
957pub trait Task: Send + Sync {
958    /// The input type for this task
959    type Input: Serialize + for<'de> Deserialize<'de> + Send;
960
961    /// The output type for this task
962    type Output: Serialize + for<'de> Deserialize<'de> + Send;
963
964    /// Execute the task with the given input
965    async fn execute(&self, input: Self::Input) -> crate::Result<Self::Output>;
966
967    /// Get the task name
968    fn name(&self) -> &str;
969}
970
971/// Serialized task ready for queue
972///
973/// # Zero-Copy Serialization Considerations
974///
975/// The current implementation uses `Vec<u8>` for the payload, which requires
976/// copying data during serialization and deserialization. For high-performance
977/// scenarios, consider these alternatives:
978///
979/// 1. **`Bytes` from `bytes` crate**: Provides cheap cloning via reference counting
980///    ```ignore
981///    use bytes::Bytes;
982///    pub payload: Bytes,
983///    ```
984///
985/// 2. **`Arc<[u8]>`**: Reference-counted slice for shared ownership
986///    ```ignore
987///    use std::sync::Arc;
988///    pub payload: Arc<[u8]>,
989///    ```
990///
991/// 3. **Borrowed payloads with lifetimes**: For truly zero-copy deserialization
992///    ```ignore
993///    #[derive(Deserialize)]
994///    pub struct SerializedTask<'a> {
995///        #[serde(borrow)]
996///        pub payload: &'a [u8],
997///    }
998///    ```
999///
1000/// Trade-offs:
1001/// - `Vec<u8>`: Simple, owned data, but requires copying
1002/// - `Bytes`: Cheap cloning, but requires external dependency
1003/// - `Arc<[u8]>`: Cheap cloning, but atomic operations have overhead
1004/// - Borrowed: True zero-copy, but lifetime complexity and limited use cases
1005///
1006/// For most use cases, the current `Vec<u8>` approach provides a good balance
1007/// of simplicity and performance. Consider alternatives only when profiling shows
1008/// serialization as a bottleneck.
1009#[derive(Debug, Clone, Serialize, Deserialize)]
1010pub struct SerializedTask {
1011    /// Task metadata
1012    pub metadata: TaskMetadata,
1013
1014    /// Serialized task payload
1015    pub payload: Vec<u8>,
1016}
1017
1018impl SerializedTask {
1019    #[inline]
1020    #[must_use]
1021    pub fn new(name: String, payload: Vec<u8>) -> Self {
1022        Self {
1023            metadata: TaskMetadata::new(name),
1024            payload,
1025        }
1026    }
1027
1028    #[inline]
1029    #[must_use]
1030    pub fn with_priority(mut self, priority: i32) -> Self {
1031        self.metadata.priority = priority;
1032        self
1033    }
1034
1035    #[inline]
1036    #[must_use]
1037    pub fn with_max_retries(mut self, max_retries: u32) -> Self {
1038        self.metadata.max_retries = max_retries;
1039        self
1040    }
1041
1042    #[inline]
1043    #[must_use]
1044    pub fn with_timeout(mut self, timeout_secs: u64) -> Self {
1045        self.metadata.timeout_secs = Some(timeout_secs);
1046        self
1047    }
1048
1049    /// Set the group ID for workflow grouping
1050    #[inline]
1051    #[must_use]
1052    pub fn with_group_id(mut self, group_id: Uuid) -> Self {
1053        self.metadata.group_id = Some(group_id);
1054        self
1055    }
1056
1057    /// Set the chord ID for barrier synchronization
1058    #[inline]
1059    #[must_use]
1060    pub fn with_chord_id(mut self, chord_id: Uuid) -> Self {
1061        self.metadata.chord_id = Some(chord_id);
1062        self
1063    }
1064
1065    /// Set the on-success chain link task name for this task
1066    #[inline]
1067    #[must_use]
1068    pub fn with_on_success_link(mut self, task_name: String) -> Self {
1069        self.metadata.on_success_link = Some(task_name);
1070        self
1071    }
1072
1073    /// Get the age of the task (time since creation)
1074    #[inline]
1075    #[must_use]
1076    pub fn age(&self) -> chrono::Duration {
1077        self.metadata.age()
1078    }
1079
1080    /// Check if the task has expired based on its timeout
1081    #[inline]
1082    #[must_use]
1083    pub fn is_expired(&self) -> bool {
1084        self.metadata.is_expired()
1085    }
1086
1087    /// Check if the task is in a terminal state (Success or Failure)
1088    #[inline]
1089    #[must_use]
1090    pub fn is_terminal(&self) -> bool {
1091        self.metadata.is_terminal()
1092    }
1093
1094    /// Check if the task is in a running or active state
1095    #[inline]
1096    #[must_use]
1097    pub fn is_active(&self) -> bool {
1098        self.metadata.is_active()
1099    }
1100
1101    /// Validate the serialized task
1102    ///
1103    /// Validates both metadata and payload constraints:
1104    /// - Delegates metadata validation to `TaskMetadata::validate()`
1105    /// - Checks payload size (must be < 1MB by default)
1106    ///
1107    /// # Errors
1108    ///
1109    /// Returns an error if validation fails.
1110    pub fn validate(&self) -> Result<(), String> {
1111        self.metadata.validate()?;
1112
1113        if self.payload.is_empty() {
1114            return Err("Task payload cannot be empty".to_string());
1115        }
1116
1117        if self.payload.len() > 1_048_576 {
1118            return Err(format!(
1119                "Task payload too large: {} bytes (max 1MB)",
1120                self.payload.len()
1121            ));
1122        }
1123
1124        Ok(())
1125    }
1126
1127    /// Validate with custom payload size limit
1128    ///
1129    /// # Errors
1130    ///
1131    /// Returns an error if validation fails.
1132    pub fn validate_with_limit(&self, max_payload_bytes: usize) -> Result<(), String> {
1133        self.metadata.validate()?;
1134
1135        if self.payload.is_empty() {
1136            return Err("Task payload cannot be empty".to_string());
1137        }
1138
1139        if self.payload.len() > max_payload_bytes {
1140            return Err(format!(
1141                "Task payload too large: {} bytes (max {} bytes)",
1142                self.payload.len(),
1143                max_payload_bytes
1144            ));
1145        }
1146
1147        Ok(())
1148    }
1149
1150    /// Check if task has a timeout configured
1151    #[inline]
1152    #[must_use]
1153    pub fn has_timeout(&self) -> bool {
1154        self.metadata.has_timeout()
1155    }
1156
1157    /// Check if task is part of a group
1158    #[inline]
1159    #[must_use]
1160    pub fn has_group_id(&self) -> bool {
1161        self.metadata.has_group_id()
1162    }
1163
1164    /// Check if task is part of a chord
1165    #[inline]
1166    #[must_use]
1167    pub fn has_chord_id(&self) -> bool {
1168        self.metadata.has_chord_id()
1169    }
1170
1171    /// Check if task has custom priority (non-zero)
1172    #[inline]
1173    #[must_use]
1174    pub fn has_priority(&self) -> bool {
1175        self.metadata.has_priority()
1176    }
1177
1178    /// Get payload size in bytes
1179    #[inline]
1180    #[must_use]
1181    pub const fn payload_size(&self) -> usize {
1182        self.payload.len()
1183    }
1184
1185    /// Check if payload is empty
1186    #[inline]
1187    #[must_use]
1188    pub fn has_empty_payload(&self) -> bool {
1189        self.payload.is_empty()
1190    }
1191
1192    /// Add a task dependency
1193    #[inline]
1194    #[must_use]
1195    pub fn with_dependency(mut self, dependency: TaskId) -> Self {
1196        self.metadata.dependencies.insert(dependency);
1197        self
1198    }
1199
1200    /// Add multiple task dependencies
1201    #[inline]
1202    #[must_use]
1203    pub fn with_dependencies(mut self, dependencies: impl IntoIterator<Item = TaskId>) -> Self {
1204        self.metadata.dependencies.extend(dependencies);
1205        self
1206    }
1207
1208    /// Check if task has any dependencies
1209    #[inline]
1210    #[must_use]
1211    pub fn has_dependencies(&self) -> bool {
1212        self.metadata.has_dependencies()
1213    }
1214
1215    /// Get the number of dependencies
1216    #[inline]
1217    #[must_use]
1218    pub fn dependency_count(&self) -> usize {
1219        self.metadata.dependency_count()
1220    }
1221
1222    /// Check if a specific task is a dependency
1223    #[inline]
1224    #[must_use]
1225    pub fn depends_on(&self, task_id: &TaskId) -> bool {
1226        self.metadata.depends_on(task_id)
1227    }
1228
1229    /// Check if task has high priority (priority > 0)
1230    #[inline]
1231    #[must_use]
1232    pub fn is_high_priority(&self) -> bool {
1233        self.metadata.is_high_priority()
1234    }
1235
1236    /// Check if task has low priority (priority < 0)
1237    #[inline]
1238    #[must_use]
1239    pub fn is_low_priority(&self) -> bool {
1240        self.metadata.is_low_priority()
1241    }
1242
1243    // ===== Convenience State Checks (Delegated) =====
1244
1245    /// Check if task is in Pending state
1246    #[inline]
1247    #[must_use]
1248    pub fn is_pending(&self) -> bool {
1249        self.metadata.is_pending()
1250    }
1251
1252    /// Check if task is in Running state
1253    #[inline]
1254    #[must_use]
1255    pub fn is_running(&self) -> bool {
1256        self.metadata.is_running()
1257    }
1258
1259    /// Check if task is in Succeeded state
1260    #[inline]
1261    #[must_use]
1262    pub fn is_succeeded(&self) -> bool {
1263        self.metadata.is_succeeded()
1264    }
1265
1266    /// Check if task is in Failed state
1267    #[inline]
1268    #[must_use]
1269    pub fn is_failed(&self) -> bool {
1270        self.metadata.is_failed()
1271    }
1272
1273    /// Check if task is in Retrying state
1274    #[inline]
1275    #[must_use]
1276    pub fn is_retrying(&self) -> bool {
1277        self.metadata.is_retrying()
1278    }
1279
1280    /// Check if task is in Reserved state
1281    #[inline]
1282    #[must_use]
1283    pub fn is_reserved(&self) -> bool {
1284        self.metadata.is_reserved()
1285    }
1286
1287    // ===== Time-related Helpers (Delegated) =====
1288
1289    /// Get remaining time before timeout
1290    #[inline]
1291    #[must_use]
1292    pub fn time_remaining(&self) -> Option<chrono::Duration> {
1293        self.metadata.time_remaining()
1294    }
1295
1296    /// Get the time elapsed since task creation
1297    #[inline]
1298    #[must_use]
1299    pub fn time_elapsed(&self) -> chrono::Duration {
1300        self.metadata.time_elapsed()
1301    }
1302
1303    // ===== Retry Helpers (Delegated) =====
1304
1305    /// Check if task can be retried
1306    #[inline]
1307    #[must_use]
1308    pub fn can_retry(&self) -> bool {
1309        self.metadata.can_retry()
1310    }
1311
1312    /// Get current retry count
1313    #[inline]
1314    #[must_use]
1315    pub const fn retry_count(&self) -> u32 {
1316        self.metadata.retry_count()
1317    }
1318
1319    /// Get remaining retry attempts
1320    #[inline]
1321    #[must_use]
1322    pub const fn retries_remaining(&self) -> u32 {
1323        self.metadata.retries_remaining()
1324    }
1325
1326    // ===== Workflow Helpers (Delegated) =====
1327
1328    /// Check if task is part of any workflow
1329    #[inline]
1330    #[must_use]
1331    pub fn is_part_of_workflow(&self) -> bool {
1332        self.metadata.is_part_of_workflow()
1333    }
1334
1335    /// Get group ID if task is part of a group
1336    #[inline]
1337    #[must_use]
1338    pub fn get_group_id(&self) -> Option<&Uuid> {
1339        self.metadata.get_group_id()
1340    }
1341
1342    /// Get chord ID if task is part of a chord
1343    #[inline]
1344    #[must_use]
1345    pub fn get_chord_id(&self) -> Option<&Uuid> {
1346        self.metadata.get_chord_id()
1347    }
1348}
1349
1350impl fmt::Display for SerializedTask {
1351    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
1352        write!(
1353            f,
1354            "SerializedTask[{}] name={} payload={}B state={}",
1355            &self.metadata.id.to_string()[..8],
1356            self.metadata.name,
1357            self.payload.len(),
1358            self.metadata.state
1359        )?;
1360        if self.metadata.has_priority() {
1361            write!(f, " priority={}", self.metadata.priority)?;
1362        }
1363        if let Some(group_id) = self.metadata.group_id {
1364            write!(f, " group={}", &group_id.to_string()[..8])?;
1365        }
1366        if let Some(chord_id) = self.metadata.chord_id {
1367            write!(f, " chord={}", &chord_id.to_string()[..8])?;
1368        }
1369        Ok(())
1370    }
1371}
1372
1373#[cfg(test)]
1374mod tests {
1375    use super::*;
1376
1377    #[test]
1378    fn test_task_metadata_creation() {
1379        let metadata = TaskMetadata::new("test_task".to_string())
1380            .with_max_retries(5)
1381            .with_timeout(60)
1382            .with_priority(10);
1383
1384        assert_eq!(metadata.name, "test_task");
1385        assert_eq!(metadata.max_retries, 5);
1386        assert_eq!(metadata.timeout_secs, Some(60));
1387        assert_eq!(metadata.priority, 10);
1388        assert_eq!(metadata.state, TaskState::Pending);
1389    }
1390
1391    #[test]
1392    fn test_task_dependencies() {
1393        let dep1 = TaskId::new_v4();
1394        let dep2 = TaskId::new_v4();
1395
1396        let metadata = TaskMetadata::new("test_task".to_string())
1397            .with_dependency(dep1)
1398            .with_dependency(dep2);
1399
1400        assert!(metadata.has_dependencies());
1401        assert_eq!(metadata.dependency_count(), 2);
1402        assert!(metadata.depends_on(&dep1));
1403        assert!(metadata.depends_on(&dep2));
1404    }
1405
1406    #[test]
1407    fn test_task_with_dependencies() {
1408        let dep1 = TaskId::new_v4();
1409        let dep2 = TaskId::new_v4();
1410        let deps = vec![dep1, dep2];
1411
1412        let metadata = TaskMetadata::new("test_task".to_string()).with_dependencies(deps);
1413
1414        assert_eq!(metadata.dependency_count(), 2);
1415        assert!(metadata.depends_on(&dep1));
1416        assert!(metadata.depends_on(&dep2));
1417    }
1418
1419    #[test]
1420    fn test_remove_dependency() {
1421        let dep1 = TaskId::new_v4();
1422        let dep2 = TaskId::new_v4();
1423
1424        let mut metadata = TaskMetadata::new("test_task".to_string())
1425            .with_dependency(dep1)
1426            .with_dependency(dep2);
1427
1428        assert_eq!(metadata.dependency_count(), 2);
1429
1430        let removed = metadata.remove_dependency(&dep1);
1431        assert!(removed);
1432        assert_eq!(metadata.dependency_count(), 1);
1433        assert!(!metadata.depends_on(&dep1));
1434        assert!(metadata.depends_on(&dep2));
1435    }
1436
1437    #[test]
1438    fn test_clear_dependencies() {
1439        let dep1 = TaskId::new_v4();
1440        let dep2 = TaskId::new_v4();
1441
1442        let mut metadata = TaskMetadata::new("test_task".to_string())
1443            .with_dependency(dep1)
1444            .with_dependency(dep2);
1445
1446        assert!(metadata.has_dependencies());
1447        metadata.clear_dependencies();
1448        assert!(!metadata.has_dependencies());
1449        assert_eq!(metadata.dependency_count(), 0);
1450    }
1451
1452    #[test]
1453    fn test_serialized_task_dependencies() {
1454        let dep1 = TaskId::new_v4();
1455        let dep2 = TaskId::new_v4();
1456
1457        let task = SerializedTask::new("test_task".to_string(), vec![1, 2, 3])
1458            .with_dependency(dep1)
1459            .with_dependency(dep2);
1460
1461        assert!(task.has_dependencies());
1462        assert_eq!(task.dependency_count(), 2);
1463        assert!(task.depends_on(&dep1));
1464        assert!(task.depends_on(&dep2));
1465    }
1466
1467    // Chain link field tests
1468    #[test]
1469    fn test_workflow_no_link_serialization_roundtrip() {
1470        let task = SerializedTask::new("my_task".to_string(), b"payload".to_vec());
1471
1472        // No on_success_link set — field must be absent after serde roundtrip
1473        let json = serde_json::to_string(&task).expect("serialize");
1474        let restored: SerializedTask = serde_json::from_str(&json).expect("deserialize");
1475
1476        assert!(
1477            restored.metadata.on_success_link.is_none(),
1478            "on_success_link should be None when never set"
1479        );
1480        // The field must be absent from the JSON (skip_serializing_if)
1481        assert!(
1482            !json.contains("on_success_link"),
1483            "on_success_link should be omitted from JSON when None"
1484        );
1485    }
1486
1487    #[test]
1488    fn test_workflow_chain_link_serialization_roundtrip() {
1489        let task = SerializedTask::new("step_a".to_string(), b"result_bytes".to_vec())
1490            .with_on_success_link("next_task".to_string());
1491
1492        assert_eq!(
1493            task.metadata.on_success_link.as_deref(),
1494            Some("next_task"),
1495            "on_success_link must be Some(\"next_task\") before serialization"
1496        );
1497
1498        let json = serde_json::to_string(&task).expect("serialize");
1499
1500        // The field must be present in the JSON
1501        assert!(
1502            json.contains("on_success_link"),
1503            "on_success_link must appear in JSON when set"
1504        );
1505        assert!(
1506            json.contains("next_task"),
1507            "link name must appear in serialized JSON"
1508        );
1509
1510        let restored: SerializedTask = serde_json::from_str(&json).expect("deserialize");
1511        assert_eq!(
1512            restored.metadata.on_success_link.as_deref(),
1513            Some("next_task"),
1514            "on_success_link must survive a serde roundtrip"
1515        );
1516        assert_eq!(
1517            restored.metadata.name, "step_a",
1518            "task name must survive a serde roundtrip"
1519        );
1520    }
1521
1522    #[test]
1523    fn test_task_metadata_with_on_success_link_builder() {
1524        let metadata =
1525            TaskMetadata::new("producer".to_string()).with_on_success_link("consumer".to_string());
1526
1527        assert_eq!(
1528            metadata.on_success_link.as_deref(),
1529            Some("consumer"),
1530            "TaskMetadata builder must set on_success_link"
1531        );
1532    }
1533
1534    #[test]
1535    fn test_serialized_task_with_on_success_link_builder() {
1536        let task = SerializedTask::new("a".to_string(), vec![1, 2, 3])
1537            .with_on_success_link("b".to_string());
1538
1539        assert_eq!(
1540            task.metadata.on_success_link.as_deref(),
1541            Some("b"),
1542            "SerializedTask builder must delegate on_success_link to metadata"
1543        );
1544    }
1545
1546    // Integration tests for full task lifecycle
1547    #[cfg(test)]
1548    mod integration_tests {
1549        use super::*;
1550        use crate::{StateHistory, TaskState};
1551
1552        #[test]
1553        fn test_complete_task_lifecycle() {
1554            // Create a task
1555            let mut task = SerializedTask::new("process_data".to_string(), vec![1, 2, 3, 4, 5])
1556                .with_priority(5)
1557                .with_max_retries(3)
1558                .with_timeout(60);
1559
1560            assert_eq!(task.metadata.state, TaskState::Pending);
1561            assert!(task.is_active());
1562            assert!(!task.is_terminal());
1563
1564            // Simulate task progression through states
1565            let mut history = StateHistory::with_initial(task.metadata.state.clone());
1566
1567            // Task is received by worker
1568            task.metadata.state = TaskState::Received;
1569            history.transition(task.metadata.state.clone());
1570
1571            // Task is reserved
1572            task.metadata.state = TaskState::Reserved;
1573            history.transition(task.metadata.state.clone());
1574
1575            // Task starts running
1576            task.metadata.state = TaskState::Running;
1577            history.transition(task.metadata.state.clone());
1578
1579            // Task completes successfully
1580            task.metadata.state = TaskState::Succeeded(vec![10, 20, 30]);
1581            history.transition(task.metadata.state.clone());
1582
1583            assert!(task.is_terminal());
1584            assert!(!task.is_active());
1585            assert_eq!(history.transition_count(), 4);
1586        }
1587
1588        #[test]
1589        fn test_task_retry_lifecycle() {
1590            let mut task =
1591                SerializedTask::new("failing_task".to_string(), vec![1, 2, 3]).with_max_retries(3);
1592
1593            let mut history = StateHistory::with_initial(TaskState::Pending);
1594
1595            // First attempt fails
1596            task.metadata.state = TaskState::Running;
1597            history.transition(task.metadata.state.clone());
1598
1599            task.metadata.state = TaskState::Failed("Network error".to_string());
1600            history.transition(task.metadata.state.clone());
1601
1602            // Check if can retry
1603            assert!(task.metadata.state.can_retry(task.metadata.max_retries));
1604
1605            // First retry
1606            task.metadata.state = TaskState::Retrying(1);
1607            history.transition(task.metadata.state.clone());
1608            assert_eq!(task.metadata.state.retry_count(), 1);
1609
1610            task.metadata.state = TaskState::Failed("Still failing".to_string());
1611            history.transition(task.metadata.state.clone());
1612
1613            // Second retry
1614            task.metadata.state = TaskState::Retrying(2);
1615            history.transition(task.metadata.state.clone());
1616            assert_eq!(task.metadata.state.retry_count(), 2);
1617
1618            // Finally succeeds
1619            task.metadata.state = TaskState::Succeeded(vec![]);
1620            history.transition(task.metadata.state.clone());
1621
1622            assert!(task.is_terminal());
1623            assert_eq!(history.transition_count(), 6);
1624        }
1625
1626        #[test]
1627        fn test_task_with_dependencies_lifecycle() {
1628            // Create parent tasks
1629            let parent1_id = TaskId::new_v4();
1630            let parent2_id = TaskId::new_v4();
1631
1632            // Create child task that depends on parents
1633            let child_task = SerializedTask::new("child_task".to_string(), vec![1, 2, 3])
1634                .with_dependency(parent1_id)
1635                .with_dependency(parent2_id)
1636                .with_priority(10);
1637
1638            assert!(child_task.has_dependencies());
1639            assert_eq!(child_task.dependency_count(), 2);
1640            assert!(child_task.depends_on(&parent1_id));
1641            assert!(child_task.depends_on(&parent2_id));
1642
1643            // Verify task properties
1644            assert_eq!(child_task.metadata.priority, 10);
1645            assert!(child_task.is_high_priority());
1646        }
1647
1648        #[test]
1649        fn test_task_serialization_roundtrip() {
1650            let original = SerializedTask::new("test_task".to_string(), vec![1, 2, 3, 4, 5])
1651                .with_priority(5)
1652                .with_max_retries(3)
1653                .with_timeout(120)
1654                .with_dependency(TaskId::new_v4());
1655
1656            // Serialize to JSON
1657            let json = serde_json::to_string(&original).expect("Failed to serialize");
1658
1659            // Deserialize back
1660            let deserialized: SerializedTask =
1661                serde_json::from_str(&json).expect("Failed to deserialize");
1662
1663            assert_eq!(deserialized.metadata.name, original.metadata.name);
1664            assert_eq!(deserialized.metadata.priority, original.metadata.priority);
1665            assert_eq!(
1666                deserialized.metadata.max_retries,
1667                original.metadata.max_retries
1668            );
1669            assert_eq!(
1670                deserialized.metadata.timeout_secs,
1671                original.metadata.timeout_secs
1672            );
1673            assert_eq!(deserialized.payload, original.payload);
1674            assert_eq!(deserialized.dependency_count(), original.dependency_count());
1675        }
1676
1677        #[test]
1678        fn test_task_validation_lifecycle() {
1679            // Valid task
1680            let valid_task = SerializedTask::new("valid_task".to_string(), vec![1, 2, 3])
1681                .with_max_retries(5)
1682                .with_timeout(30);
1683
1684            assert!(valid_task.validate().is_ok());
1685
1686            // Invalid task - empty name
1687            let mut invalid_task = SerializedTask::new(String::new(), vec![1, 2, 3]);
1688            assert!(invalid_task.metadata.validate().is_err());
1689
1690            // Invalid task - excessive retries
1691            invalid_task =
1692                SerializedTask::new("task".to_string(), vec![1, 2, 3]).with_max_retries(10000);
1693            assert!(invalid_task.metadata.validate().is_err());
1694
1695            // Invalid task - zero timeout
1696            let mut invalid_metadata = TaskMetadata::new("task".to_string());
1697            invalid_metadata.timeout_secs = Some(0);
1698            assert!(invalid_metadata.validate().is_err());
1699        }
1700
1701        #[test]
1702        fn test_task_expiration_lifecycle() {
1703            // Create task with 1 second timeout
1704            let task =
1705                SerializedTask::new("expiring_task".to_string(), vec![1, 2, 3]).with_timeout(1);
1706
1707            // Task should not be expired immediately
1708            assert!(!task.is_expired());
1709
1710            // Wait for task to expire
1711            std::thread::sleep(std::time::Duration::from_secs(2));
1712
1713            // Task should now be expired
1714            assert!(task.is_expired());
1715        }
1716
1717        #[test]
1718        fn test_workflow_with_multiple_dependencies() {
1719            use crate::TaskDag;
1720
1721            // Create a workflow: task1 -> task2 -> task3
1722            //                      \-> task4 -^
1723            let mut dag = TaskDag::new();
1724
1725            let task1 = TaskId::new_v4();
1726            let task2 = TaskId::new_v4();
1727            let task3 = TaskId::new_v4();
1728            let task4 = TaskId::new_v4();
1729
1730            dag.add_node(task1, "load_data");
1731            dag.add_node(task2, "transform_data");
1732            dag.add_node(task3, "save_results");
1733            dag.add_node(task4, "validate_data");
1734
1735            // task2 depends on task1
1736            dag.add_dependency(task2, task1).unwrap();
1737            // task4 depends on task1
1738            dag.add_dependency(task4, task1).unwrap();
1739            // task3 depends on both task2 and task4
1740            dag.add_dependency(task3, task2).unwrap();
1741            dag.add_dependency(task3, task4).unwrap();
1742
1743            // Validate DAG
1744            assert!(dag.validate().is_ok());
1745
1746            // Get execution order
1747            let order = dag.topological_sort().unwrap();
1748            assert_eq!(order.len(), 4);
1749
1750            // task1 should be first
1751            assert_eq!(order[0], task1);
1752            // task3 should be last
1753            assert_eq!(order[3], task3);
1754        }
1755
1756        #[test]
1757        fn test_task_state_history_full_lifecycle() {
1758            let mut history = StateHistory::with_initial(TaskState::Pending);
1759
1760            // Simulate complete lifecycle
1761            history.transition(TaskState::Received);
1762            history.transition(TaskState::Reserved);
1763            history.transition(TaskState::Running);
1764            history.transition(TaskState::Failed("Temporary error".to_string()));
1765            history.transition(TaskState::Retrying(1));
1766            history.transition(TaskState::Running);
1767            history.transition(TaskState::Succeeded(vec![1, 2, 3]));
1768
1769            assert_eq!(history.transition_count(), 7);
1770            assert!(history.current_state().unwrap().is_terminal());
1771
1772            // Check states we transitioned TO
1773            assert!(history.has_been_in_state("RECEIVED"));
1774            assert!(history.has_been_in_state("RESERVED"));
1775            assert!(history.has_been_in_state("RUNNING"));
1776            assert!(history.has_been_in_state("FAILURE"));
1777            assert!(history.has_been_in_state("RETRYING"));
1778            assert!(history.has_been_in_state("SUCCESS"));
1779
1780            // Current state should be SUCCESS
1781            assert_eq!(history.current_state().unwrap().name(), "SUCCESS");
1782        }
1783    }
1784}
celers_core/task.rs

celers_core/
task.rs
