scientific_workflow/study/
phase.rs

1//! First-class phase and task declarations owned by a study.
2
3use std::collections::{BTreeMap, HashSet};
4use std::fmt;
5use std::sync::Arc;
6use std::time::{Duration, Instant};
7
8use serde::de::DeserializeOwned;
9use serde_json::Value;
10use sha2::{Digest, Sha256};
11
12use super::error::StudyError;
13use super::task::{TaskContext, TaskResult, Workload};
14
15/// Stable numeric identity of one execution phase.
16#[derive(Clone, Copy, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)]
17pub struct PhaseId(u64);
18
19impl PhaseId {
20    /// Creates a phase identity suitable for command-line selections such as
21    /// `[2, 4, 5]`.
22    pub const fn new(value: u64) -> Self {
23        Self(value)
24    }
25
26    /// Returns the exact numeric phase identity.
27    pub const fn get(self) -> u64 {
28        self.0
29    }
30}
31
32impl From<u64> for PhaseId {
33    fn from(value: u64) -> Self {
34        Self::new(value)
35    }
36}
37
38impl fmt::Display for PhaseId {
39    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
40        self.0.fmt(formatter)
41    }
42}
43
44/// Stable task identity scoped to one phase.
45#[derive(Clone, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)]
46pub struct TaskId(Arc<str>);
47
48impl TaskId {
49    /// Creates a task ID. Empty or whitespace-only IDs are rejected when the
50    /// owning phase is built, keeping task construction infallible.
51    pub fn new(value: impl Into<String>) -> Self {
52        Self(value.into().into())
53    }
54
55    /// Borrows the exact ID text.
56    pub fn as_str(&self) -> &str {
57        &self.0
58    }
59}
60
61impl fmt::Display for TaskId {
62    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
63        self.0.fmt(formatter)
64    }
65}
66
67/// Exact phase-qualified task lookup key.
68#[derive(Clone, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)]
69pub struct TaskKey {
70    phase: PhaseId,
71    task: TaskId,
72}
73
74impl TaskKey {
75    /// Creates one phase-qualified task key.
76    pub fn new(phase: impl Into<PhaseId>, task: TaskId) -> Self {
77        Self {
78            phase: phase.into(),
79            task,
80        }
81    }
82
83    /// Returns the owning phase ID.
84    pub const fn phase_id(&self) -> PhaseId {
85        self.phase
86    }
87
88    /// Borrows the phase-local task ID.
89    pub fn task_id(&self) -> &TaskId {
90        &self.task
91    }
92}
93
94impl fmt::Display for TaskKey {
95    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
96        write!(formatter, "{}/{}", self.phase, self.task)
97    }
98}
99
100/// Reporting shape of one task.
101#[derive(Clone, Copy, Debug, Eq, PartialEq)]
102#[non_exhaustive]
103pub enum TaskMode {
104    /// Work that may report iterative progress.
105    Progress,
106    /// One-shot work with lifecycle status only.
107    OneShot,
108}
109
110/// First-class immutable task declaration owned by exactly one [`Phase`].
111pub struct Task {
112    key: TaskKey,
113    category: Arc<str>,
114    mode: TaskMode,
115    label: Arc<str>,
116    metadata: Arc<BTreeMap<String, Value>>,
117    workload: Option<Workload>,
118    completed: bool,
119}
120
121impl Task {
122    /// Creates a one-shot task.
123    pub fn one_shot<W>(id: impl Into<String>, label: impl Into<String>, workload: W) -> Self
124    where
125        W: FnOnce(&TaskContext) -> TaskResult + Send + 'static,
126    {
127        Self::new(
128            id,
129            label,
130            TaskMode::OneShot,
131            Some(Box::new(workload)),
132            false,
133        )
134    }
135
136    /// Creates a task that may report progress through its context.
137    pub fn progress<W>(id: impl Into<String>, label: impl Into<String>, workload: W) -> Self
138    where
139        W: FnOnce(&TaskContext) -> TaskResult + Send + 'static,
140    {
141        Self::new(
142            id,
143            label,
144            TaskMode::Progress,
145            Some(Box::new(workload)),
146            false,
147        )
148    }
149
150    /// Creates an application-verified task that is already satisfied.
151    pub fn completed(id: impl Into<String>, label: impl Into<String>) -> Self {
152        Self::new(id, label, TaskMode::OneShot, None, true)
153    }
154
155    fn new(
156        id: impl Into<String>,
157        label: impl Into<String>,
158        mode: TaskMode,
159        workload: Option<Workload>,
160        completed: bool,
161    ) -> Self {
162        Self {
163            key: TaskKey::new(0, TaskId::new(id)),
164            category: "task".into(),
165            mode,
166            label: label.into().into(),
167            metadata: Arc::new(BTreeMap::new()),
168            workload,
169            completed,
170        }
171    }
172
173    /// Sets the optional application-defined task category.
174    #[must_use]
175    pub fn category(mut self, category: impl Into<String>) -> Self {
176        self.category = category.into().into();
177        self
178    }
179
180    /// Adds application-defined metadata used for inspection and selection.
181    #[must_use]
182    pub fn metadata(mut self, key: impl Into<String>, value: impl Into<Value>) -> Self {
183        Arc::make_mut(&mut self.metadata).insert(key.into(), value.into());
184        self
185    }
186
187    /// Returns the exact phase-qualified task key.
188    pub fn key(&self) -> &TaskKey {
189        &self.key
190    }
191
192    /// Returns the phase-local task ID.
193    pub fn id(&self) -> &TaskId {
194        self.key.task_id()
195    }
196
197    /// Returns the application-defined task category.
198    pub fn category_name(&self) -> &str {
199        &self.category
200    }
201
202    /// Returns the automatically generated display label.
203    pub fn label(&self) -> &str {
204        &self.label
205    }
206
207    /// Returns whether the task reports iterative progress or one-shot status.
208    pub const fn mode(&self) -> TaskMode {
209        self.mode
210    }
211
212    /// Borrows one application-defined metadata value.
213    pub fn metadata_value(&self, key: &str) -> Option<&Value> {
214        self.metadata.get(key)
215    }
216
217    /// Borrows one required task parameter.
218    pub fn require_value(&self, key: &str) -> Result<&Value, StudyError> {
219        self.metadata_value(key)
220            .ok_or_else(|| StudyError::UnknownTaskMetadata {
221                task: self.key.to_string(),
222                key: key.to_owned(),
223            })
224    }
225
226    /// Decodes one required task parameter without first cloning its JSON tree.
227    pub fn decode_value<T>(&self, key: &str) -> Result<T, StudyError>
228    where
229        T: DeserializeOwned,
230    {
231        T::deserialize(self.require_value(key)?).map_err(|source| StudyError::DecodeTaskMetadata {
232            task: self.key.to_string(),
233            key: key.to_owned(),
234            source,
235        })
236    }
237
238    /// Iterates application-defined metadata.
239    pub fn metadata_iter(&self) -> impl Iterator<Item = (&str, &Value)> + '_ {
240        self.metadata
241            .iter()
242            .map(|(key, value)| (key.as_str(), value))
243    }
244
245    pub(crate) fn metadata_map(&self) -> Arc<BTreeMap<String, Value>> {
246        Arc::clone(&self.metadata)
247    }
248
249    pub(crate) fn take_workload(&mut self) -> Option<Workload> {
250        self.workload.take()
251    }
252
253    pub(crate) fn has_workload(&self) -> bool {
254        self.workload.is_some()
255    }
256
257    pub(crate) const fn is_completed(&self) -> bool {
258        self.completed
259    }
260
261    fn attach_to_phase(&mut self, phase: PhaseId) {
262        self.key.phase = phase;
263    }
264}
265
266impl fmt::Debug for Task {
267    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
268        formatter
269            .debug_struct("Task")
270            .field("key", &self.key)
271            .field("category", &self.category)
272            .field("label", &self.label)
273            .field("mode", &self.mode)
274            .field("metadata", &self.metadata.len())
275            .finish_non_exhaustive()
276    }
277}
278
279/// Immutable nonempty execution phase and renderer section.
280pub struct Phase {
281    id: PhaseId,
282    label: Arc<str>,
283    tasks: Vec<Task>,
284    max_active_tasks: usize,
285    prepared_task_queue_capacity: usize,
286    delay_per_task: Option<Duration>,
287    task_timeout: Option<Duration>,
288    deadline_after: Option<Duration>,
289    dependencies: Vec<PhaseId>,
290    failure_policy: PhaseFailurePolicy,
291    require_confirm: bool,
292}
293
294/// Scheduling behavior after the first workload failure in a phase.
295#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
296#[non_exhaustive]
297pub enum PhaseFailurePolicy {
298    /// Stop new work and cooperatively cancel active workloads.
299    #[default]
300    FailFast,
301    /// Stop new work but allow already active workloads to finish.
302    FinishActive,
303}
304
305impl PhaseFailurePolicy {
306    /// Returns the stable uncolored policy name used by plain output.
307    pub const fn as_str(self) -> &'static str {
308        match self {
309            Self::FailFast => "fail-fast",
310            Self::FinishActive => "finish-active",
311        }
312    }
313}
314
315impl Phase {
316    /// Begins declaring a phase. Tasks must be added before [`PhaseBuilder::build`].
317    pub fn builder(id: impl Into<PhaseId>, label: impl Into<String>) -> PhaseBuilder {
318        PhaseBuilder {
319            id: id.into(),
320            label: label.into(),
321            tasks: Vec::new(),
322            max_active_tasks: 1,
323            prepared_task_queue_capacity: 1,
324            delay_per_task: None,
325            task_timeout: None,
326            deadline_after: None,
327            dependencies: Vec::new(),
328            failure_policy: PhaseFailurePolicy::FailFast,
329            require_confirm: false,
330        }
331    }
332
333    /// Returns the stable phase identity.
334    pub const fn id(&self) -> PhaseId {
335        self.id
336    }
337
338    /// Returns the human-facing phase heading.
339    pub fn label(&self) -> &str {
340        &self.label
341    }
342
343    /// Returns every task in deterministic display/execution order.
344    pub fn tasks(&self) -> &[Task] {
345        &self.tasks
346    }
347
348    /// Returns the phase-local concurrent workload ceiling.
349    pub const fn max_active_tasks(&self) -> usize {
350        self.max_active_tasks
351    }
352
353    /// Returns the prepared-but-not-running workload ceiling.
354    pub const fn prepared_task_queue_capacity(&self) -> usize {
355        self.prepared_task_queue_capacity
356    }
357
358    /// Returns the optional minimum interval between consecutive task starts.
359    pub const fn delay_per_task(&self) -> Option<Duration> {
360        self.delay_per_task
361    }
362
363    /// Returns the optional elapsed-time limit for each running task.
364    pub const fn task_timeout(&self) -> Option<Duration> {
365        self.task_timeout
366    }
367
368    /// Returns the optional phase deadline measured from phase execution start.
369    pub const fn deadline_after(&self) -> Option<Duration> {
370        self.deadline_after
371    }
372
373    /// Returns declared predecessor phases in declaration order.
374    pub fn dependencies(&self) -> &[PhaseId] {
375        &self.dependencies
376    }
377
378    /// Returns the behavior selected for the first workload failure.
379    pub const fn failure_policy(&self) -> PhaseFailurePolicy {
380        self.failure_policy
381    }
382
383    /// Reports whether successful completion requires confirmation before the
384    /// next selected phase may start.
385    pub const fn requires_confirmation(&self) -> bool {
386        self.require_confirm
387    }
388
389    pub(crate) fn into_tasks(self) -> Vec<Task> {
390        self.tasks
391    }
392
393    /// Returns one exact phase-local task by ID.
394    pub fn task(&self, id: &TaskId) -> Option<&Task> {
395        self.tasks.iter().find(|task| task.key.task_id() == id)
396    }
397
398    /// Returns the sole task matching an exact partial selector.
399    pub fn unique_task_matching(&self, selector: &TaskSelector) -> Result<&Task, StudyError> {
400        let mut matches = self.tasks.iter().filter(|task| selector.matches(task));
401        let first = matches.next().ok_or_else(|| StudyError::TaskNotFound {
402            selector: selector.to_string(),
403        })?;
404        if let Some(second) = matches.next() {
405            return Err(StudyError::TaskSelectorAmbiguous {
406                selector: selector.to_string(),
407                first: first.key.to_string(),
408                second: second.key.to_string(),
409            });
410        }
411        Ok(first)
412    }
413}
414
415/// Builder for one nonempty first-class phase.
416pub struct PhaseBuilder {
417    id: PhaseId,
418    label: String,
419    tasks: Vec<Task>,
420    max_active_tasks: usize,
421    prepared_task_queue_capacity: usize,
422    delay_per_task: Option<Duration>,
423    task_timeout: Option<Duration>,
424    deadline_after: Option<Duration>,
425    dependencies: Vec<PhaseId>,
426    failure_policy: PhaseFailurePolicy,
427    require_confirm: bool,
428}
429
430impl PhaseBuilder {
431    /// Registers one task with this phase.
432    pub fn task(mut self, task: Task) -> Self {
433        self.tasks.push(task);
434        self
435    }
436
437    /// Registers tasks in deterministic declaration order.
438    pub fn tasks<I>(mut self, tasks: I) -> Self
439    where
440        I: IntoIterator<Item = Task>,
441    {
442        self.tasks.extend(tasks);
443        self
444    }
445
446    /// Sets the later scheduler's phase-local active workload ceiling.
447    pub fn max_active_tasks(mut self, maximum: usize) -> Self {
448        self.max_active_tasks = maximum;
449        self
450    }
451
452    /// Sets the later scheduler's prepared-work queue capacity.
453    pub fn prepared_task_queue_capacity(mut self, capacity: usize) -> Self {
454        self.prepared_task_queue_capacity = capacity;
455        self
456    }
457
458    /// Sets a minimum start-to-start interval for executable tasks.
459    ///
460    /// This policy is optional. Without this call, tasks are admitted exactly
461    /// as before. Completed tasks do not consume a delay rank.
462    pub fn delay_per_task(mut self, delay: Duration) -> Self {
463        self.delay_per_task = Some(delay);
464        self
465    }
466
467    /// Sets the maximum elapsed time for each task after it starts.
468    ///
469    /// Expiration requests cooperative cancellation; Rust workloads cannot be
470    /// forcibly terminated while they are blocked in user or system code.
471    pub fn task_timeout(mut self, timeout: Duration) -> Self {
472        self.task_timeout = Some(timeout);
473        self
474    }
475
476    /// Sets a phase-wide deadline relative to the phase execution start.
477    ///
478    /// Once reached, no additional tasks start and active tasks receive a
479    /// cooperative cancellation request.
480    pub fn deadline_after(mut self, deadline: Duration) -> Self {
481        self.deadline_after = Some(deadline);
482        self
483    }
484
485    /// Declares one phase that must be satisfied before this phase starts.
486    pub fn depends_on(mut self, dependency: impl Into<PhaseId>) -> Self {
487        self.dependencies.push(dependency.into());
488        self
489    }
490
491    /// Selects behavior after the first workload failure.
492    pub fn failure_policy(mut self, policy: PhaseFailurePolicy) -> Self {
493        self.failure_policy = policy;
494        self
495    }
496
497    /// Requires the user to type `yes` before advancing to the next phase.
498    ///
499    /// The default is `false`. This setting has no effect when this phase is
500    /// the final selected phase because there is no transition to confirm.
501    pub fn require_confirm(mut self, require: bool) -> Self {
502        self.require_confirm = require;
503        self
504    }
505
506    /// Validates and creates one immutable nonempty phase.
507    pub fn build(mut self) -> Result<Phase, StudyError> {
508        if self.label.trim().is_empty() {
509            return Err(StudyError::InvalidPhaseLabel { phase: self.id.0 });
510        }
511        if self.tasks.is_empty() {
512            return Err(StudyError::EmptyPhase { phase: self.id.0 });
513        }
514        if self.max_active_tasks == 0 {
515            return Err(StudyError::InvalidPhaseWorkloadLimit { phase: self.id.0 });
516        }
517        if self.prepared_task_queue_capacity == 0 {
518            return Err(StudyError::InvalidPhaseQueueCapacity { phase: self.id.0 });
519        }
520        for (setting, duration) in [
521            ("delay_per_task", self.delay_per_task),
522            ("task_timeout", self.task_timeout),
523            ("deadline_after", self.deadline_after),
524        ] {
525            if duration.is_some_and(|duration| {
526                duration.is_zero() || Instant::now().checked_add(duration).is_none()
527            }) {
528                return Err(StudyError::InvalidPhaseTiming {
529                    phase: self.id.0,
530                    setting,
531                });
532            }
533        }
534
535        let mut ids = HashSet::with_capacity(self.tasks.len());
536        for task in &mut self.tasks {
537            if task.key.task_id().as_str().trim().is_empty() {
538                return Err(StudyError::InvalidTaskId { phase: self.id.0 });
539            }
540            if task.category.trim().is_empty() {
541                return Err(StudyError::InvalidTaskCategory {
542                    task: task.key.task_id().to_string(),
543                });
544            }
545            if !ids.insert(task.key.task_id().clone()) {
546                return Err(StudyError::DuplicateTaskId {
547                    phase: self.id.0,
548                    task: task.key.task_id().to_string(),
549                });
550            }
551            task.attach_to_phase(self.id);
552        }
553
554        let mut dependencies = HashSet::with_capacity(self.dependencies.len());
555        for dependency in &self.dependencies {
556            if !dependencies.insert(*dependency) || *dependency == self.id {
557                return Err(StudyError::PhaseDependencyCycle { phase: self.id.0 });
558            }
559        }
560
561        Ok(Phase {
562            id: self.id,
563            label: self.label.into(),
564            tasks: self.tasks,
565            max_active_tasks: self.max_active_tasks,
566            prepared_task_queue_capacity: self.prepared_task_queue_capacity,
567            delay_per_task: self.delay_per_task,
568            task_timeout: self.task_timeout,
569            deadline_after: self.deadline_after,
570            dependencies: self.dependencies,
571            failure_policy: self.failure_policy,
572            require_confirm: self.require_confirm,
573        })
574    }
575}
576
577impl fmt::Debug for Phase {
578    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
579        formatter
580            .debug_struct("Phase")
581            .field("id", &self.id)
582            .field("label", &self.label)
583            .field("tasks", &self.tasks.len())
584            .field("max_active_tasks", &self.max_active_tasks)
585            .field(
586                "prepared_task_queue_capacity",
587                &self.prepared_task_queue_capacity,
588            )
589            .field("delay_per_task", &self.delay_per_task)
590            .field("task_timeout", &self.task_timeout)
591            .field("deadline_after", &self.deadline_after)
592            .field("dependencies", &self.dependencies)
593            .field("failure_policy", &self.failure_policy)
594            .field("require_confirm", &self.require_confirm)
595            .finish()
596    }
597}
598
599impl fmt::Debug for PhaseBuilder {
600    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
601        formatter
602            .debug_struct("PhaseBuilder")
603            .field("id", &self.id)
604            .field("label", &self.label)
605            .field("tasks", &self.tasks.len())
606            .field("max_active_tasks", &self.max_active_tasks)
607            .field(
608                "prepared_task_queue_capacity",
609                &self.prepared_task_queue_capacity,
610            )
611            .field("delay_per_task", &self.delay_per_task)
612            .field("task_timeout", &self.task_timeout)
613            .field("deadline_after", &self.deadline_after)
614            .field("dependencies", &self.dependencies)
615            .field("require_confirm", &self.require_confirm)
616            .finish_non_exhaustive()
617    }
618}
619
620/// Exact partial selector over phase, task category, and metadata.
621#[derive(Clone, Debug, Default)]
622pub struct TaskSelector {
623    phase: Option<PhaseId>,
624    category: Option<Arc<str>>,
625    metadata: Vec<(Box<str>, Value)>,
626}
627
628impl TaskSelector {
629    /// Creates an unconstrained selector.
630    pub fn new() -> Self {
631        Self::default()
632    }
633
634    /// Constrains selection to one phase.
635    pub fn phase(mut self, phase: impl Into<PhaseId>) -> Self {
636        self.phase = Some(phase.into());
637        self
638    }
639
640    /// Constrains selection to one task category.
641    pub fn category(mut self, category: impl Into<String>) -> Self {
642        self.category = Some(category.into().into());
643        self
644    }
645
646    /// Adds or replaces one exact metadata constraint.
647    pub fn metadata(mut self, key: impl Into<String>, value: impl Into<Value>) -> Self {
648        let key = key.into();
649        if let Some((_, current)) = self
650            .metadata
651            .iter_mut()
652            .find(|(candidate, _)| candidate.as_ref() == key)
653        {
654            *current = value.into();
655        } else {
656            self.metadata.push((key.into_boxed_str(), value.into()));
657        }
658        self
659    }
660
661    pub(crate) fn matches(&self, task: &Task) -> bool {
662        self.phase.is_none_or(|phase| task.key.phase == phase)
663            && self
664                .category
665                .as_deref()
666                .is_none_or(|category| task.category_name() == category)
667            && self
668                .metadata
669                .iter()
670                .all(|(key, value)| task.metadata_value(key) == Some(value))
671    }
672}
673
674impl fmt::Display for TaskSelector {
675    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
676        let mut fields = Vec::new();
677        if let Some(phase) = self.phase {
678            fields.push(format!("phase={phase}"));
679        }
680        if let Some(category) = &self.category {
681            fields.push(format!("category={category}"));
682        }
683        fields.extend(
684            self.metadata
685                .iter()
686                .map(|(key, value)| format!("{key}={}", compact_value(value))),
687        );
688        formatter.write_str(&fields.join(", "))
689    }
690}
691
692fn compact_value(value: &Value) -> String {
693    match value {
694        Value::Array(values) => format!("<array:{}:{}>", values.len(), short_hash(value)),
695        Value::Object(values) => format!("<object:{}:{}>", values.len(), short_hash(value)),
696        _ => serde_json::to_string(value)
697            .expect("serde_json::Value always serializes to valid compact JSON"),
698    }
699}
700
701fn short_hash(value: &Value) -> String {
702    let bytes = serde_json::to_vec(value)
703        .expect("serde_json::Value always serializes to valid compact JSON");
704    let digest = Sha256::digest(bytes);
705    digest[..4]
706        .iter()
707        .map(|byte| format!("{byte:02x}"))
708        .collect::<Vec<_>>()
709        .join("")
710}
scientific_workflow/study/phase.rs

scientific_workflow/study/
phase.rs
