sausage/cli/src/watch.rs

//! File watching: the host watches the whole world folder with the OS file
//! watcher (`notify`, inotify on Linux), whoever writes: actors, a 9P mount,
//! a shell or `git` on the host. Nothing outside the folder counts.
//!
//! Events are coalesced per path: once a path is quiet for `QUIET` (or has
//! kept changing for `MAX_DELAY`), it becomes one change
//! `{path, op: write | create | remove | rename, to?}` with a namespace path
//! (`/actor/...`). Then the broker
//!
//! - sends `fs_changed` to the actors that watch the path (`watch` request),
//! - starts or upgrades an actor whose `actor/<id>/actor.wasm` (or folder
//!   `actor/<id>`) changed, tested first, as at boot (`Broker::check_actor`).
//!
//! The host's temporary files (`.sausage-tmp-*`) are skipped: a rename of one
//! onto a file is a `write` of that file. If the OS drops events, the broker
//! runs a full pass and tells every watcher (`Broker::overflowed`).

use std::collections::HashMap;
use std::path::Path;
use std::sync::Arc;
use std::time::Duration;

use notify::event::{EventKind, ModifyKind, RenameMode};
use notify::{RecursiveMode, Watcher as _};
use tokio::sync::mpsc;
use tokio::time::Instant;

use crate::broker::Broker;
use crate::world::TMP_PREFIX;

/// A path that is quiet this long becomes one change.
pub const QUIET: Duration = Duration::from_millis(100);
/// A path that keeps changing becomes a change at least this often.
pub const MAX_DELAY: Duration = Duration::from_secs(1);

/// The running watcher. Dropping it stops watching.
pub struct Watcher {
    _os: notify::RecommendedWatcher,
    task: tokio::task::JoinHandle<()>,
}

impl Drop for Watcher {
    fn drop(&mut self) {
        self.task.abort();
    }
}

/// Watch the world folder of `broker` recursively. A failure to watch part
/// of it (for example too few inotify watches) is logged, and the rest is
/// watched.
pub fn start(broker: Arc<Broker>) -> notify::Result<Watcher> {
    let (tx, rx) = mpsc::unbounded_channel();
    let mut os = notify::recommended_watcher(move |event| {
        let _ = tx.send(event);
    })?;
    if let Err(e) = os.watch(&broker.world.root, RecursiveMode::Recursive) {
        broker.host_error(&format!("cannot watch all of {}: {e}", broker.world.root.display()));
    }
    let task = tokio::spawn(run(broker, rx));
    Ok(Watcher { _os: os, task })
}

/// A change waiting for its path to be quiet.
struct Pending {
    op: &'static str,
    to: Option<String>,
    first: Instant,
    last: Instant,
}

impl Pending {
    fn due(&self) -> Instant {
        (self.last + QUIET).min(self.first + MAX_DELAY)
    }
}

async fn run(broker: Arc<Broker>, mut events: mpsc::UnboundedReceiver<notify::Result<notify::Event>>) {
    let root = broker.world.actors_root();
    let mut pending: HashMap<String, Pending> = HashMap::new();
    // Actor checks and full passes. Dropping the set (when the host stops) aborts them.
    let mut tasks = tokio::task::JoinSet::new();
    loop {
        let due = pending.values().map(Pending::due).min();
        tokio::select! {
            event = events.recv() => match event {
                Some(Ok(event)) => {
                    if event.need_rescan() {
                        let broker = broker.clone();
                        tasks.spawn(async move {
                            broker.overflowed().await;
                        });
                    }
                    let now = Instant::now();
                    for (path, op, to) in changes(&root, &event) {
                        add(&mut pending, path, op, to, now);
                    }
                }
                Some(Err(e)) => broker.host_error(&format!("file watcher: {e}")),
                None => break,
            },
            _ = sleep_until(due) => {
                let now = Instant::now();
                let ready: Vec<String> = pending.iter().filter(|(_, p)| p.due() <= now).map(|(k, _)| k.clone()).collect();
                for path in ready {
                    let p = pending.remove(&path).unwrap();
                    let op = if p.op == "moved-in" { "create" } else { p.op };
                    broker.notify_watchers(&path, op, p.to.as_deref());
                    if let Some(id) = new_wasm(&path, op, p.to.as_deref()) {
                        tasks.spawn(broker.clone().check_actor(id));
                    }
                }
            }
            Some(_) = tasks.join_next() => {}
        }
    }
}

async fn sleep_until(t: Option<Instant>) {
    match t {
        Some(t) => tokio::time::sleep_until(t).await,
        None => std::future::pending().await,
    }
}

/// The namespace path (`/actor/...`) of a host path, unless it is outside
/// `actor/` or a temporary file of the host.
fn namespace(root: &Path, path: &Path) -> Option<String> {
    let rel = path.strip_prefix(root).ok()?;
    let mut out = String::from("/actor");
    for part in rel.components() {
        let name = part.as_os_str().to_str()?;
        if name.starts_with(TMP_PREFIX) {
            return None;
        }
        out.push('/');
        out.push_str(name);
    }
    Some(out)
}

/// One change: namespace path, op, and the new path of a rename.
type Change = (String, &'static str, Option<String>);

/// The changes of one OS event. inotify reports a rename three times:
/// `From` (the old name), `To` (the new name) and `Both`. `To` becomes the
/// internal op `moved-in`, which `Both` replaces.
fn changes(root: &Path, event: &notify::Event) -> Vec<Change> {
    let path = |i: usize| event.paths.get(i).and_then(|p| namespace(root, p));
    let one = |op: &'static str| -> Vec<Change> { path(0).map(|p| (p, op, None)).into_iter().collect() };
    match event.kind {
        EventKind::Create(_) => one("create"),
        EventKind::Remove(_) => one("remove"),
        EventKind::Modify(ModifyKind::Metadata(_)) => vec![],
        EventKind::Modify(ModifyKind::Name(RenameMode::From)) => one("remove"),
        EventKind::Modify(ModifyKind::Name(RenameMode::To)) => one("moved-in"),
        EventKind::Modify(ModifyKind::Name(RenameMode::Both)) => match (path(0), path(1)) {
            (Some(from), Some(to)) => vec![(from, "rename", Some(to))],
            // Renamed to a temporary name: gone.
            (Some(from), None) => vec![(from, "remove", None)],
            // A temporary file renamed onto `to`: an atomic write.
            (None, Some(to)) => vec![(to, "write", None)],
            (None, None) => vec![],
        },
        EventKind::Modify(_) => one("write"),
        _ => vec![],
    }
}

/// Coalesce a change into `pending`. The last op wins, except that a write
/// after a create stays a create. A rename replaces the `moved-in` of its
/// new name.
fn add(pending: &mut HashMap<String, Pending>, path: String, op: &'static str, to: Option<String>, now: Instant) {
    if let Some(to) = &to
        && pending.get(to).is_some_and(|p| p.op == "moved-in")
    {
        pending.remove(to);
    }
    match pending.get_mut(&path) {
        Some(p) => {
            if !(p.op == "create" && op == "write") {
                p.op = op;
                p.to = to;
            }
            p.last = now;
        }
        None => {
            pending.insert(path, Pending { op, to, first: now, last: now });
        }
    }
}

/// The actor to check when `actor/<id>/actor.wasm` or the folder `actor/<id>`
/// appeared or changed.
fn new_wasm(path: &str, op: &str, to: Option<&str>) -> Option<String> {
    let target = match op {
        "remove" => return None,
        "rename" => to?,
        _ => path,
    };
    let rest = target.strip_prefix("/actor/")?;
    let id = rest.strip_suffix("/actor.wasm").unwrap_or(rest);
    (!id.is_empty() && !id.contains('/')).then(|| id.to_string())
}

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

    #[test]
    fn wasm_and_folder_changes_name_the_actor() {
        assert_eq!(new_wasm("/actor/a/actor.wasm", "write", None).as_deref(), Some("a"));
        assert_eq!(new_wasm("/actor/a", "create", None).as_deref(), Some("a"));
        assert_eq!(new_wasm("/actor/x/y", "rename", Some("/actor/b/actor.wasm")).as_deref(), Some("b"));
        assert_eq!(new_wasm("/actor/a/actor.wasm", "remove", None), None);
        assert_eq!(new_wasm("/actor/a/versions/1/actor.wasm", "write", None), None);
        assert_eq!(new_wasm("/actor/a/public/x", "write", None), None);
        assert_eq!(new_wasm("/actor", "write", None), None);
    }

    #[test]
    fn a_rename_is_one_change_and_temp_files_are_skipped() {
        let root = Path::new("/w/actor");
        let now = Instant::now();
        let mut pending = HashMap::new();
        let ev = |kind, paths: &[&str]| {
            let mut e = notify::Event::new(kind);
            for p in paths {
                e = e.add_path(std::path::PathBuf::from(*p));
            }
            e
        };
        let name = |mode| EventKind::Modify(ModifyKind::Name(mode));
        for e in [
            ev(name(RenameMode::From), &["/w/actor/a/public/x"]),
            ev(name(RenameMode::To), &["/w/actor/a/public/y"]),
            ev(name(RenameMode::Both), &["/w/actor/a/public/x", "/w/actor/a/public/y"]),
            ev(EventKind::Create(notify::event::CreateKind::File), &["/w/actor/a/.sausage-tmp-host-1-actor.json"]),
            ev(name(RenameMode::Both), &["/w/actor/a/.sausage-tmp-host-1-actor.json", "/w/actor/a/actor.json"]),
            ev(EventKind::Create(notify::event::CreateKind::File), &["/w/.lock"]),
        ] {
            for (path, op, to) in changes(root, &e) {
                add(&mut pending, path, op, to, now);
            }
        }
        let mut got: Vec<(String, &str, Option<String>)> =
            pending.into_iter().map(|(k, p)| (k, p.op, p.to)).collect();
        got.sort();
        assert_eq!(
            got,
            vec![
                ("/actor/a/actor.json".to_string(), "write", None),
                ("/actor/a/public/x".to_string(), "rename", Some("/actor/a/public/y".to_string())),
            ]
        );
    }
}