oino_tools/
mutation_queue.rs

1//! Per-file serialisation of mutations.
2//!
3//! `write` and `edit` calls on the same file run one after another in call order; calls on
4//! different files run in parallel. The key is the real path (symlinks resolved), or the
5//! resolved path when the file does not exist yet.
6
7use std::collections::HashMap;
8use std::future::Future;
9use std::sync::{Arc, LazyLock, Mutex, PoisonError};
10
11use futures::FutureExt;
12use futures::future::{BoxFuture, Shared};
13use tokio::sync::oneshot;
14
15use crate::env::ExecutionEnv;
16use crate::node_path;
17
18type Tail = Shared<BoxFuture<'static, ()>>;
19
20/// A mutation queue keyed by file.
21#[derive(Clone, Debug, Default)]
22pub struct FileMutationQueue {
23    inner: Arc<QueueInner>,
24}
25
26#[derive(Debug, Default)]
27struct QueueInner {
28    tails: Mutex<HashMap<String, Tail>>,
29    /// Orders registrations like a registration queue, so calls queue in call order even
30    /// though resolving the key is asynchronous.
31    registration: tokio::sync::Mutex<()>,
32}
33
34static GLOBAL: LazyLock<FileMutationQueue> = LazyLock::new(FileMutationQueue::default);
35
36impl FileMutationQueue {
37    /// The process-wide queue shared by all built-in tools.
38    #[must_use]
39    pub fn global() -> Self {
40        GLOBAL.clone()
41    }
42
43    /// Runs `operation` once every earlier mutation of the same file has settled.
44    pub async fn run<T, F>(&self, env: &dyn ExecutionEnv, path: &str, operation: F) -> T
45    where
46        F: Future<Output = T>,
47    {
48        let (key, previous, release, ours) = {
49            let _registration = self.inner.registration.lock().await;
50            let resolved = node_path::resolve("/", path);
51            let key = env.realpath(&resolved).await.unwrap_or(resolved);
52            let (release, released) = oneshot::channel::<()>();
53            let ours: Tail = released.map(|_| ()).boxed().shared();
54            let previous = self.lock_tails().insert(key.clone(), ours.clone());
55            (key, previous, release, ours)
56        };
57        if let Some(previous) = previous {
58            previous.await;
59        }
60        let output = operation.await;
61        drop(release);
62        let mut tails = self.lock_tails();
63        if tails.get(&key).is_some_and(|tail| tail.ptr_eq(&ours)) {
64            tails.remove(&key);
65        }
66        output
67    }
68
69    fn lock_tails(&self) -> std::sync::MutexGuard<'_, HashMap<String, Tail>> {
70        self.inner
71            .tails
72            .lock()
73            .unwrap_or_else(PoisonError::into_inner)
74    }
75
76    #[cfg(test)]
77    fn pending(&self) -> usize {
78        self.lock_tails().len()
79    }
80}
81
82#[cfg(test)]
83mod tests {
84    use std::time::Duration;
85
86    use super::*;
87    use crate::env::LocalEnv;
88
89    #[tokio::test]
90    async fn same_file_runs_in_order() {
91        let queue = FileMutationQueue::default();
92        let env = LocalEnv::new();
93        let log = Arc::new(Mutex::new(Vec::new()));
94        let first = {
95            let (queue, log) = (queue.clone(), Arc::clone(&log));
96            async move {
97                queue
98                    .run(&env, "/nonexistent/a", async {
99                        tokio::time::sleep(Duration::from_millis(30)).await;
100                        log.lock().unwrap().push(1);
101                    })
102                    .await;
103            }
104        };
105        let second = {
106            let (queue, log) = (queue.clone(), Arc::clone(&log));
107            async move {
108                tokio::time::sleep(Duration::from_millis(5)).await;
109                queue
110                    .run(&env, "/nonexistent/a", async {
111                        log.lock().unwrap().push(2);
112                    })
113                    .await;
114            }
115        };
116        tokio::join!(first, second);
117        assert_eq!(*log.lock().unwrap(), vec![1, 2]);
118        assert_eq!(queue.pending(), 0);
119    }
120}
oino_tools/mutation_queue.rs

oino_tools/
mutation_queue.rs
