scientific_workflow/execution/
scope.rs

1use std::fs;
2use std::path::{Component, Path, PathBuf};
3use std::sync::atomic::{AtomicU64, Ordering};
4
5use crate::clock::utc_now_rfc3339;
6
7use super::error::ExecutionScopeError;
8
9static EXECUTION_SEQUENCE: AtomicU64 = AtomicU64::new(0);
10
11/// One created or reopened project-execution directory.
12#[derive(Clone, Debug, Eq, PartialEq)]
13pub struct ExecutionScope {
14    directory: PathBuf,
15    created_at_utc: Option<String>,
16}
17
18impl ExecutionScope {
19    /// Uses the recording root itself as the deterministic execution scope.
20    ///
21    /// This is intended for application-owned semantic task names where a
22    /// repeated configuration must resolve to the same output path instead of
23    /// receiving a timestamped execution wrapper.
24    pub fn open_or_create(recording_root: impl AsRef<Path>) -> Result<Self, ExecutionScopeError> {
25        let directory = recording_root.as_ref().to_path_buf();
26        fs::create_dir_all(&directory).map_err(|source| ExecutionScopeError::Io {
27            operation: "create deterministic",
28            path: directory.clone(),
29            source,
30        })?;
31        Ok(Self {
32            directory,
33            created_at_utc: None,
34        })
35    }
36
37    /// Creates a uniquely named scope beneath `recording_root`.
38    ///
39    /// The readable UTC component is supplemented by process and sequence
40    /// values. Exclusive directory creation remains the final collision check.
41    pub fn create_generated(recording_root: impl AsRef<Path>) -> Result<Self, ExecutionScopeError> {
42        let recording_root = recording_root.as_ref();
43        fs::create_dir_all(recording_root).map_err(|source| ExecutionScopeError::Io {
44            operation: "create recording root for",
45            path: recording_root.to_path_buf(),
46            source,
47        })?;
48        let created_at_utc =
49            utc_now_rfc3339().map_err(|source| ExecutionScopeError::Timestamp { source })?;
50        let compact_timestamp = compact_timestamp(&created_at_utc);
51        for _ in 0..1024 {
52            let sequence = EXECUTION_SEQUENCE.fetch_add(1, Ordering::Relaxed);
53            let name = format!(
54                "execution-{compact_timestamp}-{}-{sequence}",
55                std::process::id()
56            );
57            let directory = recording_root.join(name);
58            match fs::create_dir(&directory) {
59                Ok(()) => {
60                    return Ok(Self {
61                        directory,
62                        created_at_utc: Some(created_at_utc),
63                    });
64                }
65                Err(source) if source.kind() == std::io::ErrorKind::AlreadyExists => continue,
66                Err(source) => {
67                    return Err(ExecutionScopeError::Io {
68                        operation: "create generated",
69                        path: directory,
70                        source,
71                    });
72                }
73            }
74        }
75        Err(ExecutionScopeError::IdentityExhausted {
76            root: recording_root.to_path_buf(),
77        })
78    }
79
80    /// Creates one caller-named scope beneath `recording_root`.
81    pub fn create_named(
82        recording_root: impl AsRef<Path>,
83        name: impl Into<String>,
84    ) -> Result<Self, ExecutionScopeError> {
85        let recording_root = recording_root.as_ref();
86        let name = name.into();
87        validate_name(&name)?;
88        let created_at_utc =
89            utc_now_rfc3339().map_err(|source| ExecutionScopeError::Timestamp { source })?;
90        fs::create_dir_all(recording_root).map_err(|source| ExecutionScopeError::Io {
91            operation: "create recording root for",
92            path: recording_root.to_path_buf(),
93            source,
94        })?;
95        let directory = recording_root.join(&name);
96        fs::create_dir(&directory).map_err(|source| ExecutionScopeError::Io {
97            operation: "create named",
98            path: directory.clone(),
99            source,
100        })?;
101        Ok(Self {
102            directory,
103            created_at_utc: Some(created_at_utc),
104        })
105    }
106
107    /// Opens an existing execution scope without creating or modifying files.
108    pub fn open_existing(directory: impl Into<PathBuf>) -> Result<Self, ExecutionScopeError> {
109        let directory = directory.into();
110        let metadata = fs::metadata(&directory).map_err(|source| ExecutionScopeError::Io {
111            operation: "inspect existing",
112            path: directory.clone(),
113            source,
114        })?;
115        if !metadata.is_dir() {
116            return Err(ExecutionScopeError::Io {
117                operation: "open non-directory",
118                path: directory.clone(),
119                source: std::io::Error::new(
120                    std::io::ErrorKind::NotADirectory,
121                    "execution scope must be a directory",
122                ),
123            });
124        }
125        Ok(Self {
126            directory,
127            created_at_utc: None,
128        })
129    }
130
131    /// Returns the scope directory.
132    pub fn directory(&self) -> &Path {
133        &self.directory
134    }
135
136    /// Returns the automatically captured creation timestamp when this handle
137    /// created the scope.
138    ///
139    /// Reopened legacy scopes return `None` because no auxiliary scope metadata
140    /// is invented merely to recover a timestamp.
141    pub fn created_at_utc(&self) -> Option<&str> {
142        self.created_at_utc.as_deref()
143    }
144
145    /// Derives the absent recording path reserved for one deterministic task.
146    ///
147    /// The directory is deliberately not created: the recording writer must
148    /// retain exclusive creation and overwrite protection.
149    pub fn task_recording_directory(&self, task_ordinal: u64) -> PathBuf {
150        self.directory.join(format!("task-{task_ordinal:06}"))
151    }
152
153    /// Derives the absent recording path reserved for one semantic relative path.
154    pub fn named_task_recording_directory(
155        &self,
156        name: &str,
157    ) -> Result<PathBuf, ExecutionScopeError> {
158        validate_relative_name(name)?;
159        Ok(self.directory.join(name))
160    }
161}
162
163/// Requires a nonempty relative name containing exactly one normal component.
164fn validate_name(name: &str) -> Result<(), ExecutionScopeError> {
165    let path = Path::new(name);
166    let mut components = path.components();
167    let valid = !name.trim().is_empty()
168        && matches!(components.next(), Some(Component::Normal(_)))
169        && components.next().is_none();
170    if valid {
171        Ok(())
172    } else {
173        Err(ExecutionScopeError::InvalidName {
174            name: name.to_owned(),
175        })
176    }
177}
178
179/// Requires a nonempty relative path made only from normal components.
180fn validate_relative_name(name: &str) -> Result<(), ExecutionScopeError> {
181    let path = Path::new(name);
182    let valid = !name.trim().is_empty()
183        && path.components().next().is_some()
184        && path
185            .components()
186            .all(|component| matches!(component, Component::Normal(_)));
187    if valid {
188        Ok(())
189    } else {
190        Err(ExecutionScopeError::InvalidName {
191            name: name.to_owned(),
192        })
193    }
194}
195
196/// Removes RFC 3339 punctuation while retaining ordered UTC date/time digits.
197fn compact_timestamp(timestamp: &str) -> String {
198    timestamp
199        .chars()
200        .filter(|character| character.is_ascii_alphanumeric())
201        .collect()
202}
scientific_workflow/execution/scope.rs

scientific_workflow/execution/
scope.rs
