bio_tools/
run_log.rs

1//! For storing the input and output of runs [from any tool] to disk. This includes capturing
2//! the stdout of CLI programs in text files, capturing the time the job was run, and with what
3//! input parameters, and storing all input and output files produced by, or ingested into
4//! the tool's API. (E.g. molecule files [protein, small organic molecule etc] generated or ingested)
5
6use std::{
7    ffi::OsStr,
8    fs::{self, File, OpenOptions},
9    io::{self, BufReader, BufWriter, Read, Write},
10    path::{Path, PathBuf},
11    sync::atomic::{AtomicU64, Ordering},
12    time::{SystemTime, UNIX_EPOCH},
13};
14
15use crate::run::{CommandOutput, CommandSpec};
16
17static RUN_SEQUENCE: AtomicU64 = AtomicU64::new(0);
18
19/// Durable audit settings for one command execution.
20///
21/// Each execution gets a unique directory below `root/label`. The directory
22/// records the invocation, complete output streams, optional stdin, and
23/// before/after copies of the configured artifact paths.
24#[derive(Clone, Debug, Eq, PartialEq)]
25pub struct RunLogSpec {
26    pub root: PathBuf,
27    pub label: String,
28    pub artifacts: Vec<PathBuf>,
29}
30
31impl RunLogSpec {
32    pub fn new(root: impl Into<PathBuf>, label: impl Into<String>) -> Self {
33        Self {
34            root: root.into(),
35            label: label.into(),
36            artifacts: Vec::new(),
37        }
38    }
39
40    pub fn artifact(mut self, path: impl Into<PathBuf>) -> Self {
41        self.artifacts.push(path.into());
42        self
43    }
44
45    pub fn artifacts<I, P>(mut self, paths: I) -> Self
46    where
47        I: IntoIterator<Item = P>,
48        P: Into<PathBuf>,
49    {
50        self.artifacts.extend(paths.into_iter().map(Into::into));
51        self
52    }
53}
54
55struct ArtifactCopy {
56    source: PathBuf,
57    input: PathBuf,
58    output: PathBuf,
59}
60
61pub(crate) struct ActiveRunLog {
62    directory: PathBuf,
63    artifacts: Vec<ArtifactCopy>,
64}
65
66impl ActiveRunLog {
67    pub(crate) fn start(spec: &CommandSpec, settings: &RunLogSpec) -> io::Result<Self> {
68        let directory = unique_directory(&settings.root, &settings.label)?;
69        fs::create_dir_all(directory.join("inputs"))?;
70        fs::create_dir_all(directory.join("outputs"))?;
71
72        let mut invocation = BufWriter::new(File::create(directory.join("invocation.txt"))?);
73        writeln!(invocation, "command: {}", spec.display_command())?;
74        writeln!(invocation, "argv:")?;
75        writeln!(invocation, "  [0] {:?}", spec.program)?;
76        for (index, argument) in spec.arguments.iter().enumerate() {
77            writeln!(invocation, "  [{}] {:?}", index + 1, argument)?;
78        }
79        writeln!(
80            invocation,
81            "working directory: {}",
82            spec.current_dir
83                .as_deref()
84                .map(Path::display)
85                .map(|value| value.to_string())
86                .unwrap_or_else(|| "(inherited)".to_owned())
87        )?;
88        writeln!(
89            invocation,
90            "timeout seconds: {}",
91            spec.timeout
92                .map(|value| value.as_secs_f64().to_string())
93                .unwrap_or_else(|| "none".to_owned())
94        )?;
95        writeln!(
96            invocation,
97            "environment policy: {:?}",
98            spec.environment_policy
99        )?;
100        if spec.environment.is_empty() {
101            writeln!(invocation, "environment overrides: none")?;
102        } else {
103            writeln!(invocation, "environment overrides:")?;
104            for (name, value) in &spec.environment {
105                writeln!(invocation, "  {:?}={:?}", name, value)?;
106            }
107        }
108        invocation.flush()?;
109
110        if let Some(stdin) = &spec.stdin {
111            fs::write(directory.join("stdin.txt"), stdin)?;
112        }
113
114        let mut artifacts = Vec::new();
115        let mut manifest = BufWriter::new(File::create(directory.join("artifacts.txt"))?);
116        for (index, source) in settings.artifacts.iter().enumerate() {
117            let source = absolute_artifact(source, spec.current_dir.as_deref())?;
118            let name = source
119                .file_name()
120                .and_then(OsStr::to_str)
121                .filter(|name| !name.is_empty())
122                .unwrap_or("artifact");
123            let entry = format!("{index:02}-{}", safe_component(name));
124            let input = directory.join("inputs").join(&entry);
125            let output = directory.join("outputs").join(&entry);
126            writeln!(manifest, "{entry}: {}", source.display())?;
127            if source.exists() || fs::symlink_metadata(&source).is_ok() {
128                copy_path(&source, &input)?;
129            } else {
130                writeln!(manifest, "  absent before launch")?;
131            }
132            artifacts.push(ArtifactCopy {
133                source,
134                input,
135                output,
136            });
137        }
138        manifest.flush()?;
139
140        Ok(Self {
141            directory,
142            artifacts,
143        })
144    }
145
146    pub(crate) fn directory(&self) -> &Path {
147        &self.directory
148    }
149
150    pub(crate) fn stdout_file(&self) -> io::Result<File> {
151        File::create(self.directory.join("stdout.txt"))
152    }
153
154    pub(crate) fn stderr_file(&self) -> io::Result<File> {
155        File::create(self.directory.join("stderr.txt"))
156    }
157
158    pub(crate) fn record_start_error(&self, error: &io::Error) -> io::Result<()> {
159        fs::write(
160            self.directory.join("error.txt"),
161            format!("The command could not be started: {error}\n"),
162        )?;
163        self.write_combined_log(None, Some(&error.to_string()))
164    }
165
166    pub(crate) fn finish(&self, output: &CommandOutput, timed_out: bool) -> io::Result<()> {
167        let mut deleted = Vec::new();
168        for artifact in &self.artifacts {
169            copy_changes(
170                &artifact.source,
171                &artifact.input,
172                &artifact.output,
173                Path::new(""),
174                &mut deleted,
175            )?;
176        }
177        if !deleted.is_empty() {
178            fs::write(
179                self.directory.join("deleted_files.txt"),
180                deleted.join("\n") + "\n",
181            )?;
182        }
183
184        let status = if timed_out {
185            "timed out".to_owned()
186        } else {
187            output
188                .return_code()
189                .map(|code| format!("exit code {code}"))
190                .unwrap_or_else(|| "terminated without an exit code".to_owned())
191        };
192        fs::write(
193            self.directory.join("result.txt"),
194            format!(
195                "status: {status}\nelapsed seconds: {}\n",
196                output.elapsed.as_secs_f64()
197            ),
198        )?;
199        self.write_combined_log(Some(output), None)
200    }
201
202    fn write_combined_log(
203        &self,
204        output: Option<&CommandOutput>,
205        start_error: Option<&str>,
206    ) -> io::Result<()> {
207        let mut combined = BufWriter::new(File::create(self.directory.join("run.log"))?);
208        copy_section(
209            &mut combined,
210            "INVOCATION",
211            &self.directory.join("invocation.txt"),
212        )?;
213        if self.directory.join("stdin.txt").is_file() {
214            copy_section(&mut combined, "STDIN", &self.directory.join("stdin.txt"))?;
215        } else {
216            writeln!(
217                combined,
218                "\n===== STDIN =====\n(not supplied; the child received EOF)"
219            )?;
220        }
221        if let Some(output) = output {
222            writeln!(
223                combined,
224                "\n===== RESULT =====\nreturn code: {}\nelapsed seconds: {}",
225                output
226                    .return_code()
227                    .map(|code| code.to_string())
228                    .unwrap_or_else(|| "none".to_owned()),
229                output.elapsed.as_secs_f64()
230            )?;
231        } else if let Some(error) = start_error {
232            writeln!(combined, "\n===== RESULT =====\nstart error: {error}")?;
233        }
234        copy_section(&mut combined, "STDOUT", &self.directory.join("stdout.txt"))?;
235        copy_section(&mut combined, "STDERR", &self.directory.join("stderr.txt"))?;
236        combined.flush()
237    }
238}
239
240fn unique_directory(root: &Path, label: &str) -> io::Result<PathBuf> {
241    let tool_root = root.join(safe_component(label));
242    fs::create_dir_all(&tool_root)?;
243    let timestamp = SystemTime::now()
244        .duration_since(UNIX_EPOCH)
245        .unwrap_or_default()
246        .as_millis();
247    for _ in 0..100 {
248        let sequence = RUN_SEQUENCE.fetch_add(1, Ordering::Relaxed);
249        let path = tool_root.join(format!("{timestamp}-{}-{sequence}", std::process::id()));
250        match fs::create_dir(&path) {
251            Ok(()) => return Ok(path),
252            Err(error) if error.kind() == io::ErrorKind::AlreadyExists => continue,
253            Err(error) => return Err(error),
254        }
255    }
256    Err(io::Error::new(
257        io::ErrorKind::AlreadyExists,
258        "could not allocate a unique run-log directory",
259    ))
260}
261
262fn safe_component(value: &str) -> String {
263    let cleaned: String = value
264        .chars()
265        .map(|character| {
266            if character.is_ascii_alphanumeric() || matches!(character, '-' | '_' | '.') {
267                character
268            } else {
269                '_'
270            }
271        })
272        .collect();
273    if cleaned.is_empty() || cleaned == "." || cleaned == ".." {
274        "command".to_owned()
275    } else {
276        cleaned
277    }
278}
279
280fn absolute_artifact(path: &Path, current_dir: Option<&Path>) -> io::Result<PathBuf> {
281    if path.is_absolute() {
282        return Ok(path.to_owned());
283    }
284    if let Some(current_dir) = current_dir {
285        return Ok(current_dir.join(path));
286    }
287    Ok(std::env::current_dir()?.join(path))
288}
289
290fn copy_path(source: &Path, destination: &Path) -> io::Result<()> {
291    let metadata = fs::symlink_metadata(source)?;
292    if metadata.file_type().is_symlink() {
293        return copy_symlink(source, destination);
294    }
295    if metadata.is_dir() {
296        fs::create_dir_all(destination)?;
297        for entry in fs::read_dir(source)? {
298            let entry = entry?;
299            copy_path(&entry.path(), &destination.join(entry.file_name()))?;
300        }
301        return Ok(());
302    }
303    if metadata.is_file() {
304        if let Some(parent) = destination.parent() {
305            fs::create_dir_all(parent)?;
306        }
307        fs::copy(source, destination)?;
308    }
309    Ok(())
310}
311
312fn copy_changes(
313    source: &Path,
314    input: &Path,
315    output: &Path,
316    relative: &Path,
317    deleted: &mut Vec<String>,
318) -> io::Result<bool> {
319    let source_metadata = fs::symlink_metadata(source).ok();
320    let input_metadata = fs::symlink_metadata(input).ok();
321
322    let Some(source_metadata) = source_metadata else {
323        if input_metadata.is_some() {
324            deleted.push(relative.to_string_lossy().into_owned());
325        }
326        return Ok(false);
327    };
328
329    if source_metadata.is_dir() {
330        let mut changed = false;
331        for entry in fs::read_dir(source)? {
332            let entry = entry?;
333            let name = entry.file_name();
334            changed |= copy_changes(
335                &entry.path(),
336                &input.join(&name),
337                &output.join(&name),
338                &relative.join(&name),
339                deleted,
340            )?;
341        }
342        if input_metadata.is_some_and(|metadata| metadata.is_dir()) {
343            for entry in fs::read_dir(input)? {
344                let entry = entry?;
345                if fs::symlink_metadata(source.join(entry.file_name())).is_err() {
346                    deleted.push(
347                        relative
348                            .join(entry.file_name())
349                            .to_string_lossy()
350                            .into_owned(),
351                    );
352                }
353            }
354        }
355        return Ok(changed);
356    }
357
358    let unchanged = match input_metadata {
359        Some(metadata)
360            if metadata.file_type().is_symlink() && source_metadata.file_type().is_symlink() =>
361        {
362            fs::read_link(source)? == fs::read_link(input)?
363        }
364        Some(metadata) if metadata.is_file() && source_metadata.is_file() => {
365            files_equal(source, input)?
366        }
367        _ => false,
368    };
369    if unchanged {
370        return Ok(false);
371    }
372    copy_path(source, output)?;
373    Ok(true)
374}
375
376fn files_equal(left: &Path, right: &Path) -> io::Result<bool> {
377    if fs::metadata(left)?.len() != fs::metadata(right)?.len() {
378        return Ok(false);
379    }
380    let mut left = BufReader::new(File::open(left)?);
381    let mut right = BufReader::new(File::open(right)?);
382    let mut left_buffer = [0_u8; 64 * 1024];
383    let mut right_buffer = [0_u8; 64 * 1024];
384    loop {
385        let left_count = left.read(&mut left_buffer)?;
386        let right_count = right.read(&mut right_buffer)?;
387        if left_count != right_count || left_buffer[..left_count] != right_buffer[..right_count] {
388            return Ok(false);
389        }
390        if left_count == 0 {
391            return Ok(true);
392        }
393    }
394}
395
396fn copy_section(writer: &mut impl Write, name: &str, path: &Path) -> io::Result<()> {
397    writeln!(writer, "\n===== {name} =====")?;
398    if path.is_file() {
399        io::copy(&mut File::open(path)?, writer)?;
400    } else {
401        writeln!(writer, "(not available)")?;
402    }
403    Ok(())
404}
405
406#[cfg(unix)]
407fn copy_symlink(source: &Path, destination: &Path) -> io::Result<()> {
408    use std::os::unix::fs::symlink;
409    if let Some(parent) = destination.parent() {
410        fs::create_dir_all(parent)?;
411    }
412    symlink(fs::read_link(source)?, destination)
413}
414
415#[cfg(windows)]
416fn copy_symlink(source: &Path, destination: &Path) -> io::Result<()> {
417    use std::os::windows::fs::{symlink_dir, symlink_file};
418    if let Some(parent) = destination.parent() {
419        fs::create_dir_all(parent)?;
420    }
421    let target = fs::read_link(source)?;
422    if source.is_dir() {
423        symlink_dir(target, destination)
424    } else {
425        symlink_file(target, destination)
426    }
427}
428
429pub(crate) fn append_log_error(directory: &Path, action: &str, error: &io::Error) {
430    let _ = OpenOptions::new()
431        .create(true)
432        .append(true)
433        .open(directory.join("logging_errors.txt"))
434        .and_then(|mut file| writeln!(file, "{action}: {error}"));
435}
bio_tools/run_log.rs

bio_tools/
run_log.rs
