Skip to main content

frozen_core/memmap/
mod.rs

1//! Memory-mapped typed slot storage with per-slot concurrency and epoch durability
2//!
3//! ## Example
4//!
5//! ```
6//! use frozen_core::memmap::{MemMap, MemMapCfg};
7//! use std::time::Duration;
8//!
9//! let dir = tempfile::tempdir().unwrap();
10//! let path = dir.path().join("memmap_demo.bin");
11//!
12//! let cfg = MemMapCfg {
13//!     module_id: 1,
14//!     path,
15//!     version: 1,
16//!     initial_t: 8,
17//!     sync_interval: Some(Duration::from_millis(50)),
18//! };
19//!
20//! let memmap = MemMap::<u64>::new(cfg).unwrap();
21//! assert_eq!(memmap.length(), 8 * std::mem::size_of::<u64>());
22//!
23//! let ticket = memmap.write(&0xDEAD_BEEF_CAFE_BABE, 0).unwrap();
24//! ticket.wait().unwrap();
25//! assert!(ticket.is_durable());
26//!
27//! assert_eq!(memmap.read(0).unwrap(), 0xDEAD_BEEF_CAFE_BABE);
28//! assert!(memmap.delete().is_ok());
29//! ```
30
31use crate::{
32    ack::{AckTicket, Completion, SyncTrigger, TEpoch},
33    error::{BindModule, ErrCode, FrozenError, FrozenResult},
34    file::{File, FileCfg},
35    hints,
36};
37use core::marker::PhantomData;
38use std::sync::{Arc, Condvar, Mutex, RwLock, atomic};
39
40mod interface;
41
42#[cfg(any(target_os = "linux", target_os = "macos"))]
43mod posix;
44
45#[cfg(target_os = "windows")]
46mod windows;
47
48pub(in crate::memmap) use interface::MemMapInterface;
49
50#[cfg(any(target_os = "linux", target_os = "macos"))]
51pub(in crate::memmap) type PlatformMemMap = posix::POSIXMemMap;
52
53#[cfg(target_os = "windows")]
54pub(in crate::memmap) type PlatformMemMap = windows::WINMemMap;
55
56/// Error codes for [`MemMap`] module
57pub(in crate::memmap) mod err {
58    use super::{ErrCode, FrozenError, FrozenResult};
59
60    /// Domain Id for [`MemMap`] is **16**
61    const ERRDOMAIN: u8 = 0x10;
62
63    /// internal fuck up (hault and catch fire)
64    pub const HCF: ErrCode = ErrCode::new(0x02, "hault and catch fire");
65
66    /// unknown error (fallback)
67    pub const UNK: ErrCode = ErrCode::new(0x04, "unknown error");
68
69    /// no more memory available
70    pub const NMM: ErrCode = ErrCode::new(0x06, "not enough memory available on the device");
71
72    /// syncing error
73    pub const SYN: ErrCode = ErrCode::new(0x08, "failed to sync/flush data to storage device");
74
75    /// no write/read perm
76    pub const PRM: ErrCode = ErrCode::new(0x0A, "missing permissions for IO");
77
78    /// flush_tx error (unable to spawn)
79    pub const FXE: ErrCode = ErrCode::new(0x10, "unable to spawn flush_tx");
80
81    /// type `T` is zero sized
82    pub const ZRO: ErrCode = ErrCode::new(0x0C, "type T must not be zero sized");
83
84    /// type `T` implements drop
85    pub const DRP: ErrCode = ErrCode::new(0x12, "type T must not implement `Drop`");
86
87    /// type `T` is not 8 bytes aligned
88    pub const ALN: ErrCode = ErrCode::new(0x14, "type T must be 8-bytes aligned");
89
90    /// `size_of::<T>()` is not multiple of 8
91    pub const SZE: ErrCode = ErrCode::new(0x16, "`size_of::<T>()` must be multiple of 8 bytes");
92
93    /// type `T` is invalid
94    /// (e.g. `size_of::<T>()` exceeds page size)
95    pub const INV: ErrCode =
96        ErrCode::new(0x18, "`size_of::<T>()` must not exceed system page size");
97
98    /// parent directory does not exist
99    pub const DIR: ErrCode = ErrCode::new(0x1A, "directory does not exist");
100
101    /// invalid path to file
102    pub const PTH: ErrCode = ErrCode::new(0x1C, "invalid path to file");
103
104    /// file already exists
105    pub const EXS: ErrCode = ErrCode::new(0x1E, "file already exists");
106
107    /// file does not exist
108    pub const NEX: ErrCode = ErrCode::new(0x20, "file does not exist");
109
110    /// unable to grow memory-mapped file
111    pub const GRW: ErrCode = ErrCode::new(0x22, "unable to grow memory-mapped file");
112
113    /// state conflict
114    pub const CAS: ErrCode = ErrCode::new(0x24, "compare and swap check failed");
115
116    #[inline]
117    pub(in crate::memmap) fn raw_error<R, E: std::fmt::Display>(
118        code: ErrCode,
119        error: E,
120    ) -> FrozenResult<R> {
121        let err = FrozenError::unbound_raw(ERRDOMAIN, code, error);
122        Err(err)
123    }
124
125    #[inline]
126    pub(in crate::memmap) fn default_error<R>(code: ErrCode) -> FrozenResult<R> {
127        let err = FrozenError::unbound(ERRDOMAIN, code, "");
128        Err(err)
129    }
130
131    #[inline]
132    pub(in crate::memmap) fn make_raw_error<E: std::fmt::Display>(
133        code: ErrCode,
134        error: E,
135    ) -> FrozenError {
136        FrozenError::unbound_raw(ERRDOMAIN, code, error)
137    }
138}
139
140/// Configurations for [`MemMap`]
141///
142/// ## Example
143///
144/// ```
145/// use frozen_core::memmap::MemMapCfg;
146/// use std::time::Duration;
147///
148/// let cfg = MemMapCfg {
149///     module_id: 1,
150///     path: std::path::PathBuf::from("test.db"),
151///     version: 1,
152///     initial_t: 0x10,
153///     sync_interval: Some(Duration::from_millis(0x64)),
154/// };
155///
156/// assert_eq!(cfg.version, 1);
157/// assert_eq!(cfg.initial_t, 0x10);
158/// assert_eq!(cfg.sync_interval, Some(Duration::from_millis(0x64)));
159/// ```
160#[derive(Debug, Clone)]
161pub struct MemMapCfg {
162    /// Identifier used while error propagation
163    pub module_id: u8,
164
165    /// Base path for/of the file
166    ///
167    /// ## Notes
168    ///
169    /// - The caller must make sure that the path represents a file and all the parent directories
170    ///   included in the path do exist
171    /// - The actual file created on disk is derived by appending the [`version`](Self::version) to
172    ///   the base file name (e.g. `data.bin` with version `1` becomes `data_1.bin`)
173    pub path: std::path::PathBuf,
174
175    /// Schema or layout version of the underlying memory-mapped file
176    ///
177    /// Because the binary layout on disk is constructed according to the initial type `T` and
178    /// structural parameters, [`MemMap`] strictly requires type invariants across a file's lifetime
179    ///
180    /// When an application evolves its data structures or migrates `T`, incrementing `version`
181    /// ensures a new versioned backing file (`<name>_<ver>`) is created on disk without corrupting
182    /// or clashing with previously persisted data
183    pub version: u32,
184
185    /// Number of `T` elements to pre-allocate on fs when [`MemMap`] is initialized
186    ///
187    /// Initial file length will be `core::mem::size_of::<T>() * initial_t` (bytes)
188    pub initial_t: usize,
189
190    /// Optional interval for the background sync thread
191    ///
192    /// If `Some(duration)`, a background worker thread is spawned which flushes dirty mapped pages
193    /// at this interval; and if `None`, background sync is disabled and durability advances via
194    /// manual sync or forced tickets
195    pub sync_interval: Option<std::time::Duration>,
196}
197
198impl MemMapCfg {
199    /// Computes the versioned file path by appending `_{version}` to the base file stem
200    ///
201    /// Preserves file extension if present (e.g., `data.bin` with version `1` becomes `data_1.bin`)
202    #[inline]
203    fn versioned_path(&self) -> std::path::PathBuf {
204        let parent = self.path.parent();
205        let stem = self.path.file_stem().and_then(|s| s.to_str()).unwrap_or("");
206        let extension = self.path.extension().and_then(|e| e.to_str());
207
208        let versioned_filename = match extension {
209            Some(ext) if !ext.is_empty() => format!("{}_{}.{}", stem, self.version, ext),
210            _ => format!("{}_{}", stem, self.version),
211        };
212
213        match parent {
214            Some(p) if !p.as_os_str().is_empty() => p.join(versioned_filename),
215            _ => std::path::PathBuf::from(versioned_filename),
216        }
217    }
218}
219
220const WRITER_BIT: u32 = 0x8000_0000;
221const L1_CONTENTION: usize = 0x10;
222const L2_CONTENTION: usize = 0x20;
223
224/// Table of reader-writer spinlocks per slot for memory-mapped items
225#[derive(Debug)]
226struct SlotLocks {
227    locks: Box<[atomic::AtomicU32]>,
228}
229
230impl SlotLocks {
231    fn new(total_slots: usize) -> Self {
232        let mut slots = Vec::with_capacity(total_slots);
233        for _ in 0..total_slots {
234            slots.push(atomic::AtomicU32::new(0));
235        }
236
237        Self { locks: slots.into_boxed_slice() }
238    }
239
240    fn grow(&mut self, additional_slots: usize) {
241        if additional_slots == 0 {
242            return;
243        }
244
245        let total = self.locks.len() + additional_slots;
246        let mut slots = Vec::with_capacity(total);
247        for val in self.locks.iter() {
248            slots.push(atomic::AtomicU32::new(val.load(atomic::Ordering::Relaxed)));
249        }
250        for _ in 0..additional_slots {
251            slots.push(atomic::AtomicU32::new(0));
252        }
253
254        self.locks = slots.into_boxed_slice();
255    }
256
257    #[inline(always)]
258    fn read_lock(&self, index: usize) -> SlotReadGuard<'_> {
259        let val = &self.locks[index];
260        let mut spins = 0;
261
262        loop {
263            let state = val.load(atomic::Ordering::Relaxed);
264            if hints::likely(state & WRITER_BIT == 0)
265                && val
266                    .compare_exchange_weak(
267                        state,
268                        state + 1,
269                        atomic::Ordering::Acquire,
270                        atomic::Ordering::Relaxed,
271                    )
272                    .is_ok()
273            {
274                return SlotReadGuard(val);
275            }
276
277            if hints::likely(spins < L1_CONTENTION) {
278                std::hint::spin_loop();
279            } else if spins < L2_CONTENTION {
280                std::thread::yield_now();
281            } else {
282                let ns = 0x30 << (spins - L2_CONTENTION).min(0x0A);
283                std::thread::sleep(std::time::Duration::from_nanos(ns));
284            }
285
286            spins += 1;
287        }
288    }
289
290    #[inline(always)]
291    fn write_lock(&self, index: usize) -> SlotWriteGuard<'_> {
292        let val = &self.locks[index];
293        let mut spins = 0;
294
295        loop {
296            if val
297                .compare_exchange_weak(
298                    0,
299                    WRITER_BIT,
300                    atomic::Ordering::Acquire,
301                    atomic::Ordering::Relaxed,
302                )
303                .is_ok()
304            {
305                return SlotWriteGuard(val);
306            }
307
308            if hints::likely(spins < L1_CONTENTION) {
309                std::hint::spin_loop();
310            } else if spins < L2_CONTENTION {
311                std::thread::yield_now();
312            } else {
313                let ns = 0x30 << (spins - L2_CONTENTION).min(0x0A);
314                std::thread::sleep(std::time::Duration::from_nanos(ns));
315            }
316
317            spins += 1;
318        }
319    }
320}
321
322/// RAII read guard for a slot lock
323struct SlotReadGuard<'a>(&'a atomic::AtomicU32);
324
325impl Drop for SlotReadGuard<'_> {
326    #[inline(always)]
327    fn drop(&mut self) {
328        self.0.fetch_sub(1, atomic::Ordering::Release);
329    }
330}
331
332/// RAII write guard for a slot lock
333struct SlotWriteGuard<'a>(&'a atomic::AtomicU32);
334
335impl Drop for SlotWriteGuard<'_> {
336    #[inline(always)]
337    fn drop(&mut self) {
338        self.0.store(0, atomic::Ordering::Release);
339    }
340}
341
342/// Internal shared state for [`MemMap`]
343#[derive(Debug)]
344struct MemMapInner {
345    cfg: MemMapCfg,
346    mmap: RwLock<Option<PlatformMemMap>>,
347    slot_locks: RwLock<SlotLocks>,
348    length: atomic::AtomicUsize,
349    completion: Arc<Completion>,
350    sync_condvar: Condvar,
351    sync_mutex: Mutex<bool>,
352    shutdown: atomic::AtomicBool,
353    is_parked: atomic::AtomicBool,
354}
355
356impl SyncTrigger for MemMapInner {
357    fn trigger_sync(&self) -> FrozenResult<()> {
358        if self.cfg.sync_interval.is_some() {
359            {
360                let mut guard = self.sync_mutex.lock().unwrap_or_else(|e| e.into_inner());
361                *guard = true;
362            }
363
364            self.sync_condvar.notify_one();
365            Ok(())
366        } else {
367            self.sync_internal()
368        }
369    }
370}
371
372impl MemMapInner {
373    /// Synchronizes dirty pages to storage if current epoch exceeds durable epoch
374    fn sync_mmap_if_needed(&self, mmap: &PlatformMemMap, sync_len: usize) -> FrozenResult<()> {
375        let target_epoch = self.completion.read_current_epoch();
376        let durable_epoch = self.completion.read_durable_epoch();
377
378        if target_epoch <= durable_epoch {
379            return Ok(());
380        }
381
382        let sync_res = mmap.sync(sync_len);
383        match sync_res {
384            Ok(()) => {
385                self.completion.mark_epoch_as_durable(target_epoch);
386                self.completion.del_err();
387                self.completion.notify_all_listeners();
388
389                Ok(())
390            }
391            Err(e) => {
392                let e = e.with_module(self.cfg.module_id);
393                self.completion.set_err(e.clone());
394                self.completion.notify_all_listeners();
395
396                Err(e)
397            }
398        }
399    }
400
401    fn sync_internal(&self) -> FrozenResult<()> {
402        let guard = self.mmap.read().unwrap_or_else(|e| e.into_inner());
403        let mmap = match guard.as_ref() {
404            Some(m) => m,
405            None => return Ok(()),
406        };
407
408        let length = self.length.load(atomic::Ordering::Acquire);
409        self.sync_mmap_if_needed(mmap, length)
410    }
411}
412
413/// Custom memory-mapped typed array storage with slot-level locking and epoch durability
414///
415/// `MemMap<T>` maps an on-disk [`File`] into process virtual address space as a contiguous array of
416/// fixed-size slots of type `T`
417///
418/// It provides,
419///
420/// - **Slot Synchronization**: Per-slot reader-writer spinlocks (`SlotLocks`) protecting each index
421///   against concurrent write-write and read-write hazards with backoff spin/yield/sleep stages
422/// - **Epoch Durability**: Monotonically advancing write epoch counter incremented on each
423///   mutation, tracked via [`AckTicket`] and synchronized in the background by a dedicated worker
424/// - **Process Isolation**: Advisory kernel-level locking (`flock`/`LockFileEx`) ensuring exclusive
425///   file ownership by a single running instance
426///
427/// ## Constraints
428///
429/// - `T` must be sized and cloneable (`T: Sized + Clone`)
430/// - `T` must be 8-byte aligned (`core::mem::align_of::<T>() == 8`)
431/// - `size_of::<T>()` must be non-zero, a multiple of 8, and must not exceed the host page size
432/// - `T` must not implement `Drop` (`!core::mem::needs_drop::<T>()`)
433/// - Type layout is immutable across file lifetime; schema changes require bumping
434///   [`MemMapCfg::version`]
435///
436/// ## Example
437///
438/// ```
439/// use frozen_core::memmap::{MemMap, MemMapCfg};
440/// use std::time::Duration;
441///
442/// let dir = tempfile::tempdir().unwrap();
443/// let cfg = MemMapCfg {
444///     module_id: 1,
445///     path: dir.path().join("storage.bin"),
446///     version: 1,
447///     initial_t: 4,
448///     sync_interval: Some(Duration::from_millis(100)),
449/// };
450///
451/// let memmap = MemMap::<u64>::new(cfg).unwrap();
452/// assert_eq!(memmap.length(), 4 * std::mem::size_of::<u64>());
453///
454/// let ticket = memmap.write(&12345u64, 0).unwrap();
455/// assert_eq!(memmap.read(0).unwrap(), 12345u64);
456/// assert!(memmap.delete().is_ok());
457/// ```
458pub struct MemMap<T: Sized + Clone> {
459    cfg: MemMapCfg,
460    file: Option<File>,
461    inner: Arc<MemMapInner>,
462    bg_thread: Option<std::thread::JoinHandle<()>>,
463    _marker: PhantomData<T>,
464}
465
466impl<T> core::fmt::Debug for MemMap<T>
467where
468    T: Sized + Clone,
469{
470    fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
471        f.debug_struct("MemMap").field("cfg", &self.cfg).field("inner", &self.inner).finish()
472    }
473}
474
475unsafe impl<T> Send for MemMap<T> where T: Sized + Clone {}
476unsafe impl<T> Sync for MemMap<T> where T: Sized + Clone {}
477
478impl<T> MemMap<T>
479where
480    T: Sized + Clone,
481{
482    fn validate_t(mid: u8) -> FrozenResult<()> {
483        if core::mem::needs_drop::<T>() {
484            return err::default_error(err::DRP).with_module(mid);
485        }
486
487        let size = core::mem::size_of::<T>();
488        if size == 0 {
489            return err::default_error(err::ZRO).with_module(mid);
490        }
491
492        if size % 8 != 0 {
493            return err::default_error(err::SZE).with_module(mid);
494        }
495
496        let align = core::mem::align_of::<T>();
497        if align != 8 {
498            return err::default_error(err::ALN).with_module(mid);
499        }
500
501        let page_size = PlatformMemMap::page_size().with_module(mid)?;
502        if size > page_size {
503            return err::default_error(err::INV).with_module(mid);
504        }
505
506        Ok(())
507    }
508
509    fn validate_path(cfg: &MemMapCfg) -> FrozenResult<()> {
510        let mid = cfg.module_id;
511
512        if cfg.path.as_os_str().is_empty() {
513            return err::default_error(err::PTH).with_module(mid);
514        }
515
516        let stem = cfg.path.file_stem().and_then(|s| s.to_str()).unwrap_or("");
517        if stem.is_empty() {
518            return err::default_error(err::PTH).with_module(mid);
519        }
520
521        if cfg.path.exists() && cfg.path.is_dir() {
522            return err::default_error(err::PTH).with_module(mid);
523        }
524
525        if let Some(parent) = cfg.path.parent() {
526            if !parent.as_os_str().is_empty() && (!parent.exists() || !parent.is_dir()) {
527                return err::default_error(err::DIR).with_module(mid);
528            }
529        }
530
531        Ok(())
532    }
533
534    fn file_cfg(cfg: &MemMapCfg) -> FileCfg {
535        FileCfg {
536            module_id: cfg.module_id,
537            path: cfg.versioned_path(),
538            buffer_size: core::mem::size_of::<T>(),
539            initial_available_buffers: cfg.initial_t,
540            sync_interval: None,
541        }
542    }
543
544    fn from_components(cfg: MemMapCfg, mmap: PlatformMemMap, file: File) -> FrozenResult<Self> {
545        let length = mmap.length();
546        let completion = Arc::new(Completion::default());
547        let sync_interval = cfg.sync_interval;
548        let total_slots = length / core::mem::size_of::<T>();
549        let slot_locks = SlotLocks::new(total_slots);
550
551        let inner = Arc::new(MemMapInner {
552            cfg: cfg.clone(),
553            mmap: RwLock::new(Some(mmap)),
554            slot_locks: RwLock::new(slot_locks),
555            length: atomic::AtomicUsize::new(length),
556            completion: completion.clone(),
557            sync_condvar: Condvar::new(),
558            sync_mutex: Mutex::new(false),
559            shutdown: atomic::AtomicBool::new(false),
560            is_parked: atomic::AtomicBool::new(false),
561        });
562
563        let weak_inner: std::sync::Weak<dyn SyncTrigger> =
564            Arc::downgrade(&inner) as std::sync::Weak<dyn SyncTrigger>;
565        completion.set_sync_trigger(weak_inner);
566
567        let bg_thread = if let Some(interval) = sync_interval {
568            let worker_inner = Arc::clone(&inner);
569            let handle = std::thread::Builder::new().name("frozen-mmap-sync-worker".into()).spawn(
570                move || {
571                    let mut guard =
572                        worker_inner.sync_mutex.lock().unwrap_or_else(|e| e.into_inner());
573
574                    while !worker_inner.shutdown.load(atomic::Ordering::Acquire) {
575                        if *guard {
576                            *guard = false;
577                            drop(guard);
578
579                            let _ = worker_inner.sync_internal();
580                            guard =
581                                worker_inner.sync_mutex.lock().unwrap_or_else(|e| e.into_inner());
582                            continue;
583                        }
584
585                        let target_epoch = worker_inner.completion.read_current_epoch();
586                        let durable_epoch = worker_inner.completion.read_durable_epoch();
587
588                        if target_epoch == durable_epoch {
589                            worker_inner.is_parked.store(true, atomic::Ordering::Release);
590
591                            if worker_inner.completion.read_current_epoch() != durable_epoch {
592                                worker_inner.is_parked.store(false, atomic::Ordering::Release);
593                                continue;
594                            }
595
596                            let new_guard = worker_inner
597                                .sync_condvar
598                                .wait(guard)
599                                .unwrap_or_else(|e| e.into_inner());
600                            guard = new_guard;
601                            worker_inner.is_parked.store(false, atomic::Ordering::Release);
602
603                            if worker_inner.shutdown.load(atomic::Ordering::Acquire) || *guard {
604                                continue;
605                            }
606
607                            if worker_inner.completion.read_current_epoch()
608                                == worker_inner.completion.read_durable_epoch()
609                            {
610                                continue;
611                            }
612
613                            let (new_guard, _) = worker_inner
614                                .sync_condvar
615                                .wait_timeout(guard, interval)
616                                .unwrap_or_else(|e| e.into_inner());
617                            guard = new_guard;
618                        } else {
619                            worker_inner.is_parked.store(false, atomic::Ordering::Release);
620
621                            let (new_guard, _) = worker_inner
622                                .sync_condvar
623                                .wait_timeout(guard, interval)
624                                .unwrap_or_else(|e| e.into_inner());
625                            guard = new_guard;
626                        }
627
628                        if worker_inner.shutdown.load(atomic::Ordering::Acquire) {
629                            break;
630                        }
631
632                        *guard = false;
633                        drop(guard);
634
635                        let _ = worker_inner.sync_internal();
636                        guard = worker_inner.sync_mutex.lock().unwrap_or_else(|e| e.into_inner());
637                    }
638                },
639            );
640
641            let handle = match handle {
642                Ok(h) => h,
643                Err(spawn_err) => {
644                    let mut mmap_guard = inner.mmap.write().unwrap_or_else(|e| e.into_inner());
645                    let mut err = err::make_raw_error(err::FXE, spawn_err);
646                    if let Some(m) = mmap_guard.take() {
647                        if let Err(unmap_err) = m.unmap() {
648                            err.add_suppressed(unmap_err);
649                        }
650                    }
651                    return Err(err.with_module(cfg.module_id));
652                }
653            };
654
655            Some(handle)
656        } else {
657            None
658        };
659
660        Ok(Self { cfg, file: Some(file), inner, bg_thread, _marker: PhantomData })
661    }
662
663    /// Creates and memory-maps a new [`File`] at `cfg.versioned_path()`
664    ///
665    /// Allocates an underlying file with capacity for `initial_t * size_of::<T>()` bytes, acquires
666    /// an exclusive advisory lock on the descriptor, establishes the memory mapping, and spawns the
667    /// background worker for periodic durability flushing
668    ///
669    /// ## Constraints
670    ///
671    /// - Returns [`err::EXS`] if the versioned file already exists on the filesystem; Use
672    ///   [`MemMap::open`] or [`MemMap::open_or_create`] instead
673    /// - Returns [`err::PTH`] or [`err::DIR`] if the base path or its parent directory is invalid
674    /// - Enforces all structural type invariants for `T` (`align == 8`, `size % 8 == 0`, `size <=
675    ///   page_size`, no `Drop`)
676    /// - An exclusive advisory lock is acquired on creation; concurrent callers on the same file
677    ///   path fail
678    ///
679    /// ## Example
680    ///
681    /// ```
682    /// use frozen_core::memmap::{MemMap, MemMapCfg};
683    /// use std::time::Duration;
684    ///
685    /// let dir = tempfile::tempdir().unwrap();
686    /// let cfg = MemMapCfg {
687    ///     module_id: 1,
688    ///     path: dir.path().join("new_memmap.bin"),
689    ///     version: 1,
690    ///     initial_t: 4,
691    ///     sync_interval: Some(Duration::from_millis(100)),
692    /// };
693    ///
694    /// let mmap = MemMap::<u64>::new(cfg.clone()).unwrap();
695    /// assert_eq!(mmap.length(), std::mem::size_of::<u64>() * 4);
696    ///
697    /// // Creating another file at the same path fails
698    /// assert!(MemMap::<u64>::new(cfg).is_err());
699    /// ```
700    pub fn new(cfg: MemMapCfg) -> FrozenResult<Self> {
701        Self::validate_t(cfg.module_id)?;
702        Self::validate_path(&cfg)?;
703
704        let vpath = cfg.versioned_path();
705        if vpath.exists() {
706            return err::default_error(err::EXS).with_module(cfg.module_id);
707        }
708
709        let file_cfg = Self::file_cfg(&cfg);
710        let file = match File::new(file_cfg) {
711            Ok(f) => f,
712            Err(e) if e.reason == err::EXS.reason => {
713                return err::default_error(err::EXS).with_module(cfg.module_id);
714            }
715            Err(e) => return Err(e),
716        };
717
718        let mmap = match PlatformMemMap::new(file.fd(), file.length()) {
719            Ok(m) => m,
720            Err(mut e) => {
721                if let Err(del_err) = file.delete() {
722                    e.add_suppressed(del_err);
723                }
724
725                return Err(e.with_module(cfg.module_id));
726            }
727        };
728
729        Self::from_components(cfg, mmap, file)
730    }
731
732    /// Opens an existing memory-mapped [`File`] at `cfg.versioned_path()`
733    ///
734    /// Opens the underlying versioned file, verifies its length matches the requested layout,
735    /// acquires an exclusive advisory lock on the descriptor, maps its memory region, and spawns
736    /// the background flush worker
737    ///
738    /// ## Constraints
739    ///
740    /// - Returns [`err::NEX`] if the backing file does not exist on disk
741    /// - Returns [`err::PTH`] if the path resolves to a directory
742    /// - Fails with advisory lock error if another handle currently owns the exclusive lock
743    /// - Type `T` invariants must strictly match the binary layout originally written to the file
744    ///
745    /// ## Example
746    ///
747    /// ```
748    /// use frozen_core::memmap::{MemMap, MemMapCfg};
749    /// use std::time::Duration;
750    ///
751    /// let dir = tempfile::tempdir().unwrap();
752    /// let cfg = MemMapCfg {
753    ///     module_id: 1,
754    ///     path: dir.path().join("open_memmap.bin"),
755    ///     version: 1,
756    ///     initial_t: 4,
757    ///     sync_interval: Some(Duration::from_millis(100)),
758    /// };
759    ///
760    /// {
761    ///     let mmap = MemMap::<u64>::new(cfg.clone()).unwrap();
762    ///     assert_eq!(mmap.length(), std::mem::size_of::<u64>() * 4);
763    /// }
764    ///
765    /// let opened = MemMap::<u64>::open(cfg).unwrap();
766    /// assert_eq!(opened.length(), std::mem::size_of::<u64>() * 4);
767    /// ```
768    pub fn open(cfg: MemMapCfg) -> FrozenResult<Self> {
769        Self::validate_t(cfg.module_id)?;
770        Self::validate_path(&cfg)?;
771
772        let vpath = cfg.versioned_path();
773        if !vpath.exists() {
774            return err::default_error(err::NEX).with_module(cfg.module_id);
775        }
776        if vpath.is_dir() {
777            return err::default_error(err::PTH).with_module(cfg.module_id);
778        }
779
780        let file_cfg = Self::file_cfg(&cfg);
781        let file = File::open(file_cfg)?;
782
783        let mmap = PlatformMemMap::new(file.fd(), file.length()).with_module(cfg.module_id)?;
784
785        Self::from_components(cfg, mmap, file)
786    }
787
788    /// Opens an existing memory-mapped [`File`] or creates a new one if it does not yet exist
789    ///
790    /// Idempotent entrypoint that transparently delegates to either open or creation semantics
791    /// based on filesystem presence, maintaining consistent advisory locking and background
792    /// flushing
793    ///
794    /// ## Constraints
795    ///
796    /// - Returns [`err::PTH`] if the resolved path points to an existing directory
797    /// - Returns [`err::DIR`] if parent directory does not exist
798    /// - Fails if another handle currently owns the advisory lock on the target file
799    /// - All structural `T` invariants (`align`, `size`, `needs_drop`) are strictly validated
800    ///
801    /// ## Example
802    ///
803    /// ```
804    /// use frozen_core::memmap::{MemMap, MemMapCfg};
805    /// use std::time::Duration;
806    ///
807    /// let dir = tempfile::tempdir().unwrap();
808    /// let cfg = MemMapCfg {
809    ///     module_id: 1,
810    ///     path: dir.path().join("open_or_create_memmap.bin"),
811    ///     version: 1,
812    ///     initial_t: 2,
813    ///     sync_interval: Some(Duration::from_millis(100)),
814    /// };
815    ///
816    /// {
817    ///     let mmap = MemMap::<u64>::open_or_create(cfg.clone()).unwrap();
818    ///     assert_eq!(mmap.length(), std::mem::size_of::<u64>() * 2);
819    /// }
820    ///
821    /// let reopened = MemMap::<u64>::open_or_create(cfg).unwrap();
822    /// assert_eq!(reopened.length(), std::mem::size_of::<u64>() * 2);
823    /// ```
824    pub fn open_or_create(cfg: MemMapCfg) -> FrozenResult<Self> {
825        Self::validate_t(cfg.module_id)?;
826        Self::validate_path(&cfg)?;
827
828        let vpath = cfg.versioned_path();
829        if vpath.exists() && vpath.is_dir() {
830            return err::default_error(err::PTH).with_module(cfg.module_id);
831        }
832
833        let file_cfg = Self::file_cfg(&cfg);
834        let file = File::open_or_create(file_cfg)?;
835
836        let mmap = PlatformMemMap::new(file.fd(), file.length()).with_module(cfg.module_id)?;
837
838        Self::from_components(cfg, mmap, file)
839    }
840
841    /// Returns the system memory page size in bytes queried from kernel settings
842    ///
843    /// ## Example
844    ///
845    /// ```
846    /// use frozen_core::memmap::MemMap;
847    ///
848    /// let page_size = MemMap::<u64>::page_size().unwrap();
849    /// assert!(page_size >= 4096);
850    /// assert!(page_size.is_power_of_two());
851    /// ```
852    #[inline]
853    pub fn page_size() -> FrozenResult<usize> {
854        PlatformMemMap::page_size()
855    }
856
857    /// Returns the length of the memory mapping in bytes
858    ///
859    /// Corresponds to `initial_t * size_of::<T>()`
860    ///
861    /// ## Example
862    ///
863    /// ```
864    /// use frozen_core::memmap::{MemMap, MemMapCfg};
865    /// use std::time::Duration;
866    ///
867    /// let dir = tempfile::tempdir().unwrap();
868    /// let cfg = MemMapCfg {
869    ///     module_id: 1,
870    ///     path: dir.path().join("length_demo.bin"),
871    ///     version: 1,
872    ///     initial_t: 8,
873    ///     sync_interval: Some(Duration::from_millis(100)),
874    /// };
875    ///
876    /// let mmap = MemMap::<u64>::new(cfg).unwrap();
877    /// assert_eq!(mmap.length(), 8 * std::mem::size_of::<u64>());
878    /// ```
879    #[inline]
880    pub fn length(&self) -> usize {
881        self.inner.length.load(atomic::Ordering::Acquire)
882    }
883
884    /// Dynamically grows the memory-mapped file by `count` additional elements of type `T`
885    ///
886    /// ## Semantics
887    ///
888    /// - Halts all concurrent read and write operations across all slots for the duration of the
889    ///   growth
890    /// - Halts background durability flushing and manual sync operations
891    /// - If uncommitted dirty writes exist (`current_epoch > durable_epoch`), performs a durability
892    ///   sync before modifying file boundaries
893    /// - Explicitly unmaps the current virtual memory mapping
894    /// - Zero-extends the backing file on disk safely via [`File::grow`] and synchronizes file
895    ///   metadata
896    /// - Establishes a new memory mapping over the enlarged file range
897    /// - Dynamically expands per-slot spinlocks to accommodate new slots
898    /// - Updates tracked mapping length and advances durability epoch
899    ///
900    /// ## Crash Durability & Safety
901    ///
902    /// - Existing payload data is flushed to persistent storage before file extension, ensuring no
903    ///   data loss
904    /// - Disk block pre-allocation (`fallocate` / `f_preallocate`) precedes logical size expansion;
905    ///   out-of-space failures leave existing data untouched
906    /// - If file growth fails, the original memory mapping is automatically restored
907    ///
908    /// ## Example
909    ///
910    /// ```
911    /// use frozen_core::memmap::{MemMap, MemMapCfg};
912    /// use std::time::Duration;
913    ///
914    /// let dir = tempfile::tempdir().unwrap();
915    /// let cfg = MemMapCfg {
916    ///     module_id: 1,
917    ///     path: dir.path().join("grow_demo.bin"),
918    ///     version: 1,
919    ///     initial_t: 2,
920    ///     sync_interval: None,
921    /// };
922    ///
923    /// let mmap = MemMap::<u64>::new(cfg).unwrap();
924    /// assert_eq!(mmap.length(), 2 * std::mem::size_of::<u64>());
925    /// mmap.write(&111u64, 0).unwrap();
926    ///
927    /// mmap.grow(3).unwrap();
928    /// assert_eq!(mmap.length(), 5 * std::mem::size_of::<u64>());
929    /// assert_eq!(mmap.read(0).unwrap(), 111u64);
930    /// assert_eq!(mmap.read(4).unwrap(), 0u64);
931    /// ```
932    pub fn grow(&self, count: usize) -> FrozenResult<()> {
933        let mid = self.cfg.module_id;
934        if count == 0 {
935            return Ok(());
936        }
937
938        let size = core::mem::size_of::<T>();
939        let len_to_add = match count.checked_mul(size) {
940            Some(len) => len,
941            None => {
942                return err::raw_error(err::GRW, "file growth calculation overflowed usize")
943                    .with_module(mid);
944            }
945        };
946
947        let file = match self.file.as_ref() {
948            Some(f) => f,
949            None => return err::default_error(err::INV).with_module(mid),
950        };
951
952        // Halt all read and write ops by taking exclusive write lock on slot_locks
953        let mut slot_locks_guard = self.inner.slot_locks.write().unwrap_or_else(|e| e.into_inner());
954
955        // Halt all sync ops (including background sync worker) by taking exclusive write lock on mmap
956        let mut mmap_guard = self.inner.mmap.write().unwrap_or_else(|e| e.into_inner());
957
958        let mmap = match mmap_guard.as_ref() {
959            Some(m) => m,
960            None => return err::default_error(err::INV).with_module(mid),
961        };
962
963        let curr_len = self.inner.length.load(atomic::Ordering::Acquire);
964        let new_length = match curr_len.checked_add(len_to_add) {
965            Some(len) => len,
966            None => {
967                return err::raw_error(err::GRW, "file growth calculation overflowed usize")
968                    .with_module(mid);
969            }
970        };
971
972        // Pre-growth sync: only if needed (curr_epoch > durable_epoch)
973        self.inner.sync_mmap_if_needed(mmap, curr_len)?;
974
975        let old_mmap = match mmap_guard.take() {
976            Some(m) => m,
977            None => return err::default_error(err::INV).with_module(mid),
978        };
979
980        if let Err(e) = old_mmap.unmap().with_module(mid) {
981            if let Ok(restored) = PlatformMemMap::new(file.fd(), curr_len).with_module(mid) {
982                *mmap_guard = Some(restored);
983            }
984            return Err(e);
985        }
986
987        if let Err(e) = file.grow(count).with_module(mid) {
988            if let Ok(restored) = PlatformMemMap::new(file.fd(), curr_len).with_module(mid) {
989                *mmap_guard = Some(restored);
990            }
991            return Err(e);
992        }
993
994        // Flush new file metadata to disk for crash safety
995        if let Err(e) = file.sync().with_module(mid) {
996            if let Ok(restored) = PlatformMemMap::new(file.fd(), new_length).with_module(mid) {
997                *mmap_guard = Some(restored);
998                self.inner.length.store(new_length, atomic::Ordering::Release);
999                slot_locks_guard.grow(count);
1000            }
1001            return Err(e);
1002        }
1003
1004        let new_mmap = match PlatformMemMap::new(file.fd(), new_length).with_module(mid) {
1005            Ok(m) => m,
1006            Err(e) => {
1007                if let Ok(restored) = PlatformMemMap::new(file.fd(), curr_len).with_module(mid) {
1008                    *mmap_guard = Some(restored);
1009                }
1010                return Err(e);
1011            }
1012        };
1013
1014        *mmap_guard = Some(new_mmap);
1015        slot_locks_guard.grow(count);
1016        self.inner.length.store(new_length, atomic::Ordering::Release);
1017
1018        self.inner.completion.increment_current_epoch();
1019
1020        Ok(())
1021    }
1022
1023    /// Returns a reference to the active [`MemMapCfg`] configuration
1024    ///
1025    /// ## Example
1026    ///
1027    /// ```
1028    /// use frozen_core::memmap::{MemMap, MemMapCfg};
1029    /// use std::time::Duration;
1030    ///
1031    /// let dir = tempfile::tempdir().unwrap();
1032    /// let cfg = MemMapCfg {
1033    ///     module_id: 1,
1034    ///     path: dir.path().join("cfg_demo.bin"),
1035    ///     version: 2,
1036    ///     initial_t: 4,
1037    ///     sync_interval: Some(Duration::from_millis(100)),
1038    /// };
1039    ///
1040    /// let mmap = MemMap::<u64>::new(cfg).unwrap();
1041    /// assert_eq!(mmap.cfg().version, 2);
1042    /// assert_eq!(mmap.cfg().initial_t, 4);
1043    /// ```
1044    #[inline]
1045    pub fn cfg(&self) -> &MemMapCfg {
1046        &self.cfg
1047    }
1048
1049    /// Fetch the latest assigned durability epoch
1050    ///
1051    /// The current epoch monotonically increments on each [`MemMap::write`] call
1052    ///
1053    /// ## Example
1054    ///
1055    /// ```
1056    /// use frozen_core::memmap::{MemMap, MemMapCfg};
1057    /// use std::time::Duration;
1058    ///
1059    /// let dir = tempfile::tempdir().unwrap();
1060    /// let cfg = MemMapCfg {
1061    ///     module_id: 1,
1062    ///     path: dir.path().join("epoch_demo.bin"),
1063    ///     version: 1,
1064    ///     initial_t: 4,
1065    ///     sync_interval: Some(Duration::from_millis(100)),
1066    /// };
1067    ///
1068    /// let mmap = MemMap::<u64>::new(cfg).unwrap();
1069    /// assert_eq!(mmap.current_epoch(), 0);
1070    /// mmap.write(&999u64, 0).unwrap();
1071    /// assert_eq!(mmap.current_epoch(), 1);
1072    /// ```
1073    #[inline]
1074    pub fn current_epoch(&self) -> TEpoch {
1075        self.inner.completion.read_current_epoch()
1076    }
1077
1078    /// Fetch the latest confirmed durable epoch flushed to persistent storage
1079    ///
1080    /// ## Example
1081    ///
1082    /// ```
1083    /// use frozen_core::memmap::{MemMap, MemMapCfg};
1084    /// use std::time::Duration;
1085    ///
1086    /// let dir = tempfile::tempdir().unwrap();
1087    /// let cfg = MemMapCfg {
1088    ///     module_id: 1,
1089    ///     path: dir.path().join("durable_demo.bin"),
1090    ///     version: 1,
1091    ///     initial_t: 4,
1092    ///     sync_interval: Some(Duration::from_millis(100)),
1093    /// };
1094    ///
1095    /// let mmap = MemMap::<u64>::new(cfg).unwrap();
1096    /// let ticket = mmap.write(&100u64, 0).unwrap();
1097    /// mmap.sync().unwrap();
1098    /// assert_eq!(mmap.durable_epoch(), ticket.epoch());
1099    /// ```
1100    #[inline]
1101    pub fn durable_epoch(&self) -> TEpoch {
1102        self.inner.completion.read_durable_epoch()
1103    }
1104
1105    /// Synchronize dirty pages and file metadata to stable storage
1106    ///
1107    /// Performs a synchronous two-phase flush,
1108    ///
1109    /// - Flushes modified virtual memory pages to host cache via [`PlatformMemMap::sync`]
1110    /// - Advances the durable epoch to the current write epoch and awakens all pending listeners
1111    ///
1112    /// Returns `Ok(())` immediately if no writes have occurred since the last sync
1113    ///
1114    /// ## Example
1115    ///
1116    /// ```
1117    /// use frozen_core::memmap::{MemMap, MemMapCfg};
1118    ///
1119    /// let dir = tempfile::tempdir().unwrap();
1120    /// let cfg = MemMapCfg {
1121    ///     module_id: 1,
1122    ///     path: dir.path().join("sync_demo.bin"),
1123    ///     version: 1,
1124    ///     initial_t: 4,
1125    ///     sync_interval: None,
1126    /// };
1127    ///
1128    /// let mmap = MemMap::<u64>::new(cfg).unwrap();
1129    /// let ticket = mmap.write(&0xCAFE_BABE, 1).unwrap();
1130    /// assert!(!ticket.is_durable());
1131    ///
1132    /// mmap.sync().unwrap();
1133    /// assert!(ticket.is_durable());
1134    /// ```
1135    #[inline]
1136    pub fn sync(&self) -> FrozenResult<()> {
1137        self.inner.sync_internal()
1138    }
1139
1140    /// Reads an element of type `T` from the memory mapping at the specified `index`
1141    ///
1142    /// ## Constraints
1143    ///
1144    /// - `index * size_of::<T>()` must not overflow `usize`, otherwise returns [`err::INV`]
1145    /// - Reading past the mapped memory region returns [`err::HCF`]
1146    ///
1147    /// ## Example
1148    ///
1149    /// ```
1150    /// use frozen_core::memmap::{MemMap, MemMapCfg};
1151    /// use std::time::Duration;
1152    ///
1153    /// let dir = tempfile::tempdir().unwrap();
1154    /// let cfg = MemMapCfg {
1155    ///     module_id: 1,
1156    ///     path: dir.path().join("read_test.bin"),
1157    ///     version: 1,
1158    ///     initial_t: 4,
1159    ///     sync_interval: Some(Duration::from_millis(100)),
1160    /// };
1161    ///
1162    /// let mmap = MemMap::<u64>::new(cfg).unwrap();
1163    /// mmap.write(&42u64, 0).unwrap();
1164    /// assert_eq!(mmap.read(0).unwrap(), 42u64);
1165    /// ```
1166    #[inline(always)]
1167    pub fn read(&self, index: usize) -> FrozenResult<T> {
1168        let mid = self.cfg.module_id;
1169        let size = core::mem::size_of::<T>();
1170
1171        let offset = match index.checked_mul(size) {
1172            Some(off) => off,
1173            None => return err::default_error(err::INV).with_module(mid),
1174        };
1175
1176        let slot_locks_guard = self.inner.slot_locks.read().unwrap_or_else(|e| e.into_inner());
1177
1178        if hints::unlikely(offset.checked_add(size).is_none_or(|end| end > self.length())) {
1179            return err::default_error(err::HCF).with_module(mid);
1180        }
1181
1182        let _slot_guard = slot_locks_guard.read_lock(index);
1183
1184        let guard = self.inner.mmap.read().unwrap_or_else(|e| e.into_inner());
1185        let mmap = match guard.as_ref() {
1186            Some(m) => m,
1187            None => return err::default_error(err::INV).with_module(mid),
1188        };
1189
1190        let ptr = unsafe { mmap.as_ptr::<T>(offset) };
1191        let val = unsafe { ptr.read() };
1192        drop(guard);
1193        drop(_slot_guard);
1194        drop(slot_locks_guard);
1195        Ok(val)
1196    }
1197
1198    /// Writes an element `val` of type `T` into the memory mapping at the specified `index`
1199    ///
1200    /// Returns an [`AckTicket`] representing the durability acknowledgement of this write operation
1201    ///
1202    /// ## Constraints
1203    ///
1204    /// - `index * size_of::<T>()` must not overflow `usize`, otherwise returns [`err::INV`]
1205    /// - Writing past the mapped memory region returns [`err::HCF`]
1206    ///
1207    /// ## Example
1208    ///
1209    /// ```
1210    /// use frozen_core::memmap::{MemMap, MemMapCfg};
1211    /// use std::time::Duration;
1212    ///
1213    /// let dir = tempfile::tempdir().unwrap();
1214    /// let cfg = MemMapCfg {
1215    ///     module_id: 1,
1216    ///     path: dir.path().join("write_test.bin"),
1217    ///     version: 1,
1218    ///     initial_t: 4,
1219    ///     sync_interval: Some(Duration::from_millis(100)),
1220    /// };
1221    ///
1222    /// let mmap = MemMap::<u64>::new(cfg).unwrap();
1223    /// let ticket = mmap.write(&100u64, 2).unwrap();
1224    /// assert_eq!(mmap.read(2).unwrap(), 100u64);
1225    /// ```
1226    #[inline(always)]
1227    pub fn write(&self, val: &T, index: usize) -> FrozenResult<AckTicket> {
1228        let mid = self.cfg.module_id;
1229        let size = core::mem::size_of::<T>();
1230
1231        let offset = match index.checked_mul(size) {
1232            Some(off) => off,
1233            None => return err::default_error(err::INV).with_module(mid),
1234        };
1235
1236        let slot_locks_guard = self.inner.slot_locks.read().unwrap_or_else(|e| e.into_inner());
1237
1238        if hints::unlikely(offset.checked_add(size).is_none_or(|end| end > self.length())) {
1239            return err::default_error(err::HCF).with_module(mid);
1240        }
1241
1242        let _slot_guard = slot_locks_guard.write_lock(index);
1243
1244        let guard = self.inner.mmap.read().unwrap_or_else(|e| e.into_inner());
1245        let mmap = match guard.as_ref() {
1246            Some(m) => m,
1247            None => return err::default_error(err::INV).with_module(mid),
1248        };
1249
1250        let ptr = unsafe { mmap.as_mut_ptr::<T>(offset) };
1251        unsafe { core::ptr::write(ptr, val.clone()) };
1252        drop(guard);
1253        drop(_slot_guard);
1254        drop(slot_locks_guard);
1255
1256        let epoch = self.inner.completion.increment_current_epoch();
1257
1258        if self.inner.is_parked.load(atomic::Ordering::Acquire) {
1259            let _guard = self.inner.sync_mutex.lock().unwrap_or_else(|e| e.into_inner());
1260            if self.inner.is_parked.swap(false, atomic::Ordering::AcqRel) {
1261                self.inner.sync_condvar.notify_one();
1262            }
1263        }
1264
1265        Ok(AckTicket::new(epoch, self.inner.completion.clone()))
1266    }
1267
1268    #[cfg(test)]
1269    pub(crate) fn is_parked(&self) -> bool {
1270        self.inner.is_parked.load(atomic::Ordering::Acquire)
1271    }
1272
1273    /// Deletes the [`MemMap`] and underlying [`File`] entry from the storage device
1274    ///
1275    /// Consumes `self` by value to release resources in deterministic order,
1276    ///
1277    /// - Signals the background flush worker to stop and waits for its thread join
1278    /// - Unmaps the active memory mapping from virtual address space
1279    /// - Closes and unlinks the backing versioned file from the filesystem
1280    ///
1281    /// ## Constraints
1282    ///
1283    /// - Consumes `self` by value; after deletion, the instance can no longer be accessed
1284    /// - Returns [`err::INV`] if the underlying file handle was already released
1285    ///
1286    /// ## Example
1287    ///
1288    /// ```
1289    /// use frozen_core::memmap::{MemMap, MemMapCfg};
1290    /// use std::time::Duration;
1291    ///
1292    /// let dir = tempfile::tempdir().unwrap();
1293    /// let path = dir.path().join("delete_demo.bin");
1294    /// let versioned_path = dir.path().join("delete_demo_1.bin");
1295    /// let cfg = MemMapCfg {
1296    ///     module_id: 1,
1297    ///     path,
1298    ///     version: 1,
1299    ///     initial_t: 4,
1300    ///     sync_interval: Some(Duration::from_millis(100)),
1301    /// };
1302    ///
1303    /// let mmap = MemMap::<u64>::new(cfg).unwrap();
1304    /// assert!(versioned_path.exists());
1305    ///
1306    /// mmap.delete().unwrap();
1307    /// assert!(!versioned_path.exists());
1308    /// ```
1309    pub fn delete(mut self) -> FrozenResult<()> {
1310        let mid = self.cfg.module_id;
1311        self.inner.shutdown.store(true, atomic::Ordering::Release);
1312        {
1313            let mut guard = self.inner.sync_mutex.lock().unwrap_or_else(|e| e.into_inner());
1314            *guard = true;
1315        }
1316        self.inner.sync_condvar.notify_all();
1317
1318        if let Some(handle) = self.bg_thread.take() {
1319            let _ = handle.join();
1320        }
1321
1322        let mut guard = self.inner.mmap.write().unwrap_or_else(|e| e.into_inner());
1323        let unmap_res =
1324            if let Some(mmap) = guard.take() { mmap.unmap().with_module(mid) } else { Ok(()) };
1325
1326        let file = match self.file.take() {
1327            Some(f) => f,
1328            None => {
1329                if let Err(e) = unmap_res {
1330                    return Err(e);
1331                }
1332                return err::default_error(err::INV).with_module(mid);
1333            }
1334        };
1335
1336        let delete_res = file.delete().with_module(mid);
1337
1338        match (unmap_res, delete_res) {
1339            (Err(mut unmap_err), Err(del_err)) => {
1340                unmap_err.add_suppressed(del_err);
1341                Err(unmap_err)
1342            }
1343            (Err(unmap_err), Ok(())) => Err(unmap_err),
1344            (Ok(()), Err(del_err)) => Err(del_err),
1345            (Ok(()), Ok(())) => Ok(()),
1346        }
1347    }
1348}
1349
1350impl<T> MemMap<T>
1351where
1352    T: Sized + Clone + PartialEq,
1353{
1354    /// Conditionally writes `val` into slot `index` if and only if the current value matches
1355    /// `expected`
1356    ///
1357    /// ## Semantics
1358    ///
1359    /// Acquires the exclusive slot lock for `index` and verifies that the current slot value equals
1360    /// `expected`; if equal, writes `val`, advances the epoch, and returns an
1361    /// [`AckTicket`]. If the current value differs from `expected`, the operation immediately
1362    /// aborts without modifying the slot, returning [`err::CAS`] so the caller can re-read and
1363    /// retry
1364    ///
1365    /// ## Constraints
1366    ///
1367    /// - `index * size_of::<T>()` must not overflow `usize`, otherwise returns [`err::INV`]
1368    /// - Writing past the mapped memory region returns [`err::HCF`]
1369    /// - Returns [`err::CAS`] if `expected != current`
1370    ///
1371    /// ## Example
1372    ///
1373    /// ```
1374    /// use frozen_core::memmap::{MemMap, MemMapCfg};
1375    /// use std::time::Duration;
1376    ///
1377    /// let dir = tempfile::tempdir().unwrap();
1378    /// let cfg = MemMapCfg {
1379    ///     module_id: 1,
1380    ///     path: dir.path().join("cas_demo.bin"),
1381    ///     version: 1,
1382    ///     initial_t: 4,
1383    ///     sync_interval: Some(Duration::from_millis(100)),
1384    /// };
1385    ///
1386    /// let mmap = MemMap::<u64>::new(cfg).unwrap();
1387    /// mmap.write(&100u64, 0).unwrap();
1388    ///
1389    /// // Successful CAS: 100 -> 200
1390    /// let ticket = mmap.compare_and_write(&100u64, &200u64, 0).unwrap();
1391    /// assert_eq!(mmap.read(0).unwrap(), 200u64);
1392    ///
1393    /// // Failed CAS: expected 100, but slot is 200
1394    /// assert!(mmap.compare_and_write(&100u64, &300u64, 0).is_err());
1395    /// assert_eq!(mmap.read(0).unwrap(), 200u64);
1396    /// ```
1397    pub fn compare_and_write(
1398        &self,
1399        expected: &T,
1400        val: &T,
1401        index: usize,
1402    ) -> FrozenResult<AckTicket> {
1403        let mid = self.cfg.module_id;
1404        let size = core::mem::size_of::<T>();
1405
1406        let offset = match index.checked_mul(size) {
1407            Some(off) => off,
1408            None => return err::default_error(err::INV).with_module(mid),
1409        };
1410
1411        let slot_locks_guard = self.inner.slot_locks.read().unwrap_or_else(|e| e.into_inner());
1412
1413        if hints::unlikely(offset.checked_add(size).is_none_or(|end| end > self.length())) {
1414            return err::default_error(err::HCF).with_module(mid);
1415        }
1416
1417        let _slot_guard = slot_locks_guard.write_lock(index);
1418
1419        let guard = self.inner.mmap.read().unwrap_or_else(|e| e.into_inner());
1420        let mmap = match guard.as_ref() {
1421            Some(m) => m,
1422            None => return err::default_error(err::INV).with_module(mid),
1423        };
1424
1425        let ptr = unsafe { mmap.as_mut_ptr::<T>(offset) };
1426        let current = unsafe { ptr.read() };
1427        if current != *expected {
1428            drop(guard);
1429            drop(_slot_guard);
1430            drop(slot_locks_guard);
1431            return err::raw_error(
1432                err::CAS,
1433                format!("state conflict at index {}: slot data has changed", index),
1434            )
1435            .with_module(mid);
1436        }
1437
1438        unsafe { core::ptr::write(ptr, val.clone()) };
1439        drop(guard);
1440        drop(_slot_guard);
1441        drop(slot_locks_guard);
1442
1443        let epoch = self.inner.completion.increment_current_epoch();
1444
1445        if self.inner.is_parked.load(atomic::Ordering::Acquire) {
1446            let _guard = self.inner.sync_mutex.lock().unwrap_or_else(|e| e.into_inner());
1447            if self.inner.is_parked.swap(false, atomic::Ordering::AcqRel) {
1448                self.inner.sync_condvar.notify_one();
1449            }
1450        }
1451
1452        Ok(AckTicket::new(epoch, self.inner.completion.clone()))
1453    }
1454
1455    /// Creates a new [`MemMapTransaction`] context for grouping multiple conditional writes into a
1456    /// single atomic epoch
1457    ///
1458    /// ## Overview
1459    ///
1460    /// `MemMapTransaction` allows grouping multiple slot mutations into a single atomic operation:
1461    ///
1462    /// - **Atomicity**: All writes succeed or fail together
1463    /// - **Optimistic Concurrency Control (CAS)**: Each write asserts that its target slot
1464    ///   currently matches `expected`; if any slot differs at commit time, all changes abort and no
1465    ///   slots are modified
1466    /// - **Single Epoch**: All committed writes share a single durability epoch and durable window
1467    /// - **Deadlock Freedom**: Slot locks are acquired in strictly ascending index order
1468    ///
1469    /// ## Example
1470    ///
1471    /// ```
1472    /// use frozen_core::memmap::{MemMap, MemMapCfg};
1473    /// use std::time::Duration;
1474    ///
1475    /// let dir = tempfile::tempdir().unwrap();
1476    /// let cfg = MemMapCfg {
1477    ///     module_id: 1,
1478    ///     path: dir.path().join("tx_demo.bin"),
1479    ///     version: 1,
1480    ///     initial_t: 4,
1481    ///     sync_interval: Some(Duration::from_millis(100)),
1482    /// };
1483    ///
1484    /// let mmap = MemMap::<u64>::new(cfg).unwrap();
1485    ///
1486    /// let mut tx = mmap.new_tx();
1487    /// tx.write(&0u64, &10u64, 0).unwrap();
1488    /// tx.write(&0u64, &20u64, 1).unwrap();
1489    ///
1490    /// let ticket = tx.commit().unwrap();
1491    /// assert_eq!(mmap.read(0).unwrap(), 10u64);
1492    /// assert_eq!(mmap.read(1).unwrap(), 20u64);
1493    /// ```
1494    #[inline]
1495    pub fn new_tx(&self) -> MemMapTransaction<'_, T> {
1496        MemMapTransaction { memmap: self, ops: Vec::new() }
1497    }
1498}
1499
1500/// Context for grouping multiple conditional writes into a single atomic operation
1501///
1502/// Multi-slot mutations are staged in `MemMapTransaction` with optimistic concurrency
1503/// preconditions. At [`commit`](Self::commit) time, all target slot locks are acquired in strictly
1504/// ascending order, each slot is verified against its `expected` value, and if all match, all
1505/// updates are applied under a single epoch
1506///
1507/// If any slot fails verification, the transaction aborts with zero mutations, returning
1508/// [`err::CAS`] with the index of the first conflicting slot so the caller can retry
1509///
1510/// ## Example
1511///
1512/// ```
1513/// use frozen_core::memmap::{MemMap, MemMapCfg};
1514/// use std::time::Duration;
1515///
1516/// let dir = tempfile::tempdir().unwrap();
1517/// let cfg = MemMapCfg {
1518///     module_id: 1,
1519///     path: dir.path().join("tx_struct_demo.bin"),
1520///     version: 1,
1521///     initial_t: 4,
1522///     sync_interval: Some(Duration::from_millis(100)),
1523/// };
1524///
1525/// let mmap = MemMap::<u64>::new(cfg).unwrap();
1526/// let mut tx = mmap.new_tx();
1527/// tx.write(&0u64, &42u64, 0).unwrap();
1528/// tx.write(&0u64, &84u64, 2).unwrap();
1529///
1530/// let ticket = tx.commit().unwrap();
1531/// assert_eq!(mmap.read(0).unwrap(), 42u64);
1532/// assert_eq!(mmap.read(2).unwrap(), 84u64);
1533/// ```
1534#[derive(Debug)]
1535pub struct MemMapTransaction<'a, T: Sized + Clone + PartialEq> {
1536    memmap: &'a MemMap<T>,
1537    ops: Vec<(usize, T, T)>,
1538}
1539
1540impl<'a, T> MemMapTransaction<'a, T>
1541where
1542    T: Sized + Clone + PartialEq,
1543{
1544    /// Stages a conditional write of `new` at `index` requiring `expected` as the current slot
1545    /// value
1546    ///
1547    /// ## Requirements
1548    ///
1549    /// - Writes must be staged in strictly increasing index order (`index > previous_index`)
1550    /// - Duplicate indices in the same transaction return an error
1551    /// - `index` must not exceed mapping bounds
1552    ///
1553    /// ## Example
1554    ///
1555    /// ```
1556    /// use frozen_core::memmap::{MemMap, MemMapCfg};
1557    /// use std::time::Duration;
1558    ///
1559    /// let dir = tempfile::tempdir().unwrap();
1560    /// let cfg = MemMapCfg {
1561    ///     module_id: 1,
1562    ///     path: dir.path().join("tx_write_demo.bin"),
1563    ///     version: 1,
1564    ///     initial_t: 4,
1565    ///     sync_interval: Some(Duration::from_millis(100)),
1566    /// };
1567    ///
1568    /// let mmap = MemMap::<u64>::new(cfg).unwrap();
1569    /// let mut tx = mmap.new_tx();
1570    /// tx.write(&0u64, &111u64, 0).unwrap();
1571    /// tx.write(&0u64, &222u64, 1).unwrap();
1572    ///
1573    /// // Staging an out-of-order or duplicate index fails
1574    /// assert!(tx.write(&0u64, &333u64, 1).is_err());
1575    /// assert!(tx.write(&0u64, &333u64, 0).is_err());
1576    /// ```
1577    #[inline(always)]
1578    pub fn write(&mut self, expected: &T, new: &T, index: usize) -> FrozenResult<&mut Self> {
1579        let mid = self.memmap.cfg.module_id;
1580        if let Some(err) = self.memmap.inner.completion.get_err() {
1581            return Err(err);
1582        }
1583
1584        if let Some((last_idx, _, _)) = self.ops.last() {
1585            if index <= *last_idx {
1586                return err::raw_error(
1587                    err::HCF,
1588                    "tx writes must be strictly ordered, with no more then single ops on given index",
1589                )
1590                .with_module(mid);
1591            }
1592        }
1593
1594        let size = core::mem::size_of::<T>();
1595        let offset = match index.checked_mul(size) {
1596            Some(off) => off,
1597            None => return err::default_error(err::INV).with_module(mid),
1598        };
1599
1600        if hints::unlikely(offset.checked_add(size).is_none_or(|end| end > self.memmap.length())) {
1601            return err::default_error(err::HCF).with_module(mid);
1602        }
1603
1604        self.ops.push((index, expected.clone(), new.clone()));
1605        Ok(self)
1606    }
1607
1608    /// Commits all staged writes atomically under a single durability epoch
1609    ///
1610    /// Acquires exclusive locks for all affected slots in order; if all slots match their expected
1611    /// values, updates are applied, the durable epoch advances once, and an [`AckTicket`] is
1612    /// returned
1613    ///
1614    /// If any slot differs from `expected`, all locks are released without modifying any data, and
1615    /// [`err::CAS`] is returned with context identifying the conflicting slot index
1616    ///
1617    /// ## Example
1618    ///
1619    /// ```
1620    /// use frozen_core::memmap::{MemMap, MemMapCfg};
1621    /// use std::time::Duration;
1622    ///
1623    /// let dir = tempfile::tempdir().unwrap();
1624    /// let cfg = MemMapCfg {
1625    ///     module_id: 1,
1626    ///     path: dir.path().join("tx_commit_demo.bin"),
1627    ///     version: 1,
1628    ///     initial_t: 4,
1629    ///     sync_interval: Some(Duration::from_millis(100)),
1630    /// };
1631    ///
1632    /// let mmap = MemMap::<u64>::new(cfg).unwrap();
1633    /// let mut tx = mmap.new_tx();
1634    /// tx.write(&0u64, &100u64, 0).unwrap();
1635    /// tx.write(&0u64, &200u64, 1).unwrap();
1636    ///
1637    /// let ticket = tx.commit().unwrap();
1638    /// assert_eq!(mmap.read(0).unwrap(), 100u64);
1639    /// assert_eq!(mmap.read(1).unwrap(), 200u64);
1640    /// ```
1641    pub fn commit(self) -> FrozenResult<AckTicket> {
1642        let mid = self.memmap.cfg.module_id;
1643        if self.ops.is_empty() {
1644            return err::raw_error(err::INV, "empty transaction").with_module(mid);
1645        }
1646
1647        if let Some(err) = self.memmap.inner.completion.get_err() {
1648            return Err(err);
1649        }
1650
1651        let size = core::mem::size_of::<T>();
1652        let slot_locks_guard =
1653            self.memmap.inner.slot_locks.read().unwrap_or_else(|e| e.into_inner());
1654        let current_length = self.memmap.length();
1655
1656        // Verify bounds under slot_locks read guard
1657        for (index, _, _) in &self.ops {
1658            let offset = match index.checked_mul(size) {
1659                Some(off) => off,
1660                None => return err::default_error(err::INV).with_module(mid),
1661            };
1662            if hints::unlikely(offset.checked_add(size).is_none_or(|end| end > current_length)) {
1663                return err::default_error(err::HCF).with_module(mid);
1664            }
1665        }
1666
1667        // Acquire all slot write locks in strictly ascending index order
1668        let mut slot_guards = Vec::with_capacity(self.ops.len());
1669        for (index, _, _) in &self.ops {
1670            slot_guards.push(slot_locks_guard.write_lock(*index));
1671        }
1672
1673        let guard = self.memmap.inner.mmap.read().unwrap_or_else(|e| e.into_inner());
1674        let mmap = match guard.as_ref() {
1675            Some(m) => m,
1676            None => return err::default_error(err::INV).with_module(mid),
1677        };
1678
1679        // Phase 1: State Precondition Verification (CAS)
1680        for (index, expected, _) in &self.ops {
1681            let offset = *index * size;
1682            let ptr = unsafe { mmap.as_ptr::<T>(offset) };
1683            let current = unsafe { ptr.read() };
1684            if current != *expected {
1685                drop(guard);
1686                drop(slot_guards);
1687                drop(slot_locks_guard);
1688                return err::raw_error(
1689                    err::CAS,
1690                    format!("state conflict at index {}: slot data has changed", index),
1691                )
1692                .with_module(mid);
1693            }
1694        }
1695
1696        // Phase 2: Apply mutations
1697        for (index, _, new) in self.ops {
1698            let offset = index * size;
1699            let ptr = unsafe { mmap.as_mut_ptr::<T>(offset) };
1700            unsafe { core::ptr::write(ptr, new) };
1701        }
1702
1703        drop(guard);
1704        drop(slot_guards);
1705        drop(slot_locks_guard);
1706
1707        let epoch = self.memmap.inner.completion.increment_current_epoch();
1708
1709        if self.memmap.inner.is_parked.load(atomic::Ordering::Acquire) {
1710            let _guard = self.memmap.inner.sync_mutex.lock().unwrap_or_else(|e| e.into_inner());
1711            if self.memmap.inner.is_parked.swap(false, atomic::Ordering::AcqRel) {
1712                self.memmap.inner.sync_condvar.notify_one();
1713            }
1714        }
1715
1716        Ok(AckTicket::new(epoch, self.memmap.inner.completion.clone()))
1717    }
1718
1719    /// Returns the number of staged operations in this transaction
1720    ///
1721    /// ## Example
1722    ///
1723    /// ```
1724    /// use frozen_core::memmap::{MemMap, MemMapCfg};
1725    ///
1726    /// let dir = tempfile::tempdir().unwrap();
1727    /// let cfg = MemMapCfg {
1728    ///     module_id: 1,
1729    ///     path: dir.path().join("tx_len_demo.bin"),
1730    ///     version: 1,
1731    ///     initial_t: 4,
1732    ///     sync_interval: None,
1733    /// };
1734    ///
1735    /// let mmap = MemMap::<u64>::new(cfg).unwrap();
1736    /// let mut tx = mmap.new_tx();
1737    /// assert_eq!(tx.len(), 0);
1738    /// tx.write(&0u64, &10u64, 0).unwrap();
1739    /// assert_eq!(tx.len(), 1);
1740    /// ```
1741    #[inline]
1742    pub fn len(&self) -> usize {
1743        self.ops.len()
1744    }
1745
1746    /// Returns `true` if no operations have been staged in this transaction
1747    ///
1748    /// ## Example
1749    ///
1750    /// ```
1751    /// use frozen_core::memmap::{MemMap, MemMapCfg};
1752    ///
1753    /// let dir = tempfile::tempdir().unwrap();
1754    /// let cfg = MemMapCfg {
1755    ///     module_id: 1,
1756    ///     path: dir.path().join("tx_empty_demo.bin"),
1757    ///     version: 1,
1758    ///     initial_t: 4,
1759    ///     sync_interval: None,
1760    /// };
1761    ///
1762    /// let mmap = MemMap::<u64>::new(cfg).unwrap();
1763    /// let tx = mmap.new_tx();
1764    /// assert!(tx.is_empty());
1765    /// ```
1766    #[inline]
1767    pub fn is_empty(&self) -> bool {
1768        self.ops.is_empty()
1769    }
1770}
1771
1772impl<T> Drop for MemMap<T>
1773where
1774    T: Sized + Clone,
1775{
1776    fn drop(&mut self) {
1777        self.inner.shutdown.store(true, atomic::Ordering::Release);
1778        {
1779            let mut guard = self.inner.sync_mutex.lock().unwrap_or_else(|e| e.into_inner());
1780            *guard = true;
1781        }
1782        self.inner.sync_condvar.notify_all();
1783
1784        if let Some(handle) = self.bg_thread.take() {
1785            let _ = handle.join();
1786        }
1787
1788        let _ = self.inner.sync_internal();
1789    }
1790}
1791
1792#[cfg(test)]
1793mod tests {
1794    use super::*;
1795    use std::time::Duration;
1796
1797    const MID: u8 = 0;
1798    const INIT_T: usize = 4;
1799
1800    #[derive(Debug, Clone)]
1801    #[repr(C, align(8))]
1802    struct ValidT {
1803        a: u64,
1804        b: u64,
1805    }
1806
1807    #[derive(Debug, Clone)]
1808    #[repr(C)]
1809    struct BadAlignT {
1810        a: u32,
1811        b: u32,
1812    }
1813
1814    #[derive(Debug, Clone)]
1815    #[repr(C)]
1816    struct BadSizeT {
1817        a: [u8; 0x0C],
1818    }
1819
1820    #[derive(Debug, Clone)]
1821    #[repr(C, align(8))]
1822    struct DropT(u64);
1823
1824    impl Drop for DropT {
1825        fn drop(&mut self) {}
1826    }
1827
1828    #[derive(Debug, Clone)]
1829    #[repr(C, align(8))]
1830    struct ZstT;
1831
1832    #[derive(Debug, Clone)]
1833    #[repr(C, align(8))]
1834    struct OversizedT {
1835        data: [u8; 0x20000],
1836    }
1837
1838    fn tmp_cfg() -> (tempfile::TempDir, MemMapCfg) {
1839        let dir = tempfile::tempdir().unwrap();
1840        let path = dir.path().join("tmp_memmap");
1841        let cfg = MemMapCfg {
1842            module_id: MID,
1843            path,
1844            version: 1,
1845            initial_t: INIT_T,
1846            sync_interval: Some(Duration::from_millis(0x64)),
1847        };
1848
1849        (dir, cfg)
1850    }
1851
1852    #[test]
1853    fn ok_validate_t() {
1854        assert!(MemMap::<u64>::validate_t(MID).is_ok());
1855        assert!(MemMap::<ValidT>::validate_t(MID).is_ok());
1856    }
1857
1858    #[test]
1859    fn err_validate_t_invariants() {
1860        assert_eq!(MemMap::<DropT>::validate_t(MID).unwrap_err().reason, err::DRP.reason);
1861        assert_eq!(MemMap::<BadAlignT>::validate_t(MID).unwrap_err().reason, err::ALN.reason);
1862        assert_eq!(MemMap::<ZstT>::validate_t(MID).unwrap_err().reason, err::ZRO.reason);
1863        assert_eq!(MemMap::<BadSizeT>::validate_t(MID).unwrap_err().reason, err::SZE.reason);
1864        assert_eq!(MemMap::<OversizedT>::validate_t(MID).unwrap_err().reason, err::INV.reason);
1865    }
1866
1867    #[test]
1868    fn ok_new_sets_file_sync_interval_none() {
1869        let (_dir, cfg) = tmp_cfg();
1870
1871        let mmap = MemMap::<u64>::new(cfg.clone()).unwrap();
1872        assert_eq!(mmap.length(), core::mem::size_of::<u64>() * INIT_T);
1873        assert_eq!(mmap.cfg().sync_interval, Some(Duration::from_millis(0x64)));
1874    }
1875
1876    #[test]
1877    fn err_new_when_file_exists() {
1878        let (_dir, cfg) = tmp_cfg();
1879        let _mmap = MemMap::<u64>::new(cfg.clone()).unwrap();
1880        let err = MemMap::<u64>::new(cfg).unwrap_err();
1881        assert_eq!(err.reason, err::EXS.reason);
1882    }
1883
1884    #[test]
1885    fn err_new_when_t_invalid() {
1886        let (_dir, cfg) = tmp_cfg();
1887        assert_eq!(MemMap::<DropT>::new(cfg.clone()).unwrap_err().reason, err::DRP.reason);
1888        assert!(!cfg.versioned_path().exists());
1889    }
1890
1891    #[test]
1892    fn ok_open() {
1893        let (_dir, cfg) = tmp_cfg();
1894        {
1895            let mmap = MemMap::<u64>::new(cfg.clone()).unwrap();
1896            assert_eq!(mmap.length(), core::mem::size_of::<u64>() * INIT_T);
1897        }
1898
1899        let opened = MemMap::<u64>::open(cfg).unwrap();
1900        assert_eq!(opened.length(), core::mem::size_of::<u64>() * INIT_T);
1901    }
1902
1903    #[test]
1904    fn err_open_when_file_missing() {
1905        let (_dir, cfg) = tmp_cfg();
1906        let err = MemMap::<u64>::open(cfg).unwrap_err();
1907        assert_eq!(err.reason, err::NEX.reason);
1908    }
1909
1910    #[test]
1911    fn ok_open_or_create() {
1912        let (_dir, cfg) = tmp_cfg();
1913        {
1914            let mmap = MemMap::<u64>::open_or_create(cfg.clone()).unwrap();
1915            assert_eq!(mmap.length(), core::mem::size_of::<u64>() * INIT_T);
1916        }
1917
1918        let reopened = MemMap::<u64>::open_or_create(cfg).unwrap();
1919        assert_eq!(reopened.length(), core::mem::size_of::<u64>() * INIT_T);
1920    }
1921
1922    #[test]
1923    fn err_path_when_dir_missing() {
1924        let dir = tempfile::tempdir().unwrap();
1925        let non_existent_dir = dir.path().join("missing_dir_xyz");
1926        let cfg = MemMapCfg {
1927            module_id: MID,
1928            path: non_existent_dir.join("file.bin"),
1929            version: 1,
1930            initial_t: INIT_T,
1931            sync_interval: Some(Duration::from_millis(0x64)),
1932        };
1933
1934        let err_new = MemMap::<u64>::new(cfg.clone()).unwrap_err();
1935        assert_eq!(err_new.reason, err::DIR.reason);
1936
1937        let err_open = MemMap::<u64>::open(cfg.clone()).unwrap_err();
1938        assert_eq!(err_open.reason, err::DIR.reason);
1939
1940        let err_open_or_create = MemMap::<u64>::open_or_create(cfg).unwrap_err();
1941        assert_eq!(err_open_or_create.reason, err::DIR.reason);
1942    }
1943
1944    #[test]
1945    fn err_path_when_parent_is_not_dir() {
1946        let dir = tempfile::tempdir().unwrap();
1947        let file_path = dir.path().join("regular_file");
1948        std::fs::write(&file_path, b"test").unwrap();
1949
1950        let cfg = MemMapCfg {
1951            module_id: MID,
1952            path: file_path.join("subfile.bin"),
1953            version: 1,
1954            initial_t: INIT_T,
1955            sync_interval: Some(Duration::from_millis(0x64)),
1956        };
1957
1958        let err_new = MemMap::<u64>::new(cfg.clone()).unwrap_err();
1959        assert_eq!(err_new.reason, err::DIR.reason);
1960
1961        let err_open = MemMap::<u64>::open(cfg.clone()).unwrap_err();
1962        assert_eq!(err_open.reason, err::DIR.reason);
1963
1964        let err_open_or_create = MemMap::<u64>::open_or_create(cfg).unwrap_err();
1965        assert_eq!(err_open_or_create.reason, err::DIR.reason);
1966    }
1967
1968    #[test]
1969    fn err_path_when_empty() {
1970        let cfg = MemMapCfg {
1971            module_id: MID,
1972            path: std::path::PathBuf::from(""),
1973            version: 1,
1974            initial_t: INIT_T,
1975            sync_interval: Some(Duration::from_millis(0x64)),
1976        };
1977
1978        let err_new = MemMap::<u64>::new(cfg.clone()).unwrap_err();
1979        assert_eq!(err_new.reason, err::PTH.reason);
1980
1981        let err_open = MemMap::<u64>::open(cfg.clone()).unwrap_err();
1982        assert_eq!(err_open.reason, err::PTH.reason);
1983
1984        let err_open_or_create = MemMap::<u64>::open_or_create(cfg).unwrap_err();
1985        assert_eq!(err_open_or_create.reason, err::PTH.reason);
1986    }
1987
1988    #[test]
1989    fn err_path_when_is_directory() {
1990        let dir = tempfile::tempdir().unwrap();
1991        let cfg = MemMapCfg {
1992            module_id: MID,
1993            path: dir.path().to_path_buf(),
1994            version: 1,
1995            initial_t: INIT_T,
1996            sync_interval: Some(Duration::from_millis(0x64)),
1997        };
1998
1999        let err_new = MemMap::<u64>::new(cfg.clone()).unwrap_err();
2000        assert_eq!(err_new.reason, err::PTH.reason);
2001
2002        let err_open = MemMap::<u64>::open(cfg.clone()).unwrap_err();
2003        assert_eq!(err_open.reason, err::PTH.reason);
2004
2005        let err_open_or_create = MemMap::<u64>::open_or_create(cfg).unwrap_err();
2006        assert_eq!(err_open_or_create.reason, err::PTH.reason);
2007    }
2008
2009    #[test]
2010    fn err_path_when_no_filename() {
2011        let cfg = MemMapCfg {
2012            module_id: MID,
2013            path: std::path::PathBuf::from("/"),
2014            version: 1,
2015            initial_t: INIT_T,
2016            sync_interval: Some(Duration::from_millis(0x64)),
2017        };
2018
2019        let err_new = MemMap::<u64>::new(cfg.clone()).unwrap_err();
2020        assert_eq!(err_new.reason, err::PTH.reason);
2021
2022        let err_open = MemMap::<u64>::open(cfg.clone()).unwrap_err();
2023        assert_eq!(err_open.reason, err::PTH.reason);
2024
2025        let err_open_or_create = MemMap::<u64>::open_or_create(cfg).unwrap_err();
2026        assert_eq!(err_open_or_create.reason, err::PTH.reason);
2027    }
2028
2029    #[test]
2030    fn ok_delete() {
2031        let (_dir, cfg) = tmp_cfg();
2032        let path = cfg.versioned_path();
2033        let mmap = MemMap::<u64>::new(cfg).unwrap();
2034        assert!(path.exists());
2035        assert!(mmap.delete().is_ok());
2036        assert!(!path.exists());
2037    }
2038
2039    #[test]
2040    fn ok_versioned_path_resolution() {
2041        let cfg = MemMapCfg {
2042            module_id: MID,
2043            path: std::path::PathBuf::from("data.bin"),
2044            version: 1,
2045            initial_t: 2,
2046            sync_interval: Some(Duration::from_millis(0x64)),
2047        };
2048        assert_eq!(cfg.versioned_path(), std::path::PathBuf::from("data_1.bin"));
2049
2050        let cfg = MemMapCfg {
2051            module_id: MID,
2052            path: std::path::PathBuf::from("datastore"),
2053            version: 2,
2054            initial_t: 2,
2055            sync_interval: Some(Duration::from_millis(0x64)),
2056        };
2057        assert_eq!(cfg.versioned_path(), std::path::PathBuf::from("datastore_2"));
2058
2059        let cfg = MemMapCfg {
2060            module_id: MID,
2061            path: std::path::PathBuf::from("archive.tar.gz"),
2062            version: 3,
2063            initial_t: 2,
2064            sync_interval: Some(Duration::from_millis(0x64)),
2065        };
2066        assert_eq!(cfg.versioned_path(), std::path::PathBuf::from("archive.tar_3.gz"));
2067
2068        let cfg = MemMapCfg {
2069            module_id: MID,
2070            path: std::path::PathBuf::from("/var/data/store.db"),
2071            version: 0x0A,
2072            initial_t: 2,
2073            sync_interval: Some(Duration::from_millis(0x64)),
2074        };
2075        assert_eq!(cfg.versioned_path(), std::path::PathBuf::from("/var/data/store_10.db"));
2076
2077        let cfg = MemMapCfg {
2078            module_id: MID,
2079            path: std::path::PathBuf::from("dir/sub/file.dat"),
2080            version: 0,
2081            initial_t: 2,
2082            sync_interval: Some(Duration::from_millis(0x64)),
2083        };
2084        assert_eq!(cfg.versioned_path(), std::path::PathBuf::from("dir/sub/file_0.dat"));
2085
2086        let cfg = MemMapCfg {
2087            module_id: MID,
2088            path: std::path::PathBuf::from(".env"),
2089            version: 5,
2090            initial_t: 2,
2091            sync_interval: Some(Duration::from_millis(0x64)),
2092        };
2093        assert_eq!(cfg.versioned_path(), std::path::PathBuf::from(".env_5"));
2094
2095        let cfg = MemMapCfg {
2096            module_id: MID,
2097            path: std::path::PathBuf::from(".config.json"),
2098            version: 1,
2099            initial_t: 2,
2100            sync_interval: Some(Duration::from_millis(0x64)),
2101        };
2102        assert_eq!(cfg.versioned_path(), std::path::PathBuf::from(".config_1.json"));
2103    }
2104
2105    #[test]
2106    fn ok_versioned_path_and_multiversion() {
2107        let dir = tempfile::tempdir().unwrap();
2108        let base_path = dir.path().join("store.dat");
2109
2110        let cfg_v1 = MemMapCfg {
2111            module_id: MID,
2112            path: base_path.clone(),
2113            version: 1,
2114            initial_t: 2,
2115            sync_interval: Some(Duration::from_millis(0x64)),
2116        };
2117        assert_eq!(cfg_v1.versioned_path(), dir.path().join("store_1.dat"));
2118
2119        let cfg_v2 = MemMapCfg {
2120            module_id: MID,
2121            path: base_path.clone(),
2122            version: 2,
2123            initial_t: 4,
2124            sync_interval: Some(Duration::from_millis(0x64)),
2125        };
2126        assert_eq!(cfg_v2.versioned_path(), dir.path().join("store_2.dat"));
2127
2128        // Both versions can coexist independently on disk
2129        let mmap_v1 = MemMap::<u64>::new(cfg_v1.clone()).unwrap();
2130        let mmap_v2 = MemMap::<ValidT>::new(cfg_v2.clone()).unwrap();
2131
2132        assert!(cfg_v1.versioned_path().exists());
2133        assert!(cfg_v2.versioned_path().exists());
2134        assert_eq!(mmap_v1.length(), core::mem::size_of::<u64>() * 2);
2135        assert_eq!(mmap_v2.length(), core::mem::size_of::<ValidT>() * 4);
2136
2137        assert!(mmap_v1.delete().is_ok());
2138        assert!(!cfg_v1.versioned_path().exists());
2139        assert!(cfg_v2.versioned_path().exists());
2140
2141        assert!(mmap_v2.delete().is_ok());
2142        assert!(!cfg_v2.versioned_path().exists());
2143    }
2144
2145    #[test]
2146    fn ok_write_read_single() {
2147        let (_dir, cfg) = tmp_cfg();
2148        let mmap = MemMap::<u64>::new(cfg).unwrap();
2149
2150        assert_eq!(mmap.read(0).unwrap(), 0);
2151        let ticket = mmap.write(&0xDEAD_BEEF_CAFE_BABE, 0).unwrap();
2152        assert_eq!(mmap.read(0).unwrap(), 0xDEAD_BEEF_CAFE_BABE);
2153        assert_eq!(ticket.epoch(), 1);
2154    }
2155
2156    #[test]
2157    fn ok_write_read_multiple_indices() {
2158        let (_dir, cfg) = tmp_cfg();
2159        let mmap = MemMap::<u64>::new(cfg).unwrap();
2160
2161        for i in 0..INIT_T {
2162            let val = (i as u64 + 1) * 0x1111_2222_3333_4444;
2163            mmap.write(&val, i).unwrap();
2164        }
2165
2166        for i in 0..INIT_T {
2167            let expected = (i as u64 + 1) * 0x1111_2222_3333_4444;
2168            assert_eq!(mmap.read(i).unwrap(), expected);
2169        }
2170    }
2171
2172    #[test]
2173    fn ok_write_read_struct() {
2174        let (_dir, cfg) = tmp_cfg();
2175        let mmap = MemMap::<ValidT>::new(cfg).unwrap();
2176
2177        let val0 = ValidT { a: 0x0A, b: 0x14 };
2178        let val1 = ValidT { a: 0x64, b: 0xC8 };
2179
2180        mmap.write(&val0, 0).unwrap();
2181        mmap.write(&val1, 1).unwrap();
2182
2183        let r0 = mmap.read(0).unwrap();
2184        let r1 = mmap.read(1).unwrap();
2185
2186        assert_eq!(r0.a, 0x0A);
2187        assert_eq!(r0.b, 0x14);
2188        assert_eq!(r1.a, 0x64);
2189        assert_eq!(r1.b, 0xC8);
2190    }
2191
2192    #[test]
2193    fn err_write_read_out_of_bounds() {
2194        let (_dir, cfg) = tmp_cfg();
2195        let mmap = MemMap::<u64>::new(cfg).unwrap();
2196
2197        let err_r = mmap.read(INIT_T).unwrap_err();
2198        assert_eq!(err_r.reason, err::HCF.reason);
2199
2200        let err_w = mmap.write(&0x3039, INIT_T).unwrap_err();
2201        assert_eq!(err_w.reason, err::HCF.reason);
2202
2203        let err_r_large = mmap.read(0x64).unwrap_err();
2204        assert_eq!(err_r_large.reason, err::HCF.reason);
2205
2206        let err_w_large = mmap.write(&0x3039, 0x64).unwrap_err();
2207        assert_eq!(err_w_large.reason, err::HCF.reason);
2208    }
2209
2210    #[test]
2211    fn err_write_read_overflow_index() {
2212        let (_dir, cfg) = tmp_cfg();
2213        let mmap = MemMap::<u64>::new(cfg).unwrap();
2214
2215        let err_r = mmap.read(usize::MAX).unwrap_err();
2216        assert_eq!(err_r.reason, err::INV.reason);
2217
2218        let err_w = mmap.write(&0x3039, usize::MAX).unwrap_err();
2219        assert_eq!(err_w.reason, err::INV.reason);
2220    }
2221
2222    #[test]
2223    fn ok_durability_across_reopen() {
2224        let (_dir, cfg) = tmp_cfg();
2225        {
2226            let mmap = MemMap::<u64>::new(cfg.clone()).unwrap();
2227            mmap.write(&0xCAFE_BABE_1234_5678, 1).unwrap();
2228            mmap.inner.sync_internal().unwrap();
2229        }
2230
2231        let reopened = MemMap::<u64>::open(cfg).unwrap();
2232        assert_eq!(reopened.read(1).unwrap(), 0xCAFE_BABE_1234_5678);
2233    }
2234
2235    #[test]
2236    fn ok_background_sync_advances_epoch() {
2237        let (_dir, mut cfg) = tmp_cfg();
2238        cfg.sync_interval = Some(Duration::from_millis(0x19));
2239
2240        let mmap = MemMap::<u64>::new(cfg).unwrap();
2241        let ticket = mmap.write(&0x1234, 0).unwrap();
2242        let epoch = ticket.wait().unwrap();
2243
2244        assert_eq!(epoch, ticket.epoch());
2245        assert!(ticket.is_durable());
2246        assert!(mmap.durable_epoch() >= ticket.epoch());
2247    }
2248
2249    #[test]
2250    fn ok_ticket_force_wakes_bg_worker() {
2251        let (_dir, mut cfg) = tmp_cfg();
2252        // Long interval to prove force wakes up immediately
2253        cfg.sync_interval = Some(Duration::from_secs(0x0A));
2254
2255        let mmap = MemMap::<u64>::new(cfg).unwrap();
2256        let ticket = mmap.write(&0xABCD, 0).unwrap();
2257        assert!(!ticket.is_durable());
2258
2259        let start = std::time::Instant::now();
2260        let epoch = ticket.force().unwrap();
2261        let elapsed = start.elapsed();
2262
2263        assert_eq!(epoch, ticket.epoch());
2264        assert!(ticket.is_durable());
2265        assert!(elapsed < Duration::from_secs(2));
2266    }
2267
2268    #[test]
2269    fn ok_drop_persists_unflushed_writes() {
2270        let (_dir, mut cfg) = tmp_cfg();
2271        cfg.sync_interval = Some(Duration::from_secs(0x1E));
2272
2273        {
2274            let mmap = MemMap::<u64>::new(cfg.clone()).unwrap();
2275            let _ticket = mmap.write(&0xBEEF_CAFE, 0).unwrap();
2276            drop(mmap);
2277        }
2278
2279        let reopened = MemMap::<u64>::open(cfg).unwrap();
2280        assert_eq!(reopened.read(0).unwrap(), 0xBEEF_CAFE);
2281    }
2282
2283    #[test]
2284    fn ok_concurrent_slot_locking() {
2285        let (_dir, mut cfg) = tmp_cfg();
2286        cfg.initial_t = 8;
2287        cfg.sync_interval = Some(Duration::from_secs(0x3C));
2288
2289        let mmap = Arc::new(MemMap::<u64>::new(cfg).unwrap());
2290        let barrier = Arc::new(std::sync::Barrier::new(8));
2291        let mut handles = Vec::new();
2292
2293        // 4 writer threads, each writing to index 0 repeatedly
2294        for t in 0..4 {
2295            let mmap_clone = Arc::clone(&mmap);
2296            let barrier_clone = Arc::clone(&barrier);
2297            handles.push(std::thread::spawn(move || {
2298                barrier_clone.wait();
2299                for i in 0..0x1F4 {
2300                    let val = ((t + 1) as u64) << 0x20 | (i as u64);
2301                    let _ = mmap_clone.write(&val, 0).unwrap();
2302                }
2303            }));
2304        }
2305
2306        // 4 reader threads, reading from index 0 repeatedly
2307        for _ in 0..4 {
2308            let mmap_clone = Arc::clone(&mmap);
2309            let barrier_clone = Arc::clone(&barrier);
2310            handles.push(std::thread::spawn(move || {
2311                barrier_clone.wait();
2312                for _ in 0..0x3E8 {
2313                    let val = mmap_clone.read(0).unwrap();
2314                    let thread_id = (val >> 0x20) as usize;
2315                    assert!(thread_id <= 4, "read value must be valid without torn reads");
2316                }
2317            }));
2318        }
2319
2320        for h in handles {
2321            h.join().unwrap();
2322        }
2323
2324        // Parallel reads across different and same indices
2325        let barrier2 = Arc::new(std::sync::Barrier::new(4));
2326        let mut read_handles = Vec::new();
2327        for _ in 0..4 {
2328            let mmap_clone = Arc::clone(&mmap);
2329            let barrier_clone = Arc::clone(&barrier2);
2330            read_handles.push(std::thread::spawn(move || {
2331                barrier_clone.wait();
2332                for _ in 0..0x1F4 {
2333                    let _ = mmap_clone.read(0).unwrap();
2334                }
2335            }));
2336        }
2337
2338        for h in read_handles {
2339            h.join().unwrap();
2340        }
2341    }
2342
2343    #[test]
2344    fn err_concurrent_new_same_path() {
2345        let (_dir, cfg) = tmp_cfg();
2346        let barrier = Arc::new(std::sync::Barrier::new(2));
2347
2348        let t1 = {
2349            let cfg = cfg.clone();
2350            let b = Arc::clone(&barrier);
2351            std::thread::spawn(move || {
2352                b.wait();
2353                MemMap::<u64>::new(cfg)
2354            })
2355        };
2356
2357        let t2 = {
2358            let b = Arc::clone(&barrier);
2359            std::thread::spawn(move || {
2360                b.wait();
2361                MemMap::<u64>::new(cfg)
2362            })
2363        };
2364
2365        let r1 = t1.join().unwrap();
2366        let r2 = t2.join().unwrap();
2367
2368        match (r1, r2) {
2369            (Ok(m), Err(e)) | (Err(e), Ok(m)) => {
2370                assert_eq!(e.reason, err::EXS.reason);
2371                assert!(m.delete().is_ok());
2372            }
2373            (Ok(_), Ok(_)) => {
2374                panic!("Concurrent MemMap::new on identical path must not both succeed");
2375            }
2376            (Err(e1), Err(e2)) => {
2377                panic!("Unexpected error on both threads: {:?}, {:?}", e1, e2);
2378            }
2379        }
2380    }
2381
2382    #[test]
2383    fn err_concurrent_open_same_path() {
2384        let (_dir, cfg) = tmp_cfg();
2385        {
2386            let m = MemMap::<u64>::new(cfg.clone()).unwrap();
2387            assert_eq!(m.length(), core::mem::size_of::<u64>() * INIT_T);
2388        }
2389
2390        let barrier = Arc::new(std::sync::Barrier::new(2));
2391
2392        let t1 = {
2393            let cfg = cfg.clone();
2394            let b = Arc::clone(&barrier);
2395            std::thread::spawn(move || {
2396                b.wait();
2397                MemMap::<u64>::open(cfg)
2398            })
2399        };
2400
2401        let t2 = {
2402            let b = Arc::clone(&barrier);
2403            std::thread::spawn(move || {
2404                b.wait();
2405                MemMap::<u64>::open(cfg)
2406            })
2407        };
2408
2409        let r1 = t1.join().unwrap();
2410        let r2 = t2.join().unwrap();
2411
2412        match (r1, r2) {
2413            (Ok(m), Err(e)) | (Err(e), Ok(m)) => {
2414                assert_eq!(e.domain, 0x08, "Error domain must be File (0x08)");
2415                assert_eq!(
2416                    e.reason, 0x1C,
2417                    "Concurrent open should yield LCK (0x1C) from kernel flock: {:?}",
2418                    e
2419                );
2420                assert!(m.delete().is_ok());
2421            }
2422            (Ok(_), Ok(_)) => {
2423                panic!("Exclusive advisory lock violation: both concurrent open calls succeeded");
2424            }
2425            (Err(e1), Err(e2)) => {
2426                panic!("Unexpected failure of both threads: {:?}, {:?}", e1, e2);
2427            }
2428        }
2429    }
2430
2431    #[test]
2432    fn ok_reopen_after_drop_releases_lock() {
2433        let (_dir, cfg) = tmp_cfg();
2434        let m1 = MemMap::<u64>::new(cfg.clone()).unwrap();
2435        m1.write(&0x1122_3344, 0).unwrap();
2436        drop(m1);
2437
2438        let m2 = MemMap::<u64>::open(cfg).unwrap();
2439        assert_eq!(m2.read(0).unwrap(), 0x1122_3344);
2440        assert!(m2.delete().is_ok());
2441    }
2442
2443    #[test]
2444    fn ok_delete_and_recreate_cycle() {
2445        let (_dir, cfg) = tmp_cfg();
2446        for i in 0..4 {
2447            let mmap = MemMap::<u64>::new(cfg.clone()).unwrap();
2448            let val = (i as u64 + 1) * 0xAA55;
2449            mmap.write(&val, 0).unwrap();
2450            assert_eq!(mmap.read(0).unwrap(), val);
2451            assert!(mmap.delete().is_ok());
2452            assert!(!cfg.versioned_path().exists());
2453        }
2454    }
2455
2456    #[test]
2457    fn ok_concurrent_distinct_slots_no_contention() {
2458        let (_dir, mut cfg) = tmp_cfg();
2459        const THREADS: usize = 8;
2460        cfg.initial_t = THREADS;
2461        cfg.sync_interval = Some(Duration::from_secs(0x3C));
2462
2463        let mmap = Arc::new(MemMap::<u64>::new(cfg).unwrap());
2464        let barrier = Arc::new(std::sync::Barrier::new(THREADS));
2465        let mut handles = Vec::with_capacity(THREADS);
2466
2467        for t in 0..THREADS {
2468            let mmap_clone = Arc::clone(&mmap);
2469            let barrier_clone = Arc::clone(&barrier);
2470            handles.push(std::thread::spawn(move || {
2471                barrier_clone.wait();
2472                for i in 0..0x3E8 {
2473                    let val = (t as u64) << 0x20 | (i as u64);
2474                    let _ = mmap_clone.write(&val, t).unwrap();
2475                    let read_val = mmap_clone.read(t).unwrap();
2476                    assert_eq!(read_val >> 0x20, t as u64);
2477                }
2478            }));
2479        }
2480
2481        for h in handles {
2482            h.join().unwrap();
2483        }
2484
2485        for t in 0..THREADS {
2486            let val = mmap.read(t).unwrap();
2487            assert_eq!(val >> 0x20, t as u64);
2488        }
2489    }
2490
2491    #[test]
2492    fn ok_stress_interleaved_reads_writes() {
2493        let (_dir, mut cfg) = tmp_cfg();
2494        const NUM_SLOTS: usize = 0x10;
2495        const NUM_THREADS: usize = 8;
2496        cfg.initial_t = NUM_SLOTS;
2497        cfg.sync_interval = Some(Duration::from_millis(0x32));
2498
2499        let mmap = Arc::new(MemMap::<u64>::new(cfg).unwrap());
2500        let barrier = Arc::new(std::sync::Barrier::new(NUM_THREADS));
2501        let mut handles = Vec::with_capacity(NUM_THREADS);
2502
2503        for t in 0..NUM_THREADS {
2504            let mmap_clone = Arc::clone(&mmap);
2505            let barrier_clone = Arc::clone(&barrier);
2506            handles.push(std::thread::spawn(move || {
2507                barrier_clone.wait();
2508                for i in 0..0x5DC {
2509                    let slot = (t + i) % NUM_SLOTS;
2510                    if i % 2 == 0 {
2511                        let val = ((t + 1) as u64) << 0x20 | (i as u64);
2512                        let _ = mmap_clone.write(&val, slot).unwrap();
2513                    } else {
2514                        let _ = mmap_clone.read(slot).unwrap();
2515                    }
2516                }
2517            }));
2518        }
2519
2520        for h in handles {
2521            h.join().unwrap();
2522        }
2523
2524        assert!(mmap.inner.sync_internal().is_ok());
2525    }
2526
2527    #[test]
2528    fn ok_concurrent_writers_with_background_sync() {
2529        let (_dir, mut cfg) = tmp_cfg();
2530        cfg.initial_t = 8;
2531        cfg.sync_interval = Some(Duration::from_millis(5));
2532
2533        let mmap = Arc::new(MemMap::<u64>::new(cfg).unwrap());
2534        let barrier = Arc::new(std::sync::Barrier::new(4));
2535        let mut handles = Vec::new();
2536
2537        for t in 0..4 {
2538            let mmap_clone = Arc::clone(&mmap);
2539            let barrier_clone = Arc::clone(&barrier);
2540            handles.push(std::thread::spawn(move || {
2541                barrier_clone.wait();
2542                let mut tickets = Vec::with_capacity(0x32);
2543                for i in 0..0x32 {
2544                    let val = (t as u64) * 0x3E8 + i as u64;
2545                    tickets.push(mmap_clone.write(&val, t).unwrap());
2546                    std::thread::sleep(Duration::from_millis(1));
2547                }
2548                for ticket in tickets {
2549                    let _ = ticket.wait().unwrap();
2550                    assert!(ticket.is_durable());
2551                }
2552            }));
2553        }
2554
2555        for h in handles {
2556            h.join().unwrap();
2557        }
2558
2559        assert!(mmap.durable_epoch() >= mmap.inner.completion.read_current_epoch());
2560    }
2561
2562    #[test]
2563    fn ok_concurrent_ticket_force_stress() {
2564        let (_dir, mut cfg) = tmp_cfg();
2565        cfg.initial_t = 8;
2566        cfg.sync_interval = Some(Duration::from_secs(0x1E));
2567
2568        let mmap = Arc::new(MemMap::<u64>::new(cfg).unwrap());
2569        let barrier = Arc::new(std::sync::Barrier::new(4));
2570        let mut handles = Vec::new();
2571
2572        for t in 0..4 {
2573            let mmap_clone = Arc::clone(&mmap);
2574            let barrier_clone = Arc::clone(&barrier);
2575            handles.push(std::thread::spawn(move || {
2576                barrier_clone.wait();
2577                for i in 0..0x14 {
2578                    let val = (t as u64) * 0x64 + i as u64;
2579                    let ticket = mmap_clone.write(&val, t).unwrap();
2580                    let durable_epoch = ticket.force().unwrap();
2581                    assert!(ticket.is_durable());
2582                    assert!(durable_epoch >= ticket.epoch());
2583                }
2584            }));
2585        }
2586
2587        for h in handles {
2588            h.join().unwrap();
2589        }
2590    }
2591
2592    #[test]
2593    fn ok_exact_boundary_indices() {
2594        let (_dir, cfg) = tmp_cfg();
2595        let mmap = MemMap::<u64>::new(cfg).unwrap();
2596
2597        assert!(mmap.write(&0xF00D, INIT_T - 1).is_ok());
2598        assert_eq!(mmap.read(INIT_T - 1).unwrap(), 0xF00D);
2599
2600        assert_eq!(mmap.write(&0xBAD, INIT_T).unwrap_err().reason, err::HCF.reason);
2601        assert_eq!(mmap.read(INIT_T).unwrap_err().reason, err::HCF.reason);
2602    }
2603
2604    #[test]
2605    fn ok_large_slot_count_initialization() {
2606        let (_dir, mut cfg) = tmp_cfg();
2607        const LARGE_COUNT: usize = 0x2710;
2608        cfg.initial_t = LARGE_COUNT;
2609
2610        let mmap = MemMap::<u64>::new(cfg).unwrap();
2611        assert_eq!(mmap.length(), LARGE_COUNT * core::mem::size_of::<u64>());
2612
2613        assert!(mmap.write(&1, 0).is_ok());
2614        assert!(mmap.write(&0x1388, 0x1388).is_ok());
2615        assert!(mmap.write(&0x270F, LARGE_COUNT - 1).is_ok());
2616
2617        assert_eq!(mmap.read(0).unwrap(), 1);
2618        assert_eq!(mmap.read(0x1388).unwrap(), 0x1388);
2619        assert_eq!(mmap.read(LARGE_COUNT - 1).unwrap(), 0x270F);
2620    }
2621
2622    #[test]
2623    fn ok_new_with_sync_interval_none() {
2624        let (_dir, mut cfg) = tmp_cfg();
2625        cfg.sync_interval = None;
2626
2627        let mmap = MemMap::<u64>::new(cfg).unwrap();
2628        assert!(mmap.bg_thread.is_none());
2629
2630        let ticket = mmap.write(&0x55AA, 0).unwrap();
2631        assert!(!ticket.is_durable());
2632
2633        mmap.sync().unwrap();
2634        assert!(ticket.is_durable());
2635        assert_eq!(mmap.read(0).unwrap(), 0x55AA);
2636    }
2637
2638    #[test]
2639    fn ok_ticket_force_with_sync_interval_none() {
2640        let (_dir, mut cfg) = tmp_cfg();
2641        cfg.sync_interval = None;
2642
2643        let mmap = MemMap::<u64>::new(cfg).unwrap();
2644        assert!(mmap.bg_thread.is_none());
2645
2646        let ticket = mmap.write(&0x1122, 1).unwrap();
2647        assert!(!ticket.is_durable());
2648
2649        let epoch = ticket.force().unwrap();
2650        assert_eq!(epoch, ticket.epoch());
2651        assert!(ticket.is_durable());
2652        assert!(mmap.durable_epoch() >= ticket.epoch());
2653    }
2654
2655    #[test]
2656    fn ok_drop_with_sync_interval_none() {
2657        let (_dir, mut cfg) = tmp_cfg();
2658        cfg.sync_interval = None;
2659
2660        {
2661            let mmap = MemMap::<u64>::new(cfg.clone()).unwrap();
2662            assert!(mmap.bg_thread.is_none());
2663            let _ticket = mmap.write(&0xCAFE_F00D, 0).unwrap();
2664            drop(mmap);
2665        }
2666
2667        let reopened = MemMap::<u64>::open(cfg).unwrap();
2668        assert_eq!(reopened.read(0).unwrap(), 0xCAFE_F00D);
2669    }
2670
2671    #[test]
2672    fn ok_bg_worker_parks_when_idle_and_unparks_on_write() {
2673        let (_dir, mut cfg) = tmp_cfg();
2674        cfg.sync_interval = Some(Duration::from_millis(0x0F));
2675
2676        let mmap = MemMap::<u64>::new(cfg).unwrap();
2677
2678        // Initially without writes, the worker thread should park within a short delay
2679        let start = std::time::Instant::now();
2680        while !mmap.is_parked() && start.elapsed() < Duration::from_secs(1) {
2681            std::thread::sleep(Duration::from_millis(5));
2682        }
2683        assert!(mmap.is_parked());
2684
2685        // First write should unpark the worker
2686        let ticket = mmap.write(&0x9999, 0).unwrap();
2687        assert!(!mmap.is_parked());
2688
2689        // Ticket should become durable via the unparked background thread
2690        let epoch = ticket.wait().unwrap();
2691        assert_eq!(epoch, ticket.epoch());
2692        assert!(ticket.is_durable());
2693        assert_eq!(mmap.read(0).unwrap(), 0x9999);
2694
2695        // After flushing, worker should re-park when no new writes arrive
2696        let start = std::time::Instant::now();
2697        while !mmap.is_parked() && start.elapsed() < Duration::from_secs(1) {
2698            std::thread::sleep(Duration::from_millis(5));
2699        }
2700        assert!(mmap.is_parked());
2701    }
2702
2703    #[test]
2704    fn ok_batched_writes_while_active_remain_unparked() {
2705        let (_dir, mut cfg) = tmp_cfg();
2706        cfg.sync_interval = Some(Duration::from_millis(0x1E));
2707
2708        let mmap = MemMap::<u64>::new(cfg).unwrap();
2709
2710        // Wait until parked
2711        let start = std::time::Instant::now();
2712        while !mmap.is_parked() && start.elapsed() < Duration::from_secs(1) {
2713            std::thread::sleep(Duration::from_millis(5));
2714        }
2715        assert!(mmap.is_parked());
2716
2717        // First write unparks
2718        let t1 = mmap.write(&0x1111, 0).unwrap();
2719        assert!(!mmap.is_parked());
2720
2721        // Subsequent rapid writes find worker already unparked (is_parked == false)
2722        let t2 = mmap.write(&0x2222, 1).unwrap();
2723        let t3 = mmap.write(&0x3333, 2).unwrap();
2724        assert!(!mmap.is_parked());
2725
2726        t1.wait().unwrap();
2727        t2.wait().unwrap();
2728        t3.wait().unwrap();
2729
2730        assert!(t1.is_durable());
2731        assert!(t2.is_durable());
2732        assert!(t3.is_durable());
2733        assert_eq!(mmap.read(0).unwrap(), 0x1111);
2734        assert_eq!(mmap.read(1).unwrap(), 0x2222);
2735        assert_eq!(mmap.read(2).unwrap(), 0x3333);
2736    }
2737
2738    #[test]
2739    fn ok_concurrent_stress_park_unpark_no_lost_wakeups() {
2740        let (_dir, mut cfg) = tmp_cfg();
2741        cfg.initial_t = 0x10;
2742        cfg.sync_interval = Some(Duration::from_millis(5));
2743
2744        let mmap = Arc::new(MemMap::<u64>::new(cfg).unwrap());
2745        const THREADS: usize = 4;
2746        const WRITES_PER_THREAD: usize = 0x19;
2747        let mut handles = Vec::with_capacity(THREADS);
2748
2749        for t in 0..THREADS {
2750            let mmap_clone = Arc::clone(&mmap);
2751            handles.push(std::thread::spawn(move || {
2752                let mut tickets = Vec::with_capacity(WRITES_PER_THREAD);
2753                for i in 0..WRITES_PER_THREAD {
2754                    let val = (t as u64) << 0x20 | (i as u64);
2755                    let slot = t;
2756                    let ticket = mmap_clone.write(&val, slot).unwrap();
2757                    tickets.push(ticket);
2758
2759                    // Alternate between short sleep (allowing worker to park) and rapid writes
2760                    if i % 5 == 0 {
2761                        std::thread::sleep(Duration::from_millis(0x0F));
2762                    }
2763                }
2764
2765                for ticket in tickets {
2766                    ticket.wait().unwrap();
2767                    assert!(ticket.is_durable());
2768                }
2769            }));
2770        }
2771
2772        for h in handles {
2773            h.join().unwrap();
2774        }
2775
2776        assert!(mmap.durable_epoch() >= mmap.inner.completion.read_current_epoch());
2777    }
2778
2779    #[test]
2780    fn ok_spurious_wakeup_resumes_park_immediately() {
2781        let (_dir, mut cfg) = tmp_cfg();
2782        cfg.sync_interval = Some(Duration::from_secs(0x0A)); // Long interval
2783
2784        let mmap = MemMap::<u64>::new(cfg).unwrap();
2785
2786        // Wait until parked
2787        let start = std::time::Instant::now();
2788        while !mmap.is_parked() && start.elapsed() < Duration::from_secs(1) {
2789            std::thread::sleep(Duration::from_millis(5));
2790        }
2791        assert!(mmap.is_parked());
2792
2793        // Send spurious condvar notification without any writes
2794        mmap.inner.sync_condvar.notify_one();
2795        std::thread::sleep(Duration::from_millis(0x0A));
2796
2797        // Because no writes occurred, worker should detect spurious wakeup and re-park immediately rather than sleeping
2798        // for the 10-second interval
2799        let start = std::time::Instant::now();
2800        while !mmap.is_parked() && start.elapsed() < Duration::from_millis(0x1F4) {
2801            std::thread::sleep(Duration::from_millis(5));
2802        }
2803        assert!(mmap.is_parked());
2804    }
2805
2806    #[test]
2807    fn ok_delete_cleans_up_file_when_mmap_already_unmapped() {
2808        let (_dir, cfg) = tmp_cfg();
2809        let path = cfg.versioned_path();
2810        let mmap = MemMap::<u64>::new(cfg).unwrap();
2811        assert!(path.exists());
2812
2813        // Simulate mmap having been taken/unmapped prior to delete
2814        {
2815            let mut guard = mmap.inner.mmap.write().unwrap();
2816            let _ = guard.take();
2817        }
2818
2819        // delete should still successfully clean up the backing file on disk
2820        assert!(mmap.delete().is_ok());
2821        assert!(!path.exists());
2822    }
2823
2824    mod memmap_grow {
2825        use super::*;
2826
2827        #[test]
2828        fn ok_grow_updates_length() {
2829            let (_dir, cfg) = tmp_cfg();
2830            let mmap = MemMap::<u64>::new(cfg).unwrap();
2831            let initial_len = INIT_T * core::mem::size_of::<u64>();
2832            assert_eq!(mmap.length(), initial_len);
2833
2834            mmap.grow(4).unwrap();
2835            assert_eq!(mmap.length(), initial_len + 4 * core::mem::size_of::<u64>());
2836        }
2837
2838        #[test]
2839        fn ok_grow_zero_count() {
2840            let (_dir, cfg) = tmp_cfg();
2841            let mmap = MemMap::<u64>::new(cfg).unwrap();
2842            let initial_len = mmap.length();
2843
2844            mmap.grow(0).unwrap();
2845            assert_eq!(mmap.length(), initial_len);
2846        }
2847
2848        #[test]
2849        fn ok_grow_extends_with_zero() {
2850            let (_dir, cfg) = tmp_cfg();
2851            let mmap = MemMap::<u64>::new(cfg).unwrap();
2852            mmap.grow(3).unwrap();
2853
2854            for idx in INIT_T..(INIT_T + 3) {
2855                assert_eq!(mmap.read(idx).unwrap(), 0u64);
2856            }
2857        }
2858
2859        #[test]
2860        fn ok_grow_preserves_existing_data() {
2861            let (_dir, cfg) = tmp_cfg();
2862            let mmap = MemMap::<u64>::new(cfg).unwrap();
2863
2864            for i in 0..INIT_T {
2865                mmap.write(&(0xCAFE_0000 + i as u64), i).unwrap();
2866            }
2867
2868            mmap.grow(4).unwrap();
2869
2870            for i in 0..INIT_T {
2871                assert_eq!(mmap.read(i).unwrap(), 0xCAFE_0000 + i as u64);
2872            }
2873
2874            for i in INIT_T..(INIT_T + 4) {
2875                mmap.write(&(0xBEEF_0000 + i as u64), i).unwrap();
2876                assert_eq!(mmap.read(i).unwrap(), 0xBEEF_0000 + i as u64);
2877            }
2878        }
2879
2880        #[test]
2881        fn ok_grow_syncs_dirty_pages_when_curr_epoch_gt_durable() {
2882            let (_dir, mut cfg) = tmp_cfg();
2883            cfg.sync_interval = None; // Manual sync mode
2884
2885            let mmap = MemMap::<u64>::new(cfg).unwrap();
2886            let ticket = mmap.write(&0x1234_5678u64, 1).unwrap();
2887
2888            assert!(!ticket.is_durable());
2889            assert!(mmap.current_epoch() > mmap.durable_epoch());
2890
2891            // grow should trigger durability sync before re-mapping
2892            mmap.grow(2).unwrap();
2893
2894            assert!(ticket.is_durable());
2895            assert!(mmap.durable_epoch() >= ticket.epoch());
2896        }
2897
2898        #[test]
2899        fn ok_grow_when_already_durable() {
2900            let (_dir, mut cfg) = tmp_cfg();
2901            cfg.sync_interval = None;
2902
2903            let mmap = MemMap::<u64>::new(cfg).unwrap();
2904            let ticket = mmap.write(&0x9999u64, 0).unwrap();
2905            mmap.sync().unwrap();
2906            assert!(ticket.is_durable());
2907            assert_eq!(mmap.current_epoch(), mmap.durable_epoch());
2908
2909            mmap.grow(2).unwrap();
2910            assert_eq!(mmap.read(0).unwrap(), 0x9999u64);
2911        }
2912
2913        #[test]
2914        fn err_grow_overflow_usize() {
2915            let (_dir, cfg) = tmp_cfg();
2916            let mmap = MemMap::<u64>::new(cfg).unwrap();
2917
2918            let err = mmap.grow(usize::MAX).unwrap_err();
2919            assert_eq!(err.reason, err::GRW.reason);
2920        }
2921
2922        #[test]
2923        fn ok_grow_multiple_times() {
2924            let (_dir, cfg) = tmp_cfg();
2925            let mmap = MemMap::<u64>::new(cfg).unwrap();
2926
2927            mmap.write(&0x0Au64, 0).unwrap();
2928            mmap.grow(2).unwrap();
2929            mmap.write(&0x14u64, INIT_T).unwrap();
2930            mmap.grow(3).unwrap();
2931            mmap.write(&0x1Eu64, INIT_T + 2).unwrap();
2932
2933            assert_eq!(mmap.read(0).unwrap(), 0x0Au64);
2934            assert_eq!(mmap.read(INIT_T).unwrap(), 0x14u64);
2935            assert_eq!(mmap.read(INIT_T + 2).unwrap(), 0x1Eu64);
2936            assert_eq!(mmap.length(), (INIT_T + 5) * core::mem::size_of::<u64>());
2937        }
2938
2939        #[test]
2940        fn ok_concurrent_grow_and_reads_writes() {
2941            let (_dir, mut cfg) = tmp_cfg();
2942            cfg.sync_interval = Some(Duration::from_millis(0x0A));
2943            let mmap = Arc::new(MemMap::<u64>::new(cfg).unwrap());
2944
2945            let num_writers = 4;
2946            let mut handles = Vec::new();
2947
2948            for w in 0..num_writers {
2949                let m = Arc::clone(&mmap);
2950                handles.push(std::thread::spawn(move || {
2951                    for i in 0..0x32 {
2952                        let _ = m.write(&((w * 0x3E8 + i) as u64), w % INIT_T);
2953                        let _ = m.read(w % INIT_T);
2954                    }
2955                }));
2956            }
2957
2958            let m_grower = Arc::clone(&mmap);
2959            let grower_handle = std::thread::spawn(move || {
2960                for _ in 0..3 {
2961                    std::thread::sleep(Duration::from_millis(5));
2962                    m_grower.grow(2).unwrap();
2963                }
2964            });
2965
2966            for h in handles {
2967                h.join().unwrap();
2968            }
2969            grower_handle.join().unwrap();
2970
2971            assert_eq!(mmap.length(), (INIT_T + 6) * core::mem::size_of::<u64>());
2972        }
2973    }
2974
2975    mod thread_failure_lifecycle {
2976        use super::*;
2977
2978        #[test]
2979        fn err_new_failure_paths_spawn_no_sync_thread() {
2980            let (_dir, cfg) = tmp_cfg();
2981            // Pre-create file to force EXS error
2982            std::fs::write(cfg.versioned_path(), b"preexisting").unwrap();
2983
2984            let res = MemMap::<u64>::new(cfg);
2985            assert!(res.is_err());
2986            // No MemMap instance was created, and no background sync thread was spawned
2987        }
2988
2989        #[test]
2990        fn err_open_failure_paths_spawn_no_sync_thread() {
2991            let (_dir, mut cfg) = tmp_cfg();
2992            cfg.path = std::path::PathBuf::from("/non/existent/path/memmap.bin");
2993
2994            let res = MemMap::<u64>::open(cfg);
2995            assert!(res.is_err());
2996            // No MemMap instance was created, and no background sync thread was spawned
2997        }
2998
2999        #[test]
3000        fn ok_thread_shutdown_on_panic_unwinding() {
3001            let (_dir, cfg) = tmp_cfg();
3002            let thread_inner = Arc::new(atomic::AtomicBool::new(false));
3003            let thread_inner_clone = Arc::clone(&thread_inner);
3004
3005            // Spawn worker thread that panics with MemMap in scope
3006            let handle = std::thread::spawn(move || {
3007                let mmap = MemMap::<u64>::new(cfg).unwrap();
3008                let _inner = Arc::clone(&mmap.inner);
3009                thread_inner_clone.store(true, atomic::Ordering::Release);
3010                // Trigger panic while mmap is in scope
3011                panic!("intentional test panic to trigger Drop unwinding");
3012            });
3013
3014            assert!(handle.join().is_err());
3015            assert!(thread_inner.load(atomic::Ordering::Acquire));
3016            // Stack unwinding drops MemMap, setting shutdown = true and joining the background worker
3017        }
3018
3019        #[test]
3020        fn ok_thread_shutdown_in_delete_even_if_already_unmapped() {
3021            let (_dir, cfg) = tmp_cfg();
3022            let mmap = MemMap::<u64>::new(cfg).unwrap();
3023
3024            // Simulate mmap having been taken/unmapped prior to delete
3025            {
3026                let mut guard = mmap.inner.mmap.write().unwrap();
3027                let _ = guard.take();
3028            }
3029
3030            // delete shuts down and joins bg_thread before performing unmap and file deletion
3031            assert!(mmap.delete().is_ok());
3032        }
3033
3034        #[test]
3035        fn ok_grow_failure_keeps_thread_healthy() {
3036            let (_dir, cfg) = tmp_cfg();
3037            let mmap = MemMap::<u64>::new(cfg).unwrap();
3038
3039            assert!(mmap.grow(usize::MAX).is_err());
3040
3041            // The instance and background sync worker remain healthy
3042            let ticket = mmap.write(&0x2Au64, 0).unwrap();
3043            mmap.sync().unwrap();
3044            assert!(ticket.is_durable());
3045            assert_eq!(mmap.read(0).unwrap(), 0x2Au64);
3046        }
3047    }
3048
3049    mod memmap_transactions {
3050        use super::*;
3051
3052        #[test]
3053        fn ok_single_slot_cas_success_and_conflict() {
3054            let (_dir, cfg) = tmp_cfg();
3055            let mmap = MemMap::<u64>::new(cfg).unwrap();
3056
3057            assert_eq!(mmap.read(0).unwrap(), 0);
3058
3059            let t1 = mmap.compare_and_write(&0, &0x64, 0).unwrap();
3060            assert_eq!(mmap.read(0).unwrap(), 0x64);
3061            assert_eq!(mmap.current_epoch(), 1);
3062
3063            let err = mmap.compare_and_write(&0, &0xC8, 0).unwrap_err();
3064            assert_eq!(err.reason, err::CAS.reason);
3065            assert!(err.context.contains("slot data has changed"));
3066            assert_eq!(mmap.read(0).unwrap(), 0x64);
3067            assert_eq!(mmap.current_epoch(), 1); // Epoch does not advance on failure
3068
3069            let t2 = mmap.compare_and_write(&0x64, &0xC8, 0).unwrap();
3070            assert_eq!(mmap.read(0).unwrap(), 0xC8);
3071            assert_eq!(mmap.current_epoch(), 2);
3072            assert!(t2.epoch() > t1.epoch());
3073        }
3074
3075        #[test]
3076        fn ok_tx_multiple_writes_single_epoch() {
3077            let (_dir, cfg) = tmp_cfg();
3078            let mmap = MemMap::<u64>::new(cfg).unwrap();
3079
3080            let epoch_before = mmap.current_epoch();
3081
3082            let mut tx = mmap.new_tx();
3083            assert!(tx.is_empty());
3084            tx.write(&0, &0x6F, 0).unwrap();
3085            tx.write(&0, &0xDE, 1).unwrap();
3086            tx.write(&0, &0x14D, 2).unwrap();
3087            assert_eq!(tx.len(), 3);
3088
3089            let ticket = tx.commit().unwrap();
3090            let epoch_after = mmap.current_epoch();
3091
3092            // Single epoch increment for all 3 writes
3093            assert_eq!(epoch_after, epoch_before + 1);
3094            assert_eq!(ticket.epoch(), epoch_after);
3095
3096            assert_eq!(mmap.read(0).unwrap(), 0x6F);
3097            assert_eq!(mmap.read(1).unwrap(), 0xDE);
3098            assert_eq!(mmap.read(2).unwrap(), 0x14D);
3099
3100            ticket.wait().unwrap();
3101            assert!(ticket.is_durable());
3102        }
3103
3104        #[test]
3105        fn err_tx_cas_conflict_aborts_all_with_zero_side_effects() {
3106            let (_dir, cfg) = tmp_cfg();
3107            let mmap = MemMap::<u64>::new(cfg).unwrap();
3108
3109            // Prepare slot 1 with a different value
3110            mmap.write(&0x3E7, 1).unwrap();
3111            let epoch_before = mmap.current_epoch();
3112
3113            let mut tx = mmap.new_tx();
3114            tx.write(&0, &0x0A, 0).unwrap(); // Expected 0, currently 0 (would succeed)
3115            tx.write(&0, &0x14, 1).unwrap(); // Expected 0, currently 999 (will fail!)
3116            tx.write(&0, &0x1E, 2).unwrap(); // Expected 0, currently 0
3117
3118            let err = tx.commit().unwrap_err();
3119            assert_eq!(err.reason, err::CAS.reason);
3120            assert!(err.context.contains("state conflict at index 1"));
3121
3122            // Verify NONE of the slots were modified (all-or-nothing rollback)
3123            assert_eq!(mmap.read(0).unwrap(), 0);
3124            assert_eq!(mmap.read(1).unwrap(), 0x3E7);
3125            assert_eq!(mmap.read(2).unwrap(), 0);
3126
3127            // Epoch must NOT have advanced
3128            assert_eq!(mmap.current_epoch(), epoch_before);
3129        }
3130
3131        #[test]
3132        fn ok_tx_retry_after_conflict() {
3133            let (_dir, cfg) = tmp_cfg();
3134            let mmap = MemMap::<u64>::new(cfg).unwrap();
3135
3136            // Simulate concurrent write modifying slot 1
3137            mmap.write(&0x2A, 1).unwrap();
3138
3139            // First attempt assuming slot 1 is 0 fails
3140            let mut tx1 = mmap.new_tx();
3141            tx1.write(&0, &0x64, 0).unwrap();
3142            tx1.write(&0, &0xC8, 1).unwrap();
3143            assert_eq!(tx1.commit().unwrap_err().reason, err::CAS.reason);
3144
3145            // Caller observes current slot 1 value is 42, retries transaction with updated precondition
3146            let curr_val = mmap.read(1).unwrap();
3147            assert_eq!(curr_val, 0x2A);
3148
3149            let mut tx2 = mmap.new_tx();
3150            tx2.write(&0, &0x64, 0).unwrap();
3151            tx2.write(&0x2A, &0xC8, 1).unwrap();
3152            let ticket = tx2.commit().unwrap();
3153
3154            assert_eq!(mmap.read(0).unwrap(), 0x64);
3155            assert_eq!(mmap.read(1).unwrap(), 0xC8);
3156            ticket.wait().unwrap();
3157            assert!(ticket.is_durable());
3158        }
3159
3160        #[test]
3161        fn err_tx_out_of_order() {
3162            let (_dir, cfg) = tmp_cfg();
3163            let mmap = MemMap::<u64>::new(cfg).unwrap();
3164
3165            let mut tx = mmap.new_tx();
3166            tx.write(&0, &0x0A, 2).unwrap();
3167            let err = tx.write(&0, &0x14, 1).unwrap_err();
3168            assert_eq!(err.reason, err::HCF.reason);
3169            assert!(err.context.contains("strictly ordered"));
3170        }
3171
3172        #[test]
3173        fn err_tx_duplicate_index() {
3174            let (_dir, cfg) = tmp_cfg();
3175            let mmap = MemMap::<u64>::new(cfg).unwrap();
3176
3177            let mut tx = mmap.new_tx();
3178            tx.write(&0, &0x0A, 1).unwrap();
3179            let err = tx.write(&0, &0x14, 1).unwrap_err();
3180            assert_eq!(err.reason, err::HCF.reason);
3181            assert!(err.context.contains("strictly ordered"));
3182        }
3183
3184        #[test]
3185        fn err_tx_out_of_bounds() {
3186            let (_dir, cfg) = tmp_cfg();
3187            let mmap = MemMap::<u64>::new(cfg).unwrap();
3188
3189            let mut tx = mmap.new_tx();
3190            let err = tx.write(&0, &0x0A, INIT_T).unwrap_err();
3191            assert_eq!(err.reason, err::HCF.reason);
3192        }
3193
3194        #[test]
3195        fn err_tx_empty_commit() {
3196            let (_dir, cfg) = tmp_cfg();
3197            let mmap = MemMap::<u64>::new(cfg).unwrap();
3198
3199            let tx = mmap.new_tx();
3200            let err = tx.commit().unwrap_err();
3201            assert_eq!(err.reason, err::INV.reason);
3202            assert!(err.context.contains("empty transaction"));
3203        }
3204
3205        #[test]
3206        fn ok_tx_discard_on_drop() {
3207            let (_dir, cfg) = tmp_cfg();
3208            let mmap = MemMap::<u64>::new(cfg).unwrap();
3209
3210            let epoch_before = mmap.current_epoch();
3211            {
3212                let mut tx = mmap.new_tx();
3213                tx.write(&0, &0x309, 0).unwrap();
3214                tx.write(&0, &0x378, 1).unwrap();
3215                // tx dropped without calling commit()
3216            }
3217
3218            assert_eq!(mmap.read(0).unwrap(), 0);
3219            assert_eq!(mmap.read(1).unwrap(), 0);
3220            assert_eq!(mmap.current_epoch(), epoch_before);
3221        }
3222
3223        #[test]
3224        fn ok_tx_concurrent_non_overlapping() {
3225            let (_dir, mut cfg) = tmp_cfg();
3226            cfg.initial_t = 0x10;
3227            let mmap = Arc::new(MemMap::<u64>::new(cfg).unwrap());
3228
3229            let mut handles = Vec::new();
3230            for thread_id in 0..4u64 {
3231                let m = mmap.clone();
3232                handles.push(std::thread::spawn(move || {
3233                    let base = (thread_id as usize) * 4;
3234                    let mut tx = m.new_tx();
3235                    tx.write(&0, &(thread_id * 0x0A + 1), base).unwrap();
3236                    tx.write(&0, &(thread_id * 0x0A + 2), base + 1).unwrap();
3237                    tx.write(&0, &(thread_id * 0x0A + 3), base + 2).unwrap();
3238                    tx.write(&0, &(thread_id * 0x0A + 4), base + 3).unwrap();
3239                    let ticket = tx.commit().unwrap();
3240                    ticket.wait().unwrap();
3241                }));
3242            }
3243
3244            for h in handles {
3245                h.join().unwrap();
3246            }
3247
3248            for thread_id in 0..4u64 {
3249                let base = (thread_id as usize) * 4;
3250                assert_eq!(mmap.read(base).unwrap(), thread_id * 0x0A + 1);
3251                assert_eq!(mmap.read(base + 1).unwrap(), thread_id * 0x0A + 2);
3252                assert_eq!(mmap.read(base + 2).unwrap(), thread_id * 0x0A + 3);
3253                assert_eq!(mmap.read(base + 3).unwrap(), thread_id * 0x0A + 4);
3254            }
3255        }
3256
3257        #[test]
3258        fn ok_tx_durability_across_reopen() {
3259            let (_dir, cfg) = tmp_cfg();
3260
3261            {
3262                let mmap = MemMap::<u64>::new(cfg.clone()).unwrap();
3263                let mut tx = mmap.new_tx();
3264                tx.write(&0, &0xAAAA_BBBB, 0).unwrap();
3265                tx.write(&0, &0xCCCC_DDDD, 1).unwrap();
3266                let ticket = tx.commit().unwrap();
3267                ticket.wait().unwrap();
3268                assert!(ticket.is_durable());
3269            }
3270
3271            {
3272                let reopened = MemMap::<u64>::open(cfg).unwrap();
3273                assert_eq!(reopened.read(0).unwrap(), 0xAAAA_BBBB);
3274                assert_eq!(reopened.read(1).unwrap(), 0xCCCC_DDDD);
3275            }
3276        }
3277
3278        #[test]
3279        fn ok_tx_unparks_idle_bg_worker() {
3280            let (_dir, cfg) = tmp_cfg();
3281            let mmap = MemMap::<u64>::new(cfg).unwrap();
3282
3283            // Wait for worker to park
3284            let mut parked = false;
3285            for _ in 0..0x64 {
3286                if mmap.is_parked() {
3287                    parked = true;
3288                    break;
3289                }
3290                std::thread::sleep(Duration::from_millis(0x0A));
3291            }
3292            assert!(parked);
3293
3294            let mut tx = mmap.new_tx();
3295            tx.write(&0, &0x22B, 0).unwrap();
3296            let ticket = tx.commit().unwrap();
3297
3298            // Background worker was unparked and should make ticket durable
3299            ticket.wait().unwrap();
3300            assert!(ticket.is_durable());
3301            assert_eq!(mmap.read(0).unwrap(), 0x22B);
3302        }
3303    }
3304
3305    mod memmap_page_size {
3306        use super::*;
3307
3308        #[test]
3309        fn ok_page_size_returns_valid_power_of_two() {
3310            let page_size = MemMap::<u64>::page_size().unwrap();
3311            assert!(page_size >= 0x1000);
3312            assert!(page_size.is_power_of_two());
3313        }
3314
3315        #[test]
3316        fn ok_page_size_consistency() {
3317            let ps1 = MemMap::<u64>::page_size().unwrap();
3318            let ps2 = MemMap::<u64>::page_size().unwrap();
3319            let ps3 = MemMap::<ValidT>::page_size().unwrap();
3320            assert_eq!(ps1, ps2);
3321            assert_eq!(ps1, ps3);
3322        }
3323
3324        #[test]
3325        fn ok_page_size_across_threads() {
3326            let expected = MemMap::<u64>::page_size().unwrap();
3327            let mut handles = Vec::new();
3328
3329            for _ in 0..8 {
3330                handles.push(std::thread::spawn(move || {
3331                    let ps = MemMap::<u64>::page_size().unwrap();
3332                    assert_eq!(ps, expected);
3333                }));
3334            }
3335
3336            for h in handles {
3337                h.join().unwrap();
3338            }
3339        }
3340    }
3341}