coven_foundation/files/
reached.rs

1//! Where a test hears that file work reached a path: about to open it, or
2//! finding its lock held, by another holder or by an instance of this
3//! process that is closing, and about to wait for it. A stalled disk, such as a
4//! FIFO in a file's place, then holds the work there, and the test knows it
5//! is held rather than giving it time to get there.
6
7use crate::locks::lock_unpoisoned;
8use std::path::{Path, PathBuf};
9use std::sync::Mutex;
10
11type Watch = (PathBuf, tokio::sync::oneshot::Sender<()>);
12
13static WATCHED: Mutex<Vec<Watch>> = Mutex::new(Vec::new());
14
15/// Told once file work reaches `path`: about to open it, or finding its
16/// lock held and about to wait for it.
17pub fn when_reached(path: &Path) -> tokio::sync::oneshot::Receiver<()> {
18    let (told, telling) = tokio::sync::oneshot::channel();
19    lock_unpoisoned(&WATCHED).push((resolved(path), told));
20    telling
21}
22
23/// File work reached `path`: tell whoever waits for that.
24pub(super) fn reached(path: &Path) {
25    let path = resolved(path);
26    let mut watched = lock_unpoisoned(&WATCHED);
27    let (found, rest): (Vec<Watch>, Vec<Watch>) =
28        std::mem::take(&mut *watched).into_iter().partition(|(watched, _)| *watched == path);
29    *watched = rest;
30    for (_, told) in found {
31        #[expect(
32            clippy::let_underscore_must_use,
33            reason = "a test that stopped listening has no use for it"
34        )]
35        let _ = told.send(());
36    }
37}
38
39/// `path` under its parent's canonical path, as the work and the test may
40/// spell one folder differently.
41fn resolved(path: &Path) -> PathBuf {
42    match (path.parent().and_then(|parent| std::fs::canonicalize(parent).ok()), path.file_name()) {
43        (Some(parent), Some(name)) => parent.join(name),
44        _ => path.to_owned(),
45    }
46}
coven_foundation/files/reached.rs

coven_foundation/files/
reached.rs
