scientific_workflow/runtime/
task.rs

1//! Task-local execution context and workload contract.
2
3use std::error::Error;
4
5use serde::de::DeserializeOwned;
6use serde_json::Value;
7
8use super::error::RuntimeError;
9use super::phase::{Task, TaskDisplayKind, TaskId, TaskKey};
10use super::reporting::{ActivityTask, TaskProgress};
11use crate::configuration::TaskConfig;
12
13/// Error-erased result returned by one task-owned workload.
14pub type TaskResult = Result<(), Box<dyn Error + Send + Sync + 'static>>;
15
16pub(crate) type Workload = Box<dyn FnOnce(&TaskContext) -> TaskResult + Send + 'static>;
17
18enum TaskHandle {
19    Progress(TaskProgress),
20    Activity(ActivityTask),
21}
22
23/// Read-only task identity/configuration plus reporting and cancellation.
24///
25/// The context deliberately has no filesystem, storage, artifact, network,
26/// subprocess, or machine-resource operations. The workload owns all such
27/// effects directly.
28pub struct TaskContext {
29    task: Task,
30    handle: TaskHandle,
31}
32
33impl TaskContext {
34    pub(crate) fn progress(task: Task, progress: TaskProgress) -> Self {
35        Self {
36            task,
37            handle: TaskHandle::Progress(progress),
38        }
39    }
40
41    pub(crate) fn activity(task: Task, activity: ActivityTask) -> Self {
42        Self {
43            task,
44            handle: TaskHandle::Activity(activity),
45        }
46    }
47
48    /// Borrows the complete immutable task declaration.
49    pub fn task(&self) -> &Task {
50        &self.task
51    }
52
53    /// Borrows the exact phase-qualified task key.
54    pub fn key(&self) -> &TaskKey {
55        self.task.key()
56    }
57
58    /// Borrows the phase-local task ID.
59    pub fn id(&self) -> &TaskId {
60        self.task.id()
61    }
62
63    /// Borrows the task kind/namespace.
64    pub fn kind(&self) -> &str {
65        self.task.kind()
66    }
67
68    /// Borrows the configuration from which this task was generated.
69    pub fn configuration(&self) -> &TaskConfig {
70        self.task.configuration()
71    }
72
73    /// Borrows one fixed, swept, or explicit task parameter.
74    pub fn value(&self, key: &str) -> Option<&Value> {
75        self.task.value(key)
76    }
77
78    /// Decodes one required task parameter.
79    pub fn decode_value<T>(&self, key: &str) -> Result<T, RuntimeError>
80    where
81        T: DeserializeOwned,
82    {
83        self.task.decode_value(key)
84    }
85
86    /// Borrows iterative progress for a progress task.
87    pub fn progress_handle(&self) -> Option<&TaskProgress> {
88        match &self.handle {
89            TaskHandle::Progress(progress) => Some(progress),
90            TaskHandle::Activity(_) => None,
91        }
92    }
93
94    /// Sets or replaces the target iteration of a progress task.
95    pub fn set_target_iteration(&self, target: u64) -> Result<(), RuntimeError> {
96        self.required_progress()?.set_target_iteration(target)
97    }
98
99    /// Synchronizes a progress task to the authoritative scientific iteration.
100    pub fn set_iteration(&self, iteration: u64) -> Result<(), RuntimeError> {
101        self.required_progress()?.set_iteration(iteration)
102    }
103
104    /// Synchronizes iteration and reports whether work should continue.
105    pub fn should_continue(&self, iteration: u64) -> Result<bool, RuntimeError> {
106        self.required_progress()?.should_continue(iteration)
107    }
108
109    /// Borrows lifecycle-only reporting for an activity task.
110    pub fn activity_handle(&self) -> Option<&ActivityTask> {
111        match &self.handle {
112            TaskHandle::Progress(_) => None,
113            TaskHandle::Activity(activity) => Some(activity),
114        }
115    }
116
117    /// Reports whether the runtime requested cooperative cancellation.
118    pub fn is_cancelled(&self) -> bool {
119        match &self.handle {
120            TaskHandle::Progress(progress) => progress.is_cancelled(),
121            TaskHandle::Activity(activity) => activity.is_cancelled(),
122        }
123    }
124
125    /// Updates the human-readable task detail.
126    pub fn set_detail(&self, detail: impl Into<String>) {
127        let detail = detail.into();
128        match &self.handle {
129            TaskHandle::Progress(progress) => progress.set_detail(detail),
130            TaskHandle::Activity(activity) => activity.set_detail(detail),
131        }
132    }
133
134    /// Sends a task-scoped display message.
135    pub fn report(&self, message: impl Into<String>) -> Result<(), RuntimeError> {
136        let message = message.into();
137        match &self.handle {
138            TaskHandle::Progress(progress) => progress.report(message),
139            TaskHandle::Activity(activity) => activity.report(message),
140        }
141    }
142
143    pub(crate) fn complete(self) -> Result<(), RuntimeError> {
144        match self.handle {
145            TaskHandle::Progress(progress) => progress.complete(None),
146            TaskHandle::Activity(activity) => {
147                activity.complete();
148                Ok(())
149            }
150        }
151    }
152
153    pub(crate) fn fail(self, reason: impl Into<String>) {
154        let reason = reason.into();
155        match self.handle {
156            TaskHandle::Progress(progress) => progress.fail(reason),
157            TaskHandle::Activity(activity) => activity.fail(reason),
158        }
159    }
160
161    pub(crate) fn cancel(self, reason: impl Into<String>) {
162        let reason = reason.into();
163        match self.handle {
164            TaskHandle::Progress(progress) => progress.cancel(reason),
165            TaskHandle::Activity(activity) => activity.cancel(reason),
166        }
167    }
168
169    fn required_progress(&self) -> Result<&TaskProgress, RuntimeError> {
170        self.progress_handle()
171            .ok_or_else(|| RuntimeError::ManagedTaskKindMismatch {
172                task: self.key().to_string(),
173                requested: "progress",
174                actual: match self.task.display_kind() {
175                    TaskDisplayKind::Progress => "progress",
176                    TaskDisplayKind::Activity => "activity",
177                },
178            })
179    }
180}
181
182impl std::fmt::Debug for TaskContext {
183    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
184        formatter
185            .debug_struct("TaskContext")
186            .field("key", self.key())
187            .field("kind", &self.kind())
188            .field("display_kind", &self.task.display_kind())
189            .finish_non_exhaustive()
190    }
191}
scientific_workflow/runtime/task.rs

scientific_workflow/runtime/
task.rs
