pushkin_daemon/

warm.rs

//! Warm per-file state + watcher (spec §8.1): verdicts memoized by
//! (path, content-hash), invalidated by keyed path events, cleared
//! wholesale on manifest change. The memo NEVER re-decides — a hit
//! returns the stored envelope, a miss runs the same `check_write` the
//! cold path runs, so warm and cold are identical by construction.

use notify::Watcher as _;
use pushkin_core::envelope::CheckResult;
use pushkin_core::manifest::Manifest;
use pushkin_core::pipeline::{check_write, WriteRequest};
use std::collections::HashMap;
use std::path::Path;
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{Arc, Mutex, RwLock};

/// FNV-1a over the write content — the project's stable-hash convention
/// (spec §7.4 chose it for nudge arms because std hashers are not
/// contractually stable; the memo key reuses that decision).
fn fnv1a(bytes: &[u8]) -> u64 {
    let mut hash: u64 = 0xcbf2_9ce4_8422_2325;
    for byte in bytes {
        hash ^= u64::from(*byte);
        hash = hash.wrapping_mul(0x0000_0100_0000_01b3);
    }
    hash
}

type Memo = HashMap<String, (u64, CheckResult)>;

struct Inner {
    manifest: RwLock<Manifest>,
    memo: Mutex<Memo>,
    hits: AtomicU64,
}

/// Memoized check state shared between the accept loop and the watcher.
pub struct WarmState {
    inner: Arc<Inner>,
}

/// Keeps the filesystem watcher alive; dropping it stops watching.
pub struct WatchGuard {
    _watcher: notify::RecommendedWatcher,
}

impl WarmState {
    #[must_use]
    pub fn new(manifest: Manifest) -> Self {
        Self {
            inner: Arc::new(Inner {
                manifest: RwLock::new(manifest),
                memo: Mutex::new(HashMap::new()),
                hits: AtomicU64::new(0),
            }),
        }
    }

    /// A handle to the same underlying state (memo, manifest, hit
    /// counter are shared, not copied) — one per connection task.
    #[must_use]
    pub fn share(&self) -> Self {
        Self {
            inner: Arc::clone(&self.inner),
        }
    }

    /// Warm-or-compute a verdict. A hit returns the stored envelope for
    /// the same (path, content); a miss computes with the SAME
    /// `check_write` the cold path uses and stores the result.
    #[must_use]
    pub fn check(&self, request: &WriteRequest) -> CheckResult {
        let content_hash = fnv1a(request.content.as_bytes());
        if let Ok(memo) = self.inner.memo.lock() {
            if let Some((stored_hash, stored)) = memo.get(&request.file_path) {
                if *stored_hash == content_hash {
                    self.inner.hits.fetch_add(1, Ordering::SeqCst);
                    return stored.clone();
                }
            }
        }
        let result = match self.inner.manifest.read() {
            Ok(manifest) => check_write(&manifest, request),
            Err(poisoned) => check_write(&poisoned.into_inner(), request),
        };
        if let Ok(mut memo) = self.inner.memo.lock() {
            memo.insert(request.file_path.clone(), (content_hash, result.clone()));
        }
        result
    }

    /// Memo hits since construction.
    #[must_use]
    pub fn hits(&self) -> u64 {
        self.inner.hits.load(Ordering::SeqCst)
    }

    /// Replace the manifest and drop every memoized verdict (mappings
    /// shape every verdict; nothing memoized survives a manifest change).
    pub fn swap_manifest(&self, manifest: Manifest) {
        match self.inner.manifest.write() {
            Ok(mut slot) => *slot = manifest,
            Err(poisoned) => *poisoned.into_inner() = manifest,
        }
        if let Ok(mut memo) = self.inner.memo.lock() {
            memo.clear();
        }
    }

    /// Drop memoized verdicts for one path; untouched paths stay warm.
    pub fn invalidate_path(&self, path: &str) {
        if let Ok(mut memo) = self.inner.memo.lock() {
            memo.remove(path);
        }
    }

    /// Watch `root` recursively: a `pushkin.toml` change reloads the
    /// manifest wholesale (parse failures keep the old manifest — a
    /// half-saved edit must not tear down the gate); any other file
    /// change invalidates that file's memo entry, keyed by its
    /// root-relative path.
    ///
    /// # Errors
    /// `std::io::Error` when the watcher cannot be created or attached.
    pub fn watch(&self, root: &Path) -> std::io::Result<WatchGuard> {
        let inner = Arc::clone(&self.inner);
        let root_buf = root.to_path_buf();
        let mut watcher =
            notify::recommended_watcher(move |event: Result<notify::Event, notify::Error>| {
                let Ok(event) = event else { return };
                for path in event.paths {
                    handle_change(&inner, &root_buf, &path);
                }
            })
            .map_err(std::io::Error::other)?;
        watcher
            .watch(root, notify::RecursiveMode::Recursive)
            .map_err(std::io::Error::other)?;
        Ok(WatchGuard { _watcher: watcher })
    }
}

fn handle_change(inner: &Inner, root: &Path, changed: &Path) {
    let is_manifest = changed
        .file_name()
        .is_some_and(|name| name == "pushkin.toml");
    if is_manifest {
        // Reload wholesale; a manifest that doesn't parse right now keeps
        // the previous one live (never tear down the gate on a half-save).
        let Ok(text) = std::fs::read_to_string(changed) else {
            return;
        };
        let Ok(manifest) = Manifest::parse(&text) else {
            return;
        };
        match inner.manifest.write() {
            Ok(mut slot) => *slot = manifest,
            Err(poisoned) => *poisoned.into_inner() = manifest,
        }
        if let Ok(mut memo) = inner.memo.lock() {
            memo.clear();
        }
        return;
    }
    let key = changed
        .strip_prefix(root)
        .unwrap_or(changed)
        .to_string_lossy()
        .into_owned();
    if let Ok(mut memo) = inner.memo.lock() {
        memo.remove(&key);
    }
}