pushkin-daemon 0.2.1

//! Eager epoch probe + sequential regeneration queue (spec §5.2): probe
//! headers cheaply at startup, regenerate stale artifacts one at a time
//! with a per-item timeout. Eager because lazy checks leave artifacts
//! silently stale and never run in short-lived processes; sequential to
//! bound resource use; per-item timeout so one stuck item can't block
//! the queue.

use std::path::{Path, PathBuf};
use std::time::Duration;

/// The header line every generated file carries (spec §5.2, emitted by
/// `pushkin-compiler::targets::header`).
const EPOCH_MARKER: &str = "pushkin-epoch:";

/// How much of a file the probe reads: the epoch header sits in the first
/// lines by construction; reading whole trees would defeat "cheap".
const PROBE_HEAD_BYTES: usize = 512;

/// Per-item result of a regeneration pass; the queue reports every item,
/// never silently drops one.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum RegenOutcome {
    Regenerated,
    TimedOut,
    Failed(String),
}

/// Scan a generated tree for files whose `pushkin-epoch:` header lags
/// `expected_epoch` (missing header = stale). Returns paths relative to
/// `dir`, unordered.
///
/// # Errors
/// `std::io::Error` when the tree cannot be read.
pub fn probe_stale(dir: &Path, expected_epoch: u32) -> std::io::Result<Vec<PathBuf>> {
    let mut stale = Vec::new();
    let mut pending = vec![dir.to_path_buf()];
    while let Some(current) = pending.pop() {
        for entry in std::fs::read_dir(&current)? {
            let entry = entry?;
            let path = entry.path();
            if path.is_dir() {
                pending.push(path);
            } else if file_epoch(&path)? != Some(expected_epoch) {
                if let Ok(relative) = path.strip_prefix(dir) {
                    stale.push(relative.to_path_buf());
                }
            }
        }
    }
    Ok(stale)
}

/// The epoch stamped in a file's head, or `None` when no marker is found
/// (headerless = stale by definition — a generated file we can't date is
/// a generated file we can't trust).
fn file_epoch(path: &Path) -> std::io::Result<Option<u32>> {
    use std::io::Read;
    let mut head = vec![0_u8; PROBE_HEAD_BYTES];
    let mut file = std::fs::File::open(path)?;
    let read = file.read(&mut head)?;
    head.truncate(read);
    let text = String::from_utf8_lossy(&head);
    Ok(text.lines().find_map(|line| {
        let (_, rest) = line.split_once(EPOCH_MARKER)?;
        rest.trim().parse::<u32>().ok()
    }))
}

/// Run `worker` over `items` strictly one at a time, capping each item at
/// `timeout`. A timed-out item is reported as `TimedOut` and the queue
/// moves on immediately; its worker thread is left to finish and be
/// dropped (threads cannot be force-killed safely — the leak is bounded
/// by the toolchain call it wraps, and the alternative is a wedged queue).
pub fn run_queue<W>(items: &[PathBuf], timeout: Duration, worker: W) -> Vec<(PathBuf, RegenOutcome)>
where
    W: Fn(&Path) -> Result<(), String> + Send + Sync + 'static,
{
    let worker = std::sync::Arc::new(worker);
    let mut outcomes = Vec::with_capacity(items.len());
    for item in items {
        outcomes.push((item.clone(), run_one(item, timeout, &worker)));
    }
    outcomes
}

fn run_one<W>(item: &Path, timeout: Duration, worker: &std::sync::Arc<W>) -> RegenOutcome
where
    W: Fn(&Path) -> Result<(), String> + Send + Sync + 'static,
{
    let (sender, receiver) = std::sync::mpsc::channel();
    let worker = std::sync::Arc::clone(worker);
    let path = item.to_path_buf();
    // One worker thread per item, joined via the channel within `timeout`.
    // Deliberately NOT a thread pool: sequential is the §5.2 contract.
    std::thread::spawn(move || {
        let _ = sender.send(worker(&path));
    });
    match receiver.recv_timeout(timeout) {
        Ok(Ok(())) => RegenOutcome::Regenerated,
        Ok(Err(message)) => RegenOutcome::Failed(message),
        Err(_) => RegenOutcome::TimedOut,
    }
}