solid_pod_rs_forge/
locks.rs

1//! Per-repository serialisation locks for spine writes.
2//!
3//! JSS `forge` funnels every issue-index mutation through `serializeOn`, one
4//! promise chain per `<owner>/<repo>`, so number allocation and
5//! read-modify-write cycles never race (and pulls and merges ride a second
6//! chain). [`RepoLocks`] is the async-mutex equivalent: one
7//! [`tokio::sync::Mutex`] per repository key.
8//!
9//! Lock order, when both are needed: **issue lock first, then pull lock**
10//! (the label rename and delete cascade does exactly this). Hold a lock only
11//! across load, modify and save of the index, never across a loopback fetch.
12
13use std::collections::HashMap;
14use std::sync::{Arc, Mutex};
15
16use tokio::sync::{Mutex as AsyncMutex, OwnedMutexGuard};
17
18/// A keyed set of async mutexes, one per `<owner>/<repo>`.
19///
20/// # Examples
21///
22/// ```
23/// let rt = tokio::runtime::Builder::new_current_thread().enable_all().build().unwrap();
24/// rt.block_on(async {
25///     let locks = solid_pod_rs_forge::locks::RepoLocks::default();
26///     let g = locks.lock("alice", "demo").await;
27///     // A second acquisition of the same key waits; another key does not.
28///     let other = locks.lock("alice", "other").await;
29///     drop(other);
30///     drop(g);
31/// });
32/// ```
33#[derive(Debug, Default)]
34pub struct RepoLocks {
35    map: Mutex<HashMap<String, Arc<AsyncMutex<()>>>>,
36}
37
38impl RepoLocks {
39    /// Acquire the lock for `<owner>/<repo>`, waiting for earlier holders.
40    pub async fn lock(&self, owner: &str, repo: &str) -> OwnedMutexGuard<()> {
41        let key = format!("{owner}/{repo}");
42        let m = {
43            let mut map = self
44                .map
45                .lock()
46                .unwrap_or_else(std::sync::PoisonError::into_inner);
47            // Drop entries nobody holds or waits on, so the map cannot grow
48            // without bound over a long-lived process.
49            if map.len() > 256 {
50                map.retain(|_, v| Arc::strong_count(v) > 1);
51            }
52            map.entry(key).or_default().clone()
53        };
54        m.lock_owned().await
55    }
56}
57
58/// The two lock families JSS keeps: `issueLocks` and `pullLocks`.
59#[derive(Debug, Default)]
60pub struct ForgeLocks {
61    /// Guards `issues/<owner>/<repo>.json` (labels live in the same file).
62    pub issues: RepoLocks,
63    /// Guards `pulls/<owner>/<repo>.json` and merges. Acquire after
64    /// [`ForgeLocks::issues`] when both are needed.
65    pub pulls: RepoLocks,
66}
67
68#[cfg(test)]
69mod tests {
70    use super::*;
71    use std::sync::atomic::{AtomicU32, Ordering};
72
73    #[tokio::test]
74    async fn same_key_serialises_and_other_keys_do_not() {
75        let locks = Arc::new(RepoLocks::default());
76        let counter = Arc::new(AtomicU32::new(0));
77        let g = locks.lock("a", "r").await;
78        // A different key is immediately available.
79        let other = locks.lock("a", "s").await;
80        drop(other);
81        let (l2, c2) = (locks.clone(), counter.clone());
82        let waiter = tokio::spawn(async move {
83            let _g = l2.lock("a", "r").await;
84            c2.fetch_add(1, Ordering::SeqCst);
85        });
86        tokio::time::sleep(std::time::Duration::from_millis(50)).await;
87        assert_eq!(counter.load(Ordering::SeqCst), 0, "still blocked");
88        drop(g);
89        waiter.await.unwrap();
90        assert_eq!(counter.load(Ordering::SeqCst), 1);
91    }
92
93    #[tokio::test]
94    async fn read_modify_write_does_not_lose_updates() {
95        let locks = Arc::new(RepoLocks::default());
96        let value = Arc::new(tokio::sync::Mutex::new(0u32));
97        let mut tasks = Vec::new();
98        for _ in 0..50 {
99            let (l, v) = (locks.clone(), value.clone());
100            tasks.push(tokio::spawn(async move {
101                let _g = l.lock("o", "r").await;
102                let cur = *v.lock().await;
103                tokio::task::yield_now().await;
104                *v.lock().await = cur + 1;
105            }));
106        }
107        for t in tasks {
108            t.await.unwrap();
109        }
110        assert_eq!(*value.lock().await, 50);
111    }
112}
solid_pod_rs_forge/locks.rs

solid_pod_rs_forge/
locks.rs
