sandfs 0.9.0

//! The one background process that serves every mount of a home.
//!
//! `mount` and `fork` ask it to mount a workspace over a Unix socket that only its
//! owner can reach, and start it first when none is running. Each mount runs on
//! threads of its own, and workspaces over the same commit share it (see [`Bases`]).
//! The process exits once its last mount is gone, so nothing stays running.
//! `update` asks it to move a workspace it serves onto a newer source in place, and
//! `trim` to let go of memory its mounts can do without; a mount left unused for a
//! minute is trimmed anyway.

use std::collections::HashMap;
use std::fs::File;
use std::io::{BufRead, BufReader, ErrorKind, Write};
use std::os::unix::ffi::OsStrExt;
use std::os::unix::fs::{DirBuilderExt, PermissionsExt};
use std::os::unix::net::{UnixListener, UnixStream};
use std::path::{Path, PathBuf};
use std::process::Stdio;
use std::sync::{Arc, Condvar, Mutex, OnceLock, PoisonError, Weak, mpsc};
use std::time::{Duration, Instant};

use anyhow::{Context, Result, anyhow, bail};
use nix::sys::signal::kill;
use nix::unistd::{Pid, setsid};
use serde::{Deserialize, Serialize};

use crate::Trimmed;
use crate::fs::{Replacement, Shared};
use crate::store::{Base, Bases, Home, Lease, ROOT, Record};
use crate::update::{self, Updated};

/// How long the process stays after its last mount, so a mount right after finds it.
const LINGER: Duration = Duration::from_secs(2);
/// How long a new process may take to start answering.
const START: Duration = Duration::from_secs(10);
/// How long a mount may take, which includes fetching a remote commit's folders.
const MOUNT: Duration = Duration::from_secs(120);
/// How long a mount goes unused before it is trimmed (see [`trim`]).
const IDLE: Duration = Duration::from_secs(60);
/// The memory maps kept free when taking another mount, for what mounts already served
/// need: threads that answer requests, large reads (each read of a large file maps
/// memory of its own) and git helpers. A mount takes about 10.
#[cfg(target_os = "linux")]
const MAPS_KEPT: usize = 1024;

/// A request to mount the workspace `name` at `dir`, which exists and is canonical,
/// to move the mounted workspace `name` onto a newer source, or to trim the mount of
/// `name` (every mount when it is empty). The directory goes as its bytes, since a
/// path needn't be UTF-8.
#[derive(Serialize, Deserialize)]
struct Request {
    name: String,
    dir: Vec<u8>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    update: Option<Move>,
    #[serde(default, skip_serializing_if = "std::ops::Not::not")]
    trim: bool,
}

/// The source to move a workspace onto, found by the command that asked, and whether
/// it was named (see [`update::apply`]).
#[derive(Serialize, Deserialize)]
struct Move {
    next: Record,
    given: bool,
}

impl Request {
    fn dir(&self) -> PathBuf {
        PathBuf::from(std::ffi::OsStr::from_bytes(&self.dir))
    }
}

/// The answer, once the mount is up or has failed, the workspace has moved, or the
/// mounts are trimmed.
#[derive(Serialize, Deserialize)]
struct Reply {
    error: Option<String>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    updated: Option<Updated>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    trimmed: Option<Trimmed>,
}

/// Where the background process writes what it reports.
pub fn log_path(home: &Home) -> PathBuf {
    home.run_dir().join("server.log")
}

fn socket_path(home: &Home) -> PathBuf {
    home.run_dir().join("server.sock")
}

/// The directory holding the socket, private to its owner, so nobody else can ask for
/// mounts.
fn run_dir(home: &Home) -> Result<PathBuf> {
    let dir = home.run_dir();
    std::fs::DirBuilder::new().recursive(true).mode(0o700).create(&dir)?;
    std::fs::set_permissions(&dir, std::fs::Permissions::from_mode(0o700))?;
    Ok(dir)
}

/// Mount `name` at `dir` in the home's background process, starting the process if
/// none is running, and wait until the mount is up.
pub fn mount(home: &Home, name: &str, dir: &Path) -> Result<()> {
    let dir = dir.as_os_str().as_bytes().to_vec();
    let request = serde_json::to_string(&Request { name: name.to_owned(), dir, update: None, trim: false })?;
    let started = Instant::now();
    let mut server: Option<Pid> = None;
    // A running process answers at once, and a new one within a few milliseconds.
    let mut wait = Duration::from_millis(1);
    loop {
        // A process on its way out closes connections without answering, and then
        // another is started.
        if let Ok(stream) = UnixStream::connect(socket_path(home))
            && let Some(reply) = ask(stream, &request)?
        {
            return reply.error.map_or(Ok(()), |err| Err(anyhow!(err)));
        }
        // Start a process, unless the one started before may still be coming up.
        if !server.is_some_and(|pid| kill(pid, None).is_ok()) {
            server = Some(start(home)?);
        }
        if started.elapsed() > START {
            bail!("the mount process did not start; see {}", log_path(home).display());
        }
        std::thread::sleep(wait);
        wait = (wait * 2).min(Duration::from_millis(50));
    }
}

/// Have the process serving the mounted workspace `name` move it onto `next`, which
/// names it from now on if `given`.
pub fn update(home: &Home, name: &str, next: &Record, given: bool) -> Result<Updated> {
    let update = Move { next: next.clone(), given };
    let request =
        serde_json::to_string(&Request { name: name.to_owned(), dir: Vec::new(), update: Some(update), trim: false })?;
    let unserved =
        || anyhow!("{name} is mounted by a process that can't move it; unmount it first (sandfs unmount {name})");
    let stream = UnixStream::connect(socket_path(home)).map_err(|_| unserved())?;
    let reply = ask(stream, &request)?.ok_or_else(unserved)?;
    match (reply.error, reply.updated) {
        (Some(err), _) => Err(anyhow!(err)),
        (None, Some(updated)) => Ok(updated),
        (None, None) => Err(unserved()),
    }
}

/// Have the home's process save the changes of the mounted workspace `name` (every
/// workspace it serves, when empty), then let go of what those mounts keep that they
/// can get again, and hand the memory freed back to the system. Nothing they serve
/// changes. With no process running, nothing is mounted to trim.
pub fn trim(home: &Home, name: &str) -> Result<Trimmed> {
    let request = serde_json::to_string(&Request { name: name.to_owned(), dir: Vec::new(), update: None, trim: true })?;
    let unserved = || anyhow!("{name} isn't mounted by this home's background process");
    let Ok(stream) = UnixStream::connect(socket_path(home)) else {
        return if name.is_empty() { Ok(Trimmed::default()) } else { Err(unserved()) };
    };
    match ask(stream, &request)? {
        Some(Reply { error: Some(err), .. }) => Err(anyhow!(err)),
        Some(Reply { trimmed: Some(trimmed), .. }) => Ok(trimmed),
        Some(_) => Err(anyhow!("the background process can't trim; unmount and mount again to update it")),
        None if name.is_empty() => Ok(Trimmed::default()),
        None => Err(unserved()),
    }
}

/// Trim the mount of `name`, or every mount this process serves when it is empty.
fn trim_served(name: &str) -> Result<Trimmed> {
    let served: Vec<Arc<Served>> = served_list()
        .iter()
        .filter(|(served, _)| name.is_empty() || *served == name)
        .filter_map(|(_, served)| served.upgrade())
        .collect();
    if !name.is_empty() && served.is_empty() {
        bail!("{name} isn't mounted by this home's background process");
    }
    let mut failed = None;
    for mount in &served {
        if let Err(err) = mount.shared.trim() {
            failed.get_or_insert(err);
        }
    }
    let given_back = crate::memory::give_back().map(|bytes| bytes as u64);
    match failed {
        Some(err) => Err(err.context("saving before trimming")),
        None => Ok(Trimmed { mounts: served.len(), given_back }),
    }
}

/// Trim the mounts no one has asked anything for `idle`, once each until used again.
fn trim_idle(idle: Duration) {
    let idle: Vec<Arc<Served>> =
        served_list().values().filter_map(Weak::upgrade).filter(|served| served.shared.idle_untrimmed(idle)).collect();
    if idle.is_empty() {
        return;
    }
    for mount in &idle {
        if let Err(err) = mount.shared.trim() {
            eprintln!("sandfs: trimming an unused mount: {err:#}");
        }
    }
    crate::memory::give_back();
    eprintln!("sandfs: trimmed {} mount(s) unused for {}s", idle.len(), IDLE.as_secs());
}

/// Send `request` and read the reply, or `None` if the connection closes first.
fn ask(mut stream: UnixStream, request: &str) -> Result<Option<Reply>> {
    stream.set_read_timeout(Some(MOUNT))?;
    if stream.write_all(format!("{request}\n").as_bytes()).is_err() {
        return Ok(None);
    }
    let mut line = String::new();
    match BufReader::new(stream).read_line(&mut line) {
        Ok(0) => Ok(None),
        Ok(_) => Ok(Some(serde_json::from_str(&line).context("reading the mount process's answer")?)),
        Err(err) if matches!(err.kind(), ErrorKind::WouldBlock | ErrorKind::TimedOut) => {
            bail!("timed out waiting for the mount")
        }
        Err(err) if err.kind() == ErrorKind::ConnectionReset => Ok(None),
        Err(err) => Err(err.into()),
    }
}

/// Start the background process and return its pid.
///
/// The process serves every mount of the home, whoever asked for it, so it shares as
/// little as it can with the command that happens to start it: it runs in a session of
/// its own (no terminal, no process group to be signalled with), and its parent exits
/// at once so it is adopted by init, launchd or the nearest subreaper. A job runner
/// that ends a job by killing its process group or the job's process tree then leaves
/// the other jobs' mounts alone. It keeps the caller's environment, which carries what
/// git needs to fetch (credentials, proxies, helpers), except what points git at the
/// caller's own repository.
fn start(home: &Home) -> Result<Pid> {
    run_dir(home)?;
    let log = std::fs::OpenOptions::new().create(true).append(true).open(log_path(home))?;
    let mut command = crate::relaunch::command()?;
    for (name, _) in std::env::vars_os() {
        if name.to_str().is_some_and(names_a_repository) {
            command.env_remove(name);
        }
    }
    // glibc's allocator raises the size it maps directly once a large block is freed,
    // so after the first mount ends, each new mount's 16 MiB request buffer comes
    // zeroed from memory the process keeps, and a process outliving its mounts grew by
    // hundreds of MiB as they came and went. A fixed threshold, above any one read or
    // write, keeps those buffers mapped (and untouched, so free) and lets freed memory
    // go back.
    #[cfg(target_os = "linux")]
    if std::env::var_os("MALLOC_MMAP_THRESHOLD_").is_none() {
        command.env("MALLOC_MMAP_THRESHOLD_", "4194304");
    }
    // `daemon` starts the server in a new session, prints its pid and exits.
    let output = command
        .arg("daemon")
        // Started from inside a mount, it would otherwise keep that mount busy.
        .current_dir("/")
        .env("SANDFS_HOME", home.root())
        .stdin(Stdio::null())
        .stdout(Stdio::piped())
        .stderr(log)
        .output()
        .context("starting the mount process")?;
    let pid = String::from_utf8_lossy(&output.stdout).trim().parse::<i32>().ok();
    match pid {
        Some(pid) if output.status.success() => Ok(Pid::from_raw(pid)),
        _ => bail!("the mount process did not start; see {}", log_path(home).display()),
    }
}

/// Whether `name` is one of the variables git sets or reads to find the repository it
/// works in (`git rev-parse --local-env-vars`, less the configuration ones), which a
/// caller running inside a repository or a git hook may carry. Kept, they would send
/// every later mount's git commands to that caller's repository.
fn names_a_repository(name: &str) -> bool {
    const NAMES: &[&str] = &[
        "GIT_DIR",
        "GIT_WORK_TREE",
        "GIT_IMPLICIT_WORK_TREE",
        "GIT_COMMON_DIR",
        "GIT_INDEX_FILE",
        "GIT_OBJECT_DIRECTORY",
        "GIT_ALTERNATE_OBJECT_DIRECTORIES",
        "GIT_GRAFT_FILE",
        "GIT_SHALLOW_FILE",
        "GIT_NO_REPLACE_OBJECTS",
        "GIT_REPLACE_REF_BASE",
        "GIT_NAMESPACE",
        "GIT_PREFIX",
    ];
    NAMES.contains(&name)
}

/// Start the server in a session of its own, print its pid and return, leaving it to
/// init, launchd or the nearest subreaper: `sandfs daemon`, run by [`start`].
pub fn detach() -> Result<()> {
    // Not a process group leader, since `start` gives this process none of its own,
    // so this cannot fail with EPERM.
    setsid().context("leaving the caller's session")?;
    let server = crate::relaunch::command()?
        .args(["daemon", "--serve"])
        .stdin(Stdio::null())
        .stdout(Stdio::null())
        .spawn()
        .context("starting the mount process")?;
    println!("{}", server.id());
    Ok(())
}

/// Serve mounts until none is left: the background process itself.
pub fn run(home: Home) -> Result<()> {
    let dir = run_dir(&home)?;
    let lock = File::options().write(true).create(true).truncate(false).open(dir.join("server.lock"))?;
    // One process serves the home. One that is on its way out holds the lock for a
    // moment longer; one still running keeps it, and this one leaves.
    let deadline = Instant::now() + Duration::from_secs(1);
    while lock.try_lock().is_err() {
        if Instant::now() >= deadline {
            return Ok(());
        }
        std::thread::sleep(Duration::from_millis(10));
    }
    // Whatever socket is left belongs to a process that has exited.
    let socket = socket_path(&home);
    let _ = std::fs::remove_file(&socket);
    let listener = UnixListener::bind(&socket).context("listening for mounts")?;
    crate::host::raise_open_files();
    let state = Arc::new(State::default());
    let bases = Arc::new(Bases::default());
    let accepting = state.clone();
    std::thread::spawn(move || {
        for stream in listener.incoming() {
            let Ok(stream) = stream else {
                std::thread::sleep(Duration::from_millis(10));
                continue;
            };
            // Closing: the connection drops unanswered, and its client starts a new process.
            let Some(ticket) = accepting.enter() else { break };
            let (home, bases) = (home.clone(), bases.clone());
            std::thread::spawn(move || answer(&stream, &home, &bases, ticket));
        }
    });
    std::thread::spawn(|| {
        loop {
            std::thread::sleep(IDLE / 4);
            trim_idle(IDLE);
        }
    });
    state.wait_until_idle(LINGER);
    // The socket stays: connecting to it fails once this process is gone, and the next
    // process replaces it. Removing it here could remove a successor's instead, when
    // the home was deleted and made again meanwhile, since the lock went with the old
    // home.
    Ok(())
}

/// Read one request from `stream`, start the mount and answer once it is up or failed,
/// or move the workspace and answer once it has moved.
fn answer(mut stream: &UnixStream, home: &Home, bases: &Arc<Bases>, ticket: Ticket) {
    let mut request = read_request(stream);
    if let Ok(Request { name, trim: true, .. }) = &request {
        let reply = match trim_served(name) {
            Ok(trimmed) => Reply { error: None, updated: None, trimmed: Some(trimmed) },
            Err(err) => Reply { error: Some(format!("{err:#}")), updated: None, trimmed: None },
        };
        if let Ok(reply) = serde_json::to_string(&reply) {
            let _ = stream.write_all(format!("{reply}\n").as_bytes());
        }
        return;
    }
    if let Ok(Request { name, update: Some(next), .. }) = &mut request {
        let reply = match move_served(home, bases, name, next) {
            Ok(updated) => Reply { error: None, updated: Some(updated), trimmed: None },
            Err(err) => Reply { error: Some(format!("{err:#}")), updated: None, trimmed: None },
        };
        if let Ok(reply) = serde_json::to_string(&reply) {
            let _ = stream.write_all(format!("{reply}\n").as_bytes());
        }
        return;
    }
    let dir = request.as_ref().ok().map(Request::dir);
    let mounted = request.and_then(|request| start_mount(home, bases, request, ticket));
    let up = mounted.is_ok();
    let reply = Reply { error: mounted.err().map(|err| format!("{err:#}")), updated: None, trimmed: None };
    let told =
        serde_json::to_string(&reply).is_ok_and(|reply| stream.write_all(format!("{reply}\n").as_bytes()).is_ok());
    // A requester that gave up (timed out, or was interrupted) has reported a failure,
    // so a mount that came up after all is taken down again.
    if let Some(dir) = dir.filter(|_| up && !told) {
        eprintln!("sandfs: nobody waited for the mount at {}; unmounting it", dir.display());
        let _ = crate::host::unmount(&dir, false);
    }
}

fn read_request(stream: &UnixStream) -> Result<Request> {
    stream.set_read_timeout(Some(Duration::from_secs(5)))?;
    let mut line = String::new();
    BufReader::new(stream).read_line(&mut line)?;
    serde_json::from_str(&line).context("reading a mount request")
}

/// Serve the requested mount on a thread of its own, which keeps `ticket` for as long
/// as it serves, and wait until the mount is up or has failed.
fn start_mount(home: &Home, bases: &Arc<Bases>, request: Request, ticket: Ticket) -> Result<()> {
    // Each mount's threads take memory maps, and a process that runs out of them when
    // starting a thread aborts, taking every mount it serves with it. A mount it has no
    // room for is refused instead.
    #[cfg(target_os = "linux")]
    if let Some((left, max)) = crate::host::memory_maps_left(MAPS_KEPT)
        && left < MAPS_KEPT
    {
        bail!(
            "this process serves as many mounts as its memory maps allow (vm.max_map_count = {max}); \
             unmount some, or raise the limit, such as with `sysctl -w vm.max_map_count={}`",
            max * 4
        );
    }
    let (done, outcome) = mpsc::channel();
    let up = done.clone();
    let (home, bases) = (home.clone(), bases.clone());
    std::thread::Builder::new()
        .name(format!("mount {}", request.name))
        .spawn(move || {
            let _ticket = ticket;
            let served = crate::serve(&home, &request.name, &request.dir(), &bases, move || {
                let _ = up.send(Ok(()));
            });
            if let Err(err) = &served {
                eprintln!("sandfs: {}: {err:#}", request.name);
            }
            // What the mount took is free now: hand it back to the system.
            crate::memory::give_back();
            // A mount that never came up reports why; for one that did, this goes unread.
            let _ = done.send(served);
        })
        .context("starting a mount")?;
    outcome.recv().unwrap_or_else(|_| Err(anyhow!("the mount stopped unexpectedly")))
}

/// A mount this process serves, as `update` needs it.
pub struct Served {
    shared: Arc<Shared>,
    /// The mount's use of its base, which moves to the newer base with it.
    lease: Mutex<Lease>,
    /// Held while the workspace moves, so it moves once at a time.
    moving: Mutex<()>,
    /// The copies of folders it moved off, held for files it may still have open there
    /// (see [`Home::hold_copy`]).
    retired: Mutex<Vec<File>>,
}

/// A mount listed among the ones this process serves, until dropped.
pub struct Serving {
    name: String,
    served: Arc<Served>,
}

/// List the mount of `name`, served through `shared` over the base of `lease`.
pub fn serving(name: &str, shared: Arc<Shared>, lease: Lease) -> Serving {
    let served =
        Arc::new(Served { shared, lease: Mutex::new(lease), moving: Mutex::new(()), retired: Mutex::default() });
    let mut all = served_list();
    all.insert(name.to_owned(), Arc::downgrade(&served));
    Serving { name: name.to_owned(), served }
}

/// The mounts this process serves, by workspace name, until each one's [`Serving`]
/// is dropped.
fn served_list() -> std::sync::MutexGuard<'static, HashMap<String, Weak<Served>>> {
    static SERVED: OnceLock<Mutex<HashMap<String, Weak<Served>>>> = OnceLock::new();
    SERVED.get_or_init(Mutex::default).lock().unwrap_or_else(PoisonError::into_inner)
}

impl Serving {
    /// Stop listing the mount, which has ended, and delete the copies of folders it
    /// moved off that nothing else uses, now that it reads none of their files.
    pub fn end(self, home: &Home) {
        let retired = std::mem::take(&mut *self.served.retired.lock().unwrap_or_else(PoisonError::into_inner));
        drop(self);
        if retired.is_empty() {
            return;
        }
        drop(retired);
        // Not waited for: deleting waits on whoever is making a workspace.
        let home = home.clone();
        std::thread::spawn(move || {
            if let Ok(_creating) = home.repos_lock(true) {
                let _ = home.remove_unused_copies();
            }
        });
    }
}

impl Drop for Serving {
    fn drop(&mut self) {
        let mut all = served_list();
        // Unless the workspace was mounted again since, by a mount listed in its place.
        if all.get(&self.name).is_some_and(|listed| std::ptr::eq(listed.as_ptr(), Arc::as_ptr(&self.served))) {
            all.remove(&self.name);
        }
    }
}

/// Move the workspace `name`, mounted here, onto `next` while it stays mounted: its
/// changes are saved, the workspace on disk moved as when unmounted, and the mount
/// serves it from then on, the kernel told to forget what changed.
fn move_served(home: &Home, bases: &Arc<Bases>, name: &str, next: &Move) -> Result<Updated> {
    let served = served_list().get(name).and_then(Weak::upgrade);
    let served = served.with_context(|| {
        format!("{name} is mounted by a process that can't move it; unmount it first (sandfs unmount {name})")
    })?;
    let _moving = served.moving.lock().unwrap_or_else(PoisonError::into_inner);
    // Reading the sources: `sandfs clean` waits.
    let _repos = home.repos_lock(false)?;
    let record = home.record(name)?;
    let before = serde_json::to_value(&record)?;
    let mut updated = None;
    served.shared.replace(|| {
        let moved = update::apply(home, name, next.next.clone(), next.given);
        let failed = moved.is_err();
        let changed = moved.as_ref().ok().and_then(|moved| moved.changed.clone());
        updated = Some(moved);
        // A move that failed before changing anything leaves the mount as it is.
        if failed && serde_json::to_value(home.record(name)?)? == before {
            return Ok(None);
        }
        // A copy of a folder moved off stays while files of it may be open here.
        let moved_off = match &record.base {
            Base::Dir { path } if home.holds_copy(path) && home.record(name)?.base != record.base => {
                Some(home.hold_copy(path)?)
            }
            _ => None,
        };
        let (opened, lease) = home.open_shared(name, bases)?;
        *served.lease.lock().unwrap_or_else(PoisonError::into_inner) = lease;
        served.retired.lock().unwrap_or_else(PoisonError::into_inner).extend(moved_off);
        let full = |rel: &str| format!("{ROOT}/{rel}");
        let git = matches!(opened.record.base, Base::Git { .. });
        Ok(Some(Replacement {
            times: crate::served_times(&opened),
            ws: opened.ws,
            saved: opened.saved,
            dirs: opened.restored.dirs,
            changed: changed.map(|changed| changed.iter().map(|rel| full(rel)).collect()),
            // What git keeps follows the commit, which a source's changes don't list.
            whole: if git { vec![sandfs_core::VPath::parse(&full(".git"))?] } else { Vec::new() },
        }))
    })?;
    updated.unwrap_or_else(|| Err(anyhow!("the mount stopped before {name} moved")))
}

/// How many mounts, and requests for one, the process is serving, and whether it has
/// stopped taking more.
#[derive(Default)]
struct State {
    busy: Mutex<(usize, bool)>,
    changed: Condvar,
}

/// One mount or request, counted until dropped.
struct Ticket(Arc<State>);

impl State {
    /// Count a new request, unless the process is closing.
    fn enter(self: &Arc<Self>) -> Option<Ticket> {
        let mut busy = self.busy.lock().unwrap_or_else(std::sync::PoisonError::into_inner);
        if busy.1 {
            return None;
        }
        busy.0 += 1;
        Some(Ticket(self.clone()))
    }

    /// Wait until nothing has been served for `linger`, then stop taking requests.
    fn wait_until_idle(&self, linger: Duration) {
        let mut busy = self.busy.lock().unwrap_or_else(std::sync::PoisonError::into_inner);
        loop {
            if busy.0 > 0 {
                busy = self.changed.wait_while(busy, |busy| busy.0 > 0).unwrap_or_else(|err| err.into_inner());
                continue;
            }
            let (guard, idle) = self
                .changed
                .wait_timeout_while(busy, linger, |busy| busy.0 == 0)
                .unwrap_or_else(|err| err.into_inner());
            busy = guard;
            if idle.timed_out() {
                busy.1 = true;
                return;
            }
        }
    }
}

impl Drop for Ticket {
    fn drop(&mut self) {
        let mut busy = self.0.busy.lock().unwrap_or_else(std::sync::PoisonError::into_inner);
        busy.0 -= 1;
        self.0.changed.notify_all();
    }
}

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

    #[test]
    fn requests_carry_any_directory_name() {
        let dir = PathBuf::from(std::ffi::OsStr::from_bytes(b"/tmp/caf\xe9"));
        let request =
            Request { name: "ws".into(), dir: dir.as_os_str().as_bytes().to_vec(), update: None, trim: false };
        let sent: Request = serde_json::from_str(&serde_json::to_string(&request).unwrap()).unwrap();
        assert_eq!(sent.dir(), dir);
    }

    #[test]
    fn the_process_closes_only_after_lingering_idle() {
        let state = Arc::new(State::default());
        let ticket = state.enter().unwrap();
        let waiter = {
            let state = state.clone();
            std::thread::spawn(move || {
                state.wait_until_idle(Duration::from_millis(50));
                Instant::now()
            })
        };
        std::thread::sleep(Duration::from_millis(100));
        assert!(!waiter.is_finished(), "a mount keeps the process");
        // Timed from the drop, not from when the thread started, which a busy machine
        // can delay.
        let idle = Instant::now();
        drop(ticket);
        assert!(waiter.join().unwrap() >= idle + Duration::from_millis(50), "it lingers once idle");
        assert!(state.enter().is_none(), "a closed process takes no more requests");
    }
}