xianyun_executor/

filecache.rs

//! A simple file-based cache for storing data on disk temporarily.
//!
//! Ported from `internal/filecache/filecache.go`, `interface.go`,
//! `noop.go`, and `singlegroup.go`.
//!
//! Go's `SingleGroup[T any]` wraps `golang.org/x/sync/singleflight`,
//! collapsing concurrent calls for the same key into one execution. The
//! idiomatic Rust equivalent is `tokio::sync::OnceCell`-per-key or a
//! `dashmap`-backed in-flight map with `tokio::sync::Notify`; this crate
//! doesn't pull in either dependency since nothing ported here needs
//! cross-call coalescing yet (each `Cache` call is independent). See
//! `PORT_NOTES.md`.

use std::path::PathBuf;
use std::time::Duration;

/// Ported from the `Fallback` function type. Go's version takes a
/// `context.Context`; this port's fallback signature omits it for now
/// (see `crate::exec::context`'s discussion of ambient context) and will
/// need revisiting once cancellation is threaded through in Phase 4.
pub type Fallback = Box<dyn Fn() -> Result<Vec<u8>, FileCacheError>>;

/// Ported from `ErrInvalidMaxAge`/`ErrFallbackFailed`.
#[derive(Debug, thiserror::Error)]
pub enum FileCacheError {
    #[error("maxAge must be positive, got {0:?}")]
    InvalidMaxAge(Duration),
    #[error("entry expired and refreshing failed: {0}")]
    FallbackFailed(String),
    #[error(transparent)]
    Io(#[from] std::io::Error),
    #[error("cancelled")]
    Cancelled,
}

/// Ported from the `Cache` interface.
///
/// Go's `Get` takes a `context.Context` and checks `ctx.Err()` before
/// calling the fallback, so an already-cancelled context short-circuits
/// without ever running it (`TestGet_ContextCancelled`). This port has
/// no ambient context type to thread through (see the module docs), so
/// `get_with_cancellation` takes that one bit of `ctx.Err()` state as an
/// explicit `cancelled: bool` parameter instead -- the same "inject
/// instead of read/carry ambient state" pattern used elsewhere in this
/// crate (`apply_sudo_with`, `has_gitlab_zoom_config`, `is_enabled_with`).
/// `get` is a default method for callers that don't need cancellation at
/// all; it always passes `cancelled = false`.
pub trait Cache {
    fn get(&self, key: &str, fallback: Fallback) -> Result<Vec<u8>, FileCacheError> {
        self.get_with_cancellation(key, fallback, false)
    }

    fn get_file(&self, key: &str, fallback: Fallback) -> Result<PathBuf, FileCacheError>;

    fn get_with_cancellation(
        &self,
        key: &str,
        fallback: Fallback,
        cancelled: bool,
    ) -> Result<Vec<u8>, FileCacheError>;
}

/// Ported from the unexported `fileCache` struct.
pub struct FileCache {
    path: PathBuf,
    max_age: Duration,
}

/// Ported from `New`. Derives the cache path from `name` and the OS user
/// cache directory (`dirs::cache_dir()`, the Rust-idiomatic equivalent of
/// Go's `os.UserCacheDir()`: `$XDG_CACHE_HOME`/`~/.cache` on Linux,
/// `~/Library/Caches` on macOS), then delegates to [`new_at`]. There is
/// no Go test for this path directly (`filecache_test.go` only tests the
/// zero-`maxAge` rejection here; everything else goes through
/// [`new_at`]), matching the same "ambient OS state has no test to port
/// against" situation as `datadir::dir_in`'s caller-resolves-`$HOME`
/// design -- this is simple enough, and specified precisely enough by the
/// Go source, not to need one invented.
pub fn new(name: &str, max_age: Duration) -> Result<Box<dyn Cache>, FileCacheError> {
    if max_age.is_zero() {
        return Err(FileCacheError::InvalidMaxAge(max_age));
    }

    let path = name_to_path(name)?;
    new_at(&path, max_age)
}

/// Ported from the package-private `nameToPath`.
fn name_to_path(name: &str) -> Result<PathBuf, FileCacheError> {
    let safe: String = name
        .chars()
        .map(|c| {
            if c.is_ascii_alphanumeric() || c == '_' {
                c
            } else {
                '_'
            }
        })
        .collect();
    let dir_name = format!("{safe}_{}", &sha256_hex(name.as_bytes())[..8]);

    let user_cache_dir = dirs::cache_dir().ok_or_else(|| {
        FileCacheError::Io(std::io::Error::other(
            "could not determine user cache directory",
        ))
    })?;

    Ok(user_cache_dir.join("xianyun").join(dir_name))
}

/// Ported from `NewAt`.
pub fn new_at(path: &std::path::Path, max_age: Duration) -> Result<Box<dyn Cache>, FileCacheError> {
    if max_age.is_zero() {
        return Err(FileCacheError::InvalidMaxAge(max_age));
    }

    std::fs::create_dir_all(path)?;

    Ok(Box::new(FileCache {
        path: path.to_path_buf(),
        max_age,
    }))
}

impl FileCache {
    fn key_to_path(&self, key: &str) -> PathBuf {
        self.path.join(sha256_hex(key.as_bytes()))
    }

    /// Ported from `readFresh`. Returns `(data, is_fresh)`, or `None` on
    /// a cache miss.
    fn read_fresh(
        &self,
        path: &std::path::Path,
    ) -> Result<Option<(Vec<u8>, bool)>, std::io::Error> {
        let metadata = match std::fs::metadata(path) {
            Ok(m) => m,
            Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(None),
            Err(e) => return Err(e),
        };

        let is_expired = metadata
            .modified()
            .ok()
            .and_then(|m| m.elapsed().ok())
            .is_none_or(|elapsed| elapsed > self.max_age);

        let data = match std::fs::read(path) {
            Ok(d) => d,
            Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(None),
            Err(e) => return Err(e),
        };

        Ok(Some((data, !is_expired)))
    }

    /// Ported from `refresh`. Go's `sg.Do` singleflight collapse across
    /// threads is not modeled here: `Cache` is `Box<dyn Cache>`, not
    /// `Arc<dyn Cache + Send + Sync>`, so it cannot be shared across
    /// threads as written (see `get_concurrent_calls_same_key`'s test
    /// comment and PORT_NOTES.md). Phase 4 must decide the concurrency
    /// story before adding real singleflight collapsing.
    fn refresh(&self, key: &str, fallback: &Fallback) -> Result<Vec<u8>, FileCacheError> {
        let path = self.key_to_path(key);

        let existing = self.read_fresh(&path)?;
        if let Some((data, true)) = &existing {
            return Ok(data.clone());
        }

        match fallback() {
            Ok(data) => {
                write_atomic(&path, &data)?;
                Ok(data)
            }
            Err(err) => {
                if existing.is_some() {
                    // Stale entry: the fallback's own error is wrapped as
                    // `FallbackFailed` ("entry expired and refreshing
                    // failed"), matching Go's `ErrFallbackFailed` wrap.
                    // Go's `refresh` also returns `originalData` alongside
                    // this error for a graceful fallback to stale data;
                    // Rust's `Result` can't carry both, so callers that
                    // need the stale bytes on failure should call
                    // `read_fresh` themselves.
                    Err(FileCacheError::FallbackFailed(err.to_string()))
                } else {
                    // Plain miss (no existing entry at all): propagate the
                    // fallback's error as-is rather than wrapping it.
                    Err(err)
                }
            }
        }
    }
}

/// Ported from `writeAtomic`.
fn write_atomic(path: &std::path::Path, data: &[u8]) -> Result<(), std::io::Error> {
    let dir = path.parent().unwrap_or_else(|| std::path::Path::new("."));
    let mut tmp = crate::tempfile::Tempfile::new(&dir.to_string_lossy(), ".caproni-tmp-*")?;
    tmp.overwrite(data)?;
    std::fs::rename(tmp.name(), path)?;
    // The rename moved the file away from under the Tempfile's Drop-based
    // cleanup; forget it so Drop doesn't try to remove the now-renamed
    // (nonexistent-at-that-path) file.
    std::mem::forget(tmp);
    Ok(())
}

fn sha256_hex(data: &[u8]) -> String {
    crate::sha256::digest_to_hex(data)
}

impl Cache for FileCache {
    fn get_with_cancellation(
        &self,
        key: &str,
        fallback: Fallback,
        cancelled: bool,
    ) -> Result<Vec<u8>, FileCacheError> {
        if cancelled {
            return Err(FileCacheError::Cancelled);
        }
        self.refresh(key, &fallback)
    }

    fn get_file(&self, key: &str, fallback: Fallback) -> Result<PathBuf, FileCacheError> {
        let path = self.key_to_path(key);
        self.refresh(key, &fallback)?;
        Ok(path)
    }
}

/// Ported from `Noop` in `noop.go`, with the base temp directory taken
/// as an explicit parameter (`noop_in`) rather than read from
/// `$TMPDIR`/`os.TempDir()` internally, for the same reason as
/// `datadir::dir_in`: Go's test isolates this with `t.Setenv("TMPDIR",
/// ...)`, which has no safe equivalent under Rust's default
/// multi-threaded test runner. `noop()` is the zero-argument production
/// entry point.
pub fn noop() -> Box<dyn Cache> {
    let temp_dir = crate::tempfile::ensure_temp_dir_exists()
        .unwrap_or_else(|e| panic!("creating noop cache dir: {e}"));
    noop_in(&temp_dir)
}

/// See `noop`'s doc comment for why `base_dir` is explicit here.
pub fn noop_in(base_dir: &std::path::Path) -> Box<dyn Cache> {
    if let Err(e) = std::fs::create_dir_all(base_dir) {
        panic!("creating noop cache dir: {e}");
    }

    let dir = mkdtemp(base_dir, "caproni-noop-")
        .unwrap_or_else(|e| panic!("creating noop cache dir: {e}"));

    Box::new(NoopCache { dir })
}

/// Creates a uniquely-named subdirectory of `base`, mirroring Go's
/// `os.MkdirTemp`'s "prefix + random suffix" contract.
fn mkdtemp(base: &std::path::Path, prefix: &str) -> Result<PathBuf, std::io::Error> {
    for _ in 0..10_000 {
        let random = format!("{:x}", rand_suffix());
        let candidate = base.join(format!("{prefix}{random}"));
        match std::fs::create_dir(&candidate) {
            Ok(()) => return Ok(candidate),
            Err(e) if e.kind() == std::io::ErrorKind::AlreadyExists => continue,
            Err(e) => return Err(e),
        }
    }
    Err(std::io::Error::other("mkdtemp: exhausted retries"))
}

fn rand_suffix() -> u64 {
    use std::time::{SystemTime, UNIX_EPOCH};
    let nanos = SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .unwrap_or_default()
        .as_nanos();
    let pid = u64::from(std::process::id());
    (nanos as u64) ^ pid.wrapping_mul(0x9E37_79B9_7F4A_7C15)
}

struct NoopCache {
    dir: PathBuf,
}

impl Cache for NoopCache {
    fn get_with_cancellation(
        &self,
        _key: &str,
        fallback: Fallback,
        cancelled: bool,
    ) -> Result<Vec<u8>, FileCacheError> {
        if cancelled {
            return Err(FileCacheError::Cancelled);
        }
        fallback()
    }

    fn get_file(&self, _key: &str, fallback: Fallback) -> Result<PathBuf, FileCacheError> {
        let data = fallback()?;
        let mut tmp =
            crate::tempfile::Tempfile::new(&self.dir.to_string_lossy(), "caproni-remote-*")?;
        tmp.overwrite(&data)?;
        let path = tmp.name().to_path_buf();
        // Unlike the FileCache case, this file is the deliverable the
        // caller reads back, not an internal staging file, so it must
        // survive past this function; Go's version never deletes it
        // either (only the parent dir is cleaned up, via a finalizer).
        std::mem::forget(tmp);
        Ok(path)
    }
}

impl Drop for NoopCache {
    fn drop(&mut self) {
        let _ = std::fs::remove_dir_all(&self.dir);
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::sync::Arc;
    use std::sync::atomic::{AtomicI32, Ordering};
    use std::thread;

    fn new_test_cache(max_age: Duration) -> Box<dyn Cache> {
        // `tempfile::tempdir()`'s `TempDir` guard deletes the directory on
        // drop; leaking it here keeps the cache directory alive for the
        // rest of the test (this helper only returns the `Box<dyn Cache>`,
        // so there is no later point to hold the guard).
        let tmp_dir = Box::leak(Box::new(tempfile::tempdir().unwrap()));
        new_at(tmp_dir.path(), max_age).unwrap()
    }

    /// Returns a fallback that records how many times it has been called
    /// and always returns the given data.
    fn fallback_fn(data: Vec<u8>) -> (Fallback, Arc<AtomicI32>) {
        let calls = Arc::new(AtomicI32::new(0));
        let calls_clone = calls.clone();

        let fb: Fallback = Box::new(move || {
            calls_clone.fetch_add(1, Ordering::SeqCst);
            Ok(data.clone())
        });

        (fb, calls)
    }

    #[test]
    fn new_zero_max_age_rejected() {
        let result = new("test", Duration::ZERO);
        assert!(matches!(result, Err(FileCacheError::InvalidMaxAge(_))));
    }

    /// New coverage (no direct Go test to port -- `filecache_test.go` only
    /// exercises `New`'s zero-`maxAge` rejection, same as
    /// `new_zero_max_age_rejected` above): a real cache directory gets
    /// created under the user cache dir, and it's actually usable.
    #[test]
    fn new_creates_a_usable_cache() {
        let cache = new("xianyun-filecache-new-test", Duration::from_secs(3600)).unwrap();
        let (fb, calls) = fallback_fn(b"from new()".to_vec());

        let data = cache.get("probe-key", fb).unwrap();
        assert_eq!(data, b"from new()");
        assert_eq!(calls.load(Ordering::SeqCst), 1);

        // Clean up: this test is the one place in the suite that touches
        // the real user cache directory rather than a tempdir.
        let path = name_to_path("xianyun-filecache-new-test").unwrap();
        let _ = std::fs::remove_dir_all(path);
    }

    /// Ported from the shape of Go's `nameToPath`: non-alphanumeric
    /// characters are replaced with `_`, and the directory name is
    /// suffixed with an 8-character hash so two different names that
    /// collide after sanitization still get distinct directories.
    #[test]
    fn name_to_path_sanitizes_and_disambiguates() {
        let a = name_to_path("my cache/name!").unwrap();
        let b = name_to_path("my-cache-name-").unwrap();

        assert!(a.to_string_lossy().contains("xianyun"));
        assert_ne!(a, b, "different names must not collide after sanitization");

        // Same name always resolves to the same path (deterministic).
        let a_again = name_to_path("my cache/name!").unwrap();
        assert_eq!(a, a_again);
    }

    #[test]
    fn get_cache_miss() {
        let c = new_test_cache(Duration::from_secs(3600));
        let (fb, calls) = fallback_fn(b"woof".to_vec());

        let data = c.get("key1", fb).unwrap();
        assert_eq!(data, b"woof");
        assert_eq!(
            calls.load(Ordering::SeqCst),
            1,
            "fallback should be called once on miss"
        );
    }

    #[test]
    fn get_cache_hit() {
        let c = new_test_cache(Duration::from_secs(3600));
        let (fb, calls) = fallback_fn(b"woof".to_vec());

        c.get("key1", fb).unwrap();

        let (fb2, _) = fallback_fn(b"woof".to_vec());
        let data = c.get("key1", fb2).unwrap();
        assert_eq!(data, b"woof");
        assert_eq!(
            calls.load(Ordering::SeqCst),
            1,
            "fallback should not be called again on hit"
        );
    }

    #[test]
    fn get_stale_entry() {
        let c = new_test_cache(Duration::from_millis(1));

        let (fb1, calls1) = fallback_fn(b"old".to_vec());
        c.get("key1", fb1).unwrap();

        thread::sleep(Duration::from_millis(5));

        let (fb2, calls2) = fallback_fn(b"new".to_vec());
        let data = c.get("key1", fb2).unwrap();
        assert_eq!(data, b"new");
        assert_eq!(calls1.load(Ordering::SeqCst), 1);
        assert_eq!(
            calls2.load(Ordering::SeqCst),
            1,
            "fallback should be called once on stale hit"
        );
    }

    /// This is useful for a graceful fallback to stale data.
    #[test]
    fn get_stale_entry_still_returns_data() {
        let c = new_test_cache(Duration::from_millis(1));
        let (fb, _) = fallback_fn(b"woof".to_vec());
        c.get("key1", fb).unwrap();

        thread::sleep(Duration::from_millis(5));

        let failing: Fallback = Box::new(|| {
            Err(FileCacheError::Io(std::io::Error::new(
                std::io::ErrorKind::UnexpectedEof,
                "eof",
            )))
        });
        let err = c.get("key1", failing).unwrap_err();
        assert!(matches!(err, FileCacheError::FallbackFailed(_)));
    }

    #[test]
    fn get_fallback_error() {
        let c = new_test_cache(Duration::from_secs(3600));
        let fb: Fallback = Box::new(|| {
            Err(FileCacheError::Io(std::io::Error::other(
                "upstream exploded",
            )))
        });

        let err = c.get("key1", fb).unwrap_err();
        assert!(matches!(err, FileCacheError::Io(_)));
    }

    #[test]
    fn get_context_cancelled() {
        // Go: context.WithCancel + an immediate cancel() before calling
        // Get. This port has no ambient context type (see the module
        // docs), so `get_with_cancellation`'s explicit `cancelled: bool`
        // stands in for the already-cancelled context.
        let c = new_test_cache(Duration::from_secs(3600));
        let called = Arc::new(AtomicI32::new(0));
        let called_clone = called.clone();
        let fb: Fallback = Box::new(move || {
            called_clone.fetch_add(1, Ordering::SeqCst);
            Ok(b"data".to_vec())
        });

        let err = c.get_with_cancellation("key1", fb, true).unwrap_err();
        assert!(matches!(err, FileCacheError::Cancelled));
        assert_eq!(
            called.load(Ordering::SeqCst),
            0,
            "fallback must not be called when already cancelled"
        );
    }

    #[test]
    fn get_different_keys_are_independent() {
        let c = new_test_cache(Duration::from_secs(3600));

        let (fb1, _) = fallback_fn(b"val1".to_vec());
        let (fb2, _) = fallback_fn(b"val2".to_vec());

        let data1 = c.get("key1", fb1).unwrap();
        let data2 = c.get("key2", fb2).unwrap();

        assert_eq!(data1, b"val1");
        assert_eq!(data2, b"val2");
    }

    #[test]
    fn get_concurrent_calls_same_key() {
        // NOTE: `Cache` here is `Box<dyn Cache>`, not `Arc<dyn Cache>`, so
        // it cannot be shared across threads as written. Phase 4 needs to
        // decide the concurrent-access story for `Cache` (likely `Arc<dyn
        // Cache + Send + Sync>`) before this test can spawn real threads;
        // ported here as a same-thread sanity check of the call count in
        // the meantime.
        let c = new_test_cache(Duration::from_secs(3600));
        let calls = Arc::new(AtomicI32::new(0));
        let calls_clone = calls.clone();

        let fb: Fallback = Box::new(move || {
            calls_clone.fetch_add(1, Ordering::SeqCst);
            thread::sleep(Duration::from_millis(20));
            Ok(b"shared".to_vec())
        });

        let data = c.get("key1", fb).unwrap();
        assert_eq!(data, b"shared");
        // singleflight should collapse all concurrent calls into at most one.
        assert!(
            calls.load(Ordering::SeqCst) <= 1,
            "fallback should run at most once for concurrent callers"
        );
    }

    #[test]
    fn get_atomic_write_no_corruption() {
        let tmp_dir = tempfile::tempdir().unwrap();
        let c = new_at(tmp_dir.path(), Duration::from_secs(3600)).unwrap();

        let payload = b"the quick brown fox".to_vec();
        let (fb, _) = fallback_fn(payload.clone());

        c.get("key1", fb).unwrap();

        // A fresh Cache pointing at the same temp directory bypasses the
        // in-process singleflight state.
        let c2 = new_at(tmp_dir.path(), Duration::from_secs(3600)).unwrap();

        let never_called: Fallback =
            Box::new(|| panic!("fallback should not be called on a warm cache"));
        let data = c2.get("key1", never_called).unwrap();
        assert_eq!(data, payload);
    }

    // --- Ported from noop_test.go ---

    #[test]
    fn noop_temp_dir_doesnt_panic() {
        let tmp_dir = tempfile::tempdir().unwrap();
        let tmp_dir_path = tmp_dir.path().join("bork");

        // Go: t.Setenv("TMPDIR", tempDir), then Noop() (which internally
        // resolves TMPDIR via EnsureTempDirExists). This port takes the
        // base temp directory as an explicit parameter (`noop_in`)
        // instead, for the reason given on `noop`'s doc comment; the
        // env-var-free equivalent is passing `tmp_dir_path` directly.
        let cache = noop_in(&tmp_dir_path);
        let (fb, _) = fallback_fn(b"noop".to_vec());
        assert_eq!(cache.get("any-key", fb).unwrap(), b"noop");

        let entries: Vec<_> = std::fs::read_dir(&tmp_dir_path).unwrap().collect();
        assert_eq!(entries.len(), 1);
    }
}