fxcp_core/consistency/
wal.rs

1// SPDX-License-Identifier: GPL-2.0-or-later
2// Copyright (C) 2025 Joel Wirāmu Pauling <aenertia@aenertia.net>
3//
4// fxcp-core/src/consistency/wal.rs  --  Write-ahead log for crash recovery
5
6//! In-memory write-ahead log for tracking pending operations.
7//! Enables crash recovery by replaying uncommitted operations on restart.
8
9use std::sync::atomic::{AtomicUsize, AtomicU64, Ordering};
10use std::sync::Arc;
11use crossbeam::queue::SegQueue;
12use std::time::Instant;
13use dashmap::DashMap;
14use std::sync::atomic::AtomicBool;
15use tracing::warn;
16
17static BATCH_TOKENS: AtomicU64 = AtomicU64::new(crate::constants::WAL_BATCH_TOKENS_INITIAL as u64);
18const MAX_WRITERS_BEFORE_BARRIER: usize = crate::constants::WAL_MAX_WRITERS_BEFORE_BARRIER;
19
20#[derive(Debug, Clone, Copy, PartialEq, Eq)]
21pub enum WalOpKind {
22    Write,
23    Rename,
24    Truncate,
25    Metadata,
26    Barrier,
27}
28
29#[derive(Debug)]
30/// A single entry in the write-ahead log.
31pub struct WalEntry {
32    pub id: u64,
33    pub inode: u64,
34    pub kind: WalOpKind,
35    pub seq: u64,
36    pub timestamp: Instant,
37    pub completed: AtomicBool,
38}
39
40struct InodeState {
41    active_writers: AtomicUsize,
42    pending_barriers: AtomicUsize,
43    exclusive_locked: AtomicBool,
44    _last_seq: AtomicU64,
45}
46
47/// In-memory write-ahead log with per-inode barrier/shared locking.
48pub struct InMemoryWal {
49    log_queue: SegQueue<Arc<WalEntry>>,
50    inode_states: DashMap<u64, Arc<InodeState>>,
51    global_seq: AtomicU64,
52}
53
54impl InMemoryWal {
55    /// Create a new empty WAL.
56    pub fn new() -> Arc<Self> {
57        Arc::new(Self {
58            log_queue: SegQueue::new(),
59            inode_states: DashMap::new(),
60            global_seq: AtomicU64::new(0),
61        })
62    }
63
64    /// Allocate the next monotonic sequence number.
65    pub fn get_next_seq(&self) -> u64 {
66        let seq = self.global_seq.fetch_add(1, Ordering::SeqCst);
67        if seq >= u64::MAX - crate::constants::WAL_SEQUENCE_OVERFLOW_THRESHOLD {
68            warn!("WAL sequence approaching overflow. Resetting.");
69            self.global_seq.store(1, Ordering::SeqCst);
70            return 1;
71        }
72        seq
73    }
74
75    /// Acquire a barrier or shared lock on an inode for the given operation kind.
76    pub async fn acquire_barrier(&self, inode: u64, kind: WalOpKind) -> WalGuard<'_> {
77        let state = self.inode_states.entry(inode).or_insert_with(|| Arc::new(InodeState {
78            active_writers: AtomicUsize::new(0),
79            pending_barriers: AtomicUsize::new(0),
80            exclusive_locked: AtomicBool::new(false),
81            _last_seq: AtomicU64::new(0),
82        })).clone();
83
84        let backoff = crossbeam::utils::Backoff::new();
85
86        loop {
87            // Check if exclusive locked OR if a barrier is pending (writer starvation prevention)
88            if state.exclusive_locked.load(Ordering::Acquire) {
89                backoff.snooze();
90                if backoff.is_completed() {
91                    tokio::task::yield_now().await;
92                    backoff.reset();
93                }
94                continue;
95            }
96
97            match kind {
98                WalOpKind::Write | WalOpKind::Metadata => {
99                    // Shared-lock writers: must wait if a barrier/rename is pending
100                    let pending = state.pending_barriers.load(Ordering::Acquire);
101                    if pending > 0 {
102                        // Anti-starvation check: If too many writers are active, respect the barrier
103                        let active = state.active_writers.load(Ordering::Acquire);
104                        if active >= MAX_WRITERS_BEFORE_BARRIER {
105                            backoff.snooze();
106                            if backoff.is_completed() {
107                                tokio::task::yield_now().await;
108                                backoff.reset();
109                            }
110                            continue;
111                        }
112                    }
113                    
114                    state.active_writers.fetch_add(1, Ordering::SeqCst);
115                    // Double check if exclusive lock was grabbed in the interim
116                    if state.exclusive_locked.load(Ordering::Acquire) {
117                        state.active_writers.fetch_sub(1, Ordering::SeqCst);
118                        continue;
119                    }
120                    break;
121                },
122                WalOpKind::Rename | WalOpKind::Truncate | WalOpKind::Barrier => {
123                    // Increment pending barriers count to signal writers to back off
124                    state.pending_barriers.fetch_add(1, Ordering::SeqCst);
125                    
126                    loop {
127                        if state.active_writers.load(Ordering::Acquire) == 0
128                            && state.exclusive_locked.compare_exchange(
129                                false, true, Ordering::AcqRel, Ordering::Relaxed
130                            ).is_ok() {
131                                // Re-verify: a writer may have snuck in between the
132                                // active_writers check and the CAS (TOCTOU window).
133                                if state.active_writers.load(Ordering::Acquire) != 0 {
134                                    state.exclusive_locked.store(false, Ordering::Release);
135                                    continue;
136                                }
137                                break;
138                            }
139                        
140                        backoff.snooze();
141                        if backoff.is_completed() {
142                            // Yield inside the inner loop while holding pending_barriers count high
143                            // This effectively blocks new writers, draining the pool
144                            tokio::task::yield_now().await;
145                            backoff.reset();
146                        }
147                    }
148                    // Decrement pending count once lock is acquired
149                    state.pending_barriers.fetch_sub(1, Ordering::SeqCst);
150                    break;
151                },
152            }
153        }
154
155        let id = self.get_next_seq();
156        let entry = Arc::new(WalEntry {
157            id,
158            inode,
159            kind,
160            seq: 0,
161            timestamp: Instant::now(),
162            completed: AtomicBool::new(false),
163        });
164        
165        self.log_queue.push(entry.clone());
166        self.trigger_batch_check();
167
168        WalGuard {
169            wal: self,
170            inode,
171            kind,
172            entry,
173        }
174    }
175
176    fn trigger_batch_check(&self) -> bool {
177        let tokens = BATCH_TOKENS.fetch_sub(1, Ordering::SeqCst);
178        if tokens == 0 {
179            BATCH_TOKENS.store(crate::constants::WAL_BATCH_TOKENS_INITIAL as u64, Ordering::SeqCst);
180            return true;
181        }
182        false
183    }
184
185    fn release(&self, inode: u64, kind: WalOpKind, entry: &Arc<WalEntry>) {
186        entry.completed.store(true, Ordering::Release);
187        if let Some(state) = self.inode_states.get(&inode) {
188            match kind {
189                WalOpKind::Write | WalOpKind::Metadata => {
190                    state.active_writers.fetch_sub(1, Ordering::SeqCst);
191                },
192                WalOpKind::Rename | WalOpKind::Truncate | WalOpKind::Barrier => {
193                    state.exclusive_locked.store(false, Ordering::SeqCst);
194                },
195            }
196        }
197    }
198}
199
200/// RAII guard that releases the WAL lock on drop.
201pub struct WalGuard<'a> {
202    wal: &'a InMemoryWal,
203    inode: u64,
204    kind: WalOpKind,
205    entry: Arc<WalEntry>,
206}
207
208impl<'a> Drop for WalGuard<'a> {
209    fn drop(&mut self) {
210        self.wal.release(self.inode, self.kind, &self.entry);
211    }
212}
213
214#[cfg(test)]
215#[allow(clippy::unwrap_used, clippy::expect_used)]
216mod wal_tests {
217    use super::*;
218
219    fn rt() -> tokio::runtime::Runtime {
220        tokio::runtime::Builder::new_multi_thread()
221            .worker_threads(2)
222            .enable_all()
223            .build()
224            .unwrap()
225    }
226
227    #[test]
228    fn test_shared_write_lock_permits_concurrent_access() {
229        let wal = InMemoryWal::new();
230        rt().block_on(async {
231            let guard1 = wal.acquire_barrier(1, WalOpKind::Write).await;
232            let guard2 = wal.acquire_barrier(1, WalOpKind::Write).await;
233            // Both Write guards acquired on same inode  --  shared access works
234            assert_eq!(guard1.kind, WalOpKind::Write);
235            assert_eq!(guard2.kind, WalOpKind::Write);
236            drop(guard1);
237            drop(guard2);
238        });
239    }
240
241    #[test]
242    fn test_write_lock_increments_active_writers() {
243        let wal = InMemoryWal::new();
244        rt().block_on(async {
245            let guard = wal.acquire_barrier(1, WalOpKind::Write).await;
246            // While guard is held, active_writers should be 1
247            let state = wal.inode_states.get(&1).unwrap();
248            assert_eq!(state.active_writers.load(Ordering::SeqCst), 1);
249            drop(guard);
250            // After drop, active_writers should be 0
251            let state = wal.inode_states.get(&1).unwrap();
252            assert_eq!(state.active_writers.load(Ordering::SeqCst), 0);
253        });
254    }
255
256    #[test]
257    fn test_exclusive_lock_release_cycle() {
258        let wal = InMemoryWal::new();
259        // Directly test the release mechanism for exclusive ops
260        // by manually constructing state and verifying release() clears it
261        wal.inode_states.insert(50, Arc::new(InodeState {
262            active_writers: AtomicUsize::new(0),
263            pending_barriers: AtomicUsize::new(0),
264            exclusive_locked: AtomicBool::new(true),
265            _last_seq: AtomicU64::new(0),
266        }));
267        let entry = Arc::new(WalEntry {
268            id: 1,
269            inode: 50,
270            kind: WalOpKind::Rename,
271            seq: 0,
272            timestamp: Instant::now(),
273            completed: AtomicBool::new(false),
274        });
275        // Release should clear exclusive_locked and mark entry completed
276        wal.release(50, WalOpKind::Rename, &entry);
277        let state = wal.inode_states.get(&50).unwrap();
278        assert!(!state.exclusive_locked.load(Ordering::SeqCst));
279        assert!(entry.completed.load(Ordering::SeqCst));
280    }
281
282    #[test]
283    fn test_guard_drop_releases() {
284        let wal = InMemoryWal::new();
285        rt().block_on(async {
286            let guard = wal.acquire_barrier(3, WalOpKind::Write).await;
287            let entry_clone = guard.entry.clone();
288            assert!(!entry_clone.completed.load(Ordering::SeqCst));
289            drop(guard);
290            assert!(entry_clone.completed.load(Ordering::SeqCst));
291            // After drop, state is clean  --  new Write succeeds
292            let guard2 = wal.acquire_barrier(3, WalOpKind::Write).await;
293            assert_eq!(guard2.kind, WalOpKind::Write);
294            drop(guard2);
295        });
296    }
297
298    #[test]
299    fn test_different_inodes_independent() {
300        let wal = InMemoryWal::new();
301        rt().block_on(async {
302            let guard_a = wal.acquire_barrier(10, WalOpKind::Write).await;
303            let guard_b = wal.acquire_barrier(20, WalOpKind::Write).await;
304            assert_eq!(guard_a.inode, 10);
305            assert_eq!(guard_b.inode, 20);
306            let state_a = wal.inode_states.get(&10).unwrap();
307            let state_b = wal.inode_states.get(&20).unwrap();
308            assert_eq!(state_a.active_writers.load(Ordering::SeqCst), 1);
309            assert_eq!(state_b.active_writers.load(Ordering::SeqCst), 1);
310            drop(guard_a);
311            drop(guard_b);
312        });
313    }
314
315    #[test]
316    fn test_metadata_increments_active_writers() {
317        let wal = InMemoryWal::new();
318        rt().block_on(async {
319            let guard = wal.acquire_barrier(30, WalOpKind::Metadata).await;
320            assert_eq!(guard.kind, WalOpKind::Metadata);
321            let state = wal.inode_states.get(&30).unwrap();
322            assert!(!state.exclusive_locked.load(Ordering::SeqCst));
323            assert_eq!(state.active_writers.load(Ordering::SeqCst), 1);
324            drop(guard);
325            let state = wal.inode_states.get(&30).unwrap();
326            assert_eq!(state.active_writers.load(Ordering::SeqCst), 0);
327        });
328    }
329
330    #[test]
331    fn test_metadata_respects_barriers() {
332        let wal = InMemoryWal::new();
333        rt().block_on(async {
334            let rename_guard = wal.acquire_barrier(40, WalOpKind::Rename).await;
335            {
336                let state = wal.inode_states.get(&40).unwrap();
337                assert!(state.exclusive_locked.load(Ordering::SeqCst));
338            }
339
340            drop(rename_guard);
341
342            let meta_guard = wal.acquire_barrier(40, WalOpKind::Metadata).await;
343            {
344                let state = wal.inode_states.get(&40).unwrap();
345                assert_eq!(state.active_writers.load(Ordering::SeqCst), 1);
346            }
347
348            let meta_guard2 = wal.acquire_barrier(40, WalOpKind::Metadata).await;
349            {
350                let state = wal.inode_states.get(&40).unwrap();
351                assert_eq!(state.active_writers.load(Ordering::SeqCst), 2);
352            }
353
354            drop(meta_guard);
355            drop(meta_guard2);
356            {
357                let state = wal.inode_states.get(&40).unwrap();
358                assert_eq!(state.active_writers.load(Ordering::SeqCst), 0);
359            }
360        });
361    }
362
363    #[test]
364    fn test_metadata_then_rename_sequential() {
365        let wal = InMemoryWal::new();
366        rt().block_on(async {
367            let meta_guard = wal.acquire_barrier(50, WalOpKind::Metadata).await;
368            {
369                let state = wal.inode_states.get(&50).unwrap();
370                assert_eq!(state.active_writers.load(Ordering::SeqCst), 1);
371            }
372            drop(meta_guard);
373            {
374                let state = wal.inode_states.get(&50).unwrap();
375                assert_eq!(state.active_writers.load(Ordering::SeqCst), 0);
376            }
377
378            let rename_guard = wal.acquire_barrier(50, WalOpKind::Rename).await;
379            {
380                let state = wal.inode_states.get(&50).unwrap();
381                assert!(state.exclusive_locked.load(Ordering::SeqCst));
382            }
383            drop(rename_guard);
384            {
385                let state = wal.inode_states.get(&50).unwrap();
386                assert!(!state.exclusive_locked.load(Ordering::SeqCst));
387            }
388        });
389    }
390
391    #[test]
392    fn test_get_next_seq_increments() {
393        let wal = InMemoryWal::new();
394        let s1 = wal.get_next_seq();
395        let s2 = wal.get_next_seq();
396        assert_eq!(s1, 0);
397        assert_eq!(s2, 1);
398    }
399
400    /// Exclusive (barrier) locks must never overlap with active shared (writer)
401    /// locks on the same inode  --  detects the TOCTOU between the
402    /// `active_writers == 0` check and the `exclusive_locked` CAS.
403    #[test]
404    fn test_wal_barrier_no_concurrent_writer() {
405        use std::sync::atomic::AtomicBool;
406
407        let wal = Arc::new(InMemoryWal::new());
408        let violation = Arc::new(AtomicBool::new(false));
409        let inode: u64 = 9999;
410
411        let runtime = tokio::runtime::Builder::new_multi_thread()
412            .worker_threads(8)
413            .enable_all()
414            .build()
415            .unwrap();
416
417        runtime.block_on(async {
418            let mut handles = Vec::new();
419
420            {
421                let wal = wal.clone();
422                let violation = violation.clone();
423                handles.push(tokio::spawn(async move {
424                    for _ in 0..5 {
425                        let guard = wal.acquire_barrier(inode, WalOpKind::Write).await;
426                        if wal.inode_states.get(&inode).unwrap()
427                            .exclusive_locked.load(Ordering::Acquire)
428                        {
429                            violation.store(true, Ordering::Release);
430                        }
431                        tokio::task::yield_now().await;
432                        drop(guard);
433                        tokio::task::yield_now().await;
434                    }
435                }));
436            }
437
438            {
439                let wal = wal.clone();
440                let violation = violation.clone();
441                handles.push(tokio::spawn(async move {
442                    for _ in 0..5 {
443                        let guard = wal.acquire_barrier(inode, WalOpKind::Rename).await;
444                        if wal.inode_states.get(&inode).unwrap()
445                            .active_writers.load(Ordering::Acquire) != 0
446                        {
447                            violation.store(true, Ordering::Release);
448                        }
449                        tokio::task::yield_now().await;
450                        drop(guard);
451                        tokio::task::yield_now().await;
452                    }
453                }));
454            }
455
456            for h in handles {
457                h.await.unwrap();
458            }
459        });
460
461        assert!(
462            !violation.load(Ordering::Acquire),
463            "TOCTOU violation: exclusive lock held while writers active, or writer running under exclusive lock"
464        );
465    }
466
467    #[test]
468    fn test_wal_truncate_takes_exclusive_lock() {
469        let wal = InMemoryWal::new();
470        rt().block_on(async {
471            let guard = wal.acquire_barrier(100, WalOpKind::Truncate).await;
472            assert_eq!(guard.kind, WalOpKind::Truncate);
473            let state = wal.inode_states.get(&100).unwrap();
474            assert!(state.exclusive_locked.load(Ordering::SeqCst));
475            assert_eq!(state.active_writers.load(Ordering::SeqCst), 0);
476            drop(guard);
477            let state = wal.inode_states.get(&100).unwrap();
478            assert!(!state.exclusive_locked.load(Ordering::SeqCst));
479        });
480    }
481
482    #[test]
483    fn test_wal_barrier_takes_exclusive_lock() {
484        let wal = InMemoryWal::new();
485        rt().block_on(async {
486            let guard = wal.acquire_barrier(200, WalOpKind::Barrier).await;
487            assert_eq!(guard.kind, WalOpKind::Barrier);
488            let state = wal.inode_states.get(&200).unwrap();
489            assert!(state.exclusive_locked.load(Ordering::SeqCst));
490            assert_eq!(state.active_writers.load(Ordering::SeqCst), 0);
491            drop(guard);
492            let state = wal.inode_states.get(&200).unwrap();
493            assert!(!state.exclusive_locked.load(Ordering::SeqCst));
494        });
495    }
496
497    #[test]
498    fn test_wal_sequence_overflow_resets() {
499        let wal = InMemoryWal::new();
500        // Position global_seq at the overflow boundary
501        wal.global_seq.store(u64::MAX - 10000, Ordering::SeqCst);
502        // get_next_seq() detects seq >= u64::MAX - 10000, resets to 1, returns 1
503        let seq = wal.get_next_seq();
504        assert_eq!(seq, 1, "overflow should trigger reset and return 1");
505        // After reset, global_seq is 1; subsequent calls resume from low values
506        let _seq2 = wal.get_next_seq();
507        let seq3 = wal.get_next_seq();
508        assert!(seq3 < 100, "post-overflow sequences should be small, got {}", seq3);
509    }
510}
fxcp_core/consistency/wal.rs

fxcp_core/consistency/
wal.rs
