hennery_kernel/
coordination.rs

1//! Process ownership and the owner lifecycle gate. Every writer opens through
2//! `db::open`, even before the first optional registration exists.
3
4use anyhow::{Context, Result, bail};
5use std::collections::HashMap;
6use std::os::fd::AsRawFd;
7use std::os::unix::fs::MetadataExt;
8use std::os::unix::fs::OpenOptionsExt;
9use std::path::Path;
10use std::sync::{Arc, Condvar, Mutex, OnceLock, Weak};
11
12/// Reentrant on one thread so a store can call a second module while it
13/// owns the gate. Guards cannot cross an await or move to another thread.
14#[derive(Debug, Default)]
15pub struct Gate {
16    state: Mutex<GateState>,
17    ready: Condvar,
18}
19#[derive(Debug, Default)]
20struct GateState {
21    owner: Option<std::thread::ThreadId>,
22    depth: usize,
23}
24pub struct GateGuard {
25    gate: Arc<Gate>,
26    _thread: std::marker::PhantomData<std::rc::Rc<()>>,
27}
28impl Gate {
29    pub fn lock(self: &Arc<Self>) -> GateGuard {
30        let thread = std::thread::current().id();
31        let mut state = self.state.lock().expect("lifecycle gate");
32        while state.owner.is_some_and(|owner| owner != thread) {
33            state = self.ready.wait(state).expect("lifecycle gate");
34        }
35        state.owner = Some(thread);
36        state.depth += 1;
37        GateGuard {
38            gate: self.clone(),
39            _thread: std::marker::PhantomData,
40        }
41    }
42}
43impl Drop for GateGuard {
44    fn drop(&mut self) {
45        let mut state = self.gate.state.lock().expect("lifecycle gate");
46        state.depth -= 1;
47        if state.depth == 0 {
48            state.owner = None;
49            self.gate.ready.notify_all();
50        }
51    }
52}
53
54#[derive(Debug, Default)]
55pub struct Coordination {
56    pub gate: Arc<Gate>,
57    // Pin the identity used by OWNERS even after the last SQLite connection
58    // closes. An unlinked inode must not be recycled into a live registry key.
59    _database: Option<std::fs::File>,
60    _ownership: Option<std::fs::File>,
61}
62
63impl Coordination {
64    pub(crate) fn acquire(file: std::fs::File, path: &Path) -> Result<Arc<Self>> {
65        type Registry = Mutex<HashMap<(u64, u64), Weak<Coordination>>>;
66        static OWNERS: OnceLock<Registry> = OnceLock::new();
67        let metadata = file.metadata()?;
68        anyhow::ensure!(metadata.nlink() == 1, "hard-linked databases are refused");
69        let identity = (metadata.dev(), metadata.ino());
70        let mut owners = OWNERS.get_or_init(Mutex::default).lock().expect("writer registry");
71        if let Some(owner) = owners.get(&identity).and_then(Weak::upgrade) {
72            return Ok(owner);
73        }
74        // Darwin SQLite may itself use flock on its database. Ownership
75        // therefore has a distinct private sidecar, never the SQLite file.
76        let canonical = path.canonicalize()?;
77        let mut lock_path = canonical.as_os_str().to_os_string();
78        lock_path.push(".writer.lock");
79        let lock_file = std::fs::OpenOptions::new()
80            .read(true)
81            .write(true)
82            .create(true)
83            .truncate(false)
84            .mode(0o600)
85            .custom_flags(libc::O_NOFOLLOW)
86            .open(lock_path)?;
87        anyhow::ensure!(
88            lock_file.metadata()?.nlink() == 1,
89            "hard-linked writer locks are refused"
90        );
91        // SAFETY: lock_file owns this descriptor; released when the handle closes.
92        let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5);
93        while unsafe { libc::flock(lock_file.as_raw_fd(), libc::LOCK_EX | libc::LOCK_NB) } != 0 {
94            let err = std::io::Error::last_os_error();
95            if err.raw_os_error() == Some(libc::EWOULDBLOCK) {
96                if std::time::Instant::now() < deadline {
97                    std::thread::sleep(std::time::Duration::from_millis(20));
98                    continue;
99                }
100                bail!("another process owns this database writer");
101            }
102            return Err(err).context("acquire database writer ownership");
103        }
104        let owner = Arc::new(Self {
105            gate: Arc::default(),
106            _database: Some(file),
107            _ownership: Some(lock_file),
108        });
109        owners.insert(identity, Arc::downgrade(&owner));
110        Ok(owner)
111    }
112}
113
114#[cfg(test)]
115mod tests {
116    use super::*;
117
118    fn inspect_descriptor(descriptor: std::os::fd::RawFd) -> std::io::Result<libc::stat> {
119        // SAFETY: fstat only inspects the descriptor; the output is read only
120        // when libc reports that it initialized the stat structure.
121        let mut metadata = std::mem::MaybeUninit::<libc::stat>::uninit();
122        if unsafe { libc::fstat(descriptor, metadata.as_mut_ptr()) } != 0 {
123            return Err(std::io::Error::last_os_error());
124        }
125        Ok(unsafe { metadata.assume_init() })
126    }
127
128    #[test]
129    fn registry_identity_stays_pinned_until_coordination_is_released() {
130        let dir = tempfile::tempdir().unwrap();
131        let path = dir.path().join("hennery.db");
132        let file = std::fs::File::create(&path).unwrap();
133        let descriptor = file.as_raw_fd();
134        let identity = inspect_descriptor(descriptor).unwrap();
135        let owner = Coordination::acquire(file, &path).unwrap();
136        // Model fixture teardown or removal while a recovery worker still
137        // holds coordination after the last SQLite connection has closed.
138        std::fs::remove_file(&path).unwrap();
139        let metadata = inspect_descriptor(descriptor).expect("registry identity descriptor was closed");
140        assert_eq!((metadata.st_dev, metadata.st_ino), (identity.st_dev, identity.st_ino));
141        assert_eq!(metadata.st_nlink, 0);
142        drop(owner);
143    }
144}
hennery_kernel/coordination.rs

hennery_kernel/
coordination.rs
