scirs2_io/workflow/
types.rs

1//! Auto-generated module
2//!
3//! 🤖 Generated with [SplitRS](https://github.com/cool-japan/splitrs)
4
5use crate::error::{IoError, Result};
6use crate::metadata::{Metadata, MetadataValue};
7use chrono::{DateTime, Datelike, Duration, Utc};
8use serde::{Deserialize, Serialize};
9use std::collections::{HashMap, HashSet};
10use std::path::PathBuf;
11use std::sync::{Arc, Mutex};
12
13/// Schedule configuration for periodic execution
14#[derive(Debug, Clone, Serialize, Deserialize)]
15pub struct ScheduleConfig {
16    /// Cron expression for scheduling (e.g., "0 0 * * *")
17    pub cron: Option<String>,
18    /// Fixed interval between executions
19    pub interval: Option<Duration>,
20    /// Earliest time to start execution
21    pub start_time: Option<DateTime<Utc>>,
22    /// Latest time to stop execution
23    pub end_time: Option<DateTime<Utc>>,
24}
25/// Retry policy for failed tasks
26#[derive(Debug, Clone, Serialize, Deserialize)]
27pub struct RetryPolicy {
28    /// Maximum number of retry attempts
29    pub max_retries: usize,
30    /// Initial backoff delay in seconds
31    pub backoff_seconds: u64,
32    /// Whether to use exponential backoff
33    pub exponential_backoff: bool,
34}
35/// Workflow builder
36pub struct WorkflowBuilder {
37    workflow: Workflow,
38}
39impl WorkflowBuilder {
40    /// Create a new workflow builder
41    pub fn new(id: impl Into<String>, name: impl Into<String>) -> Self {
42        Self {
43            workflow: Workflow {
44                id: id.into(),
45                name: name.into(),
46                description: None,
47                tasks: Vec::new(),
48                dependencies: HashMap::new(),
49                config: WorkflowConfig::default(),
50                metadata: Metadata::new(),
51            },
52        }
53    }
54    /// Set description
55    pub fn description(mut self, desc: impl Into<String>) -> Self {
56        self.workflow.description = Some(desc.into());
57        self
58    }
59    /// Add a task
60    pub fn add_task(mut self, task: Task) -> Self {
61        self.workflow.tasks.push(task);
62        self
63    }
64    /// Add a dependency
65    pub fn add_dependency(
66        mut self,
67        task_id: impl Into<String>,
68        depends_on: impl Into<String>,
69    ) -> Self {
70        let task_id = task_id.into();
71        let depends_on = depends_on.into();
72        self.workflow
73            .dependencies
74            .entry(task_id)
75            .or_default()
76            .push(depends_on);
77        self
78    }
79    /// Configure workflow
80    pub fn configure<F>(mut self, f: F) -> Self
81    where
82        F: FnOnce(&mut WorkflowConfig),
83    {
84        f(&mut self.workflow.config);
85        self
86    }
87    /// Build the workflow
88    pub fn build(self) -> Result<Workflow> {
89        self.validate()?;
90        Ok(self.workflow)
91    }
92    fn validate(&self) -> Result<()> {
93        if self.has_cycles() {
94            return Err(IoError::ValidationError(
95                "Workflow contains dependency cycles".to_string(),
96            ));
97        }
98        let mut ids = HashSet::new();
99        for task in &self.workflow.tasks {
100            if !ids.insert(&task.id) {
101                return Err(IoError::ValidationError(format!(
102                    "Duplicate task ID: {}",
103                    task.id
104                )));
105            }
106        }
107        for (task_id, deps) in &self.workflow.dependencies {
108            if !ids.contains(&task_id) {
109                return Err(IoError::ValidationError(format!(
110                    "Unknown task in dependencies: {}",
111                    task_id
112                )));
113            }
114            for dep in deps {
115                if !ids.contains(&dep) {
116                    return Err(IoError::ValidationError(format!(
117                        "Unknown dependency: {}",
118                        dep
119                    )));
120                }
121            }
122        }
123        Ok(())
124    }
125    fn has_cycles(&self) -> bool {
126        let mut visited = HashSet::new();
127        let mut rec_stack = HashSet::new();
128        for task in &self.workflow.tasks {
129            if !visited.contains(&task.id)
130                && self.has_cycle_dfs(&task.id, &mut visited, &mut rec_stack)
131            {
132                return true;
133            }
134        }
135        false
136    }
137    fn has_cycle_dfs(
138        &self,
139        node: &str,
140        visited: &mut HashSet<String>,
141        rec_stack: &mut HashSet<String>,
142    ) -> bool {
143        visited.insert(node.to_string());
144        rec_stack.insert(node.to_string());
145        if let Some(deps) = self.workflow.dependencies.get(node) {
146            for dep in deps {
147                if !visited.contains(dep) {
148                    if self.has_cycle_dfs(dep, visited, rec_stack) {
149                        return true;
150                    }
151                } else if rec_stack.contains(dep) {
152                    return true;
153                }
154            }
155        }
156        rec_stack.remove(node);
157        false
158    }
159}
160/// Workflow execution state
161#[derive(Debug, Clone)]
162pub struct WorkflowState {
163    /// Workflow identifier
164    pub workflowid: String,
165    /// Unique execution identifier
166    pub executionid: String,
167    /// Current workflow status
168    pub status: WorkflowStatus,
169    /// State of each task in the workflow
170    pub task_states: HashMap<String, TaskState>,
171    /// When the workflow started
172    pub start_time: Option<DateTime<Utc>>,
173    /// When the workflow completed
174    pub end_time: Option<DateTime<Utc>>,
175    /// Error message if workflow failed
176    pub error: Option<String>,
177}
178/// Workflow configuration
179#[derive(Debug, Clone, Serialize, Deserialize)]
180pub struct WorkflowConfig {
181    /// Maximum number of tasks to run in parallel
182    pub max_parallel_tasks: usize,
183    /// Retry policy for failed tasks
184    pub retry_policy: RetryPolicy,
185    /// Maximum workflow execution time
186    pub timeout: Option<Duration>,
187    /// Directory for workflow checkpoints
188    pub checkpoint_dir: Option<PathBuf>,
189    /// Notification settings
190    pub notifications: NotificationConfig,
191    /// Scheduling configuration for periodic execution
192    pub scheduling: Option<ScheduleConfig>,
193}
194/// Runtime state of a task execution
195#[derive(Debug, Clone)]
196pub struct TaskState {
197    /// Current execution status
198    pub status: TaskStatus,
199    /// When the task started executing
200    pub start_time: Option<DateTime<Utc>>,
201    /// When the task finished executing
202    pub end_time: Option<DateTime<Utc>>,
203    /// Number of execution attempts
204    pub attempts: usize,
205    /// Error message if task failed
206    pub error: Option<String>,
207    /// Task outputs as key-value pairs
208    pub outputs: HashMap<String, serde_json::Value>,
209}
210/// Resource requirements for tasks
211#[derive(Debug, Clone, Serialize, Deserialize, Default)]
212pub struct ResourceRequirements {
213    /// Number of CPU cores required
214    pub cpu_cores: Option<usize>,
215    /// Memory requirement in GB
216    pub memorygb: Option<f64>,
217    /// GPU requirements if needed
218    pub gpu: Option<GpuRequirement>,
219    /// Disk space requirement in GB
220    pub disk_space_gb: Option<f64>,
221}
222/// Notification delivery channels
223#[derive(Debug, Clone, Serialize, Deserialize)]
224#[serde(rename_all = "snake_case")]
225pub enum NotificationChannel {
226    /// Send email notifications
227    Email {
228        /// Email addresses to notify
229        to: Vec<String>,
230    },
231    /// Send webhook notifications
232    Webhook {
233        /// Webhook URL
234        url: String,
235    },
236    /// Write notifications to file
237    File {
238        /// File path for notifications
239        path: PathBuf,
240    },
241}
242/// Workflow definition
243#[derive(Debug, Clone, Serialize, Deserialize)]
244pub struct Workflow {
245    /// Unique workflow identifier
246    pub id: String,
247    /// Human-readable workflow name
248    pub name: String,
249    /// Optional workflow description
250    pub description: Option<String>,
251    /// List of tasks in the workflow
252    pub tasks: Vec<Task>,
253    /// Task dependencies (task_id -> list of prerequisite task_ids)
254    pub dependencies: HashMap<String, Vec<String>>,
255    /// Workflow configuration
256    pub config: WorkflowConfig,
257    /// Workflow metadata
258    pub metadata: Metadata,
259}
260/// Notification configuration
261#[derive(Debug, Clone, Serialize, Deserialize)]
262pub struct NotificationConfig {
263    /// Send notification on successful completion
264    pub on_success: bool,
265    /// Send notification on failure
266    pub on_failure: bool,
267    /// Send notification when workflow starts
268    pub on_start: bool,
269    /// List of notification channels to use
270    pub channels: Vec<NotificationChannel>,
271}
272/// Workflow executor
273pub struct WorkflowExecutor {
274    config: ExecutorConfig,
275    state: Arc<Mutex<HashMap<String, WorkflowState>>>,
276}
277impl WorkflowExecutor {
278    /// Create a new workflow executor
279    pub fn new(config: ExecutorConfig) -> Self {
280        Self {
281            config,
282            state: Arc::new(Mutex::new(HashMap::new())),
283        }
284    }
285    /// Execute a workflow
286    pub fn execute(&self, workflow: &Workflow) -> Result<String> {
287        let executionid = format!("{}-{}", workflow.id, Utc::now().timestamp());
288        let mut state = WorkflowState {
289            workflowid: workflow.id.clone(),
290            executionid: executionid.clone(),
291            status: WorkflowStatus::Pending,
292            task_states: HashMap::new(),
293            start_time: None,
294            end_time: None,
295            error: None,
296        };
297        for task in &workflow.tasks {
298            state.task_states.insert(
299                task.id.clone(),
300                TaskState {
301                    status: TaskStatus::Pending,
302                    start_time: None,
303                    end_time: None,
304                    attempts: 0,
305                    error: None,
306                    outputs: HashMap::new(),
307                },
308            );
309        }
310        self.state
311            .lock()
312            .expect("Operation failed")
313            .insert(executionid.clone(), state);
314        self.executeworkflow_internal(workflow.clone(), executionid.clone())?;
315        Ok(executionid)
316    }
317    /// Internal workflow execution logic
318    fn executeworkflow_internal(&self, workflow: Workflow, executionid: String) -> Result<()> {
319        {
320            let mut states = self.state.lock().expect("Operation failed");
321            if let Some(state) = states.get_mut(&executionid) {
322                state.status = WorkflowStatus::Running;
323                state.start_time = Some(Utc::now());
324            }
325        }
326        let execution_result = self.execute_tasks_in_order(&workflow, &executionid);
327        {
328            let mut states = self.state.lock().expect("Operation failed");
329            if let Some(state) = states.get_mut(&executionid) {
330                state.end_time = Some(Utc::now());
331                match execution_result {
332                    Ok(_) => state.status = WorkflowStatus::Success,
333                    Err(ref e) => {
334                        state.status = WorkflowStatus::Failed;
335                        state.error = Some(e.to_string());
336                    }
337                }
338            }
339        }
340        execution_result
341    }
342    /// Execute tasks in dependency order
343    fn execute_tasks_in_order(&self, workflow: &Workflow, executionid: &str) -> Result<()> {
344        let mut executed_tasks = HashSet::new();
345        let mut remaining_tasks: HashSet<String> =
346            workflow.tasks.iter().map(|t| t.id.clone()).collect();
347        while !remaining_tasks.is_empty() {
348            let mut tasks_to_execute = Vec::new();
349            for task_id in &remaining_tasks {
350                let can_execute = workflow
351                    .dependencies
352                    .get(task_id as &String)
353                    .is_none_or(|deps| deps.iter().all(|dep| executed_tasks.contains(dep)));
354                if can_execute {
355                    tasks_to_execute.push(task_id.clone());
356                }
357            }
358            if tasks_to_execute.is_empty() {
359                return Err(IoError::Other(
360                    "Circular dependency or unresolvable dependencies".to_string(),
361                ));
362            }
363            let batch_size = workflow
364                .config
365                .max_parallel_tasks
366                .min(tasks_to_execute.len());
367            for batch in tasks_to_execute.chunks(batch_size) {
368                for task_id in batch {
369                    let task = workflow
370                        .tasks
371                        .iter()
372                        .find(|t| &t.id == task_id)
373                        .ok_or_else(|| IoError::Other(format!("Task not found: {task_id}")))?;
374                    self.execute_single_task(task, executionid)?;
375                    executed_tasks.insert(task_id.clone());
376                    remaining_tasks.remove(task_id);
377                }
378            }
379        }
380        Ok(())
381    }
382    /// Execute a single task with retry logic
383    fn execute_single_task(&self, task: &Task, executionid: &str) -> Result<()> {
384        let mut attempt = 0;
385        let max_retries = 3;
386        loop {
387            attempt += 1;
388            {
389                let mut states = self.state.lock().expect("Operation failed");
390                if let Some(state) = states.get_mut(executionid) {
391                    if let Some(task_state) = state.task_states.get_mut(&task.id) {
392                        task_state.status = if attempt == 1 {
393                            TaskStatus::Running
394                        } else {
395                            TaskStatus::Retrying
396                        };
397                        task_state.start_time = Some(Utc::now());
398                        task_state.attempts = attempt;
399                    }
400                }
401            }
402            let result = self.execute_task_by_type(task);
403            {
404                let mut states = self.state.lock().expect("Operation failed");
405                if let Some(state) = states.get_mut(executionid) {
406                    if let Some(task_state) = state.task_states.get_mut(&task.id) {
407                        task_state.end_time = Some(Utc::now());
408                        match result {
409                            Ok(outputs) => {
410                                task_state.status = TaskStatus::Success;
411                                task_state.outputs = outputs;
412                                task_state.error = None;
413                                return Ok(());
414                            }
415                            Err(e) => {
416                                if attempt >= max_retries {
417                                    task_state.status = TaskStatus::Failed;
418                                    task_state.error = Some(e.to_string());
419                                    return Err(e);
420                                } else {
421                                    task_state.error = Some(format!("Attempt {attempt}: {e}"));
422                                }
423                            }
424                        }
425                    }
426                }
427            }
428            if attempt < max_retries {
429                std::thread::sleep(std::time::Duration::from_secs(1 << (attempt - 1)));
430            }
431        }
432    }
433    /// Execute task based on its type
434    fn execute_task_by_type(&self, task: &Task) -> Result<HashMap<String, serde_json::Value>> {
435        let mut outputs = HashMap::new();
436        match task.task_type {
437            TaskType::DataIngestion => {
438                outputs.insert("status".to_string(), serde_json::json!("completed"));
439                outputs.insert("records_processed".to_string(), serde_json::json!(1000));
440            }
441            TaskType::Transform => {
442                outputs.insert("status".to_string(), serde_json::json!("completed"));
443                outputs.insert("rows_transformed".to_string(), serde_json::json!(1000));
444            }
445            TaskType::Validation => {
446                outputs.insert("status".to_string(), serde_json::json!("completed"));
447                outputs.insert("validation_errors".to_string(), serde_json::json!(0));
448            }
449            TaskType::MLTraining => {
450                outputs.insert("status".to_string(), serde_json::json!("completed"));
451                outputs.insert("model_accuracy".to_string(), serde_json::json!(0.95));
452            }
453            TaskType::MLInference => {
454                outputs.insert("status".to_string(), serde_json::json!("completed"));
455                outputs.insert("predictions_generated".to_string(), serde_json::json!(500));
456            }
457            TaskType::Export => {
458                outputs.insert("status".to_string(), serde_json::json!("completed"));
459                outputs.insert("files_written".to_string(), serde_json::json!(1));
460            }
461            TaskType::Script => {
462                outputs.insert("status".to_string(), serde_json::json!("completed"));
463                outputs.insert("exit_code".to_string(), serde_json::json!(0));
464            }
465            TaskType::SubWorkflow => {
466                outputs.insert("status".to_string(), serde_json::json!("completed"));
467                outputs.insert(
468                    "subworkflowid".to_string(),
469                    serde_json::json!(format!("sub-{}", task.id)),
470                );
471            }
472            TaskType::Conditional => {
473                let condition_met = true;
474                outputs.insert(
475                    "condition_met".to_string(),
476                    serde_json::json!(condition_met),
477                );
478                outputs.insert("status".to_string(), serde_json::json!("completed"));
479            }
480            TaskType::Parallel => {
481                outputs.insert("status".to_string(), serde_json::json!("completed"));
482                outputs.insert("parallel_tasks_completed".to_string(), serde_json::json!(4));
483            }
484        }
485        outputs.insert("execution_time_ms".to_string(), serde_json::json!(100));
486        outputs.insert(
487            "execution_timestamp".to_string(),
488            serde_json::json!(Utc::now().to_rfc3339()),
489        );
490        Ok(outputs)
491    }
492    /// Get workflow state
493    pub fn get_state(&self, executionid: &str) -> Option<WorkflowState> {
494        self.state
495            .lock()
496            .expect("Operation failed")
497            .get(executionid)
498            .cloned()
499    }
500    /// Cancel a workflow execution
501    pub fn cancel(&self, executionid: &str) -> Result<()> {
502        let mut states = self.state.lock().expect("Operation failed");
503        if let Some(state) = states.get_mut(executionid) {
504            state.status = WorkflowStatus::Cancelled;
505            state.end_time = Some(Utc::now());
506            Ok(())
507        } else {
508            Err(IoError::Other(format!("Execution {executionid} not found")))
509        }
510    }
511}
512/// Configuration for the workflow executor
513#[derive(Debug, Clone)]
514pub struct ExecutorConfig {
515    /// Maximum number of workflows that can run concurrently
516    pub max_concurrentworkflows: usize,
517    /// Maximum time allowed for a single task execution
518    pub task_timeout: Duration,
519    /// Interval between workflow state checkpoints
520    pub checkpoint_interval: Duration,
521}
522/// GPU resource requirements for workflow execution
523#[derive(Debug, Clone, Serialize, Deserialize)]
524pub struct GpuRequirement {
525    /// Number of GPUs required
526    pub count: usize,
527    /// Memory requirement in GB per GPU
528    pub memorygb: Option<f64>,
529    /// Required CUDA compute capability (e.g., "7.5")
530    pub compute_capability: Option<String>,
531}
532/// Task types
533#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
534#[serde(rename_all = "snake_case")]
535pub enum TaskType {
536    /// Data ingestion from files or databases
537    DataIngestion,
538    /// Data transformation using pipeline
539    Transform,
540    /// Data validation
541    Validation,
542    /// Machine learning training
543    MLTraining,
544    /// Model inference
545    MLInference,
546    /// Data export
547    Export,
548    /// Custom script execution
549    Script,
550    /// Sub-workflow execution
551    SubWorkflow,
552    /// Conditional execution
553    Conditional,
554    /// Parallel execution
555    Parallel,
556}
557/// Workflow execution status
558#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
559pub enum WorkflowStatus {
560    /// Workflow is waiting to start
561    Pending,
562    /// Workflow is currently executing
563    Running,
564    /// Workflow completed successfully
565    Success,
566    /// Workflow failed with errors
567    Failed,
568    /// Workflow was cancelled by user
569    Cancelled,
570    /// Workflow is temporarily paused
571    Paused,
572}
573/// Task definition
574#[derive(Debug, Clone, Serialize, Deserialize)]
575pub struct Task {
576    /// Unique task identifier
577    pub id: String,
578    /// Human-readable task name
579    pub name: String,
580    /// Type of task to execute
581    pub task_type: TaskType,
582    /// Task-specific configuration parameters
583    pub config: serde_json::Value,
584    /// Input data identifiers
585    pub inputs: Vec<String>,
586    /// Output data identifiers
587    pub outputs: Vec<String>,
588    /// Resource requirements for this task
589    pub resources: ResourceRequirements,
590}
591/// Task execution status
592#[derive(Debug, Clone, Copy, PartialEq, Eq)]
593pub enum TaskStatus {
594    /// Task is waiting to execute
595    Pending,
596    /// Task is currently running
597    Running,
598    /// Task completed successfully
599    Success,
600    /// Task failed with errors
601    Failed,
602    /// Task was skipped due to conditions
603    Skipped,
604    /// Task is being retried after failure
605    Retrying,
606}
scirs2_io/workflow/types.rs

scirs2_io/workflow/
types.rs
