xianyun_deploy/registrycache/

mod.rs

//! Manages caproni's host-side pull-through zot registry. Owns the zot
//! binary lifecycle (start on `caproni up`, stop on the last
//! `caproni destroy`), config templating, and the flock-protected ref
//! counting that lets multiple profiles share a single zot instance.
//!
//! Ported from `internal/registrycache/registrycache.go`.

#![allow(dead_code)]

pub mod config;
pub mod lock;
pub mod paths;
pub mod process;
pub mod refs;

use std::path::PathBuf;

use thiserror::Error;

pub use config::ConfigError;
pub use paths::PathError;

/// The port zot listens on. Caproni picks a non-default to avoid
/// collisions with macOS `AirPlay` (which binds 5000). Exported so
/// drivers can compose their own mirror URL alongside their preferred
/// host name; see [`Endpoint`] and each driver's mirror-wiring code.
pub const DEFAULT_PORT: &str = "5555";

/// Errors from registry-cache orchestration.
#[derive(Debug, Error)]
pub enum RegistryCacheError {
    /// The `zot` binary is not on the user's `PATH`. Caproni does not
    /// pin zot itself; users install it from
    /// <https://github.com/project-zot/zot/releases> or run
    /// `mise use -g aqua:project-zot/zot` for a user-wide install.
    #[error("zot binary not found on PATH: {0}")]
    ZotNotInstalled(std::io::Error),
    /// No `ensure` has ever been invoked for this user, so there's no
    /// `config.json` to spawn zot with.
    #[error(
        "no persisted registry cache config; run `caproni up` in a project with \
        `cluster.registry_mirror.managed: true` first"
    )]
    NoPersistedConfig,
    /// Called with an empty upstream list.
    #[error(transparent)]
    Config(#[from] ConfigError),
    /// Failed resolving a registry-cache path.
    #[error(transparent)]
    Path(#[from] PathError),
    /// Failed acquiring the registry-cache flock.
    #[error(transparent)]
    Lock(#[from] lock::LockError),
    /// Failed a ref-marker file operation.
    #[error(transparent)]
    Refs(#[from] refs::RefsError),
    /// Failed a zot process operation.
    #[error(transparent)]
    Process(#[from] process::ProcessError),
    /// Failed creating the data directory.
    #[error("creating data dir: {0}")]
    CreateDataDir(std::io::Error),
    /// Failed writing the zot config to disk.
    #[error("writing zot config: {0}")]
    WriteConfig(std::io::Error),
    /// Failed checking whether a persisted config exists.
    #[error("checking persisted config: {0}")]
    CheckConfig(std::io::Error),
}

/// Describes the running zot from the cluster's perspective. Each
/// driver composes its own mirror URL by calling [`Endpoint::host_port`]
/// with the host name resolvable from inside its own cluster (colima
/// uses `host.docker.internal`, k3d uses `host.k3d.internal`).
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Endpoint {
    /// The port zot listens on (currently always [`DEFAULT_PORT`]).
    pub port: String,
}

impl Endpoint {
    /// Returns `"<host>:<port>"` suitable for use in a registry mirror
    /// URL.
    #[must_use]
    pub fn host_port(&self, host: &str) -> String {
        format!("{host}:{}", self.port)
    }
}

/// Captures the result of [`status`]. Powers `caproni registry status`.
#[derive(Debug, Clone, Default)]
pub struct State {
    /// Whether zot is currently running.
    pub running: bool,
    /// The zot process's PID, when running.
    pub pid: i32,
    /// The port zot listens on.
    pub port: String,
    /// Profile names holding zot up.
    pub refs: Vec<String>,
    /// Configured upstreams (parsed from the running zot config); empty
    /// if the config is absent or unreadable.
    pub upstreams: Vec<String>,
    /// Size of the data dir in bytes; zero when the dir does not exist
    /// yet.
    pub data_bytes: u64,
    /// Path to the generated zot config.
    pub config_path: PathBuf,
    /// Path to the zot blob/manifest cache.
    pub data_dir: PathBuf,
    /// Path to the zot stdout/stderr log.
    pub log_path: PathBuf,
}

fn lookup_zot() -> Result<PathBuf, RegistryCacheError> {
    // Mirrors Go's `exec.LookPath("zot")`: search `$PATH` for an
    // executable named `zot`.
    let path_var = std::env::var_os("PATH").unwrap_or_default();
    for dir in std::env::split_paths(&path_var) {
        let candidate = dir.join("zot");
        if is_executable(&candidate) {
            return Ok(candidate);
        }
    }

    Err(RegistryCacheError::ZotNotInstalled(std::io::Error::new(
        std::io::ErrorKind::NotFound,
        "zot: executable file not found in $PATH",
    )))
}

fn is_executable(path: &std::path::Path) -> bool {
    use std::os::unix::fs::PermissionsExt;

    match std::fs::metadata(path) {
        Ok(meta) => meta.is_file() && meta.permissions().mode() & 0o111 != 0,
        Err(_) => false,
    }
}

/// Starts the host-side zot if not running, registers the given
/// profile's ref, regenerates the config, and returns the endpoint
/// drivers should configure as a mirror. Idempotent. Concurrency-safe:
/// holds an exclusive flock for the duration.
///
/// `upstreams` must be non-empty.
pub fn ensure(profile: &str, upstreams: &[String]) -> Result<Endpoint, RegistryCacheError> {
    if upstreams.is_empty() {
        return Err(RegistryCacheError::Config(ConfigError::NoUpstreams));
    }

    let binary = lookup_zot()?;
    let binary = binary.to_string_lossy().into_owned();

    lock::with_lock(|| ensure_locked(&binary, profile, upstreams))
}

/// Performs the state-mutating work of [`ensure`] while the package
/// flock is held. Extracted to keep cognitive complexity low.
fn ensure_locked(
    binary: &str,
    profile: &str,
    upstreams: &[String],
) -> Result<Endpoint, RegistryCacheError> {
    let data_dir = paths::data_dir()?;
    std::fs::create_dir_all(&data_dir).map_err(RegistryCacheError::CreateDataDir)?;

    let config_path = write_config(&data_dir, upstreams)?;

    refs::add_ref(profile)?;

    start_if_not_running(binary, &config_path)?;

    Ok(Endpoint {
        port: DEFAULT_PORT.to_string(),
    })
}

/// Renders the zot config and writes it to disk. Returns the config
/// file path on success.
fn write_config(
    data_dir: &std::path::Path,
    upstreams: &[String],
) -> Result<PathBuf, RegistryCacheError> {
    // Regenerate the config every `ensure` call so an upstream change
    // in caproni.yaml is picked up on the next `caproni up`. An
    // existing zot won't see the new config until it restarts --
    // documented as a v1 limitation.
    let body = config::render_config(&data_dir.to_string_lossy(), DEFAULT_PORT, upstreams)?;

    let config_path = paths::config_path()?;
    std::fs::write(&config_path, body).map_err(RegistryCacheError::WriteConfig)?;

    Ok(config_path)
}

/// Spawns zot when no live process is detected.
fn start_if_not_running(
    binary: &str,
    config_path: &std::path::Path,
) -> Result<(), RegistryCacheError> {
    if process::is_running()? {
        return Ok(());
    }

    let args = vec![
        "serve".to_string(),
        config_path.to_string_lossy().into_owned(),
    ];
    let pid = process::spawn_process(binary, &args)?;

    tracing::info!(pid, port = DEFAULT_PORT, "registry cache zot started");

    Ok(())
}

/// Releases the profile's hold on the host-side registry cache. The
/// cache process itself is left running so the next `caproni up` (for
/// this or any profile) starts warm. Use [`stop`] to explicitly tear
/// down zot. Concurrency-safe.
pub fn ensure_stopped_for_profile(profile: &str) -> Result<(), RegistryCacheError> {
    lock::with_lock(|| {
        refs::remove_ref(profile)?;

        let remaining = refs::list_refs()?;

        tracing::debug!(
            removed_ref = profile,
            ?remaining,
            "registry cache ref released"
        );

        Ok(())
    })
}

/// Spawns the host-side zot process using the previously persisted
/// config at [`paths::config_path`]. Returns `Ok(())` immediately if
/// zot is already running. Does not touch refs -- this is a
/// process-level operation, not a profile lifecycle event.
///
/// Use this from `caproni registry start` for explicit recovery after
/// zot has died (crash, OOM, user kill). For normal lifecycle, use
/// [`ensure`].
///
/// # Errors
///
/// - [`RegistryCacheError::ZotNotInstalled`] if `zot` is not on `PATH`.
/// - [`RegistryCacheError::NoPersistedConfig`] if no [`ensure`] has
///   ever run for this user.
pub fn start() -> Result<(), RegistryCacheError> {
    let binary = lookup_zot()?;
    let binary = binary.to_string_lossy().into_owned();

    lock::with_lock(|| start_inner(&binary))
}

fn start_inner(binary: &str) -> Result<(), RegistryCacheError> {
    if process::is_running()? {
        tracing::debug!("registry cache zot already running, skipping start");
        return Ok(());
    }

    let config_path = paths::config_path()?;
    if !config_path.exists() {
        return Err(RegistryCacheError::NoPersistedConfig);
    }

    let args = vec![
        "serve".to_string(),
        config_path.to_string_lossy().into_owned(),
    ];
    let pid = process::spawn_process(binary, &args)?;

    tracing::info!(pid, "registry cache zot started");

    Ok(())
}

/// Stops the host-side zot process. Refs are preserved (callers that
/// still expect the cache running can notice via [`status`]). Cached
/// data on disk is preserved so the next [`ensure`] restarts warm.
/// Concurrency-safe.
///
/// `caproni registry stop` gates this on having zero refs; this
/// function does not. Direct callers are expected to enforce their own
/// policy.
pub fn stop() -> Result<(), RegistryCacheError> {
    lock::with_lock(|| {
        process::stop_process()?;
        tracing::info!("registry cache zot stopped");
        Ok(())
    })
}

/// Reports the current state. Read-only; safe to call without the
/// lock (concurrent mutations are short and will momentarily
/// disagree).
pub fn status() -> Result<State, RegistryCacheError> {
    let running = process::is_running()?;

    let mut state = State {
        running,
        port: DEFAULT_PORT.to_string(),
        ..Default::default()
    };

    if running && let Ok(pid) = process::read_pid() {
        state.pid = pid;
    }

    state.refs = refs::list_refs()?;

    state.config_path = paths::config_path().unwrap_or_default();
    state.data_dir = paths::data_dir().unwrap_or_default();
    state.log_path = paths::log_path().unwrap_or_default();

    state.upstreams = read_upstreams_from_config(&state.config_path);
    state.data_bytes = dir_size(&state.data_dir);

    Ok(state)
}

/// Parses the upstream hostnames out of the generated zot config JSON
/// for display. Returns an empty vec when the file is missing,
/// unreadable, or malformed -- a config-display feature shouldn't
/// surface I/O errors.
fn read_upstreams_from_config(path: &std::path::Path) -> Vec<String> {
    if path.as_os_str().is_empty() {
        return Vec::new();
    }

    let Ok(body) = std::fs::read(path) else {
        return Vec::new();
    };

    let Ok(doc) = serde_json::from_slice::<serde_json::Value>(&body) else {
        return Vec::new();
    };

    let registries = doc
        .get("extensions")
        .and_then(|v| v.get("sync"))
        .and_then(|v| v.get("registries"))
        .and_then(|v| v.as_array())
        .cloned()
        .unwrap_or_default();

    registries
        .into_iter()
        .filter_map(|r| {
            r.get("urls")
                .and_then(|u| u.as_array())
                .and_then(|arr| arr.first())
                .and_then(|u| u.as_str())
                .map(hostname_from_url)
        })
        .collect()
}

/// Reverses `upstream_url` for display: strips the scheme and undoes
/// the `docker.io` quirk so users see what they wrote in caproni.yaml.
fn hostname_from_url(u: &str) -> String {
    let host = u
        .strip_prefix("https://")
        .or_else(|| u.strip_prefix("http://"))
        .unwrap_or(u);

    if host == "registry-1.docker.io" {
        "docker.io".to_string()
    } else {
        host.to_string()
    }
}

/// Walks the directory and returns the total size of regular files in
/// bytes. Returns 0 when the directory does not exist or any error
/// occurs during the walk -- this is a best-effort display value.
fn dir_size(dir: &std::path::Path) -> u64 {
    if dir.as_os_str().is_empty() {
        return 0;
    }

    fn walk(dir: &std::path::Path, total: &mut u64) {
        let Ok(entries) = std::fs::read_dir(dir) else {
            return;
        };

        for entry in entries.flatten() {
            let Ok(file_type) = entry.file_type() else {
                continue;
            };

            if file_type.is_dir() {
                walk(&entry.path(), total);
            } else if file_type.is_file()
                && let Ok(meta) = entry.metadata()
            {
                *total += meta.len();
            }
        }
    }

    let mut total = 0;
    walk(dir, &mut total);
    total
}

/// Test-only helpers shared across every `registrycache` submodule.
///
/// The upstream Go test suite relies on `t.Setenv("HOME", ...)` plus
/// Go's default of running same-package tests sequentially (no
/// `t.Parallel()` in this package) to make mutating `$HOME`/`$PATH`
/// safe. Rust's default test harness runs tests from the same binary
/// concurrently, so we serialize every test that mutates process-wide
/// environment state behind one mutex and restore the previous value
/// afterwards.
#[cfg(test)]
pub(crate) mod test_support {
    use std::path::{Path, PathBuf};
    use std::sync::{Mutex, OnceLock};

    fn env_lock() -> &'static Mutex<()> {
        static LOCK: OnceLock<Mutex<()>> = OnceLock::new();
        LOCK.get_or_init(|| Mutex::new(()))
    }

    /// Runs `f` with `$HOME` set to a fresh temporary directory,
    /// serialized against every other `with_home`/env-mutating test in
    /// this crate. Restores the previous `$HOME` (or unsets it) on
    /// return.
    pub(crate) fn with_home<T>(f: impl FnOnce(&Path) -> T) -> T {
        let _guard = env_lock().lock().unwrap_or_else(|e| e.into_inner());

        let dir = tempfile::tempdir().unwrap();
        let previous = std::env::var_os("HOME");

        // SAFETY: serialized by `env_lock` above; no other thread in
        // this test binary observes `$HOME` while we hold the guard.
        unsafe {
            std::env::set_var("HOME", dir.path());
        }

        let result = f(dir.path());

        // SAFETY: same as above.
        unsafe {
            match &previous {
                Some(v) => std::env::set_var("HOME", v),
                None => std::env::remove_var("HOME"),
            }
        }

        result
    }

    /// Like [`with_home`], but also prepends `path_prefix` (typically a
    /// directory containing a stub binary) to `$PATH` for the duration
    /// of `f`.
    pub(crate) fn with_home_and_path<T>(path_prefix: &Path, f: impl FnOnce(&Path) -> T) -> T {
        let _guard = env_lock().lock().unwrap_or_else(|e| e.into_inner());

        let dir = tempfile::tempdir().unwrap();
        let previous_home = std::env::var_os("HOME");
        let previous_path = std::env::var_os("PATH");

        let new_path = match &previous_path {
            Some(p) => {
                let mut joined = std::ffi::OsString::from(path_prefix);
                joined.push(":");
                joined.push(p);
                joined
            }
            None => path_prefix.as_os_str().to_owned(),
        };

        // SAFETY: serialized by `env_lock` above.
        unsafe {
            std::env::set_var("HOME", dir.path());
            std::env::set_var("PATH", &new_path);
        }

        let result = f(dir.path());

        // SAFETY: same as above.
        unsafe {
            match &previous_home {
                Some(v) => std::env::set_var("HOME", v),
                None => std::env::remove_var("HOME"),
            }
            match &previous_path {
                Some(v) => std::env::set_var("PATH", v),
                None => std::env::remove_var("PATH"),
            }
        }

        result
    }

    /// Renames the stub binary produced by
    /// `process::tests::write_stub_binary`-style helpers to `zot` in
    /// place, returning its new path, so `lookup_zot` finds it.
    pub(crate) fn install_as_zot(stub: &Path) -> PathBuf {
        let zot_path = stub.with_file_name("zot");
        std::fs::rename(stub, &zot_path).unwrap();
        zot_path
    }
}

#[cfg(test)]
mod tests {
    use std::os::unix::fs::PermissionsExt;
    use std::path::Path;
    use std::time::{Duration, Instant};

    use super::*;
    use test_support::{install_as_zot, with_home, with_home_and_path};

    fn write_stub_binary() -> tempfile::TempDir {
        let dir = tempfile::tempdir().unwrap();
        let bin = dir.path().join("zot-stub");

        let script = format!(
            "#!/bin/sh\n\
            trap \"exit 0\" INT TERM\n\
            echo woof > {}/started\n\
            i=0; while [ \"$i\" -lt 60 ]; do sleep 1; i=$((i + 1)); done\n",
            dir.path().display()
        );

        std::fs::write(&bin, script).unwrap();
        std::fs::set_permissions(&bin, std::fs::Permissions::from_mode(0o755)).unwrap();

        dir
    }

    fn wait_for_process_started(bin: &Path) {
        let dir = bin.parent().unwrap();
        let started = dir.join("started");

        let deadline = Instant::now() + Duration::from_secs(5);
        while Instant::now() < deadline {
            if started.exists() {
                return;
            }
            std::thread::sleep(Duration::from_millis(10));
        }

        panic!("timed out waiting for stub process to start");
    }

    // Ported from internal/registrycache/registrycache_test.go:13 (TestEnsureWritesConfigAndStartsAndAddsRef)
    #[test]
    fn test_ensure_writes_config_and_starts_and_adds_ref() {
        let stub_dir = write_stub_binary();
        let stub = stub_dir.path().join("zot-stub");
        let zot = install_as_zot(&stub);

        with_home_and_path(zot.parent().unwrap(), |_home| {
            let ep = ensure("alpha", &["docker.io".to_string()]).unwrap();
            assert_eq!(ep.port, DEFAULT_PORT);
            assert_eq!(
                ep.host_port("host.docker.internal"),
                format!("host.docker.internal:{DEFAULT_PORT}")
            );

            wait_for_process_started(&zot);

            let config_path = paths::config_path().unwrap();
            assert!(config_path.exists());

            let refs = refs::list_refs().unwrap();
            assert_eq!(refs, vec!["alpha".to_string()]);

            let running = process::is_running().unwrap();
            assert!(running);

            let _ = process::stop_process();
        });
    }

    // Ported from internal/registrycache/registrycache_test.go:49 (TestEnsureIdempotent)
    #[test]
    fn test_ensure_idempotent() {
        let stub_dir = write_stub_binary();
        let stub = stub_dir.path().join("zot-stub");
        let zot = install_as_zot(&stub);

        with_home_and_path(zot.parent().unwrap(), |_home| {
            ensure("alpha", &["docker.io".to_string()]).unwrap();

            let pid_path = paths::pid_path().unwrap();
            let first = std::fs::read_to_string(&pid_path).unwrap();

            // Second ensure for the same profile: should not respawn.
            ensure("alpha", &["docker.io".to_string()]).unwrap();

            let second = std::fs::read_to_string(&pid_path).unwrap();
            assert_eq!(
                first, second,
                "ensure must not respawn an already-running zot"
            );

            let _ = process::stop_process();
        });
    }

    // Ported from internal/registrycache/registrycache_test.go:77 (TestEnsureFailsWhenZotMissing)
    #[test]
    fn test_ensure_fails_when_zot_missing() {
        with_home(|_home| {
            let empty_dir = tempfile::tempdir().unwrap();
            let previous_path = std::env::var_os("PATH");

            // SAFETY: serialized inside `with_home`'s guard.
            unsafe {
                std::env::set_var("PATH", empty_dir.path());
            }

            let err = ensure("alpha", &["docker.io".to_string()]).unwrap_err();
            assert!(matches!(err, RegistryCacheError::ZotNotInstalled(_)));

            // SAFETY: same as above.
            unsafe {
                match &previous_path {
                    Some(v) => std::env::set_var("PATH", v),
                    None => std::env::remove_var("PATH"),
                }
            }
        });
    }

    // Ported from internal/registrycache/registrycache_test.go:90 (TestEnsureNoUpstreamsIsError)
    #[test]
    fn test_ensure_no_upstreams_is_error() {
        with_home(|_home| {
            let err = ensure("alpha", &[]).unwrap_err();
            assert!(matches!(
                err,
                RegistryCacheError::Config(ConfigError::NoUpstreams)
            ));
        });
    }

    // Ported from internal/registrycache/registrycache_test.go:100 (TestEnsureStoppedForProfileRemovesRefButLeavesZotRunning)
    #[test]
    fn test_ensure_stopped_for_profile_removes_ref_but_leaves_zot_running() {
        let stub_dir = write_stub_binary();
        let stub = stub_dir.path().join("zot-stub");
        let zot = install_as_zot(&stub);

        with_home_and_path(zot.parent().unwrap(), |_home| {
            ensure("alpha", &["docker.io".to_string()]).unwrap();

            ensure_stopped_for_profile("alpha").unwrap();

            let refs = refs::list_refs().unwrap();
            assert!(refs.is_empty());

            let running = process::is_running().unwrap();
            assert!(running);

            let _ = process::stop_process();
        });
    }

    // Ported from internal/registrycache/registrycache_test.go:127 (TestStopKillsZotEvenWithRefs)
    #[test]
    fn test_stop_kills_zot_even_with_refs() {
        let stub_dir = write_stub_binary();
        let stub = stub_dir.path().join("zot-stub");
        let zot = install_as_zot(&stub);

        with_home_and_path(zot.parent().unwrap(), |_home| {
            ensure("alpha", &["docker.io".to_string()]).unwrap();

            stop().unwrap();

            let running = process::is_running().unwrap();
            assert!(!running);

            // Refs are not cleared by stop -- that's the diagnostic story.
            let refs = refs::list_refs().unwrap();
            assert_eq!(refs, vec!["alpha".to_string()]);
        });
    }

    // Ported from internal/registrycache/registrycache_test.go:152 (TestStatusReportsRunningAndRefs)
    #[test]
    fn test_status_reports_running_and_refs() {
        let stub_dir = write_stub_binary();
        let stub = stub_dir.path().join("zot-stub");
        let zot = install_as_zot(&stub);

        with_home_and_path(zot.parent().unwrap(), |_home| {
            ensure("alpha", &["docker.io".to_string(), "quay.io".to_string()]).unwrap();

            let state = status().unwrap();
            assert!(state.running);
            assert!(state.pid > 0);
            assert_eq!(state.port, DEFAULT_PORT);
            assert_eq!(state.refs, vec!["alpha".to_string()]);

            let _ = process::stop_process();
        });
    }

    // Ported from internal/registrycache/registrycache_test.go:175 (TestStatusReportsStopped)
    #[test]
    fn test_status_reports_stopped() {
        with_home(|_home| {
            let state = status().unwrap();
            assert!(!state.running);
            assert!(state.refs.is_empty());
        });
    }

    // Ported from internal/registrycache/registrycache_test.go:186 (TestStart_NoPersistedConfig)
    #[test]
    fn test_start_no_persisted_config() {
        let stub_dir = write_stub_binary();
        let stub = stub_dir.path().join("zot-stub");
        let zot = install_as_zot(&stub);

        with_home_and_path(zot.parent().unwrap(), |_home| {
            let err = start().unwrap_err();
            assert!(matches!(err, RegistryCacheError::NoPersistedConfig));
        });
    }

    // Ported from internal/registrycache/registrycache_test.go:200 (TestStart_SpawnsWhenConfigExists)
    #[test]
    fn test_start_spawns_when_config_exists() {
        let stub_dir = write_stub_binary();
        let stub = stub_dir.path().join("zot-stub");
        let zot = install_as_zot(&stub);

        with_home_and_path(zot.parent().unwrap(), |_home| {
            let config_path = paths::config_path().unwrap();
            std::fs::create_dir_all(config_path.parent().unwrap()).unwrap();
            std::fs::write(&config_path, r#"{"distSpecVersion":"1.1.1"}"#).unwrap();

            start().unwrap();

            let running = process::is_running().unwrap();
            assert!(running);

            // Refs are NOT touched by start.
            let refs = refs::list_refs().unwrap();
            assert!(refs.is_empty());

            let _ = process::stop_process();
        });
    }
}