khive_db/stores/blob/
gc_locks.rs

1use std::collections::HashMap;
2use std::fs;
3use std::path::{Path, PathBuf};
4use std::sync::{Arc, Mutex as StdMutex, OnceLock};
5
6use khive_storage::{SqlAccess, StorageCapability, StorageError, StorageResult};
7
8use super::{map_io_err, DATABASE_GC_LOCK_SUFFIX};
9
10pub(super) fn database_gc_lock_path(database_path: &Path) -> PathBuf {
11    let mut lock_path = database_path.as_os_str().to_os_string();
12    lock_path.push(DATABASE_GC_LOCK_SUFFIX);
13    PathBuf::from(lock_path)
14}
15
16pub(super) fn acquire_database_gc_lock(
17    database_path: Option<&Path>,
18) -> StorageResult<Option<fs::File>> {
19    let Some(database_path) = database_path else {
20        // An in-memory database cannot be shared by another process. The
21        // process-local lock below is therefore the complete owner fence.
22        return Ok(None);
23    };
24    let lock_path = database_gc_lock_path(database_path);
25    let lock_file = fs::OpenOptions::new()
26        .read(true)
27        .write(true)
28        .create(true)
29        .truncate(false)
30        .open(&lock_path)
31        .map_err(|e| map_io_err(e, "database_gc_lock_open"))?;
32    fs4::FileExt::lock(&lock_file).map_err(|e| map_io_err(e, "database_gc_lock_acquire"))?;
33    Ok(Some(lock_file))
34}
35
36/// Process-wide database owner fence for transactional blob sweeps.
37///
38/// Claims live in the database and their attachment triggers are database-global,
39/// so a root-only lock is insufficient: two differently configured roots for
40/// one database must not recover each other's live claims. File-backed pools
41/// additionally take [`acquire_database_gc_lock`] for cross-process exclusion.
42type SweepLockMap = HashMap<Option<PathBuf>, Arc<DatabaseGcProcessLock>>;
43
44#[derive(Debug, Default)]
45pub(super) struct DatabaseGcProcessLock {
46    held: StdMutex<bool>,
47    released: std::sync::Condvar,
48    #[cfg(test)]
49    waiters: std::sync::atomic::AtomicUsize,
50}
51
52impl DatabaseGcProcessLock {
53    pub(super) fn acquire(self: &Arc<Self>) -> DatabaseGcProcessGuard {
54        let mut held = self
55            .held
56            .lock()
57            .unwrap_or_else(std::sync::PoisonError::into_inner);
58        while *held {
59            #[cfg(test)]
60            self.waiters
61                .fetch_add(1, std::sync::atomic::Ordering::SeqCst);
62            held = self
63                .released
64                .wait(held)
65                .unwrap_or_else(std::sync::PoisonError::into_inner);
66            #[cfg(test)]
67            self.waiters
68                .fetch_sub(1, std::sync::atomic::Ordering::SeqCst);
69        }
70        *held = true;
71        DatabaseGcProcessGuard {
72            lock: Arc::clone(self),
73        }
74    }
75
76    pub(super) fn try_acquire(self: &Arc<Self>) -> Option<DatabaseGcProcessGuard> {
77        let mut held = self
78            .held
79            .lock()
80            .unwrap_or_else(std::sync::PoisonError::into_inner);
81        if *held {
82            return None;
83        }
84        *held = true;
85        Some(DatabaseGcProcessGuard {
86            lock: Arc::clone(self),
87        })
88    }
89}
90
91#[derive(Debug)]
92pub(super) struct DatabaseGcProcessGuard {
93    lock: Arc<DatabaseGcProcessLock>,
94}
95
96impl Drop for DatabaseGcProcessGuard {
97    fn drop(&mut self) {
98        let mut held = self
99            .lock
100            .held
101            .lock()
102            .unwrap_or_else(std::sync::PoisonError::into_inner);
103        debug_assert!(*held, "database GC process owner released twice");
104        *held = false;
105        self.lock.released.notify_one();
106    }
107}
108
109fn database_sweep_locks() -> &'static StdMutex<SweepLockMap> {
110    static REGISTRY: OnceLock<StdMutex<SweepLockMap>> = OnceLock::new();
111    REGISTRY.get_or_init(|| StdMutex::new(HashMap::new()))
112}
113
114pub(super) fn sweep_lock_for_database(database_path: Option<&Path>) -> Arc<DatabaseGcProcessLock> {
115    let key = database_path.map(Path::to_path_buf);
116    let mut locks = database_sweep_locks()
117        .lock()
118        .unwrap_or_else(std::sync::PoisonError::into_inner);
119    locks
120        .entry(key)
121        .or_insert_with(|| Arc::new(DatabaseGcProcessLock::default()))
122        .clone()
123}
124
125#[cfg(test)]
126pub(crate) fn database_gc_waiter_count(database_path: Option<&Path>) -> usize {
127    sweep_lock_for_database(database_path)
128        .waiters
129        .load(std::sync::atomic::Ordering::SeqCst)
130}
131
132/// Exclusive canonical-database ownership shared by transactional blob GC and
133/// the boot-gated V21 attachment cutover.
134///
135/// The process-local mutex is acquired first and retained while the advisory
136/// file lock is acquired on a blocking thread. Moving the owned mutex guard
137/// into that closure makes cancellation safe: dropping the outer future cannot
138/// release process ownership while a blocking advisory acquisition continues.
139pub struct DatabaseGcOwnerGuard {
140    _process_guard: DatabaseGcProcessGuard,
141    _advisory_guard: Option<fs::File>,
142    database_path: Option<PathBuf>,
143}
144
145pub(crate) fn acquire_database_gc_owner_for_path_blocking(
146    database_path: Option<PathBuf>,
147) -> StorageResult<DatabaseGcOwnerGuard> {
148    let process_guard = sweep_lock_for_database(database_path.as_deref()).acquire();
149    let advisory_guard = acquire_database_gc_lock(database_path.as_deref())?;
150    Ok(DatabaseGcOwnerGuard {
151        _process_guard: process_guard,
152        _advisory_guard: advisory_guard,
153        database_path,
154    })
155}
156
157/// Try to acquire canonical database-GC ownership without waiting.
158///
159/// This is the fail-closed bridge for the legacy raw-connection migration API:
160/// a caller may already hold an opaque pooled writer guard, so waiting here
161/// could invert the canonical owner-before-writer order used by sweeps. The
162/// production backend boot path uses the blocking helper before writer
163/// checkout instead.
164pub(crate) fn try_acquire_database_gc_owner_for_path(
165    database_path: PathBuf,
166) -> StorageResult<DatabaseGcOwnerGuard> {
167    let process_guard = sweep_lock_for_database(Some(&database_path))
168        .try_acquire()
169        .ok_or_else(|| {
170            StorageError::Internal(format!(
171                "database GC owner for {} is already held; retry schema migration through the \
172                 coordinated backend boot path",
173                database_path.display()
174            ))
175        })?;
176    let lock_path = database_gc_lock_path(&database_path);
177    let advisory_guard = fs::OpenOptions::new()
178        .read(true)
179        .write(true)
180        .create(true)
181        .truncate(false)
182        .open(&lock_path)
183        .map_err(|error| map_io_err(error, "database_gc_lock_open"))?;
184    fs4::FileExt::try_lock(&advisory_guard)
185        .map_err(|error| map_io_err(error.into(), "database_gc_lock_try_acquire"))?;
186    Ok(DatabaseGcOwnerGuard {
187        _process_guard: process_guard,
188        _advisory_guard: Some(advisory_guard),
189        database_path: Some(database_path),
190    })
191}
192
193impl std::fmt::Debug for DatabaseGcOwnerGuard {
194    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
195        formatter
196            .debug_struct("DatabaseGcOwnerGuard")
197            .field("database_path", &self.database_path)
198            .finish_non_exhaustive()
199    }
200}
201
202impl DatabaseGcOwnerGuard {
203    /// Canonical database path that keys this owner, or `None` for an
204    /// in-memory database whose process-local mutex is the complete fence.
205    pub fn database_path(&self) -> Option<&Path> {
206        self.database_path.as_deref()
207    }
208}
209
210/// Acquire the canonical database owner used by both V21 boot cutover and
211/// transactional blob sweep. Callers must retain the returned guard across
212/// every stage that must exclude the other protocol.
213pub async fn acquire_database_gc_owner(sql: &dyn SqlAccess) -> StorageResult<DatabaseGcOwnerGuard> {
214    let database_path = sql.database_path();
215    tokio::task::spawn_blocking(move || acquire_database_gc_owner_for_path_blocking(database_path))
216        .await
217        .map_err(|error| {
218            StorageError::driver(StorageCapability::Blob, "acquire_database_gc_owner", error)
219        })?
220}
khive_db/stores/blob/gc_locks.rs

khive_db/stores/blob/
gc_locks.rs
