zlayer-agent 0.13.0

//! Overlayfs-backed container rootfs (Wave 1 foundations — not yet wired).
//!
//! Today every container's rootfs is a full unpacked COPY of all of its image's
//! layers (see `runtimes::youki`'s use of
//! [`LayerUnpacker::unpack_layers_from_files`](zlayer_registry::LayerUnpacker::unpack_layers_from_files)).
//! That duplicates every layer's bytes per container and re-runs the whole
//! extraction on each `create_container`.
//!
//! This module provides the additive building blocks for an overlayfs rootfs
//! instead: each image layer is extracted ONCE into a shared, read-only
//! `lowerdir` under [`ZLayerDirs::layer_store`](zlayer_paths::ZLayerDirs::layer_store),
//! and a container's rootfs becomes an `overlay` mount stacking those shared
//! lowerdirs under a fresh per-container `upperdir` (+ `workdir`). N containers
//! of one image then share one copy of each layer and pay only the cheap mount.
//!
//! NOTHING here is wired into `create_container` yet — Wave 2 does that. This
//! file only defines the pieces and proves them with root-gated tests, so
//! runtime behaviour is unchanged.
//!
//! Linux-only: it relies on the kernel `overlay` filesystem and on
//! [`LayerUnpacker::new_for_overlay`](zlayer_registry::LayerUnpacker::new_for_overlay),
//! which writes `0:0` char-device / `trusted.overlay.opaque` whiteout markers
//! that need `CAP_MKNOD` / `CAP_SYS_ADMIN`.
//!
//! The `mod overlay_rootfs;` declaration in `runtimes/mod.rs` is already
//! `#[cfg(target_os = "linux")]`-gated, so this file only ever compiles on
//! Linux — no in-file `cfg` guard is needed.

use std::collections::HashMap;
use std::path::{Path, PathBuf};
use std::sync::{Arc, OnceLock};

use tokio::sync::Mutex;
use zlayer_registry::LayerUnpacker;

use crate::error::{AgentError, Result};

/// Sentinel file written beside an extracted layer's `fs/` dir once extraction
/// has fully completed and been atomically published. Its presence means the
/// layer is ready to use as a lowerdir; its absence means "extract me".
const DONE_SENTINEL: &str = ".done";

/// Sentinel for the ROOTLESS `fuse-overlayfs` extraction form. Distinct from
/// [`DONE_SENTINEL`] so the kernel `fs/` and fuse `fs-fuse/` extractions of the
/// same digest are tracked independently (their on-disk whiteout markers
/// differ).
const DONE_SENTINEL_FUSE: &str = ".done-fuse";

/// Process-global single-flight registry of per-digest extraction locks.
///
/// Mirrors the single-flight ethos of `zlayer-registry`'s image-pull registry:
/// two containers of the same image must not race to first-extract the SAME
/// layer (double work, and a half-written temp dir). Keyed by the sanitized
/// layer digest; the value is a per-digest async mutex held only for the
/// duration of that digest's first extraction. Different digests extract fully
/// in parallel.
static EXTRACT_LOCKS: OnceLock<Mutex<HashMap<String, Arc<Mutex<()>>>>> = OnceLock::new();

/// Get (or lazily create) the per-digest async lock for `key`.
async fn digest_lock(key: &str) -> Arc<Mutex<()>> {
    let registry = EXTRACT_LOCKS.get_or_init(|| Mutex::new(HashMap::new()));
    let mut guard = registry.lock().await;
    Arc::clone(
        guard
            .entry(key.to_string())
            .or_insert_with(|| Arc::new(Mutex::new(()))),
    )
}

/// Extract one image layer into the shared layer store exactly once, returning
/// the path to its read-only overlay `lowerdir` (`<layer_store>/<digest>/fs`).
///
/// `digest` is the layer's compressed digest (the dedup key); it is sanitized
/// via [`zlayer_paths::sanitize_digest`] so the store directory name matches the
/// staged blob filename used by the registry. If the `.done` sentinel already
/// exists beside the `fs/` dir, the layer is reused as-is (no re-extraction).
///
/// Concurrency: the FIRST extraction of a given digest is serialized by a
/// process-global per-digest lock (so two containers of one image don't both
/// extract it); a second waiter re-checks the sentinel after acquiring the lock
/// and short-circuits. Extraction is crash-safe: it writes into a sibling temp
/// dir, then atomically `rename(2)`s it into place and only then writes the
/// sentinel — a partially-extracted dir is never visible as `<digest>/fs`.
///
/// # Errors
///
/// Returns an error if the store dirs can't be created, the layer can't be
/// extracted (decompression / whiteout-marker failures, e.g. missing
/// `CAP_MKNOD`), or the atomic publish (rename) fails.
pub async fn extract_layer_once(
    layer_store: &Path,
    digest: &str,
    layer_file: &Path,
    media_type: &str,
) -> Result<PathBuf> {
    extract_layer_once_inner(
        layer_store,
        digest,
        layer_file,
        media_type,
        OverlayBackend::Kernel,
    )
    .await
}

/// Rootless (`fuse-overlayfs`) counterpart of [`extract_layer_once`].
///
/// Identical single-flight / atomic-publish machinery, but the layer is
/// extracted with the ROOTLESS whiteout convention (`.wh.<name>` files +
/// `user.overlay.opaque` xattr) into a SEPARATE per-digest subdir
/// (`<layer_store>/<digest>/fs-fuse`) so it can never collide with the kernel
/// path's `fs` subdir — the two on-disk whiteout representations differ, and a
/// host that flips between backends must not reuse the wrong-form layer. The
/// `.done-fuse` sentinel is likewise distinct from the kernel `.done`.
///
/// Needs no `CAP_MKNOD` / `CAP_SYS_ADMIN`.
///
/// # Errors
///
/// Same shape as [`extract_layer_once`]: store-dir creation, extraction, or the
/// atomic publish failing.
pub async fn extract_layer_once_rootless(
    layer_store: &Path,
    digest: &str,
    layer_file: &Path,
    media_type: &str,
) -> Result<PathBuf> {
    extract_layer_once_inner(
        layer_store,
        digest,
        layer_file,
        media_type,
        OverlayBackend::Fuse,
    )
    .await
}

/// Which overlay backend a layer-store extraction targets. The on-disk whiteout
/// representation differs between the two (kernel: `0:0` char devices +
/// `trusted.overlay.opaque`; fuse: `.wh.<name>` files + `user.overlay.opaque`),
/// so each backend gets its OWN per-digest subdir + sentinel and the two never
/// share bytes.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum OverlayBackend {
    /// Kernel `overlay` filesystem (root / `CAP_SYS_ADMIN`).
    Kernel,
    /// Userspace `fuse-overlayfs` (rootless).
    Fuse,
}

/// The rootfs build strategy chosen for a container, given the daemon's overlay
/// capabilities. Pure selection so the order is unit-testable without building a
/// whole runtime.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum RootfsStrategy {
    /// Kernel `overlay` mount (shared layers; needs root / `CAP_SYS_ADMIN`).
    KernelOverlay,
    /// Rootless `fuse-overlayfs` mount (shared layers; no privilege).
    FuseOverlay,
    /// Full per-container copy of every layer (no dedup; last resort).
    FullCopy,
}

/// Choose the rootfs strategy: kernel overlay first (cheapest, most native),
/// else rootless fuse-overlayfs (still shared-layer dedup), else a full copy.
/// Both overlay strategies share the per-digest layer store; only `FullCopy`
/// duplicates layer bytes per container.
#[must_use]
pub const fn select_rootfs_strategy(
    kernel_overlay_available: bool,
    fuse_overlay_available: bool,
) -> RootfsStrategy {
    if kernel_overlay_available {
        RootfsStrategy::KernelOverlay
    } else if fuse_overlay_available {
        RootfsStrategy::FuseOverlay
    } else {
        RootfsStrategy::FullCopy
    }
}

impl OverlayBackend {
    /// Per-digest subdir name holding the extracted layer for this backend.
    const fn fs_subdir(self) -> &'static str {
        match self {
            Self::Kernel => "fs",
            Self::Fuse => "fs-fuse",
        }
    }

    /// Sentinel file name (beside the subdir) marking a completed extraction.
    const fn sentinel(self) -> &'static str {
        match self {
            Self::Kernel => DONE_SENTINEL,
            Self::Fuse => DONE_SENTINEL_FUSE,
        }
    }
}

/// Shared single-flight + atomic-publish extraction for both overlay backends.
/// The only backend-specific bits are the per-digest subdir name, the sentinel
/// name, and which whiteout-preserving unpacker is used.
async fn extract_layer_once_inner(
    layer_store: &Path,
    digest: &str,
    layer_file: &Path,
    media_type: &str,
    backend: OverlayBackend,
) -> Result<PathBuf> {
    let key = zlayer_paths::sanitize_digest(digest);
    let layer_dir = layer_store.join(&key);
    let fs_dir = layer_dir.join(backend.fs_subdir());
    let sentinel = layer_dir.join(backend.sentinel());

    // Fast path: already extracted and published.
    if sentinel.exists() {
        return Ok(fs_dir);
    }

    // Serialize the first extraction of THIS (digest, backend) pair. Keying on
    // both means the kernel and fuse extraction of the same digest can proceed
    // in parallel (they write disjoint subdirs) instead of needlessly blocking
    // each other.
    let lock = digest_lock(&format!("{key}:{}", backend.fs_subdir())).await;
    let _held = lock.lock().await;

    // Re-check under the lock: a peer may have published while we waited.
    if sentinel.exists() {
        return Ok(fs_dir);
    }

    std::fs::create_dir_all(&layer_dir).map_err(|e| {
        AgentError::Configuration(format!(
            "failed to create layer-store dir {}: {e}",
            layer_dir.display()
        ))
    })?;

    // Extract into a sibling temp dir (same parent => same filesystem => the
    // publish is an atomic rename, never a cross-device copy). A leftover temp
    // from a previous crash is cleared first. The staging name is
    // backend-specific so a concurrent kernel+fuse extract of one digest can't
    // clobber each other's partial tree.
    let staging = layer_dir.join(format!("{}.partial", backend.fs_subdir()));
    if staging.exists() {
        std::fs::remove_dir_all(&staging).map_err(|e| {
            AgentError::Configuration(format!(
                "failed to clear stale staging dir {}: {e}",
                staging.display()
            ))
        })?;
    }

    match backend {
        OverlayBackend::Kernel => {
            let mut unpacker = LayerUnpacker::new_for_overlay(staging.clone());
            unpacker
                .unpack_layer_to_overlay_dir(layer_file, media_type)
                .map_err(|e| {
                    let _ = std::fs::remove_dir_all(&staging);
                    AgentError::Configuration(format!(
                        "failed to extract layer {digest} into overlay lowerdir: {e}"
                    ))
                })?;
        }
        OverlayBackend::Fuse => {
            let mut unpacker = LayerUnpacker::new_for_overlay_rootless(staging.clone());
            unpacker
                .unpack_layer_to_overlay_dir_rootless(layer_file, media_type)
                .map_err(|e| {
                    let _ = std::fs::remove_dir_all(&staging);
                    AgentError::Configuration(format!(
                        "failed to extract layer {digest} into rootless overlay lowerdir: {e}"
                    ))
                })?;
        }
    }

    // Atomic publish: rename the completed tree into place, then drop the
    // sentinel. If the target subdir somehow already exists (e.g. an aborted
    // prior rename), remove it first so the rename is clean.
    if fs_dir.exists() {
        let _ = std::fs::remove_dir_all(&fs_dir);
    }
    std::fs::rename(&staging, &fs_dir).map_err(|e| {
        let _ = std::fs::remove_dir_all(&staging);
        AgentError::Configuration(format!(
            "failed to publish extracted layer {} -> {}: {e}",
            staging.display(),
            fs_dir.display()
        ))
    })?;

    std::fs::write(&sentinel, b"").map_err(|e| {
        AgentError::Configuration(format!(
            "failed to write layer sentinel {}: {e}",
            sentinel.display()
        ))
    })?;

    Ok(fs_dir)
}

/// Mount an overlayfs rootfs at `target`, stacking `lower_dirs` under `upper`
/// (+ `work`).
///
/// `lower_dirs` MUST be in OCI base-first order (the order layers are applied
/// when merging). The overlay `lowerdir=` option lists layers TOP-first
/// (leftmost = highest priority), so this reverses the slice: the last
/// (topmost) image layer ends up leftmost. `upper` and `work` are created if
/// absent (the caller owns their lifetime — they hold the container's writes).
///
/// # Errors
///
/// Returns an error if there are no lower dirs, the upper/work dirs can't be
/// created, or the `mount(2)` fails (needs root / `CAP_SYS_ADMIN`).
pub fn mount_overlay_rootfs(
    lower_dirs: &[PathBuf],
    upper: &Path,
    work: &Path,
    target: &Path,
) -> Result<()> {
    use nix::mount::{mount, MsFlags};

    if lower_dirs.is_empty() {
        return Err(AgentError::Configuration(
            "overlay rootfs requires at least one lower dir".to_string(),
        ));
    }

    for d in [upper, work, target] {
        std::fs::create_dir_all(d).map_err(|e| {
            AgentError::Configuration(format!("failed to create overlay dir {}: {e}", d.display()))
        })?;
    }

    // overlay `lowerdir=` is TOP-first; OCI layers are base-first, so reverse.
    let lower_joined = lower_dirs
        .iter()
        .rev()
        .map(|p| p.display().to_string())
        .collect::<Vec<_>>()
        .join(":");

    let opts = format!(
        "lowerdir={lower_joined},upperdir={},workdir={}",
        upper.display(),
        work.display()
    );

    mount(
        Some("overlay"),
        target,
        Some("overlay"),
        MsFlags::empty(),
        Some(opts.as_str()),
    )
    .map_err(|e| {
        AgentError::Configuration(format!(
            "failed to mount overlay rootfs at {} (opts: {opts}): {e}",
            target.display()
        ))
    })?;

    Ok(())
}

/// Mount a ROOTLESS `fuse-overlayfs` rootfs at `target`, stacking `lower_dirs`
/// under `upper` (+ `work`) by spawning the `fuse-overlayfs` userspace binary.
///
/// Like [`mount_overlay_rootfs`] but uses the FUSE backend, so it needs neither
/// `CAP_SYS_ADMIN` nor `CAP_MKNOD` — only the `fuse-overlayfs` binary and
/// `/dev/fuse`. The lowerdirs MUST have been extracted with the ROOTLESS
/// whiteout convention (see [`extract_layer_once_rootless`]).
///
/// Lowerdir ordering matches the kernel path: `lower_dirs` is OCI base-first and
/// the `lowerdir=` option is TOP-first, so the slice is reversed.
///
/// Lifecycle: `fuse-overlayfs` self-daemonizes (the backing FUSE daemon
/// reparents to PID 1) and the spawned process returns promptly, so the mount
/// survives the short-lived `zlayer runtime create` subprocess that calls this
/// — we wait on the spawned process (it exits as soon as the daemon is up) and
/// do NOT hold a child handle. The `-o noacl` option avoids ACL-passthrough
/// surprises on backing filesystems that don't support them.
///
/// # Errors
///
/// Returns an error if there are no lower dirs, the upper/work/target dirs can't
/// be created, the `fuse-overlayfs` binary can't be located or spawned, or it
/// exits non-zero.
pub fn mount_fuse_overlay_rootfs(
    fuse_overlayfs_bin: &Path,
    lower_dirs: &[PathBuf],
    upper: &Path,
    work: &Path,
    target: &Path,
) -> Result<()> {
    if lower_dirs.is_empty() {
        return Err(AgentError::Configuration(
            "fuse overlay rootfs requires at least one lower dir".to_string(),
        ));
    }

    for d in [upper, work, target] {
        std::fs::create_dir_all(d).map_err(|e| {
            AgentError::Configuration(format!(
                "failed to create fuse overlay dir {}: {e}",
                d.display()
            ))
        })?;
    }

    // `lowerdir=` is TOP-first; OCI layers are base-first, so reverse.
    let lower_joined = lower_dirs
        .iter()
        .rev()
        .map(|p| p.display().to_string())
        .collect::<Vec<_>>()
        .join(":");

    let opts = format!(
        "lowerdir={lower_joined},upperdir={},workdir={},noacl",
        upper.display(),
        work.display()
    );

    let output = std::process::Command::new(fuse_overlayfs_bin)
        .arg("-o")
        .arg(&opts)
        .arg(target)
        .stdin(std::process::Stdio::null())
        .output()
        .map_err(|e| {
            AgentError::Configuration(format!(
                "failed to spawn fuse-overlayfs ({}) for {}: {e}",
                fuse_overlayfs_bin.display(),
                target.display()
            ))
        })?;

    if !output.status.success() {
        let stderr = String::from_utf8_lossy(&output.stderr);
        return Err(AgentError::Configuration(format!(
            "fuse-overlayfs mount at {} failed (opts: {opts}): status {}, stderr: {}",
            target.display(),
            output.status,
            stderr.trim()
        )));
    }

    Ok(())
}

/// Unmount an overlay rootfs previously mounted at `target`, handling BOTH the
/// kernel `overlay` backend and the userspace `fuse.fuse-overlayfs` backend.
///
/// The mount's filesystem TYPE (read from `/proc/mounts`, the 3rd field)
/// decides the teardown: a `fuse.*` mount is unmounted with `fusermount -u`
/// (the only way to release a FUSE mount without `CAP_SYS_ADMIN`); a kernel
/// `overlay` mount uses `umount(2)` with an `MNT_DETACH` lazy-unmount fallback
/// on `EBUSY`. A target that is not currently a mountpoint is an idempotent
/// no-op (covers the plain-copy fallback bundles).
///
/// # Errors
///
/// Returns an error only if the relevant unmount path fails for a reason other
/// than "not mounted".
pub fn unmount_overlay_rootfs(target: &Path) -> Result<()> {
    // Idempotent: nothing mounted here.
    let Some(fstype) = mount_fstype(target) else {
        return Ok(());
    };

    if fstype.starts_with("fuse") {
        return unmount_fuse_overlay_rootfs(target);
    }

    unmount_kernel_overlay_rootfs(target)
}

/// Unmount a kernel `overlay` mount: eager `umount(2)`, lazy `MNT_DETACH` on
/// `EBUSY`, treating "not a mount" as idempotent success.
fn unmount_kernel_overlay_rootfs(target: &Path) -> Result<()> {
    use nix::errno::Errno;
    use nix::mount::{umount, umount2, MntFlags};

    match umount(target) {
        Ok(()) | Err(Errno::EINVAL | Errno::ENOENT) => Ok(()),
        Err(Errno::EBUSY) => match umount2(target, MntFlags::MNT_DETACH) {
            Ok(()) | Err(Errno::EINVAL | Errno::ENOENT) => Ok(()),
            Err(e) => Err(AgentError::Configuration(format!(
                "failed to lazily unmount overlay rootfs at {}: {e}",
                target.display()
            ))),
        },
        Err(e) => Err(AgentError::Configuration(format!(
            "failed to unmount overlay rootfs at {}: {e}",
            target.display()
        ))),
    }
}

/// Unmount a `fuse.fuse-overlayfs` mount via `fusermount3 -u` (falling back to
/// `fusermount -u`). This is the privilege-free way to release a FUSE mount and
/// reap its backing daemon. If the helper exits non-zero but the target is no
/// longer a mountpoint (raced with another unmount), it's treated as success.
///
/// **`EBUSY` retry + lazy fallback.** A FUSE unmount races the kernel and the
/// `fuse-overlayfs` daemon: right after the last user of the merged tree goes
/// away, the kernel may still hold a transient reference (a just-closed fd, an
/// in-flight `getdents`, a child that hasn't fully reaped) and `fusermount -u`
/// reports `Device or resource busy`. This is not a real failure — the
/// reference drains within milliseconds. So we retry the plain `-u` a few times
/// with a short backoff, and if it is *still* busy on the final attempt we issue
/// a lazy unmount (`-u -z`, the FUSE analogue of `umount2(MNT_DETACH)`): the
/// mount is detached from the namespace immediately and the backing daemon reaps
/// once the last reference drops. A target that is no longer a mountpoint at any
/// point is idempotent success. This mirrors the eager-then-lazy strategy
/// [`unmount_kernel_overlay_rootfs`] uses for kernel `overlay` mounts and makes
/// teardown robust under load (concurrent container stops, prune sweeps).
fn unmount_fuse_overlay_rootfs(target: &Path) -> Result<()> {
    /// How many times to retry a plain `-u` that reports the mount busy before
    /// escalating to a lazy `-u -z` detach.
    const MAX_BUSY_RETRIES: u32 = 5;
    /// Base backoff between busy retries (grows linearly: 20ms, 40ms, …).
    const BUSY_BACKOFF: std::time::Duration = std::time::Duration::from_millis(20);

    let Some(fusermount) = crate::capability::fusermount_binary() else {
        return Err(AgentError::Configuration(format!(
            "cannot unmount fuse overlay rootfs at {}: neither fusermount3 nor fusermount on PATH",
            target.display()
        )));
    };

    // Helper: run `fusermount <extra-args> -u <target>` once.
    let run = |lazy: bool| -> Result<std::process::Output> {
        let mut cmd = std::process::Command::new(&fusermount);
        cmd.arg("-u");
        if lazy {
            // `-z` = lazy/detach unmount (supported by both fusermount and
            // fusermount3); detach now, reap the daemon when refs drain.
            cmd.arg("-z");
        }
        cmd.arg(target)
            .stdin(std::process::Stdio::null())
            .output()
            .map_err(|e| {
                AgentError::Configuration(format!(
                    "failed to spawn {} -u {}: {e}",
                    fusermount.display(),
                    target.display()
                ))
            })
    };

    let mut last_stderr = String::new();
    for attempt in 0..=MAX_BUSY_RETRIES {
        let output = run(false)?;
        // Success, or the target is already not a mountpoint (raced with another
        // unmount / the daemon already exited) — both are clean.
        if output.status.success() || mount_fstype(target).is_none() {
            return Ok(());
        }

        last_stderr = String::from_utf8_lossy(&output.stderr).trim().to_string();
        let busy = last_stderr.to_ascii_lowercase().contains("busy");
        if !busy {
            // A non-busy failure won't clear by waiting; report it now.
            return Err(AgentError::Configuration(format!(
                "fusermount -u failed for {}: status {}, stderr: {}",
                target.display(),
                output.status,
                last_stderr
            )));
        }

        // Busy: back off and retry the eager unmount, unless we've exhausted
        // the retry budget (then fall through to the lazy detach below).
        if attempt < MAX_BUSY_RETRIES {
            std::thread::sleep(BUSY_BACKOFF * (attempt + 1));
        }
    }

    // Still busy after every eager retry: lazy-detach so teardown completes.
    let lazy = run(true)?;
    if lazy.status.success() || mount_fstype(target).is_none() {
        tracing::debug!(
            target = %target.display(),
            "fuse overlay rootfs was busy; completed teardown via lazy (-z) unmount",
        );
        return Ok(());
    }

    let lazy_stderr = String::from_utf8_lossy(&lazy.stderr);
    Err(AgentError::Configuration(format!(
        "fusermount -u failed for {} after {} busy retries (last stderr: {}); \
         lazy -uz also failed: status {}, stderr: {}",
        target.display(),
        MAX_BUSY_RETRIES,
        last_stderr,
        lazy.status,
        lazy_stderr.trim()
    )))
}

/// The filesystem TYPE (3rd `/proc/mounts` field, e.g. `overlay` or
/// `fuse.fuse-overlayfs`) of the mount whose mountpoint (2nd field) equals
/// `path`, or `None` if `path` is not currently a mountpoint.
///
/// Octal escapes (`\040` for space, etc.) in the mountpoint field are left as-is
/// because overlay rootfs targets under the data dir never contain such
/// characters. When a path is mounted more than once (stacked), the LAST
/// matching line wins — that's the mount currently visible at the path, the one
/// a single `umount`/`fusermount -u` pops.
fn mount_fstype(path: &Path) -> Option<String> {
    let mounts = std::fs::read_to_string("/proc/mounts").ok()?;
    let want = path.to_string_lossy();
    let mut found: Option<String> = None;
    for line in mounts.lines() {
        let mut fields = line.split_whitespace();
        let _src = fields.next();
        let mountpoint = fields.next();
        let fstype = fields.next();
        if mountpoint == Some(want.as_ref()) {
            if let Some(ty) = fstype {
                found = Some(ty.to_string());
            }
        }
    }
    found
}

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

    /// Uncompressed-tar OCI layer media type (zlayer-agent has no gzip dep, so
    /// the tests build plain tars and pass this).
    const TAR_MT: &str = "application/vnd.oci.image.layer.v1.tar";

    /// Root-gated helper: skip (and say why) when not uid 0, since extraction
    /// (mknod) and the overlay mount need `CAP_MKNOD` / `CAP_SYS_ADMIN`.
    fn require_root_or_skip(test: &str) -> bool {
        if zlayer_paths::is_root() {
            return true;
        }
        eprintln!("skipping {test}: requires root (CAP_MKNOD / CAP_SYS_ADMIN)");
        false
    }

    /// Build a plain (uncompressed) layer tar in memory.
    fn tar_layer(files: &[(&str, &[u8])]) -> Vec<u8> {
        let mut builder = tar::Builder::new(Vec::new());
        for (path, content) in files {
            let mut h = tar::Header::new_gnu();
            h.set_path(path).unwrap();
            h.set_size(content.len() as u64);
            h.set_mode(0o644);
            h.set_cksum();
            builder.append(&h, *content).unwrap();
        }
        builder.into_inner().unwrap()
    }

    #[tokio::test]
    async fn extract_layer_once_is_idempotent_and_writes_sentinel() {
        if !require_root_or_skip("extract_layer_once_is_idempotent_and_writes_sentinel") {
            return;
        }
        let tmp = tempfile::tempdir().unwrap();
        let store = tmp.path().join("layers");
        let layer = tar_layer(&[("etc/hostname", b"host\n")]);
        let layer_file = tmp.path().join("layer0.tar");
        std::fs::write(&layer_file, &layer).unwrap();

        let digest = "sha256:deadbeef";
        let fs1 = extract_layer_once(&store, digest, &layer_file, TAR_MT)
            .await
            .expect("first extract");
        assert!(fs1.join("etc/hostname").exists(), "layer content extracted");
        let sentinel = store
            .join(zlayer_paths::sanitize_digest(digest))
            .join(".done");
        assert!(sentinel.exists(), ".done sentinel must be written");

        // Second call short-circuits to the same path and leaves no staging.
        let fs2 = extract_layer_once(&store, digest, &layer_file, TAR_MT)
            .await
            .expect("second extract (idempotent)");
        assert_eq!(fs1, fs2);
        assert!(
            !fs1.with_file_name("fs.partial").exists(),
            "no leftover staging dir"
        );
    }

    #[tokio::test]
    async fn mount_overlay_merges_lowers_and_writes_land_in_upper() {
        if !require_root_or_skip("mount_overlay_merges_lowers_and_writes_land_in_upper") {
            return;
        }
        let tmp = tempfile::tempdir().unwrap();
        let store = tmp.path().join("layers");

        // Two synthetic layers: base provides `base.txt`; top provides
        // `top.txt` and overrides `shared.txt`.
        let base = tar_layer(&[("base.txt", b"base"), ("shared.txt", b"from-base")]);
        let top = tar_layer(&[("top.txt", b"top"), ("shared.txt", b"from-top")]);
        let base_file = tmp.path().join("base.tar");
        let top_file = tmp.path().join("top.tar");
        std::fs::write(&base_file, &base).unwrap();
        std::fs::write(&top_file, &top).unwrap();

        let base_fs = extract_layer_once(&store, "sha256:base", &base_file, TAR_MT)
            .await
            .unwrap();
        let top_fs = extract_layer_once(&store, "sha256:top", &top_file, TAR_MT)
            .await
            .unwrap();

        // Base-first order; mount_overlay_rootfs reverses for `lowerdir=`.
        let lowers = vec![base_fs, top_fs];
        let upper = tmp.path().join("upper");
        let work = tmp.path().join("work");
        let merged = tmp.path().join("merged");

        mount_overlay_rootfs(&lowers, &upper, &work, &merged).expect("overlay mount");

        // Merged view: both files visible, the TOP layer wins for `shared.txt`.
        assert_eq!(std::fs::read(merged.join("base.txt")).unwrap(), b"base");
        assert_eq!(std::fs::read(merged.join("top.txt")).unwrap(), b"top");
        assert_eq!(
            std::fs::read(merged.join("shared.txt")).unwrap(),
            b"from-top",
            "topmost layer must win in the merged overlay view"
        );

        // A write into the merged view must land in the upperdir, not a lower.
        std::fs::write(merged.join("written.txt"), b"hi").unwrap();
        assert_eq!(
            std::fs::read(upper.join("written.txt")).unwrap(),
            b"hi",
            "writes through the overlay must materialize in the upperdir"
        );

        // Unmount and confirm /proc/mounts no longer lists the target.
        unmount_overlay_rootfs(&merged).expect("unmount");
        let mounts = std::fs::read_to_string("/proc/mounts").unwrap_or_default();
        let merged_str = merged.display().to_string();
        assert!(
            !mounts.lines().any(|l| l.contains(&merged_str)),
            "target must no longer appear in /proc/mounts after unmount"
        );
    }

    #[tokio::test]
    async fn unmount_is_idempotent_on_unmounted_target() {
        // No root needed: unmounting a never-mounted dir must be a no-op.
        let tmp = tempfile::tempdir().unwrap();
        let target = tmp.path().join("not-a-mount");
        std::fs::create_dir_all(&target).unwrap();
        unmount_overlay_rootfs(&target).expect("unmount of non-mount is success");
    }

    #[test]
    fn mount_requires_at_least_one_lower() {
        let tmp = tempfile::tempdir().unwrap();
        let err = mount_overlay_rootfs(
            &[],
            &tmp.path().join("u"),
            &tmp.path().join("w"),
            &tmp.path().join("m"),
        )
        .expect_err("empty lowers must error");
        assert!(err.to_string().contains("at least one lower"));
    }

    #[test]
    fn selection_order_kernel_then_fuse_then_copy() {
        // Kernel wins whenever available, regardless of fuse.
        assert_eq!(
            select_rootfs_strategy(true, true),
            RootfsStrategy::KernelOverlay
        );
        assert_eq!(
            select_rootfs_strategy(true, false),
            RootfsStrategy::KernelOverlay
        );
        // Fuse is chosen only when kernel is unavailable but fuse is — the exact
        // rootless case this feature adds.
        assert_eq!(
            select_rootfs_strategy(false, true),
            RootfsStrategy::FuseOverlay
        );
        // Neither overlay backend → full per-container copy.
        assert_eq!(
            select_rootfs_strategy(false, false),
            RootfsStrategy::FullCopy
        );
    }

    #[test]
    fn fuse_mount_requires_at_least_one_lower() {
        let tmp = tempfile::tempdir().unwrap();
        let err = mount_fuse_overlay_rootfs(
            std::path::Path::new("/nonexistent/fuse-overlayfs"),
            &[],
            &tmp.path().join("u"),
            &tmp.path().join("w"),
            &tmp.path().join("m"),
        )
        .expect_err("empty lowers must error before any spawn");
        assert!(err.to_string().contains("at least one lower"));
    }

    /// The two backends key into DISJOINT per-digest subdirs (`fs` vs `fs-fuse`)
    /// and sentinels (`.done` vs `.done-fuse`), so a kernel and a rootless
    /// extract of the same digest never collide. No privilege needed — the fuse
    /// extract writes `.wh.` files + `user.*` xattrs.
    #[tokio::test]
    async fn kernel_and_fuse_layer_store_dirs_do_not_collide() {
        let tmp = tempfile::tempdir().unwrap();
        let store = tmp.path().join("layers");
        let layer = tar_layer(&[("etc/hostname", b"host\n")]);
        let layer_file = tmp.path().join("layer.tar");
        std::fs::write(&layer_file, &layer).unwrap();
        let digest = "sha256:cafe";
        let key = zlayer_paths::sanitize_digest(digest);

        // Rootless extract works without root.
        let fuse_fs = extract_layer_once_rootless(&store, digest, &layer_file, TAR_MT)
            .await
            .expect("rootless extract");
        assert_eq!(fuse_fs, store.join(&key).join("fs-fuse"));
        assert!(fuse_fs.join("etc/hostname").exists());
        assert!(
            store.join(&key).join(".done-fuse").exists(),
            ".done-fuse sentinel written"
        );
        // The kernel `fs/` form must NOT have been created by the fuse extract.
        assert!(
            !store.join(&key).join("fs").exists(),
            "fuse extract must not populate the kernel `fs` subdir"
        );
        assert!(!store.join(&key).join(".done").exists());

        // Idempotent: a second rootless extract short-circuits, no leftover stage.
        let fuse_fs2 = extract_layer_once_rootless(&store, digest, &layer_file, TAR_MT)
            .await
            .expect("second rootless extract");
        assert_eq!(fuse_fs, fuse_fs2);
        assert!(!store.join(&key).join("fs-fuse.partial").exists());
    }

    /// Locate the `fuse-overlayfs` binary, or `None` to skip the live tests.
    fn fuse_overlayfs_bin_or_skip(test: &str) -> Option<std::path::PathBuf> {
        let path_var = std::env::var("PATH").unwrap_or_default();
        for dir in path_var.split(':').filter(|d| !d.is_empty()) {
            let candidate = std::path::Path::new(dir).join("fuse-overlayfs");
            if candidate.exists() {
                return Some(candidate);
            }
        }
        eprintln!("skipping {test}: fuse-overlayfs not installed (PATH scan found nothing)");
        None
    }

    /// End-to-end ROOTLESS path (no root, no `CAP_SYS_ADMIN`): extract two layers
    /// with the rootless whiteout convention, mount them via `fuse-overlayfs`,
    /// assert the merged view (topmost layer wins, whiteout hides a lower file),
    /// assert writes land in the upperdir, then `fusermount -u` and confirm the
    /// mount is gone. Skips with a message when `fuse-overlayfs` is absent.
    #[tokio::test]
    async fn fuse_overlay_end_to_end_rootless_merge_write_unmount() {
        let Some(bin) =
            fuse_overlayfs_bin_or_skip("fuse_overlay_end_to_end_rootless_merge_write_unmount")
        else {
            return;
        };
        // /dev/fuse must be usable; otherwise the mount cannot back itself.
        if std::fs::OpenOptions::new()
            .read(true)
            .write(true)
            .open("/dev/fuse")
            .is_err()
        {
            eprintln!("skipping: /dev/fuse not openable");
            return;
        }

        let tmp = tempfile::tempdir().unwrap();
        let store = tmp.path().join("layers");

        // base: provides base.txt, shared.txt, and gone.txt.
        // top: provides top.txt, overrides shared.txt, and whites out gone.txt.
        let base = tar_layer(&[
            ("base.txt", b"base"),
            ("shared.txt", b"from-base"),
            ("gone.txt", b"doomed"),
        ]);
        let mut top_builder = tar::Builder::new(Vec::new());
        for (p, c) in [("top.txt", &b"top"[..]), ("shared.txt", &b"from-top"[..])] {
            let mut h = tar::Header::new_gnu();
            h.set_path(p).unwrap();
            h.set_size(c.len() as u64);
            h.set_mode(0o644);
            h.set_cksum();
            top_builder.append(&h, c).unwrap();
        }
        // Whiteout gone.txt (rootless `.wh.gone.txt` file).
        let mut wh = tar::Header::new_gnu();
        wh.set_path(".wh.gone.txt").unwrap();
        wh.set_size(0);
        wh.set_mode(0o644);
        wh.set_cksum();
        top_builder.append(&wh, std::io::empty()).unwrap();
        let top = top_builder.into_inner().unwrap();

        let base_file = tmp.path().join("base.tar");
        let top_file = tmp.path().join("top.tar");
        std::fs::write(&base_file, &base).unwrap();
        std::fs::write(&top_file, &top).unwrap();

        let base_fs = extract_layer_once_rootless(&store, "sha256:base", &base_file, TAR_MT)
            .await
            .unwrap();
        let top_fs = extract_layer_once_rootless(&store, "sha256:top", &top_file, TAR_MT)
            .await
            .unwrap();

        let lowers = vec![base_fs, top_fs]; // base-first
        let upper = tmp.path().join("upper");
        let work = tmp.path().join("work");
        let merged = tmp.path().join("merged");

        mount_fuse_overlay_rootfs(&bin, &lowers, &upper, &work, &merged)
            .expect("rootless fuse-overlayfs mount");

        // Merged view assertions.
        assert_eq!(std::fs::read(merged.join("base.txt")).unwrap(), b"base");
        assert_eq!(std::fs::read(merged.join("top.txt")).unwrap(), b"top");
        assert_eq!(
            std::fs::read(merged.join("shared.txt")).unwrap(),
            b"from-top",
            "topmost layer wins"
        );
        assert!(
            !merged.join("gone.txt").exists(),
            "rootless .wh. whiteout must hide the lower gone.txt"
        );

        // Write goes to the upperdir.
        std::fs::write(merged.join("written.txt"), b"hi").unwrap();
        assert_eq!(
            std::fs::read(upper.join("written.txt")).unwrap(),
            b"hi",
            "writes materialize in the upperdir"
        );

        // /proc/mounts reports the fuse type, and unmount uses fusermount.
        assert!(
            mount_fstype(&merged).is_some_and(|t| t.starts_with("fuse")),
            "merged must be a fuse.* mount"
        );
        unmount_overlay_rootfs(&merged).expect("fusermount -u via unmount_overlay_rootfs");
        assert!(
            mount_fstype(&merged).is_none(),
            "target must be unmounted after unmount_overlay_rootfs"
        );
    }
}