genegraph_storage/
commit.rs

1//! Commit serialization for metadata read-modify-write cycles and dataset
2//! writes.
3//!
4//! Base concurrency-safety model borrowed from duva's actor design
5//! ([`<https://github.com/Migorithm/duva>`]): every storage instance has a single
6//! logical *commit actor*. All metadata mutations (the `load_metadata →
7//! mutate → save_metadata` cycle performed by every `save_*` call) are
8//! serialized through it, so concurrent writers cannot interleave their
9//! cycles and lose each other's registry entries (lost update) — the same
10//! reason duva routes writes through one actor mailbox instead of shared
11//! mutable state.
12//!
13//! The same mailbox shape extends to Lance dataset writes
14//! ([`with_dataset_write_lock`]): manifest-version allocation and the
15//! commit-point publish of one dataset directory are serialized, so two
16//! concurrent overwrites cannot mint the same `N.manifest` (#95).
17//!
18//! Both registries hold **weak** references (#98): a mailbox stays alive
19//! only while some caller holds its `Arc` (i.e. while a commit cycle or
20//! dataset write is in flight), and dead entries are swept on insert.
21//! Instances that churn (create/drop thousands of collections) keep the
22//! registries bounded.
23//!
24//! Durability of the commit itself is the tmp + fsync + rename discipline
25//! ([`crate::generations::write_json_atomic`] for metadata, the same
26//! sequence inside the lancefmt writer for data/txn/manifest files):
27//! readers never observe a half-written commit pointer, and a read after a
28//! completed commit observes its effects (read-your-own-writes).
29//! Cross-process arbitration stays with the transactional-generations work
30//! (#93/#81-P5).
31
32use std::collections::HashMap;
33use std::path::Path;
34use std::sync::{Arc, Mutex as StdMutex, OnceLock, Weak};
35
36use tokio::sync::Mutex;
37
38use crate::{StorageError, StorageResult};
39
40/// Commit-actor mailboxes, keyed by metadata path (weak-valued, #98).
41static COMMIT_LOCKS: OnceLock<StdMutex<HashMap<String, Weak<Mutex<()>>>>> = OnceLock::new();
42/// Dataset-write mailboxes, keyed by dataset dir (weak-valued, #98).
43static DATASET_LOCKS: OnceLock<StdMutex<HashMap<String, Weak<StdMutex<()>>>>> = OnceLock::new();
44
45/// Shared weak-registry lookup (#98): reuse the live mailbox for `key` if
46/// one exists, otherwise sweep dead entries and insert `fresh`.
47pub(crate) fn weak_lookup<T>(
48    registry: &'static OnceLock<StdMutex<HashMap<String, Weak<T>>>>,
49    key: String,
50    fresh: Arc<T>,
51) -> Arc<T> {
52    let mut map = registry
53        .get_or_init(|| StdMutex::new(HashMap::new()))
54        .lock()
55        .unwrap_or_else(|poisoned| poisoned.into_inner());
56    if let Some(weak) = map.get(&key)
57        && let Some(strong) = weak.upgrade()
58    {
59        return strong;
60    }
61    map.retain(|_, weak| weak.strong_count() > 0);
62    let arc = Arc::clone(&fresh);
63    map.insert(key, Arc::downgrade(&fresh));
64    arc
65}
66
67/// One commit-actor mailbox per metadata path.
68fn lock_for(metadata_path: &Path) -> Arc<Mutex<()>> {
69    let key = metadata_path.to_string_lossy().to_string();
70    weak_lookup(&COMMIT_LOCKS, key, Arc::new(Mutex::new(())))
71}
72
73/// One dataset-write mailbox per dataset directory.
74fn dataset_lock_for(dataset_dir: &Path) -> Arc<StdMutex<()>> {
75    let key = dataset_dir.to_string_lossy().to_string();
76    weak_lookup(&DATASET_LOCKS, key, Arc::new(StdMutex::new(())))
77}
78
79/// Runs `commit` (a full metadata read-modify-write cycle) under the
80/// instance's commit actor: at most one cycle runs at a time for the same
81/// metadata path.
82pub(crate) async fn with_commit_actor<T, F, Fut>(
83    metadata_path: &Path,
84    commit: F,
85) -> StorageResult<T>
86where
87    F: FnOnce() -> Fut,
88    Fut: std::future::Future<Output = StorageResult<T>>,
89{
90    let mailbox = lock_for(metadata_path);
91    let _guard = mailbox.lock().await;
92    commit().await
93}
94
95/// Runs `write` under the dataset's write mailbox. `write` must be a
96/// non-async closure: the whole dataset write — version allocation through
97/// commit-point publish — happens under the lock.
98pub(crate) fn with_dataset_write_lock<T>(
99    dataset_dir: &Path,
100    write: impl FnOnce() -> StorageResult<T>,
101) -> StorageResult<T> {
102    let mailbox = dataset_lock_for(dataset_dir);
103    let _guard = mailbox
104        .lock()
105        .map_err(|_| StorageError::InvalidState("dataset write lock poisoned".into()))?;
106    write()
107}
108
109/// Test hook: (commit-registry size, dataset-registry size).
110#[cfg(test)]
111pub(crate) fn registry_sizes() -> (usize, usize) {
112    let commit = COMMIT_LOCKS
113        .get()
114        .map(|m| m.lock().unwrap().len())
115        .unwrap_or(0);
116    let dataset = DATASET_LOCKS
117        .get()
118        .map(|m| m.lock().unwrap().len())
119        .unwrap_or(0);
120    (commit, dataset)
121}
genegraph_storage/commit.rs

genegraph_storage/
commit.rs
