clusterflux-public/crates/clusterflux-node/src/task_artifacts.rs

use std::collections::BTreeSet;
use std::fs::{self, OpenOptions};
use std::io::{Read, Seek, SeekFrom, Write};
use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicU64, Ordering};
use std::time::{SystemTime, UNIX_EPOCH};

use clusterflux_core::{ArtifactId, Digest, NodeId, TaskInstanceId, VfsManifest, VfsOverlay};
use sha2::{Digest as _, Sha256};

const MAX_NODE_ARTIFACT_BYTES: u64 = 256 * 1024 * 1024;
const MAX_NODE_ARTIFACT_CHUNK_BYTES: u64 = 256 * 1024;
const DEFAULT_MAX_NODE_ARTIFACT_STORE_BYTES: u64 = 4 * 1024 * 1024 * 1024;
const DEFAULT_MAX_NODE_ARTIFACT_COUNT: usize = 1_024;
const DEFAULT_MAX_NODE_ARTIFACT_AGE_SECONDS: u64 = 7 * 24 * 60 * 60;
const DEFAULT_NODE_ARTIFACT_RESTART_PIN_SECONDS: u64 = 60 * 60;

#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub(crate) struct NodeArtifactRetentionLimits {
    pub(crate) max_bytes: u64,
    pub(crate) max_count: usize,
    pub(crate) max_age_seconds: u64,
    pub(crate) restart_pin_seconds: u64,
}

impl Default for NodeArtifactRetentionLimits {
    fn default() -> Self {
        Self {
            max_bytes: DEFAULT_MAX_NODE_ARTIFACT_STORE_BYTES,
            max_count: DEFAULT_MAX_NODE_ARTIFACT_COUNT,
            max_age_seconds: DEFAULT_MAX_NODE_ARTIFACT_AGE_SECONDS,
            restart_pin_seconds: DEFAULT_NODE_ARTIFACT_RESTART_PIN_SECONDS,
        }
    }
}

impl NodeArtifactRetentionLimits {
    pub(crate) fn from_environment() -> Result<Self, String> {
        Ok(Self {
            max_bytes: environment_u64(
                "CLUSTERFLUX_NODE_ARTIFACT_MAX_BYTES",
                DEFAULT_MAX_NODE_ARTIFACT_STORE_BYTES,
            )?,
            max_count: usize::try_from(environment_u64(
                "CLUSTERFLUX_NODE_ARTIFACT_MAX_COUNT",
                DEFAULT_MAX_NODE_ARTIFACT_COUNT as u64,
            )?)
            .map_err(|_| "CLUSTERFLUX_NODE_ARTIFACT_MAX_COUNT does not fit usize".to_owned())?,
            max_age_seconds: environment_u64(
                "CLUSTERFLUX_NODE_ARTIFACT_MAX_AGE_SECONDS",
                DEFAULT_MAX_NODE_ARTIFACT_AGE_SECONDS,
            )?,
            restart_pin_seconds: environment_u64(
                "CLUSTERFLUX_NODE_ARTIFACT_RESTART_PIN_SECONDS",
                DEFAULT_NODE_ARTIFACT_RESTART_PIN_SECONDS,
            )?,
        })
    }
}

#[derive(Clone, Debug, Default, PartialEq, Eq)]
pub(crate) struct NodeArtifactGcReport {
    pub(crate) retained_count: usize,
    pub(crate) retained_bytes: u64,
    pub(crate) evicted: Vec<ArtifactId>,
    pub(crate) limits_satisfied: bool,
}

#[derive(Clone, Debug, PartialEq, Eq)]
pub(crate) struct RetainedArtifact {
    pub(crate) id: ArtifactId,
    pub(crate) digest: Digest,
    pub(crate) size_bytes: u64,
    pub(crate) path: PathBuf,
}

#[derive(Clone, Debug)]
pub(crate) struct NodeArtifactStore {
    root: PathBuf,
}

impl NodeArtifactStore {
    pub(crate) fn for_runtime(project_root: Option<&Path>, node: &str) -> Result<Self, String> {
        let base = project_root
            .map(Path::to_path_buf)
            .unwrap_or_else(|| PathBuf::from("."));
        let node = node
            .chars()
            .map(|character| {
                if character.is_ascii_alphanumeric() || "._-".contains(character) {
                    character
                } else {
                    '_'
                }
            })
            .collect::<String>();
        let root = base.join(".clusterflux").join("node-artifacts").join(node);
        if fs::symlink_metadata(&root)
            .map(|metadata| metadata.file_type().is_symlink())
            .unwrap_or(false)
        {
            return Err("node artifact store root must not be a symbolic link".to_owned());
        }
        fs::create_dir_all(&root).map_err(|error| error.to_string())?;
        #[cfg(unix)]
        {
            use std::os::unix::fs::PermissionsExt;
            fs::set_permissions(&root, fs::Permissions::from_mode(0o700))
                .map_err(|error| error.to_string())?;
        }
        let root = root.canonicalize().map_err(|error| error.to_string())?;
        Ok(Self { root })
    }

    pub(crate) fn retain_output_file(
        &self,
        output_root: &Path,
        relative_path: &str,
    ) -> Result<RetainedArtifact, String> {
        let source = confined_output_file(output_root, relative_path)?;
        let name = relative_path.replace('/', "_");
        validate_artifact_component(&name, "artifact name", 240)?;
        let temporary = self.root.join(format!(
            ".flush.{}.{}.tmp",
            std::process::id(),
            NEXT_FILE_ID.fetch_add(1, Ordering::Relaxed)
        ));
        let retained = (|| {
            let mut input = open_regular_readonly(&source, "task output")?;
            let metadata = input.metadata().map_err(|error| error.to_string())?;
            if metadata.len() > MAX_NODE_ARTIFACT_BYTES {
                return Err(format!(
                    "task output is {} bytes; node artifact limit is {MAX_NODE_ARTIFACT_BYTES} bytes",
                    metadata.len()
                ));
            }
            let mut output = create_new_private_file(&temporary)?;
            let (digest, size) = copy_and_hash(&mut input, &mut output, MAX_NODE_ARTIFACT_BYTES)?;
            output.sync_all().map_err(|error| error.to_string())?;
            let digest_hex = digest
                .as_str()
                .strip_prefix("sha256:")
                .ok_or_else(|| "task output digest is not SHA-256".to_owned())?;
            let id = ArtifactId::new(format!("{name}-{digest_hex}"));
            let path = self.root.join(id.as_str());
            match fs::hard_link(&temporary, &path) {
                Ok(()) => {
                    fs::remove_file(&temporary).map_err(|error| error.to_string())?;
                    Ok(RetainedArtifact {
                        id,
                        digest,
                        size_bytes: size,
                        path,
                    })
                }
                Err(error) if error.kind() == std::io::ErrorKind::AlreadyExists => {
                    let existing = self
                        .metadata(&id)?
                        .ok_or_else(|| "retained artifact disappeared during flush".to_owned())?;
                    if existing.digest != digest || existing.size_bytes != size {
                        return Err("retained artifact path has conflicting bytes".to_owned());
                    }
                    Ok(existing)
                }
                Err(error) => Err(error.to_string()),
            }
        })();
        let _ = fs::remove_file(&temporary);
        retained
    }

    pub(crate) fn materialize_into_output(
        &self,
        artifact: &clusterflux_core::ArtifactHandle,
        output_root: &Path,
        relative_path: &str,
    ) -> Result<PathBuf, String> {
        let retained = self
            .metadata(&artifact.id)?
            .ok_or_else(|| format!("retained artifact `{}` is unavailable", artifact.id))?;
        if retained.digest != artifact.digest || retained.size_bytes != artifact.size_bytes {
            return Err(
                "artifact handle digest/size does not match retained node metadata".to_owned(),
            );
        }
        let target = confined_output_target(output_root, relative_path)?;
        let copied = (|| {
            let mut output = create_new_private_file(&target)?;
            let mut input = open_regular_readonly(&retained.path, "retained artifact")?;
            let (digest, size_bytes) =
                copy_and_hash(&mut input, &mut output, MAX_NODE_ARTIFACT_BYTES)?;
            output.sync_all().map_err(|error| error.to_string())?;
            Ok::<_, String>((digest, size_bytes))
        })();
        let (digest, size_bytes) = match copied {
            Ok(copied) => copied,
            Err(error) => {
                let _ = fs::remove_file(&target);
                return Err(error);
            }
        };
        if size_bytes != artifact.size_bytes || digest != artifact.digest {
            let _ = fs::remove_file(&target);
            return Err("materialized artifact digest/size changed during local copy".to_owned());
        }
        Ok(target)
    }

    pub(crate) fn metadata(&self, id: &ArtifactId) -> Result<Option<RetainedArtifact>, String> {
        validate_artifact_component(id.as_str(), "artifact id", 256)?;
        let path = self.root.join(id.as_str());
        let metadata = match fs::symlink_metadata(&path) {
            Ok(metadata) => metadata,
            Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(None),
            Err(error) => return Err(error.to_string()),
        };
        if metadata.file_type().is_symlink() || !metadata.is_file() {
            return Err("retained artifact path is not a regular non-symlink file".to_owned());
        }
        if metadata.len() > MAX_NODE_ARTIFACT_BYTES {
            return Err(format!(
                "retained artifact exceeds node artifact limit of {MAX_NODE_ARTIFACT_BYTES} bytes"
            ));
        }
        let mut file = open_regular_readonly(&path, "retained artifact")?;
        let (digest, size_bytes) = hash_reader(&mut file, MAX_NODE_ARTIFACT_BYTES)?;
        Ok(Some(RetainedArtifact {
            id: id.clone(),
            digest,
            size_bytes,
            path,
        }))
    }

    #[cfg(test)]
    pub(crate) fn read_verified(
        &self,
        id: &ArtifactId,
        expected_digest: &Digest,
        expected_size_bytes: u64,
    ) -> Result<Vec<u8>, String> {
        let retained = self
            .metadata(id)?
            .ok_or_else(|| format!("retained artifact `{id}` no longer exists on this node"))?;
        if &retained.digest != expected_digest || retained.size_bytes != expected_size_bytes {
            return Err(format!(
                "retained artifact `{id}` does not match coordinator digest/size metadata"
            ));
        }
        fs::read(retained.path).map_err(|error| error.to_string())
    }

    pub(crate) fn read_verified_chunk(
        &self,
        id: &ArtifactId,
        expected_digest: &Digest,
        expected_size_bytes: u64,
        offset: u64,
        max_chunk_bytes: u64,
    ) -> Result<Vec<u8>, String> {
        validate_artifact_component(id.as_str(), "artifact id", 256)?;
        let digest_hex = expected_digest
            .as_str()
            .strip_prefix("sha256:")
            .ok_or_else(|| "artifact transfer expected digest is not SHA-256".to_owned())?;
        if !id.as_str().ends_with(digest_hex) {
            return Err(format!(
                "retained artifact `{id}` is not the content-addressed object requested by the coordinator"
            ));
        }
        let path = self.root.join(id.as_str());
        let mut file = open_regular_readonly(&path, "retained artifact")
            .map_err(|_| format!("retained artifact `{id}` no longer exists on this node"))?;
        let actual_size = file.metadata().map_err(|error| error.to_string())?.len();
        if actual_size != expected_size_bytes {
            return Err(format!(
                "retained artifact `{id}` does not match coordinator size metadata"
            ));
        }
        if offset > actual_size {
            return Err(format!(
                "artifact transfer offset {offset} exceeds retained artifact size {actual_size}"
            ));
        }
        file.seek(SeekFrom::Start(offset))
            .map_err(|error| error.to_string())?;
        let length = max_chunk_bytes
            .min(MAX_NODE_ARTIFACT_CHUNK_BYTES)
            .min(actual_size.saturating_sub(offset));
        let mut bytes = vec![0; usize::try_from(length).map_err(|error| error.to_string())?];
        file.read_exact(&mut bytes)
            .map_err(|error| error.to_string())?;
        Ok(bytes)
    }

    pub(crate) fn garbage_collect(
        &self,
        limits: NodeArtifactRetentionLimits,
        pinned: &BTreeSet<ArtifactId>,
        now_epoch_seconds: u64,
    ) -> Result<NodeArtifactGcReport, String> {
        let mut entries = Vec::new();
        for entry in fs::read_dir(&self.root).map_err(|error| error.to_string())? {
            let entry = entry.map_err(|error| error.to_string())?;
            let name = entry
                .file_name()
                .into_string()
                .map_err(|_| "retained artifact filename is not UTF-8".to_owned())?;
            if name.starts_with('.') {
                continue;
            }
            validate_artifact_component(&name, "artifact id", 256)?;
            let metadata = entry.metadata().map_err(|error| error.to_string())?;
            if !entry
                .file_type()
                .map_err(|error| error.to_string())?
                .is_file()
                || !metadata.is_file()
            {
                continue;
            }
            entries.push(StoredArtifactEntry {
                id: ArtifactId::new(name),
                path: entry.path(),
                size_bytes: metadata.len(),
                modified_epoch_seconds: metadata
                    .modified()
                    .ok()
                    .and_then(system_time_epoch_seconds)
                    .unwrap_or_default(),
            });
        }
        entries.sort_by(|left, right| {
            left.modified_epoch_seconds
                .cmp(&right.modified_epoch_seconds)
                .then_with(|| left.id.cmp(&right.id))
        });

        let age_cutoff = now_epoch_seconds.saturating_sub(limits.max_age_seconds);
        let mut retained_count = entries.len();
        let mut retained_bytes = entries
            .iter()
            .try_fold(0_u64, |total, entry| total.checked_add(entry.size_bytes))
            .ok_or_else(|| "node artifact byte accounting overflowed".to_owned())?;
        let mut evicted = Vec::new();
        for entry in &entries {
            let expired = entry.modified_epoch_seconds <= age_cutoff;
            let over_count = retained_count > limits.max_count;
            let over_bytes = retained_bytes > limits.max_bytes;
            if !(expired || over_count || over_bytes) || pinned.contains(&entry.id) {
                continue;
            }
            match fs::remove_file(&entry.path) {
                Ok(()) => {
                    retained_count = retained_count.saturating_sub(1);
                    retained_bytes = retained_bytes.saturating_sub(entry.size_bytes);
                    evicted.push(entry.id.clone());
                }
                Err(error) if error.kind() == std::io::ErrorKind::NotFound => {
                    retained_count = retained_count.saturating_sub(1);
                    retained_bytes = retained_bytes.saturating_sub(entry.size_bytes);
                    evicted.push(entry.id.clone());
                }
                Err(error) => return Err(error.to_string()),
            }
        }
        let retained_ids = self.artifact_ids()?.into_iter().collect::<BTreeSet<_>>();
        let retained_has_expired = entries.iter().any(|entry| {
            retained_ids.contains(&entry.id) && entry.modified_epoch_seconds <= age_cutoff
        });
        Ok(NodeArtifactGcReport {
            retained_count,
            retained_bytes,
            evicted,
            limits_satisfied: retained_count <= limits.max_count
                && retained_bytes <= limits.max_bytes
                && !retained_has_expired,
        })
    }

    pub(crate) fn artifact_ids(&self) -> Result<Vec<ArtifactId>, String> {
        let mut ids = Vec::new();
        for entry in fs::read_dir(&self.root).map_err(|error| error.to_string())? {
            let entry = entry.map_err(|error| error.to_string())?;
            if !entry
                .file_type()
                .map_err(|error| error.to_string())?
                .is_file()
            {
                continue;
            }
            let name = entry
                .file_name()
                .into_string()
                .map_err(|_| "retained artifact filename is not UTF-8".to_owned())?;
            if name.starts_with('.') {
                continue;
            }
            validate_artifact_component(&name, "artifact id", 256)?;
            ids.push(ArtifactId::new(name));
        }
        ids.sort();
        Ok(ids)
    }
}

#[derive(Clone, Debug)]
struct StoredArtifactEntry {
    id: ArtifactId,
    path: PathBuf,
    size_bytes: u64,
    modified_epoch_seconds: u64,
}

pub(crate) fn retained_result_artifact(
    project_root: Option<&Path>,
    node: &str,
    result: Option<&clusterflux_core::TaskBoundaryValue>,
) -> Result<Option<RetainedArtifact>, String> {
    let Some(clusterflux_core::TaskBoundaryValue::Artifact(handle)) = result else {
        return Ok(None);
    };
    NodeArtifactStore::for_runtime(project_root, node)?
        .metadata(&handle.id)?
        .ok_or_else(|| {
            format!(
                "task returned artifact `{}` without flushing a file on this node",
                handle.id
            )
        })
        .and_then(|retained| {
            if retained.digest != handle.digest || retained.size_bytes != handle.size_bytes {
                return Err("task artifact handle does not match retained file metadata".to_owned());
            }
            Ok(Some(retained))
        })
}

static NEXT_FILE_ID: AtomicU64 = AtomicU64::new(1);
static NEXT_OUTPUT_ROOT_ID: AtomicU64 = AtomicU64::new(1);

pub(crate) fn task_output_root(
    project_root: Option<&Path>,
    node: &str,
    task: &TaskInstanceId,
) -> Result<PathBuf, String> {
    let base = project_root
        .map(Path::to_path_buf)
        .unwrap_or_else(|| PathBuf::from("."));
    let safe = |value: &str| {
        value
            .chars()
            .map(|character| {
                if character.is_ascii_alphanumeric() || "._-".contains(character) {
                    character
                } else {
                    '_'
                }
            })
            .collect::<String>()
    };
    let parent = base
        .join(".clusterflux")
        .join("task-outputs")
        .join(safe(node));
    fs::create_dir_all(&parent).map_err(|error| error.to_string())?;
    let now_nanos = SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .map(|duration| duration.as_nanos())
        .unwrap_or_default();
    let root = parent.join(format!(
        "{}.{}.{}.{}",
        safe(task.as_str()),
        std::process::id(),
        now_nanos,
        NEXT_OUTPUT_ROOT_ID.fetch_add(1, Ordering::Relaxed)
    ));
    fs::create_dir(&root).map_err(|error| error.to_string())?;
    #[cfg(unix)]
    {
        use std::os::unix::fs::PermissionsExt;
        fs::set_permissions(&root, fs::Permissions::from_mode(0o700))
            .map_err(|error| error.to_string())?;
    }
    root.canonicalize().map_err(|error| error.to_string())
}

pub(crate) fn clean_stale_task_output_roots(
    project_root: Option<&Path>,
    node: &str,
) -> Result<usize, String> {
    let base = project_root
        .map(Path::to_path_buf)
        .unwrap_or_else(|| PathBuf::from("."));
    let node = node
        .chars()
        .map(|character| {
            if character.is_ascii_alphanumeric() || "._-".contains(character) {
                character
            } else {
                '_'
            }
        })
        .collect::<String>();
    let parent = base.join(".clusterflux").join("task-outputs").join(node);
    let parent_metadata = match fs::symlink_metadata(&parent) {
        Ok(metadata) => metadata,
        Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(0),
        Err(error) => return Err(error.to_string()),
    };
    if parent_metadata.file_type().is_symlink() || !parent_metadata.is_dir() {
        return Err("task-output store root must be a non-symlink directory".to_owned());
    }
    let mut removed = 0;
    for entry in fs::read_dir(&parent).map_err(|error| error.to_string())? {
        let entry = entry.map_err(|error| error.to_string())?;
        let file_type = entry.file_type().map_err(|error| error.to_string())?;
        if file_type.is_dir() {
            fs::remove_dir_all(entry.path()).map_err(|error| error.to_string())?;
        } else {
            fs::remove_file(entry.path()).map_err(|error| error.to_string())?;
        }
        removed += 1;
    }
    Ok(removed)
}

fn confined_output_file(output_root: &Path, relative_path: &str) -> Result<PathBuf, String> {
    validate_relative_path(relative_path)?;
    let root = output_root
        .canonicalize()
        .map_err(|error| error.to_string())?;
    reject_symlink_components(&root, relative_path, true)?;
    let path = root.join(relative_path);
    let metadata = fs::symlink_metadata(&path).map_err(|error| error.to_string())?;
    if metadata.file_type().is_symlink() {
        return Err("task output publication rejects symbolic links".to_owned());
    }
    let canonical = path.canonicalize().map_err(|error| error.to_string())?;
    if !canonical.starts_with(&root) {
        return Err("task output path escapes its bounded output root".to_owned());
    }
    Ok(canonical)
}

fn confined_output_target(output_root: &Path, relative_path: &str) -> Result<PathBuf, String> {
    validate_relative_path(relative_path)?;
    let root = output_root
        .canonicalize()
        .map_err(|error| error.to_string())?;
    let path = root.join(relative_path);
    let parent = path
        .parent()
        .ok_or_else(|| "task output target has no parent".to_owned())?;
    create_confined_directories(&root, relative_path)?;
    reject_symlink_components(&root, relative_path, false)?;
    let parent = parent.canonicalize().map_err(|error| error.to_string())?;
    if !parent.starts_with(&root) {
        return Err("task output target escapes its bounded output root".to_owned());
    }
    Ok(path)
}

fn create_confined_directories(root: &Path, relative_path: &str) -> Result<(), String> {
    let components = relative_path.split('/').collect::<Vec<_>>();
    let mut current = root.to_path_buf();
    for component in components.iter().take(components.len().saturating_sub(1)) {
        current.push(component);
        match fs::symlink_metadata(&current) {
            Ok(metadata) if metadata.file_type().is_symlink() => {
                return Err("task output target rejects symbolic-link directories".to_owned())
            }
            Ok(metadata) if !metadata.is_dir() => {
                return Err("task output target parent is not a directory".to_owned())
            }
            Ok(_) => {}
            Err(error) if error.kind() == std::io::ErrorKind::NotFound => {
                fs::create_dir(&current).map_err(|error| error.to_string())?;
                #[cfg(unix)]
                {
                    use std::os::unix::fs::PermissionsExt;
                    fs::set_permissions(&current, fs::Permissions::from_mode(0o700))
                        .map_err(|error| error.to_string())?;
                }
            }
            Err(error) => return Err(error.to_string()),
        }
    }
    Ok(())
}

fn reject_symlink_components(
    root: &Path,
    relative_path: &str,
    include_leaf: bool,
) -> Result<(), String> {
    let components = relative_path.split('/').collect::<Vec<_>>();
    let count = if include_leaf {
        components.len()
    } else {
        components.len().saturating_sub(1)
    };
    let mut current = root.to_path_buf();
    for component in components.iter().take(count) {
        current.push(component);
        let metadata = fs::symlink_metadata(&current).map_err(|error| error.to_string())?;
        if metadata.file_type().is_symlink() {
            return Err("task output path rejects symbolic-link components".to_owned());
        }
    }
    Ok(())
}

fn open_regular_readonly(path: &Path, label: &str) -> Result<fs::File, String> {
    let metadata = fs::symlink_metadata(path).map_err(|error| error.to_string())?;
    if metadata.file_type().is_symlink() || !metadata.is_file() {
        return Err(format!("{label} must be a regular non-symlink file"));
    }
    let mut options = OpenOptions::new();
    options.read(true);
    #[cfg(unix)]
    {
        use std::os::unix::fs::OpenOptionsExt;
        options.custom_flags(libc::O_NOFOLLOW | libc::O_CLOEXEC);
    }
    let file = options.open(path).map_err(|error| error.to_string())?;
    if !file
        .metadata()
        .map_err(|error| error.to_string())?
        .is_file()
    {
        return Err(format!("{label} must remain a regular file while open"));
    }
    Ok(file)
}

fn create_new_private_file(path: &Path) -> Result<fs::File, String> {
    let mut options = OpenOptions::new();
    options.write(true).create_new(true);
    #[cfg(unix)]
    {
        use std::os::unix::fs::OpenOptionsExt;
        options
            .mode(0o600)
            .custom_flags(libc::O_NOFOLLOW | libc::O_CLOEXEC);
    }
    options.open(path).map_err(|error| error.to_string())
}

fn hash_reader(reader: &mut impl Read, limit: u64) -> Result<(Digest, u64), String> {
    let mut sink = std::io::sink();
    copy_and_hash(reader, &mut sink, limit)
}

fn copy_and_hash(
    reader: &mut impl Read,
    writer: &mut impl Write,
    limit: u64,
) -> Result<(Digest, u64), String> {
    let mut hasher = Sha256::new();
    let mut size_bytes = 0_u64;
    let mut buffer = [0_u8; 64 * 1024];
    loop {
        let read = reader
            .read(&mut buffer)
            .map_err(|error| error.to_string())?;
        if read == 0 {
            break;
        }
        size_bytes = size_bytes
            .checked_add(read as u64)
            .ok_or_else(|| "artifact size accounting overflowed".to_owned())?;
        if size_bytes > limit {
            return Err(format!(
                "artifact exceeds bounded size limit of {limit} bytes"
            ));
        }
        hasher.update(&buffer[..read]);
        writer
            .write_all(&buffer[..read])
            .map_err(|error| error.to_string())?;
    }
    let digest_hex = format!("{:x}", hasher.finalize());
    Ok((Digest::from_sha256_hex(&digest_hex)?, size_bytes))
}

fn environment_u64(name: &str, default: u64) -> Result<u64, String> {
    match std::env::var(name) {
        Ok(value) => value
            .parse::<u64>()
            .map_err(|_| format!("{name} must be an unsigned integer")),
        Err(std::env::VarError::NotPresent) => Ok(default),
        Err(error) => Err(format!("read {name}: {error}")),
    }
}

fn system_time_epoch_seconds(time: SystemTime) -> Option<u64> {
    time.duration_since(UNIX_EPOCH)
        .ok()
        .map(|duration| duration.as_secs())
}

pub(crate) fn current_epoch_seconds() -> u64 {
    system_time_epoch_seconds(SystemTime::now()).unwrap_or_default()
}

fn validate_relative_path(path: &str) -> Result<(), String> {
    if path.is_empty()
        || path.len() > 240
        || path.starts_with('/')
        || path.starts_with('\\')
        || path.split('/').any(|component| {
            component.is_empty()
                || component == "."
                || component == ".."
                || !component
                    .chars()
                    .all(|character| character.is_ascii_alphanumeric() || "._-".contains(character))
        })
    {
        return Err("task output path must be a safe relative path".to_owned());
    }
    Ok(())
}

fn validate_artifact_component(value: &str, label: &str, max_len: usize) -> Result<(), String> {
    if value.trim().is_empty()
        || value.len() > max_len
        || !value
            .chars()
            .all(|character| character.is_ascii_alphanumeric() || "._-".contains(character))
    {
        return Err(format!("{label} is invalid"));
    }
    Ok(())
}

pub(crate) struct TaskArtifactStore {
    overlay: VfsOverlay,
}

impl TaskArtifactStore {
    pub(crate) fn new(task: TaskInstanceId, node: NodeId) -> Self {
        Self {
            overlay: VfsOverlay::new(task, node),
        }
    }

    pub(crate) fn overlay_mut(&mut self) -> &mut VfsOverlay {
        &mut self.overlay
    }

    pub(crate) fn flush(mut self) -> VfsManifest {
        self.overlay.flush()
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    fn retain_fixture(
        store: &NodeArtifactStore,
        workspace: &Path,
        name: &str,
        bytes: &[u8],
    ) -> RetainedArtifact {
        let output = workspace.join("output");
        fs::create_dir_all(&output).unwrap();
        let path = output.join(name);
        if let Some(parent) = path.parent() {
            fs::create_dir_all(parent).unwrap();
        }
        fs::write(path, bytes).unwrap();
        store.retain_output_file(&output, name).unwrap()
    }

    #[test]
    fn node_artifact_store_retains_real_content_addressed_bytes() {
        let temp = tempfile::tempdir().unwrap();
        let store = NodeArtifactStore::for_runtime(Some(temp.path()), "node/a").unwrap();
        let retained = retain_fixture(&store, temp.path(), "app.tar.zst", b"real artifact bytes");

        assert_eq!(retained.digest, Digest::sha256(b"real artifact bytes"));
        assert_eq!(retained.size_bytes, 19);
        assert_eq!(fs::read(&retained.path).unwrap(), b"real artifact bytes");
        assert_eq!(store.metadata(&retained.id).unwrap().unwrap(), retained);
    }

    #[test]
    fn node_artifact_store_rejects_path_traversal() {
        let temp = tempfile::tempdir().unwrap();
        let store = NodeArtifactStore::for_runtime(Some(temp.path()), "node-a").unwrap();
        let output = temp.path().join("output");
        fs::create_dir_all(&output).unwrap();

        assert!(store.retain_output_file(&output, "../escape").is_err());
        assert!(store.metadata(&ArtifactId::from("../escape")).is_err());
    }

    #[test]
    fn node_artifact_store_reads_only_digest_verified_bytes() {
        let temp = tempfile::tempdir().unwrap();
        let store = NodeArtifactStore::for_runtime(Some(temp.path()), "node-a").unwrap();
        let retained = retain_fixture(&store, temp.path(), "result.bin", b"verified bytes");

        assert_eq!(
            store
                .read_verified(&retained.id, &retained.digest, retained.size_bytes)
                .unwrap(),
            b"verified bytes"
        );
        assert!(store
            .read_verified(&retained.id, &Digest::sha256("wrong"), retained.size_bytes)
            .is_err());
        assert_eq!(store.artifact_ids().unwrap(), vec![retained.id]);
    }

    #[test]
    fn node_artifact_store_reads_bounded_transfer_chunks() {
        let temp = tempfile::tempdir().unwrap();
        let store = NodeArtifactStore::for_runtime(Some(temp.path()), "node-a").unwrap();
        let retained = retain_fixture(&store, temp.path(), "large.bin", b"0123456789");

        assert_eq!(
            store
                .read_verified_chunk(&retained.id, &retained.digest, retained.size_bytes, 3, 4)
                .unwrap(),
            b"3456"
        );
        assert!(store
            .read_verified_chunk(&retained.id, &retained.digest, retained.size_bytes, 11, 4)
            .is_err());
    }

    #[test]
    fn downstream_materialization_copies_exact_digest_verified_bytes() {
        let temp = tempfile::tempdir().unwrap();
        let store = NodeArtifactStore::for_runtime(Some(temp.path()), "node-a").unwrap();
        let retained = retain_fixture(&store, temp.path(), "build/app", b"real compiler output");
        let consumer = temp.path().join("consumer");
        fs::create_dir_all(&consumer).unwrap();
        let handle = clusterflux_core::ArtifactHandle {
            id: retained.id,
            digest: retained.digest,
            size_bytes: retained.size_bytes,
        };

        let path = store
            .materialize_into_output(&handle, &consumer, "package/app")
            .unwrap();

        assert_eq!(path, consumer.join("package/app"));
        assert_eq!(fs::read(path).unwrap(), b"real compiler output");
    }

    #[test]
    fn materialization_rejects_a_corrupt_handle_and_removes_partial_output() {
        let temp = tempfile::tempdir().unwrap();
        let store = NodeArtifactStore::for_runtime(Some(temp.path()), "node-a").unwrap();
        let retained = retain_fixture(&store, temp.path(), "app", b"artifact");
        let consumer = temp.path().join("consumer");
        fs::create_dir_all(&consumer).unwrap();
        let handle = clusterflux_core::ArtifactHandle {
            id: retained.id,
            digest: Digest::sha256("forged"),
            size_bytes: retained.size_bytes,
        };

        assert!(store
            .materialize_into_output(&handle, &consumer, "app")
            .is_err());
        assert!(!consumer.join("app").exists());
    }

    #[cfg(unix)]
    #[test]
    fn publication_and_materialization_reject_symlink_components() {
        use std::os::unix::fs::symlink;

        let temp = tempfile::tempdir().unwrap();
        let store = NodeArtifactStore::for_runtime(Some(temp.path()), "node-a").unwrap();
        let output = temp.path().join("output");
        let outside = temp.path().join("outside");
        fs::create_dir_all(&output).unwrap();
        fs::create_dir_all(&outside).unwrap();
        fs::write(outside.join("secret"), b"outside").unwrap();
        symlink(outside.join("secret"), output.join("linked-file")).unwrap();
        symlink(&outside, output.join("linked-dir")).unwrap();

        assert!(store.retain_output_file(&output, "linked-file").is_err());
        assert!(store
            .retain_output_file(&output, "linked-dir/secret")
            .is_err());

        let retained = retain_fixture(&store, temp.path(), "safe", b"safe");
        let handle = clusterflux_core::ArtifactHandle {
            id: retained.id,
            digest: retained.digest,
            size_bytes: retained.size_bytes,
        };
        assert!(store
            .materialize_into_output(&handle, &output, "linked-dir/copied")
            .is_err());
        assert!(!outside.join("copied").exists());
    }

    #[test]
    fn transfer_chunks_are_hard_bounded_by_the_node() {
        let temp = tempfile::tempdir().unwrap();
        let store = NodeArtifactStore::for_runtime(Some(temp.path()), "node-a").unwrap();
        let bytes = vec![7_u8; (MAX_NODE_ARTIFACT_CHUNK_BYTES + 10) as usize];
        let retained = retain_fixture(&store, temp.path(), "large", &bytes);

        let chunk = store
            .read_verified_chunk(
                &retained.id,
                &retained.digest,
                retained.size_bytes,
                0,
                u64::MAX,
            )
            .unwrap();

        assert_eq!(chunk.len() as u64, MAX_NODE_ARTIFACT_CHUNK_BYTES);
    }

    #[test]
    fn garbage_collection_obeys_count_bytes_age_and_pins() {
        let temp = tempfile::tempdir().unwrap();
        let store = NodeArtifactStore::for_runtime(Some(temp.path()), "node-a").unwrap();
        let pinned = retain_fixture(&store, temp.path(), "pinned", b"12345");
        let disposable = retain_fixture(&store, temp.path(), "disposable", b"67890");
        let pins = BTreeSet::from([pinned.id.clone()]);

        let report = store
            .garbage_collect(
                NodeArtifactRetentionLimits {
                    max_bytes: 5,
                    max_count: 1,
                    max_age_seconds: u64::MAX,
                    restart_pin_seconds: 60,
                },
                &pins,
                current_epoch_seconds(),
            )
            .unwrap();

        assert_eq!(report.evicted, vec![disposable.id]);
        assert!(report.limits_satisfied);
        assert_eq!(store.artifact_ids().unwrap(), vec![pinned.id.clone()]);

        let pinned_over_limit = store
            .garbage_collect(
                NodeArtifactRetentionLimits {
                    max_bytes: 0,
                    max_count: 0,
                    max_age_seconds: 0,
                    restart_pin_seconds: 60,
                },
                &pins,
                current_epoch_seconds().saturating_add(1),
            )
            .unwrap();
        assert!(pinned_over_limit.evicted.is_empty());
        assert!(!pinned_over_limit.limits_satisfied);
        assert!(pinned.path.exists());
    }

    #[test]
    fn task_output_roots_are_unique_and_stale_roots_are_cleaned_on_node_start() {
        let temp = tempfile::tempdir().unwrap();
        let task = TaskInstanceId::from("task:1");
        let first = task_output_root(Some(temp.path()), "node-a", &task).unwrap();
        let second = task_output_root(Some(temp.path()), "node-a", &task).unwrap();
        fs::write(first.join("stale"), b"one").unwrap();
        fs::write(second.join("stale"), b"two").unwrap();

        assert_ne!(first, second);
        assert_eq!(
            clean_stale_task_output_roots(Some(temp.path()), "node-a").unwrap(),
            2
        );
        assert!(!first.exists());
        assert!(!second.exists());
    }
}