char/avril

git clone https://git.t4t.associates/char/avril

Charlotte Somrearrange libs into lib and extensions into extsf4a0f5b2026-10-19

use super::paths::{resolve, tree};
use super::*;

use std::collections::HashSet;
use tokio::sync::Notify;

const MAX_ROOTS: usize = 128;
const MAX_QUEUED_EVENTS: usize = 256;
const MAX_QUEUED_PATHS: usize = 4096;

const CREATE: u8 = 0;
const MODIFY: u8 = 1;
const REMOVE: u8 = 2;
const RENAME: u8 = 3;
const RESCAN: u8 = 4;

type WatchEvent = (u8, Vec<String>);

#[derive(Clone)]
struct Root {
    target: PathTarget,
    path: PathBuf,
    directory: bool,
    recursive: bool,
    // A file root is watched through its parent directory, so replacing the
    // file (the usual atomic save) or recreating it stays observable at the
    // same path instead of following the old inode.
    watch_path: PathBuf,
}

#[derive(Clone, Debug, PartialEq, Eq)]
struct QueuedEvent {
    kind: u8,
    paths: Vec<PathBuf>,
    ends: Vec<usize>,
}

#[derive(Default)]
struct WatchState {
    queue: VecDeque<QueuedEvent>,
    closed: bool,
    terminal: Option<(io::ErrorKind, String)>,
    lossy: bool,
    live: Vec<bool>,
    // Root endings discarded with an overflowed queue, applied when the
    // replacement rescan is delivered instead.
    pending_ends: Vec<usize>,
    epoch: u32,
}

struct Shared {
    caps: Arc<Capabilities>,
    roots: Vec<Root>,
    state: Mutex<WatchState>,
    notify: Notify,
}

struct WatchResource {
    shared: Arc<Shared>,
    #[cfg_attr(
        not(any(
            target_os = "linux",
            target_os = "android",
            target_os = "macos",
            target_os = "windows"
        )),
        allow(dead_code)
    )]
    native: Mutex<Option<notify::RecommendedWatcher>>,
}

impl WatchResource {
    fn close(&self) {
        {
            let mut state = self.shared.state.lock().unwrap();
            state.closed = true;
            state.queue.clear();
            state.lossy = false;
            state.epoch = state.epoch.wrapping_add(1);
        }
        self.shared.notify.notify_waiters();
        let native = self.native.lock().unwrap().take();
        drop(native);
        self.shared.notify.notify_waiters();
    }
}

struct Watchers {
    caps: Arc<Capabilities>,
    resources: Mutex<HashMap<u64, Arc<WatchResource>>>,
    next: AtomicU64,
}

impl Drop for Watchers {
    fn drop(&mut self) {
        for resource in self
            .resources
            .get_mut()
            .unwrap()
            .drain()
            .map(|(_, resource)| resource)
        {
            resource.close();
        }
    }
}

impl Watchers {
    fn resource(&self, id: u64) -> io::Result<Arc<WatchResource>> {
        self.resources
            .lock()
            .unwrap()
            .get(&id)
            .cloned()
            .ok_or_else(bad_resource)
    }

    fn create(&self, labels: Vec<String>, recursive: bool) -> io::Result<u64> {
        if labels.is_empty() || labels.len() > MAX_ROOTS {
            return Err(io::Error::new(
                io::ErrorKind::InvalidInput,
                "Expected between one and 128 filesystem watch paths",
            ));
        }
        let roots = roots(&self.caps, labels, recursive)?;
        let live = vec![true; roots.len()];
        let shared = Arc::new(Shared {
            caps: self.caps.clone(),
            roots,
            state: Mutex::new(WatchState { live, ..WatchState::default() }),
            notify: Notify::new(),
        });
        let native = register(&shared)?;
        check_roots_unchanged(&shared)?;
        let id = self
            .next
            .fetch_update(Ordering::AcqRel, Ordering::Acquire, |id| {
                (id <= super::MAX_ID).then_some(id + 1)
            })
            .map_err(|_| io::Error::other("Filesystem watcher identifiers exhausted"))?;
        self.resources.lock().unwrap().insert(
            id,
            Arc::new(WatchResource {
                shared,
                native: Mutex::new(Some(native)),
            }),
        );
        Ok(id)
    }

    fn close(&self, id: u64, quiet: bool) -> io::Result<bool> {
        let resource = self.resources.lock().unwrap().remove(&id);
        match resource {
            Some(resource) => {
                resource.close();
                Ok(true)
            }
            None if quiet => Ok(true),
            None => Err(bad_resource()),
        }
    }
}

fn roots(caps: &Capabilities, labels: Vec<String>, recursive: bool) -> io::Result<Vec<Root>> {
    let mut roots = Vec::new();
    let mut seen = HashSet::new();
    for label in labels {
        let target = resolve(caps, &label, false, false)?;
        let path = target.as_path().to_owned();
        if !seen.insert(path.clone()) {
            continue;
        }
        let (directory, name) = parent(&path)?;
        let metadata = directory.symlink_metadata(&name)?;
        if metadata.is_symlink() {
            return Err(io::Error::new(
                io::ErrorKind::WouldBlock,
                "Filesystem watch root changed while it was registered",
            ));
        }
        let (directory, watch_path) = if metadata.is_dir() {
            (true, path.clone())
        } else if metadata.is_file() {
            let parent = path.parent().ok_or_else(|| {
                io::Error::new(io::ErrorKind::InvalidInput, "Watched file has no parent")
            })?;
            (false, parent.to_owned())
        } else {
            return Err(io::Error::new(
                io::ErrorKind::Unsupported,
                "Only regular files and directories can be watched",
            ));
        };
        if directory && recursive {
            tree(caps, &target, &label, false)?;
        }
        roots.push(Root {
            target,
            path,
            directory,
            recursive: directory && recursive,
            watch_path,
        });
    }
    Ok(roots)
}

fn check_roots_unchanged(shared: &Shared) -> io::Result<()> {
    for root in &shared.roots {
        let current = PathTarget::for_creation(&root.path).map_err(|_| {
            io::Error::new(
                io::ErrorKind::WouldBlock,
                "Filesystem watch root changed while it was registered",
            )
        })?;
        if current.as_path() != root.path {
            return Err(io::Error::new(
                io::ErrorKind::WouldBlock,
                "Filesystem watch root changed while it was registered",
            ));
        }
        let (directory, name) = parent(&root.path)?;
        let metadata = directory.symlink_metadata(&name)?;
        if metadata.is_symlink() || metadata.is_dir() != root.directory {
            return Err(io::Error::new(
                io::ErrorKind::WouldBlock,
                "Filesystem watch root changed while it was registered",
            ));
        }
    }
    Ok(())
}

#[cfg(any(
    target_os = "linux",
    target_os = "android",
    target_os = "macos",
    target_os = "windows"
))]
fn register(shared: &Arc<Shared>) -> io::Result<notify::RecommendedWatcher> {
    use notify::Watcher;

    let weak = Arc::downgrade(shared);
    let handler = move |result: notify::Result<notify::Event>| {
        if let Some(shared) = weak.upgrade() {
            shared.handle(result);
        }
    };
    let config = notify::Config::default().with_follow_symlinks(false);
    let mut watcher =
        <notify::RecommendedWatcher as Watcher>::new(handler, config).map_err(watch_error)?;
    // Roots can share a registration path (a watched file and its parent
    // directory); the recursive registration must win when both are present.
    let mut registrations: HashMap<PathBuf, bool> = HashMap::new();
    for root in &shared.roots {
        let recursive = root.directory && root.recursive;
        registrations
            .entry(root.watch_path.clone())
            .and_modify(|existing| *existing |= recursive)
            .or_insert(recursive);
    }
    for (path, recursive) in registrations {
        let mode = if recursive {
            notify::RecursiveMode::Recursive
        } else {
            notify::RecursiveMode::NonRecursive
        };
        watcher.watch(&path, mode).map_err(watch_error)?;
    }
    Ok(watcher)
}

#[cfg(not(any(
    target_os = "linux",
    target_os = "android",
    target_os = "macos",
    target_os = "windows"
)))]
fn register(_shared: &Arc<Shared>) -> io::Result<notify::RecommendedWatcher> {
    Err(io::Error::new(
        io::ErrorKind::Unsupported,
        "Filesystem watching is not supported on this platform",
    ))
}

fn watch_error(error: notify::Error) -> io::Error {
    let message = error.to_string();
    match error.kind {
        notify::ErrorKind::PathNotFound | notify::ErrorKind::WatchNotFound => {
            io::Error::new(io::ErrorKind::NotFound, message)
        }
        notify::ErrorKind::Io(inner) => io::Error::new(inner.kind(), message),
        notify::ErrorKind::InvalidConfig(_) => io::Error::new(io::ErrorKind::InvalidInput, message),
        notify::ErrorKind::MaxFilesWatch => io::Error::new(io::ErrorKind::WouldBlock, message),
        notify::ErrorKind::Generic(_) => io::Error::other(message),
    }
}

impl Shared {
    fn handle(self: &Arc<Self>, result: notify::Result<notify::Event>) {
        match result {
            Ok(event) => {
                if event.need_rescan() {
                    self.enqueue(QueuedEvent {
                        kind: RESCAN,
                        paths: self.event_roots(&event.paths),
                        ends: Vec::new(),
                    });
                    return;
                }
                let Some(kind) = map_kind(&event.kind, event.paths.len()) else {
                    return;
                };
                if event.paths.is_empty() {
                    return;
                }
                let mut paths = Vec::new();
                let mut unknown = false;
                for path in event.paths {
                    if self.root_index(&path).is_some() || kind == RENAME {
                        paths.push(path);
                    } else if self.in_scope(&path) {
                        // The backend also sees siblings of a watched file and,
                        // on some platforms, descendants of a non-recursive
                        // directory. Those are in scope but not reportable.
                        continue;
                    } else {
                        unknown = true;
                    }
                }
                if unknown && kind != RENAME {
                    self.enqueue(QueuedEvent {
                        kind: RESCAN,
                        paths: self.roots.iter().map(|root| root.path.clone()).collect(),
                        ends: Vec::new(),
                    });
                    return;
                }
                if paths.is_empty() {
                    return;
                }
                let ends = self.event_ends(kind, &paths);
                self.enqueue(QueuedEvent { kind, paths, ends });
            }
            Err(error) => {
                let error = watch_error(error);
                let mut state = self.state.lock().unwrap();
                if !state.closed && state.terminal.is_none() {
                    state.terminal = Some((error.kind(), error.to_string()));
                    drop(state);
                    self.notify.notify_waiters();
                }
            }
        }
    }

    fn event_roots(&self, paths: &[PathBuf]) -> Vec<PathBuf> {
        let mut roots = Vec::new();
        for path in paths {
            if let Some(index) = self.root_index(path) {
                let root = &self.roots[index];
                if !roots.contains(&root.path) {
                    roots.push(root.path.clone());
                }
            }
        }
        if roots.is_empty() {
            roots = self.roots.iter().map(|root| root.path.clone()).collect();
        }
        roots
    }

    fn event_ends(&self, kind: u8, paths: &[PathBuf]) -> Vec<usize> {
        let mut ends = Vec::new();
        if kind != REMOVE && kind != RENAME {
            return ends;
        }
        for (index, root) in self.roots.iter().enumerate() {
            if !root.directory {
                // A watched file is anchored at its path, not its inode, and
                // may be removed and recreated without ending the watch.
                continue;
            }
            let ends_at_root = if kind == REMOVE {
                paths.iter().any(|path| path == &root.path)
            } else {
                paths.first() == Some(&root.path)
            };
            if ends_at_root {
                ends.push(index);
            }
        }
        ends
    }

    fn root_index(&self, path: &Path) -> Option<usize> {
        self.roots
            .iter()
            .enumerate()
            .filter(|(_, root)| {
                if !root.directory {
                    return path == root.path;
                }
                if path == root.path {
                    return true;
                }
                let Ok(relative) = path.strip_prefix(&root.path) else {
                    return false;
                };
                root.recursive || relative.components().count() == 1
            })
            .max_by_key(|(_, root)| root.path.components().count())
            .map(|(index, _)| index)
    }

    fn in_scope(&self, path: &Path) -> bool {
        self.roots
            .iter()
            .any(|root| path.starts_with(&root.watch_path))
    }

    fn enqueue(&self, event: QueuedEvent) {
        let mut state = self.state.lock().unwrap();
        if state.closed {
            return;
        }
        if state.lossy {
            state.pending_ends.extend(event.ends.iter().copied());
            return;
        }
        if event.kind == RESCAN {
            if let Some(existing) = state.queue.iter_mut().find(|queued| queued.kind == RESCAN) {
                for path in event.paths {
                    if !existing.paths.contains(&path) {
                        existing.paths.push(path);
                    }
                }
                drop(state);
                self.notify.notify_waiters();
                return;
            }
        } else if event.kind != RENAME
            && state.queue.back().is_some_and(|last| {
                last.kind == event.kind && last.ends.is_empty() && event.ends.is_empty()
            })
        {
            let queued_paths = state
                .queue
                .iter()
                .map(|queued| queued.paths.len())
                .sum::<usize>();
            let new_paths = {
                let last = state.queue.back().unwrap();
                event
                    .paths
                    .iter()
                    .filter(|path| !last.paths.contains(path))
                    .count()
            };
            if queued_paths + new_paths <= MAX_QUEUED_PATHS {
                let last = state.queue.back_mut().unwrap();
                for path in event.paths {
                    if !last.paths.contains(&path) {
                        last.paths.push(path);
                    }
                }
                drop(state);
                self.notify.notify_waiters();
                return;
            }
        }
        let queued_paths = state
            .queue
            .iter()
            .map(|queued| queued.paths.len())
            .sum::<usize>();
        if state.queue.len() >= MAX_QUEUED_EVENTS
            || queued_paths + event.paths.len() > MAX_QUEUED_PATHS
        {
            let paths = self
                .roots
                .iter()
                .enumerate()
                .filter(|(index, _)| state.live.get(*index).copied().unwrap_or(false))
                .map(|(_, root)| root.path.clone())
                .collect::<Vec<_>>();
            let mut discarded: Vec<usize> = state
                .queue
                .drain(..)
                .flat_map(|queued| queued.ends)
                .collect();
            discarded.extend(event.ends.iter().copied());
            state.pending_ends.extend(discarded);
            state.queue.push_back(QueuedEvent {
                kind: RESCAN,
                paths,
                ends: Vec::new(),
            });
            state.lossy = true;
        } else {
            state.queue.push_back(event);
        }
        drop(state);
        self.notify.notify_waiters();
    }

    async fn next(&self) -> io::Result<Option<WatchEvent>> {
        loop {
            let notified = self.notify.notified();
            tokio::pin!(notified);
            notified.as_mut().enable();
            let candidate = {
                let mut state = self.state.lock().unwrap();
                if state.closed {
                    return Ok(None);
                }
                if let Some(mut event) = state.queue.pop_front() {
                    if event.kind == RESCAN {
                        state.lossy = false;
                        event.ends.append(&mut state.pending_ends);
                    }
                    Some(event)
                } else if let Some((kind, message)) = state.terminal.take() {
                    state.closed = true;
                    return Err(io::Error::new(kind, message));
                } else if state.live.iter().all(|live| !*live) {
                    state.closed = true;
                    return Ok(None);
                } else {
                    None
                }
            };
            if let Some(candidate) = candidate {
                let ends = candidate.ends.clone();
                let event = self.authorize(candidate);
                {
                    let mut state = self.state.lock().unwrap();
                    for index in ends {
                        if let Some(live) = state.live.get_mut(index) {
                            *live = false;
                        }
                    }
                }
                if let Some(event) = event {
                    return Ok(Some(event));
                }
                if !self.any_root_authorized() {
                    let mut state = self.state.lock().unwrap();
                    state.closed = true;
                    return Err(denied(
                        &self
                            .roots
                            .first()
                            .map(|root| root.path.display().to_string())
                            .unwrap_or_default(),
                        false,
                    ));
                }
                continue;
            }
            notified.await;
        }
    }

    fn authorize(&self, event: QueuedEvent) -> Option<WatchEvent> {
        if event.kind == RESCAN {
            let mut paths = Vec::new();
            for path in event.paths {
                let Some(index) = self.root_index(&path) else {
                    continue;
                };
                if !self.state.lock().unwrap().live[index] {
                    continue;
                }
                let root = &self.roots[index];
                if self.caps.check_read(&root.target).is_err() {
                    continue;
                }
                if let Ok(path) = path.into_os_string().into_string()
                    && !paths.contains(&path)
                {
                    paths.push(path);
                }
            }
            return (!paths.is_empty()).then_some((RESCAN, paths));
        }

        let mut paths = Vec::new();
        for path in &event.paths {
            if !self.path_authorized(path) {
                continue;
            }
            match path.clone().into_os_string().into_string() {
                Ok(path) => {
                    if !paths.contains(&path) {
                        paths.push(path);
                    }
                }
                Err(_) => return self.rescan_for(event.paths.first()),
            }
        }
        if paths.is_empty() {
            return None;
        }
        let kind = if event.kind == RENAME && paths.len() == 1 {
            let retained = &paths[0];
            if event.paths.first().and_then(|from| from.to_str()) == Some(retained.as_str()) {
                REMOVE
            } else {
                CREATE
            }
        } else {
            event.kind
        };
        Some((kind, paths))
    }

    fn rescan_for(&self, _path: Option<&PathBuf>) -> Option<WatchEvent> {
        let mut paths = Vec::new();
        for root in &self.roots {
            if self.caps.check_read(&root.target).is_err() {
                continue;
            }
            if let Ok(path) = root.path.clone().into_os_string().into_string() {
                paths.push(path);
            }
        }
        (!paths.is_empty()).then_some((RESCAN, paths))
    }

    fn path_authorized(&self, path: &Path) -> bool {
        let Some(index) = self.root_index(path) else {
            return false;
        };
        let root = &self.roots[index];
        if !self.state.lock().unwrap().live[index] {
            return false;
        }
        let mut target = root.target.clone();
        if path != root.path {
            let Ok(relative) = path.strip_prefix(&root.path) else {
                return false;
            };
            for component in relative.components() {
                let Component::Normal(name) = component else {
                    return false;
                };
                let Ok(child) = target.child(name) else {
                    return false;
                };
                target = child;
            }
        }
        self.caps.check_read(&target).is_ok()
    }

    fn any_root_authorized(&self) -> bool {
        let state = self.state.lock().unwrap();
        self.roots.iter().enumerate().any(|(index, root)| {
            state.live.get(index).copied().unwrap_or(false)
                && self.caps.check_read(&root.target).is_ok()
        })
    }

    async fn wait_epoch(&self, epoch: u32) -> bool {
        loop {
            let notified = self.notify.notified();
            tokio::pin!(notified);
            notified.as_mut().enable();
            {
                let state = self.state.lock().unwrap();
                if state.closed || state.epoch != epoch {
                    return false;
                }
            }
            notified.await;
        }
    }

    fn bump_epoch(&self) -> u32 {
        let epoch = {
            let mut state = self.state.lock().unwrap();
            state.epoch = state.epoch.wrapping_add(1);
            state.epoch
        };
        self.notify.notify_waiters();
        epoch
    }
}

fn map_kind(kind: &notify::EventKind, path_count: usize) -> Option<u8> {
    use notify::event::{EventKind, ModifyKind, RenameMode};

    match kind {
        EventKind::Access(_) => None,
        EventKind::Create(_) => Some(CREATE),
        EventKind::Modify(ModifyKind::Name(RenameMode::Both)) if path_count == 2 => Some(RENAME),
        EventKind::Modify(ModifyKind::Name(RenameMode::From)) => Some(REMOVE),
        EventKind::Modify(ModifyKind::Name(RenameMode::To)) => Some(CREATE),
        EventKind::Modify(_) => Some(MODIFY),
        EventKind::Remove(_) => Some(REMOVE),
        EventKind::Any | EventKind::Other => Some(MODIFY),
    }
}

pub(super) fn ops(files: Arc<Files>) -> Vec<Op> {
    let watchers = Arc::new(Watchers {
        caps: files.caps.clone(),
        resources: Mutex::new(HashMap::new()),
        next: AtomicU64::new(1),
    });
    let create = watchers.clone();
    let next = watchers.clone();
    let next_unref = watchers.clone();
    let keepalive = watchers.clone();
    let reference = watchers.clone();
    let close = watchers.clone();
    vec![
        Op::new(
            "op_file_watch",
            move |(paths, recursive): (Vec<String>, bool)| {
                Ok::<_, Infallible>(response(create.create(paths, recursive)))
            },
        ),
        Op::asynchronous("op_file_watch_next", move |id: u64| {
            let resource = next.resource(id);
            async move { Ok::<_, Infallible>(receive_watch(resource).await) }
        }),
        Op::asynchronous_unreferenced("op_file_watch_next_unref", move |id: u64| {
            let resource = next_unref.resource(id);
            async move { Ok::<_, Infallible>(receive_watch(resource).await) }
        }),
        Op::asynchronous("op_file_watch_keepalive", move |(id, epoch): (u64, u32)| {
            let resource = keepalive.resource(id);
            async move {
                let result = match resource {
                    Ok(resource) => Ok(resource.shared.wait_epoch(epoch).await),
                    Err(error) => Err(error),
                };
                Ok::<_, Infallible>(response(result))
            }
        }),
        Op::new("op_file_watch_ref", move |id: u64| {
            let result = reference
                .resource(id)
                .map(|resource| resource.shared.bump_epoch());
            Ok::<_, Infallible>(response(result))
        }),
        Op::new("op_file_watch_close", move |(id, quiet): (u64, bool)| {
            Ok::<_, Infallible>(response(close.close(id, quiet)))
        }),
    ]
}

async fn receive_watch(resource: io::Result<Arc<WatchResource>>) -> Response<Option<WatchEvent>> {
    match resource {
        Ok(resource) => response(resource.shared.next().await),
        Err(error) => response(Err(error)),
    }
}

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

    #[test]
    fn maps_notify_kinds_to_public_kinds() {
        use notify::event::{CreateKind, EventKind, ModifyKind, RemoveKind, RenameMode};

        assert_eq!(
            map_kind(&EventKind::Create(CreateKind::File), 1),
            Some(CREATE)
        );
        assert_eq!(
            map_kind(&EventKind::Remove(RemoveKind::File), 1),
            Some(REMOVE)
        );
        assert_eq!(
            map_kind(&EventKind::Modify(ModifyKind::Any), 1),
            Some(MODIFY)
        );
        assert_eq!(
            map_kind(&EventKind::Modify(ModifyKind::Name(RenameMode::Both)), 2),
            Some(RENAME)
        );
        assert_eq!(
            map_kind(&EventKind::Modify(ModifyKind::Name(RenameMode::Both)), 1),
            Some(MODIFY)
        );
        assert_eq!(
            map_kind(&EventKind::Modify(ModifyKind::Name(RenameMode::From)), 1),
            Some(REMOVE)
        );
        assert_eq!(
            map_kind(&EventKind::Modify(ModifyKind::Name(RenameMode::To)), 1),
            Some(CREATE)
        );
    }

    fn shared(root: &Path) -> Arc<Shared> {
        Arc::new(Shared {
            caps: Capabilities::new(avril_caps::Initial::new(false), root.to_path_buf()),
            roots: vec![Root {
                target: PathTarget::resolve(root).unwrap(),
                path: root.to_path_buf(),
                directory: true,
                recursive: true,
                watch_path: root.to_path_buf(),
            }],
            state: Mutex::new(WatchState {
                live: vec![true],
                ..WatchState::default()
            }),
            notify: Notify::new(),
        })
    }

    #[test]
    fn coalesces_adjacent_events_without_reordering_kinds() {
        let directory = tempfile::tempdir().unwrap();
        let root = directory.path();
        let shared = shared(root);
        shared.enqueue(QueuedEvent {
            kind: MODIFY,
            paths: vec![root.join("a")],
            ends: Vec::new(),
        });
        shared.enqueue(QueuedEvent {
            kind: MODIFY,
            paths: vec![root.join("b"), root.join("a")],
            ends: Vec::new(),
        });
        shared.enqueue(QueuedEvent {
            kind: CREATE,
            paths: vec![root.join("c")],
            ends: Vec::new(),
        });
        let state = shared.state.lock().unwrap();
        assert_eq!(state.queue.len(), 2);
        assert_eq!(state.queue[0].paths, [root.join("a"), root.join("b")]);
        assert_eq!(state.queue[1].kind, CREATE);
    }

    #[test]
    fn queue_overflow_collapses_to_one_rescan() {
        let directory = tempfile::tempdir().unwrap();
        let root = directory.path();
        let shared = shared(root);
        for index in 0..=MAX_QUEUED_PATHS {
            shared.enqueue(QueuedEvent {
                kind: MODIFY,
                paths: vec![root.join(index.to_string())],
                ends: Vec::new(),
            });
        }
        {
            let state = shared.state.lock().unwrap();
            assert!(state.lossy);
            assert_eq!(state.queue.len(), 1);
            assert_eq!(state.queue[0].kind, RESCAN);
            assert_eq!(state.queue[0].paths, [root.to_path_buf()]);
        }
        shared.enqueue(QueuedEvent {
            kind: CREATE,
            paths: vec![root.join("late")],
            ends: Vec::new(),
        });
        assert_eq!(shared.state.lock().unwrap().queue.len(), 1);
    }

    #[test]
    fn overflow_preserves_pending_root_ends() {
        let directory = tempfile::tempdir().unwrap();
        let root = directory.path();
        let shared = shared(root);
        shared.enqueue(QueuedEvent {
            kind: REMOVE,
            paths: vec![root.to_path_buf()],
            ends: vec![0],
        });
        for index in 0..=MAX_QUEUED_PATHS {
            shared.enqueue(QueuedEvent {
                kind: MODIFY,
                paths: vec![root.join(index.to_string())],
                ends: Vec::new(),
            });
        }
        let state = shared.state.lock().unwrap();
        assert_eq!(state.queue[0].kind, RESCAN);
        assert_eq!(state.pending_ends, [0]);
    }

    #[test]
    fn rename_with_an_unauthorized_destination_becomes_remove() {
        let directory = tempfile::tempdir().unwrap();
        let root = directory.path();
        let shared = shared(root);
        let secret = root.join("secret");
        shared
            .caps
            .pledge(vec![(
                avril_caps::Category::Read,
                None,
                vec![secret.display().to_string()],
            )])
            .unwrap();
        let event = shared.authorize(QueuedEvent {
            kind: RENAME,
            paths: vec![root.join("old"), secret.join("new")],
            ends: Vec::new(),
        });
        assert_eq!(
            event,
            Some((REMOVE, vec![root.join("old").display().to_string()]))
        );
    }
}