scientific_workflow/
artifact.rs

1//! Generic content-addressed input artifacts inside an execution scope.
2
3use std::fs::{self, File, OpenOptions};
4use std::io::Write;
5use std::path::{Component, Path, PathBuf};
6use std::sync::atomic::{AtomicU64, Ordering};
7
8use serde::{Deserialize, Serialize};
9use sha2::{Digest, Sha256};
10use thiserror::Error;
11
12use crate::execution::ExecutionScope;
13
14static TEMPORARY_SEQUENCE: AtomicU64 = AtomicU64::new(0);
15
16/// Exact identity and execution-relative location of immutable bytes.
17#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
18#[serde(deny_unknown_fields)]
19pub struct ArtifactDescriptor {
20    sha256: String,
21    path: String,
22}
23
24impl ArtifactDescriptor {
25    /// Returns the lowercase SHA-256 identity of the exact bytes.
26    pub fn sha256(&self) -> &str {
27        &self.sha256
28    }
29
30    /// Returns the artifact path relative to its execution directory.
31    pub fn path(&self) -> &str {
32        &self.path
33    }
34}
35
36/// Whether publishing created new bytes or reused identical existing bytes.
37#[derive(Clone, Copy, Debug, Eq, Hash, PartialEq)]
38pub enum ArtifactDisposition {
39    Created,
40    Reused,
41}
42
43/// Result of atomically publishing immutable content.
44#[derive(Clone, Debug, Eq, PartialEq)]
45pub struct PersistedArtifact {
46    descriptor: ArtifactDescriptor,
47    disposition: ArtifactDisposition,
48}
49
50/// Verified canonical path and exact immutable bytes.
51#[derive(Clone, Debug, Eq, PartialEq)]
52pub struct VerifiedArtifact {
53    path: PathBuf,
54    bytes: Vec<u8>,
55}
56
57impl VerifiedArtifact {
58    /// Returns the canonical verified filesystem path.
59    pub fn path(&self) -> &Path {
60        &self.path
61    }
62
63    /// Borrows the bytes whose digest has been verified.
64    pub fn bytes(&self) -> &[u8] {
65        &self.bytes
66    }
67
68    /// Transfers ownership of the verified bytes.
69    pub fn into_bytes(self) -> Vec<u8> {
70        self.bytes
71    }
72}
73
74impl PersistedArtifact {
75    /// Borrows the immutable identity and relative path.
76    pub const fn descriptor(&self) -> &ArtifactDescriptor {
77        &self.descriptor
78    }
79
80    /// Reports whether this call created or reused the destination.
81    pub const fn disposition(&self) -> ArtifactDisposition {
82        self.disposition
83    }
84
85    /// Transfers ownership of the descriptor.
86    pub fn into_descriptor(self) -> ArtifactDescriptor {
87        self.descriptor
88    }
89}
90
91/// Atomically publishes exact bytes beneath `scope/inputs` using their SHA-256.
92///
93/// `stem` and `extension` describe representation only; callers retain all
94/// domain interpretation. Both must be nonempty safe filename fragments.
95pub fn persist_artifact(
96    scope: &ExecutionScope,
97    stem: &str,
98    extension: &str,
99    bytes: &[u8],
100) -> Result<PersistedArtifact, ArtifactError> {
101    validate_fragment("stem", stem, true)?;
102    validate_fragment("extension", extension, false)?;
103    let digest = sha256_hex(bytes);
104    let file_name = format!("{stem}-{digest}.{extension}");
105    let relative_path = format!("inputs/{file_name}");
106    let inputs = scope.directory().join("inputs");
107    fs::create_dir_all(&inputs).map_err(|source| ArtifactError::Io {
108        operation: "create artifact input directory",
109        path: inputs.clone(),
110        source,
111    })?;
112    let destination = inputs.join(file_name);
113
114    if destination.exists() {
115        verify_existing(&destination, bytes, &digest)?;
116        return Ok(persisted(
117            digest,
118            relative_path,
119            ArtifactDisposition::Reused,
120        ));
121    }
122
123    let temporary = create_complete_temporary(&inputs, &digest, bytes)?;
124    let disposition = match fs::hard_link(&temporary, &destination) {
125        Ok(()) => ArtifactDisposition::Created,
126        Err(source) if source.kind() == std::io::ErrorKind::AlreadyExists => {
127            let verification = verify_existing(&destination, bytes, &digest);
128            remove_published_temporary(&temporary)?;
129            verification?;
130            ArtifactDisposition::Reused
131        }
132        Err(source) => {
133            remove_temporary(&temporary);
134            return Err(ArtifactError::Io {
135                operation: "publish artifact",
136                path: destination,
137                source,
138            });
139        }
140    };
141    if temporary.exists() {
142        remove_published_temporary(&temporary)?;
143    }
144    sync_directory(&inputs)?;
145    Ok(persisted(digest, relative_path, disposition))
146}
147
148/// Reads immutable bytes after path-containment and exact-digest verification.
149pub fn load_verified_artifact(
150    execution_directory: impl AsRef<Path>,
151    descriptor: &ArtifactDescriptor,
152) -> Result<VerifiedArtifact, ArtifactLoadError> {
153    validate_descriptor(descriptor)?;
154    let execution_directory =
155        fs::canonicalize(execution_directory.as_ref()).map_err(|source| ArtifactLoadError::Io {
156            operation: "resolve execution directory",
157            path: execution_directory.as_ref().to_path_buf(),
158            source,
159        })?;
160    let relative = Path::new(descriptor.path());
161    if relative.as_os_str().is_empty()
162        || relative
163            .components()
164            .any(|component| !matches!(component, Component::Normal(_)))
165    {
166        return Err(ArtifactLoadError::InvalidDescriptor {
167            reason: "artifact path must be a nonempty normalized relative path".to_owned(),
168        });
169    }
170    let unresolved = execution_directory.join(relative);
171    let path = fs::canonicalize(&unresolved).map_err(|source| ArtifactLoadError::Io {
172        operation: "resolve artifact",
173        path: unresolved,
174        source,
175    })?;
176    if !path.starts_with(&execution_directory) {
177        return Err(ArtifactLoadError::InvalidDescriptor {
178            reason: "artifact path resolves outside the execution directory".to_owned(),
179        });
180    }
181    let bytes = fs::read(&path).map_err(|source| ArtifactLoadError::Io {
182        operation: "read artifact",
183        path: path.clone(),
184        source,
185    })?;
186    let actual = sha256_hex(&bytes);
187    if actual != descriptor.sha256 {
188        return Err(ArtifactLoadError::DigestMismatch {
189            path,
190            expected: descriptor.sha256.clone(),
191            actual,
192        });
193    }
194    Ok(VerifiedArtifact { path, bytes })
195}
196
197fn persisted(sha256: String, path: String, disposition: ArtifactDisposition) -> PersistedArtifact {
198    PersistedArtifact {
199        descriptor: ArtifactDescriptor { sha256, path },
200        disposition,
201    }
202}
203
204fn validate_fragment(
205    kind: &'static str,
206    value: &str,
207    allow_hyphen: bool,
208) -> Result<(), ArtifactError> {
209    let valid = !value.is_empty()
210        && value.bytes().all(|byte| {
211            byte.is_ascii_alphanumeric() || byte == b'_' || (allow_hyphen && byte == b'-')
212        });
213    if valid {
214        Ok(())
215    } else {
216        Err(ArtifactError::InvalidFragment {
217            kind,
218            value: value.to_owned(),
219        })
220    }
221}
222
223fn validate_descriptor(descriptor: &ArtifactDescriptor) -> Result<(), ArtifactLoadError> {
224    if descriptor.sha256.len() != 64
225        || !descriptor
226            .sha256
227            .bytes()
228            .all(|byte| byte.is_ascii_digit() || (b'a'..=b'f').contains(&byte))
229    {
230        return Err(ArtifactLoadError::InvalidDescriptor {
231            reason: "artifact SHA-256 must contain exactly 64 lowercase hexadecimal digits"
232                .to_owned(),
233        });
234    }
235    Ok(())
236}
237
238fn create_complete_temporary(
239    directory: &Path,
240    digest: &str,
241    bytes: &[u8],
242) -> Result<PathBuf, ArtifactError> {
243    for _ in 0..1024 {
244        let sequence = TEMPORARY_SEQUENCE.fetch_add(1, Ordering::Relaxed);
245        let path = directory.join(format!(
246            ".artifact-{digest}-{}-{sequence}.tmp",
247            std::process::id()
248        ));
249        let mut file = match OpenOptions::new().write(true).create_new(true).open(&path) {
250            Ok(file) => file,
251            Err(source) if source.kind() == std::io::ErrorKind::AlreadyExists => continue,
252            Err(source) => {
253                return Err(ArtifactError::Io {
254                    operation: "create temporary artifact",
255                    path,
256                    source,
257                });
258            }
259        };
260        if let Err(source) = file.write_all(bytes).and_then(|()| file.sync_all()) {
261            drop(file);
262            remove_temporary(&path);
263            return Err(ArtifactError::Io {
264                operation: "write temporary artifact",
265                path,
266                source,
267            });
268        }
269        return Ok(path);
270    }
271    Err(ArtifactError::TemporaryIdentityExhausted {
272        directory: directory.to_path_buf(),
273    })
274}
275
276fn verify_existing(path: &Path, expected: &[u8], digest: &str) -> Result<(), ArtifactError> {
277    let actual = fs::read(path).map_err(|source| ArtifactError::Io {
278        operation: "read existing artifact",
279        path: path.to_path_buf(),
280        source,
281    })?;
282    if actual == expected {
283        Ok(())
284    } else {
285        Err(ArtifactError::DigestCollision {
286            digest: digest.to_owned(),
287            path: path.to_path_buf(),
288        })
289    }
290}
291
292fn sha256_hex(bytes: &[u8]) -> String {
293    let digest = Sha256::digest(bytes);
294    let mut encoded = String::with_capacity(digest.len() * 2);
295    for byte in digest {
296        use std::fmt::Write as _;
297        write!(encoded, "{byte:02x}").expect("writing into a String cannot fail");
298    }
299    encoded
300}
301
302fn sync_directory(path: &Path) -> Result<(), ArtifactError> {
303    File::open(path)
304        .and_then(|directory| directory.sync_all())
305        .map_err(|source| ArtifactError::Io {
306            operation: "synchronize artifact input directory",
307            path: path.to_path_buf(),
308            source,
309        })
310}
311
312fn remove_temporary(path: &Path) {
313    let _ = fs::remove_file(path);
314}
315
316fn remove_published_temporary(path: &Path) -> Result<(), ArtifactError> {
317    fs::remove_file(path).map_err(|source| ArtifactError::Io {
318        operation: "remove temporary artifact",
319        path: path.to_path_buf(),
320        source,
321    })
322}
323
324#[derive(Debug, Error)]
325#[non_exhaustive]
326pub enum ArtifactError {
327    #[error("invalid artifact {kind} `{value}`")]
328    InvalidFragment { kind: &'static str, value: String },
329    #[error("failed to {operation} at `{path}`")]
330    Io {
331        operation: &'static str,
332        path: PathBuf,
333        #[source]
334        source: std::io::Error,
335    },
336    #[error("artifact digest collision for `{digest}` at `{path}`")]
337    DigestCollision { digest: String, path: PathBuf },
338    #[error("could not allocate a temporary artifact beneath `{directory}`")]
339    TemporaryIdentityExhausted { directory: PathBuf },
340}
341
342#[derive(Debug, Error)]
343#[non_exhaustive]
344pub enum ArtifactLoadError {
345    #[error("invalid artifact descriptor: {reason}")]
346    InvalidDescriptor { reason: String },
347    #[error("failed to {operation} at `{path}`")]
348    Io {
349        operation: &'static str,
350        path: PathBuf,
351        #[source]
352        source: std::io::Error,
353    },
354    #[error("artifact `{path}` has SHA-256 `{actual}`, but metadata declares `{expected}`")]
355    DigestMismatch {
356        path: PathBuf,
357        expected: String,
358        actual: String,
359    },
360}
scientific_workflow/artifact.rs

scientific_workflow/
artifact.rs
