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

scientific_workflow/study/
task.rs
