scientific_workflow/execution/
scope.rs1use std::fs;
8use std::path::{Component, Path, PathBuf};
9use std::sync::atomic::{AtomicU64, Ordering};
10
11use crate::clock::utc_now_rfc3339;
12
13use super::error::ExecutionScopeError;
14
15static EXECUTION_SEQUENCE: AtomicU64 = AtomicU64::new(0);
16
17#[derive(Clone, Debug, Eq, PartialEq)]
19pub struct ExecutionScope {
20 directory: PathBuf,
21 created_at_utc: Option<String>,
22}
23
24impl ExecutionScope {
25 pub fn open_or_create(recording_root: impl AsRef<Path>) -> Result<Self, ExecutionScopeError> {
31 let directory = recording_root.as_ref().to_path_buf();
32 fs::create_dir_all(&directory).map_err(|source| ExecutionScopeError::Io {
33 operation: "create deterministic",
34 path: directory.clone(),
35 source,
36 })?;
37 Ok(Self {
38 directory,
39 created_at_utc: None,
40 })
41 }
42
43 pub fn create_generated(recording_root: impl AsRef<Path>) -> Result<Self, ExecutionScopeError> {
48 let recording_root = recording_root.as_ref();
49 fs::create_dir_all(recording_root).map_err(|source| ExecutionScopeError::Io {
50 operation: "create recording root for",
51 path: recording_root.to_path_buf(),
52 source,
53 })?;
54 let created_at_utc =
55 utc_now_rfc3339().map_err(|source| ExecutionScopeError::Timestamp { source })?;
56 let compact_timestamp = compact_timestamp(&created_at_utc);
57 for _ in 0..1024 {
58 let sequence = EXECUTION_SEQUENCE.fetch_add(1, Ordering::Relaxed);
59 let name = format!(
60 "execution-{compact_timestamp}-{}-{sequence}",
61 std::process::id()
62 );
63 let directory = recording_root.join(name);
64 match fs::create_dir(&directory) {
65 Ok(()) => {
66 return Ok(Self {
67 directory,
68 created_at_utc: Some(created_at_utc),
69 });
70 }
71 Err(source) if source.kind() == std::io::ErrorKind::AlreadyExists => continue,
72 Err(source) => {
73 return Err(ExecutionScopeError::Io {
74 operation: "create generated",
75 path: directory,
76 source,
77 });
78 }
79 }
80 }
81 Err(ExecutionScopeError::IdentityExhausted {
82 root: recording_root.to_path_buf(),
83 })
84 }
85
86 pub fn create_named(
88 recording_root: impl AsRef<Path>,
89 name: impl Into<String>,
90 ) -> Result<Self, ExecutionScopeError> {
91 let recording_root = recording_root.as_ref();
92 let name = name.into();
93 validate_name(&name)?;
94 let created_at_utc =
95 utc_now_rfc3339().map_err(|source| ExecutionScopeError::Timestamp { source })?;
96 fs::create_dir_all(recording_root).map_err(|source| ExecutionScopeError::Io {
97 operation: "create recording root for",
98 path: recording_root.to_path_buf(),
99 source,
100 })?;
101 let directory = recording_root.join(&name);
102 fs::create_dir(&directory).map_err(|source| ExecutionScopeError::Io {
103 operation: "create named",
104 path: directory.clone(),
105 source,
106 })?;
107 Ok(Self {
108 directory,
109 created_at_utc: Some(created_at_utc),
110 })
111 }
112
113 pub fn open_existing(directory: impl Into<PathBuf>) -> Result<Self, ExecutionScopeError> {
115 let directory = directory.into();
116 let metadata = fs::metadata(&directory).map_err(|source| ExecutionScopeError::Io {
117 operation: "inspect existing",
118 path: directory.clone(),
119 source,
120 })?;
121 if !metadata.is_dir() {
122 return Err(ExecutionScopeError::Io {
123 operation: "open non-directory",
124 path: directory,
125 source: std::io::Error::new(
126 std::io::ErrorKind::NotADirectory,
127 "execution scope must be a directory",
128 ),
129 });
130 }
131 Ok(Self {
132 directory,
133 created_at_utc: None,
134 })
135 }
136
137 pub fn directory(&self) -> &Path {
139 &self.directory
140 }
141
142 pub fn created_at_utc(&self) -> Option<&str> {
148 self.created_at_utc.as_deref()
149 }
150
151 pub fn task_recording_directory(&self, ordinal: u64) -> PathBuf {
156 self.directory.join(format!("task-{ordinal:06}"))
157 }
158
159 pub fn named_task_recording_directory(
161 &self,
162 name: &str,
163 ) -> Result<PathBuf, ExecutionScopeError> {
164 validate_relative_name(name)?;
165 Ok(self.directory.join(name))
166 }
167}
168
169fn validate_name(name: &str) -> Result<(), ExecutionScopeError> {
171 validate_path_components(name, false)
172}
173
174fn validate_relative_name(name: &str) -> Result<(), ExecutionScopeError> {
176 validate_path_components(name, true)
177}
178
179fn validate_path_components(name: &str, allow_relative: bool) -> Result<(), ExecutionScopeError> {
180 let path = Path::new(name);
181 let mut components = path.components();
182 let valid = !name.trim().is_empty()
183 && matches!(components.next(), Some(Component::Normal(_)))
184 && if allow_relative {
185 components.all(|component| matches!(component, Component::Normal(_)))
186 } else {
187 components.next().is_none()
188 };
189 if valid {
190 Ok(())
191 } else {
192 Err(ExecutionScopeError::InvalidName {
193 name: name.to_owned(),
194 })
195 }
196}
197
198fn compact_timestamp(timestamp: &str) -> String {
200 timestamp
201 .chars()
202 .filter(|character| character.is_ascii_alphanumeric())
203 .collect()
204}