subetha_cxc/
lazy_config.rs

1//! `LazyConfig<T>` - thundering-herd-proof distributed config fetch.
2//!
3//! Composite primitive built on `SharedOnceCell<T>` + (optionally)
4//! the BackgroundScheduler's Pass dispatch. Guarantees that the
5//! config-fetch closure runs EXACTLY ONCE across all participating
6//! processes - no matter how many of them call `get_or_fetch`
7//! concurrently. The losers of the CAS race spin until the winner
8//! publishes, then read the canonical value.
9//!
10//! # Why this exists
11//!
12//! Distributed services often want shared config (Consul, etcd,
13//! DNS, an internal config service) loaded into every process once.
14//! Naive implementations have every process independently fetch:
15//! N processes -> N backend requests at startup, a classic
16//! thundering herd against the config service. With LazyConfig,
17//! one process fetches; all others see the result.
18//!
19//! # Two access modes
20//!
21//! 1. **Local fetcher**: `get_or_fetch(|| { ... })` runs the
22//!    closure in the calling process if it wins the CAS; losers
23//!    block until the winner publishes.
24//!
25//! 2. **Scheduler-dispatched fetcher** (extension): submit a Pass
26//!    via the BackgroundScheduler with a registered closure_id.
27//!    Any process registered for that closure_id can serve as the
28//!    fetcher; the result is published via the SharedOnceCell. This
29//!    allows the fetch to happen in a process dedicated to that
30//!    role (e.g., a privileged process with network access) while
31//!    other processes only block on the cell.
32//!
33//! Both modes share the same underlying CAS protocol, so the
34//! thundering-herd-prevention property holds either way.
35
36use std::path::Path;
37use std::sync::Arc;
38
39use crate::shared_once_cell::{SharedOnceCell, SharedOnceError};
40
41#[derive(Debug, Clone, Copy, PartialEq, Eq)]
42pub enum LazyConfigError {
43    Once(SharedOnceError),
44}
45
46impl From<SharedOnceError> for LazyConfigError {
47    fn from(e: SharedOnceError) -> Self { Self::Once(e) }
48}
49
50pub struct LazyConfig<T: Copy + Send + Sync + 'static> {
51    cell: Arc<SharedOnceCell<T>>,
52    header_sidecar: subetha_core::HandshakeHeader,
53    ring_sidecar: Box<subetha_core::ObservationRing>,
54}
55
56impl<T: Copy + Send + Sync + 'static> subetha_sidecar::AdaptiveInstance for LazyConfig<T> {
57    fn header(&self) -> &subetha_core::HandshakeHeader { &self.header_sidecar }
58    fn ring(&self) -> &subetha_core::ObservationRing { &self.ring_sidecar }
59    fn make_policy(&self) -> Box<dyn subetha_sidecar::Policy> {
60        Box::new(subetha_sidecar::NoMigrationPolicy)
61    }
62}
63
64impl<T: Copy + Send + Sync + 'static> LazyConfig<T> {
65    pub fn create(path: impl AsRef<Path>) -> Result<Self, LazyConfigError> {
66        Ok(Self {
67            cell: Arc::new(SharedOnceCell::create(path)?),
68            header_sidecar: subetha_core::HandshakeHeader::new(),
69            ring_sidecar: Box::new(subetha_core::ObservationRing::new()),
70        })
71    }
72
73    pub fn open(path: impl AsRef<Path>) -> Result<Self, LazyConfigError> {
74        Ok(Self {
75            cell: Arc::new(SharedOnceCell::open(path)?),
76            header_sidecar: subetha_core::HandshakeHeader::new(),
77            ring_sidecar: Box::new(subetha_core::ObservationRing::new()),
78        })
79    }
80
81    /// Fast path: returns the value when loaded, otherwise None.
82    pub fn try_get(&self) -> Option<T> {
83        let r = self.cell.get();
84        self.ring_sidecar.push_op(
85            crate::sidecar_ops::lazy_config::OP_GET,
86            if r.is_none() { 2 } else { 0 },
87        );
88        r
89    }
90
91    /// True when the config has been loaded.
92    pub fn is_loaded(&self) -> bool {
93        self.cell.is_initialized()
94    }
95
96    /// Get the value if cached. Otherwise, race to be the canonical
97    /// fetcher: the CAS winner runs `fetcher` and publishes; CAS
98    /// losers block until the winner publishes, then return the
99    /// canonical value.
100    ///
101    /// Across all processes mapping this file, `fetcher` runs at
102    /// most once per process AND only one process's result becomes
103    /// canonical. (In practice with the CAS-then-fetch protocol
104    /// from `SharedOnceCell::get_or_init`, the winner is the
105    /// only one to actually run the fetcher.)
106    pub fn get_or_fetch<F: FnOnce() -> T>(&self, fetcher: F) -> T {
107        let was_loaded = self.cell.is_initialized();
108        let v = self.cell.get_or_init(fetcher);
109        self.ring_sidecar.push_op(
110            crate::sidecar_ops::lazy_config::OP_FETCH,
111            if was_loaded { 0 } else { 1 }, // cold-fetch path
112        );
113        v
114    }
115
116    /// Force-set the value without going through a fetcher. Useful
117    /// for testing or for an admin-set-config workflow. Returns
118    /// `true` if this caller's value became canonical (it won the
119    /// CAS), `false` when the cell was already initialised.
120    pub fn force_set(&self, value: T) -> bool {
121        let ok = self.cell.set(value);
122        self.ring_sidecar.push_op(
123            crate::sidecar_ops::lazy_config::OP_FETCH,
124            if ok { 0 } else { 1 }, // lost the race
125        );
126        ok
127    }
128
129    pub fn flush(&self) -> Result<(), LazyConfigError> {
130        Ok(self.cell.flush()?)
131    }
132
133    /// Non-blocking flush: schedules a writeback via the OS.
134    /// Note: Windows is only partially async (sync to page cache,
135    /// not to disk).
136    pub fn flush_async(&self) -> Result<(), LazyConfigError> {
137        Ok(self.cell.flush_async()?)
138    }
139}
140
141#[cfg(test)]
142mod tests {
143    use super::*;
144    use std::sync::atomic::{AtomicU32, Ordering};
145    use std::thread;
146    use std::time::Duration;
147
148    fn tmp(name: &str) -> std::path::PathBuf {
149        let mut p = std::env::temp_dir();
150        let pid = std::process::id();
151        p.push(format!("subetha-lazyconf-{name}-{pid}.bin"));
152        p
153    }
154
155    #[test]
156    fn unloaded_try_get_returns_none() {
157        let p = tmp("unloaded");
158        let c: LazyConfig<u64> = LazyConfig::create(&p).unwrap();
159        assert_eq!(c.try_get(), None);
160        assert!(!c.is_loaded());
161        std::fs::remove_file(&p).ok();
162    }
163
164    #[test]
165    fn first_fetch_loads_value() {
166        let p = tmp("first");
167        let c: LazyConfig<u64> = LazyConfig::create(&p).unwrap();
168        let v = c.get_or_fetch(|| 12345);
169        assert_eq!(v, 12345);
170        assert!(c.is_loaded());
171        assert_eq!(c.try_get(), Some(12345));
172        std::fs::remove_file(&p).ok();
173    }
174
175    #[test]
176    fn second_fetch_returns_cached() {
177        let p = tmp("second");
178        let c: LazyConfig<u64> = LazyConfig::create(&p).unwrap();
179        // Prime the cache; this fetcher must run.
180        let _primed = c.get_or_fetch(|| 100);
181        // Second fetcher must not run; if it ran, the panic fires.
182        let v = c.get_or_fetch(|| panic!("must not run on loaded config"));
183        assert_eq!(v, 100);
184        std::fs::remove_file(&p).ok();
185    }
186
187    #[test]
188    fn fetch_runs_at_most_once_under_concurrency() {
189        let p = tmp("concurrent");
190        let c: LazyConfig<u64> = LazyConfig::create(&p).unwrap();
191        let cell = c.cell.clone();
192        let runs = Arc::new(AtomicU32::new(0));
193        let mut handles = vec![];
194        for _ in 0..16 {
195            let cell = cell.clone();
196            let runs = runs.clone();
197            handles.push(thread::spawn(move || {
198                cell.get_or_init(|| {
199                    runs.fetch_add(1, Ordering::AcqRel);
200                    // Simulate slow fetch so workers actually race.
201                    thread::sleep(Duration::from_millis(5));
202                    9999u64
203                })
204            }));
205        }
206        let results: Vec<u64> = handles.into_iter().map(|h| h.join().unwrap()).collect();
207        assert!(results.iter().all(|v| *v == 9999));
208        assert_eq!(runs.load(Ordering::Acquire), 1,
209                   "fetcher must run EXACTLY once across 16 concurrent callers");
210        std::fs::remove_file(&p).ok();
211    }
212
213    #[test]
214    fn cross_handle_load_visible() {
215        let p = tmp("cross-handle");
216        let a: LazyConfig<u64> = LazyConfig::create(&p).unwrap();
217        let b: LazyConfig<u64> = LazyConfig::open(&p).unwrap();
218        // Process A fetches; B sees the result without running.
219        let v_a = a.get_or_fetch(|| 7777);
220        let v_b = b.get_or_fetch(|| panic!("must not run after A loaded"));
221        assert_eq!(v_a, 7777);
222        assert_eq!(v_b, 7777);
223        assert!(b.is_loaded());
224        std::fs::remove_file(&p).ok();
225    }
226
227    #[test]
228    fn force_set_wins_first_then_loses() {
229        let p = tmp("force-set");
230        let c: LazyConfig<u64> = LazyConfig::create(&p).unwrap();
231        assert!(c.force_set(42));
232        assert!(!c.force_set(99));
233        assert_eq!(c.try_get(), Some(42));
234        std::fs::remove_file(&p).ok();
235    }
236
237    #[test]
238    fn disk_persistence_loaded_config_survives_reopen() {
239        let p = tmp("disk-persist");
240        {
241            let c: LazyConfig<u64> = LazyConfig::create(&p).unwrap();
242            let _primed = c.get_or_fetch(|| 8888);
243            c.flush().unwrap();
244        }
245        let c2: LazyConfig<u64> = LazyConfig::open(&p).unwrap();
246        assert!(c2.is_loaded());
247        assert_eq!(c2.try_get(), Some(8888));
248        // Subsequent fetch returns cached value.
249        let v = c2.get_or_fetch(|| panic!("must not run on already-loaded config"));
250        assert_eq!(v, 8888);
251        std::fs::remove_file(&p).ok();
252    }
253
254    #[test]
255    fn struct_config_round_trip() {
256        #[derive(Clone, Copy, Debug, PartialEq)]
257        #[repr(C)]
258        struct ConfigBytes { max_conns: u32, ttl_ms: u32, debug: u32 }
259        let p = tmp("struct-config");
260        let c: LazyConfig<ConfigBytes> = LazyConfig::create(&p).unwrap();
261        let loaded = c.get_or_fetch(|| ConfigBytes {
262            max_conns: 256,
263            ttl_ms: 60_000,
264            debug: 1,
265        });
266        assert_eq!(loaded, ConfigBytes { max_conns: 256, ttl_ms: 60_000, debug: 1 });
267        std::fs::remove_file(&p).ok();
268    }
269}
subetha_cxc/lazy_config.rs

subetha_cxc/
lazy_config.rs
