Skip to main content

frozen_core/file/
mod.rs

1//! Custom implementation of `std::fs::File` with fixed-size buffer addressing, background
2//! synchronization, and epoch-based durability tracking
3//!
4//! ## Example
5//!
6//! ```
7//! use frozen_core::file::{File, FileCfg};
8//!
9//! const MID: u8 = 0;
10//!
11//! let dir = tempfile::tempdir().unwrap();
12//! let path = dir.path().join("tmp_file");
13//!
14//! let cfg = FileCfg {
15//!     module_id: MID,
16//!     buffer_size: 0x10,
17//!     path: path.clone(),
18//!     initial_available_buffers: 4,
19//!     sync_interval: None,
20//! };
21//!
22//! let file = File::new(cfg.clone()).unwrap();
23//! assert_eq!(file.length(), 0x10 * 4);
24//!
25//! let data = [1u8; 0x10];
26//! let ticket = file.write(&data, 0).unwrap();
27//! file.sync().unwrap();
28//! assert!(ticket.is_durable());
29//!
30//! let mut buf = [0u8; 0x10];
31//! file.read(&mut buf, 0).unwrap();
32//! assert_eq!(buf, data);
33//!
34//! assert!(File::new(cfg.clone()).is_err());
35//!
36//! assert!(file.delete().is_ok());
37//! assert!(!path.exists());
38//! ```
39
40// TODO: Tackle durability verification for uncommitted/non-durable reads internally in the future TODO: Tackle
41// coarse-grained RwLock write lock in sync_internal which blocks concurrent pread and pwrite ops during disk flushes
42// CRITICAL: On EIO during sync(), kernel clears dirty bits and consumes writeback errors (f_wb_err); subsequent sync()
43// calls return Ok(()), causing del_err() to falsely report lost data as durable (ref: PostgreSQL Bug #15096 /
44// fsyncgate); sync failures must poison the state or panic rather than retrying blindly
45
46mod interface;
47
48#[cfg(any(target_os = "linux", target_os = "macos"))]
49mod posix;
50
51#[cfg(target_os = "windows")]
52mod windows;
53
54use crate::{
55    ack::{AckTicket, Completion, SyncTrigger, TEpoch},
56    error::{BindModule, ErrCode, FrozenError, FrozenResult},
57};
58use interface::FileInterface;
59use std::sync::{Arc, Condvar, Mutex, RwLock, atomic};
60
61#[cfg(any(target_os = "linux", target_os = "macos"))]
62pub(in crate::file) type PlatformFile = posix::POSIXFile;
63
64#[cfg(target_os = "windows")]
65pub(in crate::file) type PlatformFile = windows::WINFile;
66
67/// Error codes for [`File`] module
68pub(in crate::file) mod err {
69    use super::{ErrCode, FrozenError, FrozenResult};
70
71    /// Domain Id for [`File`] is **8**
72    const ERRDOMAIN: u8 = 0x08;
73
74    /// internal fuck up (hault and catch fire)
75    pub const HCF: ErrCode = ErrCode::new(0x02, "hault and catch fire");
76
77    /// unknown error (fallback)
78    pub const UNK: ErrCode = ErrCode::new(0x04, "unknown error");
79
80    /// no more space available
81    pub const NSP: ErrCode = ErrCode::new(0x08, "not enough space available on the storage device");
82
83    /// syncing error
84    pub const SYN: ErrCode = ErrCode::new(0x0A, "failed to sync/flush data to storage device");
85
86    /// no write perm
87    pub const WRT: ErrCode = ErrCode::new(0x0C, "missing permissions for write");
88
89    /// no read perm
90    pub const RED: ErrCode = ErrCode::new(0x0E, "missing permissions for read");
91
92    /// invalid file path
93    pub const INV: ErrCode = ErrCode::new(0x10, "invalid path to file");
94
95    /// corrupted file
96    pub const CPT: ErrCode = ErrCode::new(0x12, "file is either invalid or corrupted");
97
98    /// unable to grow
99    pub const GRW: ErrCode = ErrCode::new(0x14, "unable to zero-extend file");
100
101    /// locks exhausted (can happen on systems w/ NFS)
102    pub const LEX: ErrCode =
103        ErrCode::new(0x18, "failed to obtain lock, as no more locks available");
104
105    /// no write/read permission
106    pub const PRM: ErrCode = ErrCode::new(0x1A, "missing permissions for IO");
107
108    /// unable to obtain exclusive lock
109    pub const LCK: ErrCode =
110        ErrCode::new(0x1C, "failed to obtain exclusive lock as file may already opened");
111
112    /// file already exists
113    pub const EXS: ErrCode = ErrCode::new(0x1E, "file already exists");
114
115    #[inline]
116    pub(in crate::file) fn raw_error<R, E: std::fmt::Display>(
117        code: ErrCode,
118        error: E,
119    ) -> FrozenResult<R> {
120        let err = FrozenError::unbound_raw(ERRDOMAIN, code, error);
121        Err(err)
122    }
123
124    #[inline]
125    pub(in crate::file) fn default_error<R>(code: ErrCode) -> FrozenResult<R> {
126        let err = FrozenError::unbound(ERRDOMAIN, code, "");
127        Err(err)
128    }
129
130    #[inline]
131    pub(in crate::file) fn make_error(code: ErrCode) -> FrozenError {
132        FrozenError::unbound(ERRDOMAIN, code, "")
133    }
134
135    #[inline]
136    pub(in crate::file) fn make_raw_error<E: std::fmt::Display>(
137        code: ErrCode,
138        error: E,
139    ) -> FrozenError {
140        FrozenError::unbound_raw(ERRDOMAIN, code, error)
141    }
142}
143
144/// File descriptor of [`File`]
145#[cfg(any(target_os = "linux", target_os = "macos"))]
146pub type FileId = libc::c_int;
147
148/// File handle of [`File`]
149///
150/// ## NOTE
151///
152/// On Windows, kernel HANDLE's are pointer sized but are stored as `isize` for atomic access; while
153/// the `INVALID_HANDLE_VALUE` maps to `-1isize`, which is [`windows::CLOSED_HANDLE`]
154#[cfg(target_os = "windows")]
155pub type FileId = isize;
156
157/// Configurations for [`File`]
158///
159/// ## Example
160///
161/// ```
162/// use frozen_core::file::FileCfg;
163/// use std::time::Duration;
164///
165/// let cfg = FileCfg {
166///     module_id: 1,
167///     path: std::path::PathBuf::from("test.db"),
168///     buffer_size: 0x100,
169///     initial_available_buffers: 0x10,
170///     sync_interval: Some(Duration::from_millis(0x64)),
171/// };
172///
173/// assert_eq!(cfg.buffer_size, 0x100);
174/// assert_eq!(cfg.initial_available_buffers, 0x10);
175/// assert_eq!(cfg.sync_interval, Some(Duration::from_millis(0x64)));
176/// ```
177#[derive(Debug, Clone)]
178pub struct FileCfg {
179    /// Identifier used while error propagation
180    pub module_id: u8,
181
182    /// Absolute path for/of the file
183    ///
184    /// *NOTE:* The caller must make sure that the path represents a file and all the parent
185    /// directories included in the path do exists
186    pub path: std::path::PathBuf,
187
188    /// Size (in bytes) of a single chunk in file
189    ///
190    /// A chunk is a small fixed size allocation and addressing unit used by [`File`] for all the
191    /// write/read ops
192    ///
193    /// These ops are operated by index of the chunk and not the offset of the byte
194    ///
195    /// *NOTE:* Chunk size when power of 2, is cache efficient and good for performance
196    pub buffer_size: usize,
197
198    /// Number of chunks to pre-allocate on fs when [`File`] is initialized
199    ///
200    /// Initial file length will be `buffer_size * initial_available_buffers` (bytes)
201    pub initial_available_buffers: usize,
202
203    /// Optional interval for the background sync thread
204    ///
205    /// If `Some(duration)`, a background worker thread is spawned which flushes dirty pages at this
206    /// interval; and if `None`, background sync is disabled and durability advances via manual sync
207    /// or forced tickets
208    pub sync_interval: Option<std::time::Duration>,
209}
210
211#[derive(Debug)]
212struct FileInner {
213    cfg: FileCfg,
214    file: RwLock<Option<PlatformFile>>,
215    current_length: atomic::AtomicUsize,
216    completion: Arc<Completion>,
217    sync_condvar: Condvar,
218    sync_mutex: Mutex<bool>,
219    shutdown: atomic::AtomicBool,
220}
221
222impl SyncTrigger for FileInner {
223    fn trigger_sync(&self) -> FrozenResult<()> {
224        if self.cfg.sync_interval.is_some() {
225            {
226                let mut guard = self.sync_mutex.lock().unwrap_or_else(|e| e.into_inner());
227                *guard = true;
228            }
229
230            self.sync_condvar.notify_one();
231            Ok(())
232        } else {
233            self.sync_internal()
234        }
235    }
236}
237
238impl FileInner {
239    fn sync_internal(&self) -> FrozenResult<()> {
240        let guard = self.file.write().unwrap_or_else(|e| e.into_inner());
241        let file = match guard.as_ref() {
242            Some(f) => f,
243            None => return Ok(()),
244        };
245
246        let target_epoch = self.completion.read_current_epoch();
247        let durable_epoch = self.completion.read_durable_epoch();
248
249        if target_epoch == durable_epoch {
250            return Ok(());
251        }
252
253        match file.sync() {
254            Ok(()) => {
255                self.completion.mark_epoch_as_durable(target_epoch);
256                self.completion.del_err();
257                self.completion.notify_all_listeners();
258
259                Ok(())
260            }
261            Err(e) => {
262                let e = e.with_module(self.cfg.module_id);
263                self.completion.set_err(e.clone());
264                self.completion.notify_all_listeners();
265
266                Err(e)
267            }
268        }
269    }
270}
271
272/// Custom implementation of `std::fs::File`
273#[derive(Debug)]
274pub struct File {
275    cfg: FileCfg,
276    inner: Arc<FileInner>,
277    bg_thread: Option<std::thread::JoinHandle<()>>,
278}
279
280unsafe impl Send for File {}
281unsafe impl Sync for File {}
282
283impl File {
284    fn from_platform_file(
285        cfg: FileCfg,
286        file: PlatformFile,
287        current_length: usize,
288    ) -> FrozenResult<Self> {
289        let completion = Arc::new(Completion::default());
290        let sync_interval = cfg.sync_interval;
291
292        let inner = Arc::new(FileInner {
293            cfg: cfg.clone(),
294            file: RwLock::new(Some(file)),
295            current_length: atomic::AtomicUsize::new(current_length),
296            completion: completion.clone(),
297            sync_condvar: Condvar::new(),
298            sync_mutex: Mutex::new(false),
299            shutdown: atomic::AtomicBool::new(false),
300        });
301
302        let weak_inner: std::sync::Weak<dyn SyncTrigger> =
303            Arc::downgrade(&inner) as std::sync::Weak<dyn SyncTrigger>;
304        completion.set_sync_trigger(weak_inner);
305
306        let bg_thread = if let Some(interval) = sync_interval {
307            let worker_inner = Arc::clone(&inner);
308            let handle =
309                std::thread::Builder::new().name("frozen-sync-worker".into()).spawn(move || {
310                    let mut guard =
311                        worker_inner.sync_mutex.lock().unwrap_or_else(|e| e.into_inner());
312
313                    while !worker_inner.shutdown.load(atomic::Ordering::Acquire) {
314                        if *guard {
315                            *guard = false;
316                            drop(guard);
317
318                            let _ = worker_inner.sync_internal();
319                            guard =
320                                worker_inner.sync_mutex.lock().unwrap_or_else(|e| e.into_inner());
321
322                            continue;
323                        }
324
325                        let (new_guard, _) = worker_inner
326                            .sync_condvar
327                            .wait_timeout(guard, interval)
328                            .unwrap_or_else(|e| e.into_inner());
329
330                        guard = new_guard;
331                        if worker_inner.shutdown.load(atomic::Ordering::Acquire) {
332                            break;
333                        }
334
335                        *guard = false;
336                        drop(guard);
337
338                        let _ = worker_inner.sync_internal();
339                        guard = worker_inner.sync_mutex.lock().unwrap_or_else(|e| e.into_inner());
340                    }
341                });
342
343            let handle = match handle {
344                Ok(h) => h,
345                Err(spawn_err) => {
346                    let mut file_guard = inner.file.write().unwrap_or_else(|e| e.into_inner());
347                    if let Some(f) = file_guard.take() {
348                        let mut err = err::make_raw_error(err::HCF, spawn_err);
349                        if let Err(close_err) = f.close() {
350                            err.add_suppressed(close_err);
351                        }
352
353                        return Err(err.with_module(cfg.module_id));
354                    }
355
356                    return err::raw_error(err::HCF, spawn_err).with_module(cfg.module_id);
357                }
358            };
359
360            Some(handle)
361        } else {
362            None
363        };
364
365        Ok(Self { cfg, inner, bg_thread })
366    }
367
368    /// Creates and pre-allocates a new [`File`] at `cfg.path`
369    ///
370    /// ## TOCTAU Safe
371    ///
372    /// If the file already exists, an error w/ `err::EXS` is returned to guard against TOCTOU
373    /// overwrites
374    ///
375    /// ## Exclusive Lock
376    ///
377    /// Acquires an exclusive advisory lock via `flock(LOCK_EX | LOCK_NB)` immediately after
378    /// descriptor creation
379    ///
380    /// ## Example
381    ///
382    /// ```
383    /// use frozen_core::file::{File, FileCfg};
384    ///
385    /// let dir = tempfile::tempdir().unwrap();
386    /// let cfg = FileCfg {
387    ///     module_id: 1,
388    ///     path: dir.path().join("new_file.bin"),
389    ///     buffer_size: 64,
390    ///     initial_available_buffers: 4,
391    ///     sync_interval: None,
392    /// };
393    ///
394    /// let file = File::new(cfg.clone()).unwrap();
395    /// assert_eq!(file.length(), 256);
396    ///
397    /// // Creating another file at the same path fails
398    /// assert!(File::new(cfg).is_err());
399    /// ```
400    pub fn new(cfg: FileCfg) -> FrozenResult<Self> {
401        let mid = cfg.module_id;
402        if cfg.buffer_size == 0 || cfg.initial_available_buffers == 0 {
403            return err::default_error(err::INV).with_module(mid);
404        }
405
406        let init_len = match cfg.buffer_size.checked_mul(cfg.initial_available_buffers) {
407            Some(len) => len,
408            None => return err::default_error(err::INV).with_module(mid),
409        };
410
411        let file = PlatformFile::create(&cfg.path).with_module(mid)?;
412        if let Err(mut e) = file.flock() {
413            if let Err(unlink_err) = file.unlink(&cfg.path) {
414                e.add_suppressed(unlink_err);
415            }
416
417            return Err(e.with_module(mid));
418        }
419
420        if let Err(mut e) = file.grow(0, init_len) {
421            if let Err(unlink_err) = file.unlink(&cfg.path) {
422                e.add_suppressed(unlink_err);
423            }
424
425            return Err(e.with_module(mid));
426        }
427
428        if let Err(mut e) = file.sync() {
429            if let Err(unlink_err) = file.unlink(&cfg.path) {
430                e.add_suppressed(unlink_err);
431            }
432
433            return Err(e.with_module(mid));
434        }
435
436        Self::from_platform_file(cfg, file, init_len)
437    }
438
439    /// Open an existing [`File`] while validating its layout invariants
440    ///
441    /// ## Invariants
442    ///
443    /// Following invariants are checked,
444    ///
445    /// - Current file length is at least `buffer_size * initial_available_buffers`
446    /// - Current file length is a multiple of `buffer_size`
447    ///
448    /// If any invariant is violated, the file is closed and `err::CPT` is returned
449    ///
450    /// ## Example
451    ///
452    /// ```
453    /// use frozen_core::file::{File, FileCfg};
454    ///
455    /// let dir = tempfile::tempdir().unwrap();
456    /// let cfg = FileCfg {
457    ///     module_id: 1,
458    ///     path: dir.path().join("open_file.bin"),
459    ///     buffer_size: 64,
460    ///     initial_available_buffers: 4,
461    ///     sync_interval: None,
462    /// };
463    ///
464    /// {
465    ///     let file = File::new(cfg.clone()).unwrap();
466    ///     assert_eq!(file.length(), 256);
467    /// }
468    ///
469    /// let opened = File::open(cfg).unwrap();
470    /// assert_eq!(opened.length(), 256);
471    /// ```
472    pub fn open(cfg: FileCfg) -> FrozenResult<Self> {
473        let mid = cfg.module_id;
474
475        if cfg.buffer_size == 0 || cfg.initial_available_buffers == 0 {
476            return err::default_error(err::INV).with_module(mid);
477        }
478
479        let init_len = match cfg.buffer_size.checked_mul(cfg.initial_available_buffers) {
480            Some(len) => len,
481            None => return err::default_error(err::INV).with_module(mid),
482        };
483
484        let file = PlatformFile::open(&cfg.path).with_module(mid)?;
485
486        if let Err(mut e) = file.flock() {
487            if let Err(close_err) = file.close() {
488                e.add_suppressed(close_err);
489            }
490
491            return Err(e.with_module(mid));
492        }
493
494        let curr_len = match file.length() {
495            Ok(len) => len,
496            Err(mut e) => {
497                if let Err(close_err) = file.close() {
498                    e.add_suppressed(close_err);
499                }
500
501                return Err(e.with_module(mid));
502            }
503        };
504
505        if curr_len < init_len || curr_len % cfg.buffer_size != 0 {
506            let mut e = err::make_error(err::CPT);
507            if let Err(close_err) = file.close() {
508                e.add_suppressed(close_err);
509            }
510
511            return Err(e.with_module(mid));
512        }
513
514        Self::from_platform_file(cfg, file, curr_len)
515    }
516
517    /// Open an existing [`File`] or create it if it does not yet exist
518    ///
519    /// ## Semantics
520    ///
521    /// Uses idempotent open-or-create (`OPEN_ALWAYS` on Windows, `O_CREAT` without `O_EXCL` on
522    /// POSIX)
523    ///
524    /// If the file is newly created, it is grown to `buffer_size * initial_available_buffers` and
525    /// synced; and if the file already exists, it is opened and its layout invariants are validated
526    ///
527    /// In both cases, an exclusive advisory lock is acquired on the file
528    ///
529    /// ## Example
530    ///
531    /// ```
532    /// use frozen_core::file::{File, FileCfg};
533    ///
534    /// let dir = tempfile::tempdir().unwrap();
535    /// let cfg = FileCfg {
536    ///     module_id: 1,
537    ///     path: dir.path().join("idempotent.bin"),
538    ///     buffer_size: 64,
539    ///     initial_available_buffers: 2,
540    ///     sync_interval: None,
541    /// };
542    ///
543    /// // First invocation creates and initializes the file
544    /// {
545    ///     let file = File::open_or_create(cfg.clone()).unwrap();
546    ///     assert_eq!(file.length(), 128);
547    /// }
548    ///
549    /// // Subsequent invocation opens the existing file safely
550    /// let reopened = File::open_or_create(cfg).unwrap();
551    /// assert_eq!(reopened.length(), 128);
552    /// ```
553    pub fn open_or_create(cfg: FileCfg) -> FrozenResult<Self> {
554        let mid = cfg.module_id;
555
556        if cfg.buffer_size == 0 || cfg.initial_available_buffers == 0 {
557            return err::default_error(err::INV).with_module(mid);
558        }
559
560        let init_len = match cfg.buffer_size.checked_mul(cfg.initial_available_buffers) {
561            Some(len) => len,
562            None => return err::default_error(err::INV).with_module(mid),
563        };
564
565        let file = PlatformFile::new(&cfg.path).with_module(mid)?;
566        if let Err(mut e) = file.flock() {
567            if let Err(close_err) = file.close() {
568                e.add_suppressed(close_err);
569            }
570
571            return Err(e.with_module(mid));
572        }
573
574        let curr_len = match file.length() {
575            Ok(len) => len,
576            Err(mut e) => {
577                if let Err(close_err) = file.close() {
578                    e.add_suppressed(close_err);
579                }
580
581                return Err(e.with_module(mid));
582            }
583        };
584
585        if curr_len == 0 {
586            if let Err(mut e) = file.grow(0, init_len) {
587                if let Err(close_err) = file.close() {
588                    e.add_suppressed(close_err);
589                }
590
591                return Err(e.with_module(mid));
592            }
593
594            if let Err(mut e) = file.sync() {
595                if let Err(close_err) = file.close() {
596                    e.add_suppressed(close_err);
597                }
598
599                return Err(e.with_module(mid));
600            }
601
602            Self::from_platform_file(cfg, file, init_len)
603        } else {
604            if curr_len < init_len || curr_len % cfg.buffer_size != 0 {
605                let mut e = err::make_error(err::CPT);
606                if let Err(close_err) = file.close() {
607                    e.add_suppressed(close_err);
608                }
609
610                return Err(e.with_module(mid));
611            }
612
613            Self::from_platform_file(cfg, file, curr_len)
614        }
615    }
616
617    /// Read bytes starting from buffer `index` into `buf` w/ `pread` syscall
618    ///
619    /// ## Durability Guarantee & Caller Responsibility
620    ///
621    /// It is 100% the caller's responsibility to ensure that they do not read newly written data
622    /// unless the older data has been confirmed durable
623    /// (e.g. by checking or waiting on the write's [`AckTicket`])
624    ///
625    /// ## Multiple Buffers
626    ///
627    /// The input `buf` can span across a single or multiple buffers in memory
628    ///
629    /// ## Constraints
630    ///
631    /// - `buf.len()` must be a non-zero multiple of `cfg.buffer_size`
632    /// - Reading beyond current file length will return `err::HCF`
633    ///
634    /// ## Example
635    ///
636    /// ```
637    /// use frozen_core::file::{File, FileCfg};
638    ///
639    /// let dir = tempfile::tempdir().unwrap();
640    /// let cfg = FileCfg {
641    ///     module_id: 1,
642    ///     path: dir.path().join("read_test.bin"),
643    ///     buffer_size: 16,
644    ///     initial_available_buffers: 2,
645    ///     sync_interval: None,
646    /// };
647    ///
648    /// let file = File::new(cfg).unwrap();
649    /// let data = [7u8; 16];
650    /// let ticket = file.write(&data, 0).unwrap();
651    /// ticket.force().unwrap();
652    ///
653    /// let mut out = [0u8; 16];
654    /// file.read(&mut out, 0).unwrap();
655    /// assert_eq!(out, data);
656    /// ```
657    #[inline(always)]
658    pub fn read(&self, buf: &mut [u8], index: usize) -> FrozenResult<()> {
659        let mid = self.cfg.module_id;
660        if buf.is_empty() {
661            return Ok(());
662        }
663
664        if buf.len() % self.cfg.buffer_size != 0 {
665            return err::default_error(err::INV).with_module(mid);
666        }
667
668        let offset = match index.checked_mul(self.cfg.buffer_size) {
669            Some(off) => off,
670            None => return err::default_error(err::INV).with_module(mid),
671        };
672
673        if offset.checked_add(buf.len()).is_none_or(|end| end > self.length()) {
674            return err::default_error(err::HCF).with_module(mid);
675        }
676
677        let guard = self.inner.file.read().unwrap_or_else(|e| e.into_inner());
678        let file = match guard.as_ref() {
679            Some(f) => f,
680            None => return err::default_error(err::INV).with_module(mid),
681        };
682
683        file.pread(buf, offset).with_module(mid)
684    }
685
686    /// Write bytes starting at buffer `index` from `buf` w/ `pwrite` syscall
687    ///
688    /// Returns an [`AckTicket`] representing the durability acknowledgement of this write operation
689    ///
690    /// ## Multiple Buffers
691    ///
692    /// The input `buf` can span across a single or multiple buffers in memory
693    ///
694    /// ## Constraints
695    ///
696    /// - `buf.len()` must be a non-zero multiple of `cfg.buffer_size`
697    /// - Writing beyond current file length will return `err::HCF`
698    ///
699    /// ## Example
700    ///
701    /// ```
702    /// use frozen_core::file::{File, FileCfg};
703    ///
704    /// let dir = tempfile::tempdir().unwrap();
705    /// let cfg = FileCfg {
706    ///     module_id: 1,
707    ///     path: dir.path().join("write_test.bin"),
708    ///     buffer_size: 16,
709    ///     initial_available_buffers: 2,
710    ///     sync_interval: None,
711    /// };
712    ///
713    /// let file = File::new(cfg).unwrap();
714    /// let data = [42u8; 16];
715    /// let ticket = file.write(&data, 1).unwrap();
716    /// assert_eq!(ticket.is_durable(), false);
717    ///
718    /// file.sync().unwrap();
719    /// assert!(ticket.is_durable());
720    /// ```
721    #[inline(always)]
722    pub fn write(&self, buf: &[u8], index: usize) -> FrozenResult<AckTicket> {
723        let mid = self.cfg.module_id;
724        if buf.is_empty() {
725            let current = self.inner.completion.read_current_epoch();
726            return Ok(AckTicket::new(current, self.inner.completion.clone()));
727        }
728
729        if buf.len() % self.cfg.buffer_size != 0 {
730            return err::default_error(err::INV).with_module(mid);
731        }
732
733        let offset = match index.checked_mul(self.cfg.buffer_size) {
734            Some(off) => off,
735            None => return err::default_error(err::INV).with_module(mid),
736        };
737
738        if offset.checked_add(buf.len()).is_none_or(|end| end > self.length()) {
739            return err::default_error(err::HCF).with_module(mid);
740        }
741
742        let guard = self.inner.file.read().unwrap_or_else(|e| e.into_inner());
743        let file = match guard.as_ref() {
744            Some(f) => f,
745            None => return err::default_error(err::INV).with_module(mid),
746        };
747
748        file.pwrite(buf, offset).with_module(mid)?;
749
750        let epoch = self.inner.completion.increment_current_epoch();
751        Ok(AckTicket::new(epoch, self.inner.completion.clone()))
752    }
753
754    /// Grow file size of [`File`] by given `count` of buffers
755    ///
756    /// After successful execution, updated file length will be `current_length + (count *
757    /// buffer_size)`
758    ///
759    /// ## Example
760    ///
761    /// ```
762    /// use frozen_core::file::{File, FileCfg};
763    ///
764    /// let dir = tempfile::tempdir().unwrap();
765    /// let cfg = FileCfg {
766    ///     module_id: 1,
767    ///     path: dir.path().join("grow_test.bin"),
768    ///     buffer_size: 32,
769    ///     initial_available_buffers: 2,
770    ///     sync_interval: None,
771    /// };
772    ///
773    /// let file = File::new(cfg).unwrap();
774    /// assert_eq!(file.length(), 64);
775    ///
776    /// file.grow(3).unwrap();
777    /// assert_eq!(file.length(), 160);
778    /// assert_eq!(file.total_buffers().unwrap(), 5);
779    /// ```
780    pub fn grow(&self, count: usize) -> FrozenResult<()> {
781        let mid = self.cfg.module_id;
782        if count == 0 {
783            return Ok(());
784        }
785
786        let len_to_add = match self.cfg.buffer_size.checked_mul(count) {
787            Some(len) => len,
788            None => return err::default_error(err::GRW).with_module(mid),
789        };
790
791        let guard = self.inner.file.write().unwrap_or_else(|e| e.into_inner());
792        let file = match guard.as_ref() {
793            Some(f) => f,
794            None => return err::default_error(err::INV).with_module(mid),
795        };
796
797        let curr_len = self.inner.current_length.load(atomic::Ordering::Acquire);
798        file.grow(curr_len, len_to_add).with_module(mid)?;
799
800        self.inner.current_length.fetch_add(len_to_add, atomic::Ordering::Release);
801        self.inner.completion.increment_current_epoch();
802
803        Ok(())
804    }
805
806    /// Syncs in-mem data to the storage device
807    ///
808    /// ## Example
809    ///
810    /// ```
811    /// use frozen_core::file::{File, FileCfg};
812    ///
813    /// let dir = tempfile::tempdir().unwrap();
814    /// let cfg = FileCfg {
815    ///     module_id: 1,
816    ///     path: dir.path().join("sync_test.bin"),
817    ///     buffer_size: 16,
818    ///     initial_available_buffers: 2,
819    ///     sync_interval: None,
820    /// };
821    ///
822    /// let file = File::new(cfg).unwrap();
823    /// let ticket = file.write(&[9u8; 16], 0).unwrap();
824    /// assert!(!ticket.is_durable());
825    ///
826    /// file.sync().unwrap();
827    /// assert!(ticket.is_durable());
828    /// ```
829    #[inline]
830    pub fn sync(&self) -> FrozenResult<()> {
831        self.inner.sync_internal()
832    }
833
834    /// Best-effort call to prompt kernel to start flushing dirty pages in the specified chunk range
835    ///
836    /// If `count == 0`, this call is a no-op and returns `Ok(())`
837    ///
838    /// ## Example
839    ///
840    /// ```
841    /// # #[cfg(target_os = "linux")]
842    /// # {
843    /// use frozen_core::file::{File, FileCfg};
844    ///
845    /// let dir = tempfile::tempdir().unwrap();
846    /// let cfg = FileCfg {
847    ///     module_id: 1,
848    ///     path: dir.path().join("sync_range.bin"),
849    ///     buffer_size: 64,
850    ///     initial_available_buffers: 4,
851    ///     sync_interval: None,
852    /// };
853    ///
854    /// let file = File::new(cfg).unwrap();
855    /// file.write(&[1u8; 64], 0).unwrap();
856    /// assert!(file.sync_range(0, 1).is_ok());
857    /// assert!(file.sync_range(0, 0).is_ok());
858    /// # }
859    /// ```
860    #[cfg(target_os = "linux")]
861    pub fn sync_range(&self, index: usize, count: usize) -> FrozenResult<()> {
862        if count == 0 {
863            return Ok(());
864        }
865
866        let mid = self.cfg.module_id;
867        let offset = match index.checked_mul(self.cfg.buffer_size) {
868            Some(off) => off,
869            None => return err::default_error(err::INV).with_module(mid),
870        };
871        let len_to_sync = match count.checked_mul(self.cfg.buffer_size) {
872            Some(len) => len,
873            None => return err::default_error(err::INV).with_module(mid),
874        };
875
876        let guard = self.inner.file.read().unwrap_or_else(|e| e.into_inner());
877        let file = match guard.as_ref() {
878            Some(f) => f,
879            None => return err::default_error(err::INV).with_module(mid),
880        };
881
882        file.sync_range(offset, len_to_sync).with_module(mid)
883    }
884
885    /// Fetch total available buffers in [`File`]
886    ///
887    /// ## Example
888    ///
889    /// ```
890    /// use frozen_core::file::{File, FileCfg};
891    ///
892    /// let dir = tempfile::tempdir().unwrap();
893    /// let cfg = FileCfg {
894    ///     module_id: 1,
895    ///     path: dir.path().join("buffers.bin"),
896    ///     buffer_size: 32,
897    ///     initial_available_buffers: 8,
898    ///     sync_interval: None,
899    /// };
900    ///
901    /// let file = File::new(cfg).unwrap();
902    /// assert_eq!(file.total_buffers().unwrap(), 8);
903    /// ```
904    #[inline]
905    pub fn total_buffers(&self) -> FrozenResult<usize> {
906        let curr_len = self.length();
907        let buffer_size = self.cfg.buffer_size;
908
909        if crate::hints::unlikely(curr_len % buffer_size != 0) {
910            return err::default_error(err::CPT).with_module(self.cfg.module_id);
911        }
912
913        Ok(curr_len / buffer_size)
914    }
915
916    /// Get reference to configuration of [`File`]
917    ///
918    /// ## Example
919    ///
920    /// ```
921    /// use frozen_core::file::{File, FileCfg};
922    ///
923    /// let dir = tempfile::tempdir().unwrap();
924    /// let cfg = FileCfg {
925    ///     module_id: 1,
926    ///     path: dir.path().join("cfg.bin"),
927    ///     buffer_size: 64,
928    ///     initial_available_buffers: 2,
929    ///     sync_interval: None,
930    /// };
931    ///
932    /// let file = File::new(cfg).unwrap();
933    /// assert_eq!(file.cfg().buffer_size, 64);
934    /// assert_eq!(file.cfg().initial_available_buffers, 2);
935    /// ```
936    #[inline]
937    pub fn cfg(&self) -> &FileCfg {
938        &self.cfg
939    }
940
941    /// Read current length (in bytes) of [`File`]
942    ///
943    /// ## Example
944    ///
945    /// ```
946    /// use frozen_core::file::{File, FileCfg};
947    ///
948    /// let dir = tempfile::tempdir().unwrap();
949    /// let cfg = FileCfg {
950    ///     module_id: 1,
951    ///     path: dir.path().join("len.bin"),
952    ///     buffer_size: 128,
953    ///     initial_available_buffers: 4,
954    ///     sync_interval: None,
955    /// };
956    ///
957    /// let file = File::new(cfg).unwrap();
958    /// assert_eq!(file.length(), 512);
959    /// ```
960    #[inline]
961    pub fn length(&self) -> usize {
962        self.inner.current_length.load(atomic::Ordering::Acquire)
963    }
964
965    /// Fetch the latest assigned durability epoch
966    ///
967    /// ## Example
968    ///
969    /// ```
970    /// use frozen_core::file::{File, FileCfg};
971    ///
972    /// let dir = tempfile::tempdir().unwrap();
973    /// let cfg = FileCfg {
974    ///     module_id: 1,
975    ///     path: dir.path().join("epoch.bin"),
976    ///     buffer_size: 16,
977    ///     initial_available_buffers: 2,
978    ///     sync_interval: None,
979    /// };
980    ///
981    /// let file = File::new(cfg).unwrap();
982    /// assert_eq!(file.current_epoch(), 0);
983    ///
984    /// file.write(&[1u8; 16], 0).unwrap();
985    /// assert_eq!(file.current_epoch(), 1);
986    /// ```
987    #[inline]
988    pub fn current_epoch(&self) -> TEpoch {
989        self.inner.completion.read_current_epoch()
990    }
991
992    /// Fetch the latest durable epoch
993    ///
994    /// ## Example
995    ///
996    /// ```
997    /// use frozen_core::file::{File, FileCfg};
998    ///
999    /// let dir = tempfile::tempdir().unwrap();
1000    /// let cfg = FileCfg {
1001    ///     module_id: 1,
1002    ///     path: dir.path().join("durable_epoch.bin"),
1003    ///     buffer_size: 16,
1004    ///     initial_available_buffers: 2,
1005    ///     sync_interval: None,
1006    /// };
1007    ///
1008    /// let file = File::new(cfg).unwrap();
1009    /// file.write(&[1u8; 16], 0).unwrap();
1010    /// assert_eq!(file.durable_epoch(), 0);
1011    ///
1012    /// file.sync().unwrap();
1013    /// assert_eq!(file.durable_epoch(), 1);
1014    /// ```
1015    #[inline]
1016    pub fn durable_epoch(&self) -> TEpoch {
1017        self.inner.completion.read_durable_epoch()
1018    }
1019
1020    /// Fetch reference to underlying durability [`Completion`]
1021    ///
1022    /// ## Example
1023    ///
1024    /// ```
1025    /// use frozen_core::file::{File, FileCfg};
1026    ///
1027    /// let dir = tempfile::tempdir().unwrap();
1028    /// let cfg = FileCfg {
1029    ///     module_id: 1,
1030    ///     path: dir.path().join("completion.bin"),
1031    ///     buffer_size: 16,
1032    ///     initial_available_buffers: 2,
1033    ///     sync_interval: None,
1034    /// };
1035    ///
1036    /// let file = File::new(cfg).unwrap();
1037    /// let completion = file.completion();
1038    /// assert_eq!(completion.read_current_epoch(), 0);
1039    /// ```
1040    #[inline]
1041    pub fn completion(&self) -> &Arc<Completion> {
1042        &self.inner.completion
1043    }
1044
1045    /// Check if [`File`] exists on storage device or not
1046    ///
1047    /// ## Access Semantics
1048    ///
1049    /// Uses `access(path, F_OK)` to verify the existence of the file
1050    ///
1051    /// ## Example
1052    ///
1053    /// ```
1054    /// use frozen_core::file::{File, FileCfg};
1055    ///
1056    /// let dir = tempfile::tempdir().unwrap();
1057    /// let cfg = FileCfg {
1058    ///     module_id: 1,
1059    ///     path: dir.path().join("exists.bin"),
1060    ///     buffer_size: 16,
1061    ///     initial_available_buffers: 2,
1062    ///     sync_interval: None,
1063    /// };
1064    ///
1065    /// let file = File::new(cfg).unwrap();
1066    /// assert!(file.exists().unwrap());
1067    /// ```
1068    #[inline]
1069    pub fn exists(&self) -> FrozenResult<bool> {
1070        PlatformFile::exists(&self.cfg.path).with_module(self.cfg.module_id)
1071    }
1072
1073    /// Deletes the [`File`] entry from the storage device
1074    ///
1075    /// Consumes `self` by value to prevent any concurrent or post-deletion operations
1076    ///
1077    /// Unlinks the file at `path`, closes the underlying descriptor, and syncs the parent directory
1078    /// to guarantee crash-safe durability
1079    ///
1080    /// ## Caller Responsibilities
1081    ///
1082    /// The caller must ensure that when calling `delete`, no write operations are pending for
1083    /// durability and no write operations are invoked during or after `delete` is called
1084    ///
1085    /// ## Example
1086    ///
1087    /// ```
1088    /// use frozen_core::file::{File, FileCfg};
1089    ///
1090    /// let dir = tempfile::tempdir().unwrap();
1091    /// let path = dir.path().join("delete_test.bin");
1092    /// let cfg = FileCfg {
1093    ///     module_id: 1,
1094    ///     path: path.clone(),
1095    ///     buffer_size: 16,
1096    ///     initial_available_buffers: 2,
1097    ///     sync_interval: None,
1098    /// };
1099    ///
1100    /// let file = File::new(cfg).unwrap();
1101    /// assert!(path.exists());
1102    ///
1103    /// file.delete().unwrap();
1104    /// assert!(!path.exists());
1105    /// ```
1106    pub fn delete(mut self) -> FrozenResult<()> {
1107        let mid = self.cfg.module_id;
1108        self.inner.shutdown.store(true, atomic::Ordering::Release);
1109        {
1110            let mut guard = self.inner.sync_mutex.lock().unwrap_or_else(|e| e.into_inner());
1111            *guard = true;
1112        }
1113        self.inner.sync_condvar.notify_all();
1114
1115        if let Some(handle) = self.bg_thread.take() {
1116            let _ = handle.join();
1117        }
1118
1119        let mut guard = self.inner.file.write().unwrap_or_else(|e| e.into_inner());
1120        let file = match guard.take() {
1121            Some(f) => f,
1122            None => return err::default_error(err::INV).with_module(mid),
1123        };
1124
1125        file.unlink(&self.cfg.path).with_module(mid)
1126    }
1127
1128    /// Get file descriptor or handle for [`File`]
1129    ///
1130    /// ## Example
1131    ///
1132    /// ```
1133    /// use frozen_core::file::{File, FileCfg};
1134    ///
1135    /// let dir = tempfile::tempdir().unwrap();
1136    /// let cfg = FileCfg {
1137    ///     module_id: 1,
1138    ///     path: dir.path().join("fd_test.bin"),
1139    ///     buffer_size: 16,
1140    ///     initial_available_buffers: 2,
1141    ///     sync_interval: None,
1142    /// };
1143    ///
1144    /// let file = File::new(cfg).unwrap();
1145    /// #[cfg(unix)]
1146    /// assert!(file.fd() >= 0);
1147    /// #[cfg(windows)]
1148    /// assert_ne!(file.fd(), -1isize);
1149    /// ```
1150    #[inline]
1151    pub fn fd(&self) -> FileId {
1152        let guard = self.inner.file.read().unwrap_or_else(|e| e.into_inner());
1153        match guard.as_ref() {
1154            Some(f) => f.fd(),
1155            None => PlatformFile::CLOSED_ID,
1156        }
1157    }
1158}
1159
1160impl Drop for File {
1161    fn drop(&mut self) {
1162        self.inner.shutdown.store(true, atomic::Ordering::Release);
1163        {
1164            let mut guard = self.inner.sync_mutex.lock().unwrap_or_else(|e| e.into_inner());
1165            *guard = true;
1166        }
1167        self.inner.sync_condvar.notify_all();
1168
1169        if let Some(handle) = self.bg_thread.take() {
1170            let _ = handle.join();
1171        }
1172
1173        let is_closed = {
1174            let guard = self.inner.file.read().unwrap_or_else(|e| e.into_inner());
1175            match guard.as_ref() {
1176                Some(f) => f.is_closed(),
1177                None => true,
1178            }
1179        };
1180
1181        if is_closed {
1182            return;
1183        }
1184
1185        let _ = self.inner.sync_internal();
1186    }
1187}
1188
1189#[cfg(test)]
1190mod tests {
1191    use super::*;
1192    use std::sync::{Arc, Barrier};
1193
1194    const MID: u8 = 0;
1195    const BUFFER_SIZE: usize = 0x10;
1196    const INIT_BUFFERS: usize = 4;
1197
1198    fn tmp_path() -> (tempfile::TempDir, FileCfg) {
1199        let dir = tempfile::tempdir().unwrap();
1200        let path = dir.path().join("tmp_file");
1201        let cfg = FileCfg {
1202            module_id: MID,
1203            path,
1204            buffer_size: BUFFER_SIZE,
1205            initial_available_buffers: INIT_BUFFERS,
1206            sync_interval: None,
1207        };
1208
1209        (dir, cfg)
1210    }
1211
1212    mod file_new {
1213        use super::*;
1214
1215        #[test]
1216        fn ok_new_creates_file_with_correct_length() {
1217            let (_dir, cfg) = tmp_path();
1218            let file = File::new(cfg.clone()).unwrap();
1219
1220            let expected_len = BUFFER_SIZE * INIT_BUFFERS;
1221            assert_eq!(file.length(), expected_len);
1222            assert_eq!(file.cfg().buffer_size, BUFFER_SIZE);
1223            assert_ne!(file.fd(), PlatformFile::CLOSED_ID);
1224            assert!(cfg.path.exists());
1225        }
1226
1227        #[test]
1228        fn err_new_when_already_exists() {
1229            let (_dir, cfg) = tmp_path();
1230            let file = File::new(cfg.clone()).unwrap();
1231            drop(file);
1232
1233            let err = File::new(cfg).unwrap_err();
1234            assert_eq!(err.reason, err::EXS.reason);
1235        }
1236
1237        #[test]
1238        fn err_new_invalid_cfg() {
1239            let (_dir, mut cfg) = tmp_path();
1240
1241            cfg.buffer_size = 0;
1242            let err = File::new(cfg.clone()).unwrap_err();
1243            assert_eq!(err.reason, err::INV.reason);
1244
1245            cfg.buffer_size = BUFFER_SIZE;
1246            cfg.initial_available_buffers = 0;
1247            let err = File::new(cfg).unwrap_err();
1248            assert_eq!(err.reason, err::INV.reason);
1249        }
1250
1251        #[test]
1252        fn err_new_cfg_overflow() {
1253            let (_dir, mut cfg) = tmp_path();
1254            cfg.buffer_size = usize::MAX;
1255            cfg.initial_available_buffers = 2;
1256            let err = File::new(cfg).unwrap_err();
1257            assert_eq!(err.reason, err::INV.reason);
1258        }
1259
1260        #[test]
1261        fn err_new_missing_parent_dir() {
1262            let (_dir, mut cfg) = tmp_path();
1263            cfg.path = cfg.path.join("missing/sub/dir/file.db");
1264
1265            let err = File::new(cfg).unwrap_err();
1266            assert_eq!(err.reason, err::INV.reason);
1267        }
1268
1269        #[test]
1270        fn err_new_cleans_up_on_grow_failure() {
1271            let (_dir, mut cfg) = tmp_path();
1272            // On 64-bit systems, setting buffer_size such that buffer_size * initial_buffers exceeds off_t::MAX /
1273            // i64::MAX triggers grow failure after successful creation
1274            cfg.buffer_size = 1 << 0x3F;
1275            cfg.initial_available_buffers = 1;
1276
1277            let err = File::new(cfg.clone()).unwrap_err();
1278            assert_eq!(err.reason, err::GRW.reason);
1279
1280            // File must be unlinked and not left as an orphan on disk
1281            assert!(!cfg.path.exists());
1282
1283            // A subsequent File::new with valid config must succeed and not fail with EXS
1284            cfg.buffer_size = BUFFER_SIZE;
1285            cfg.initial_available_buffers = INIT_BUFFERS;
1286            let file = File::new(cfg);
1287            assert!(file.is_ok());
1288        }
1289    }
1290
1291    mod file_open {
1292        use super::*;
1293
1294        #[test]
1295        fn ok_open_existing_valid_file() {
1296            let (_dir, cfg) = tmp_path();
1297
1298            {
1299                let file = File::new(cfg.clone()).unwrap();
1300                assert_eq!(file.length(), BUFFER_SIZE * INIT_BUFFERS);
1301            }
1302
1303            let file = File::open(cfg).unwrap();
1304            assert_eq!(file.length(), BUFFER_SIZE * INIT_BUFFERS);
1305        }
1306
1307        #[test]
1308        fn err_open_missing_file() {
1309            let (_dir, cfg) = tmp_path();
1310
1311            let err = File::open(cfg).unwrap_err();
1312            assert_eq!(err.reason, err::INV.reason);
1313        }
1314
1315        #[test]
1316        fn err_open_invalid_cfg() {
1317            let (_dir, mut cfg) = tmp_path();
1318
1319            cfg.buffer_size = 0;
1320            let err = File::open(cfg.clone()).unwrap_err();
1321            assert_eq!(err.reason, err::INV.reason);
1322
1323            cfg.buffer_size = BUFFER_SIZE;
1324            cfg.initial_available_buffers = 0;
1325            let err = File::open(cfg).unwrap_err();
1326            assert_eq!(err.reason, err::INV.reason);
1327        }
1328
1329        #[test]
1330        fn err_open_cfg_overflow() {
1331            let (_dir, mut cfg) = tmp_path();
1332            cfg.buffer_size = usize::MAX;
1333            cfg.initial_available_buffers = 2;
1334            let err = File::open(cfg).unwrap_err();
1335            assert_eq!(err.reason, err::INV.reason);
1336        }
1337
1338        #[test]
1339        fn err_open_when_file_smaller_than_init_len() {
1340            let (_dir, cfg) = tmp_path();
1341
1342            std::fs::write(&cfg.path, []).unwrap();
1343
1344            let err = File::open(cfg).unwrap_err();
1345            assert_eq!(err.reason, err::CPT.reason);
1346        }
1347
1348        #[test]
1349        fn err_open_when_file_not_buffer_multiple() {
1350            let (_dir, cfg) = tmp_path();
1351
1352            let non_aligned_len = (BUFFER_SIZE * INIT_BUFFERS) + 3;
1353            std::fs::write(&cfg.path, vec![0u8; non_aligned_len]).unwrap();
1354
1355            let err = File::open(cfg).unwrap_err();
1356            assert_eq!(err.reason, err::CPT.reason);
1357        }
1358    }
1359
1360    mod file_open_or_create {
1361        use super::*;
1362
1363        #[test]
1364        fn ok_creates_when_missing() {
1365            let (_dir, cfg) = tmp_path();
1366            assert!(!cfg.path.exists());
1367
1368            let file = File::open_or_create(cfg.clone()).unwrap();
1369            assert!(cfg.path.exists());
1370            assert_eq!(file.length(), BUFFER_SIZE * INIT_BUFFERS);
1371            assert!(file.exists().unwrap());
1372        }
1373
1374        #[test]
1375        fn ok_opens_when_existing() {
1376            let (_dir, cfg) = tmp_path();
1377            let file = File::new(cfg.clone()).unwrap();
1378            let data = [0x5Au8; BUFFER_SIZE];
1379            file.write(&data, 0).unwrap();
1380            file.sync().unwrap();
1381            drop(file);
1382
1383            let opened = File::open_or_create(cfg.clone()).unwrap();
1384            assert_eq!(opened.length(), BUFFER_SIZE * INIT_BUFFERS);
1385            let mut buf = [0u8; BUFFER_SIZE];
1386            opened.read(&mut buf, 0).unwrap();
1387            assert_eq!(buf, data);
1388        }
1389
1390        #[test]
1391        fn err_open_or_create_invalid_cfg() {
1392            let (_dir, mut cfg) = tmp_path();
1393            cfg.buffer_size = 0;
1394            let err = File::open_or_create(cfg.clone()).unwrap_err();
1395            assert_eq!(err.reason, err::INV.reason);
1396
1397            cfg.buffer_size = BUFFER_SIZE;
1398            cfg.initial_available_buffers = 0;
1399            let err = File::open_or_create(cfg).unwrap_err();
1400            assert_eq!(err.reason, err::INV.reason);
1401        }
1402
1403        #[test]
1404        fn err_open_or_create_cfg_overflow() {
1405            let (_dir, mut cfg) = tmp_path();
1406            cfg.buffer_size = usize::MAX;
1407            cfg.initial_available_buffers = 2;
1408            let err = File::open_or_create(cfg).unwrap_err();
1409            assert_eq!(err.reason, err::INV.reason);
1410        }
1411
1412        #[test]
1413        fn err_open_or_create_corrupt_existing() {
1414            let (_dir, cfg) = tmp_path();
1415            let non_aligned = (BUFFER_SIZE * INIT_BUFFERS) + 1;
1416            std::fs::write(&cfg.path, vec![0u8; non_aligned]).unwrap();
1417
1418            let err = File::open_or_create(cfg).unwrap_err();
1419            assert_eq!(err.reason, err::CPT.reason);
1420        }
1421
1422        #[test]
1423        fn err_open_or_create_when_locked() {
1424            let (_dir, cfg) = tmp_path();
1425            let file = File::open_or_create(cfg.clone()).unwrap();
1426
1427            let err = File::open_or_create(cfg).unwrap_err();
1428            assert_eq!(err.reason, err::LCK.reason);
1429            drop(file);
1430        }
1431    }
1432
1433    mod file_toctou_and_concurrency {
1434        use super::*;
1435
1436        #[test]
1437        fn err_concurrent_new_exact_same_path() {
1438            let (_dir, cfg) = tmp_path();
1439            let barrier = Arc::new(Barrier::new(2));
1440
1441            let t1 = {
1442                let cfg = cfg.clone();
1443                let b = barrier.clone();
1444                std::thread::spawn(move || {
1445                    b.wait();
1446                    File::new(cfg)
1447                })
1448            };
1449
1450            let t2 = {
1451                let b = barrier;
1452                std::thread::spawn(move || {
1453                    b.wait();
1454                    File::new(cfg)
1455                })
1456            };
1457
1458            let r1 = t1.join().unwrap();
1459            let r2 = t2.join().unwrap();
1460
1461            // NOTE:
1462            //
1463            // Due to atomic `O_CREAT | O_EXCL`, exactly one MUST succeed and the other MUST fail with EXS
1464            match (r1, r2) {
1465                (Ok(f), Err(e)) | (Err(e), Ok(f)) => {
1466                    assert_eq!(e.reason, err::EXS.reason);
1467                    assert_eq!(f.length(), BUFFER_SIZE * INIT_BUFFERS);
1468                }
1469                (Ok(_), Ok(_)) => {
1470                    panic!("TOCTOU violation: both concurrent File::new succeeded on same path!");
1471                }
1472                (Err(e1), Err(e2)) => {
1473                    panic!("Unexpected failure of both threads: {:?}, {:?}", e1, e2);
1474                }
1475            }
1476        }
1477
1478        #[test]
1479        fn err_concurrent_new_and_open() {
1480            let (_dir, cfg) = tmp_path();
1481            let barrier = Arc::new(Barrier::new(2));
1482
1483            let t_new = {
1484                let cfg = cfg.clone();
1485                let b = barrier.clone();
1486                std::thread::spawn(move || {
1487                    b.wait();
1488                    File::new(cfg)
1489                })
1490            };
1491
1492            let t_open = {
1493                let b = barrier;
1494                std::thread::spawn(move || {
1495                    b.wait();
1496                    File::open(cfg)
1497                })
1498            };
1499
1500            let r_new = t_new.join().unwrap();
1501            let r_open = t_open.join().unwrap();
1502
1503            let _file = r_new.expect("File::new should succeed");
1504
1505            // File::open either ran before creation (INV), while lock was held (LCK), or in the brief window between
1506            // creation and grow/sync (CPT)
1507            if let Err(e) = r_open {
1508                assert!(
1509                    e.reason == err::INV.reason
1510                        || e.reason == err::LCK.reason
1511                        || e.reason == err::CPT.reason,
1512                    "unexpected error: {:?}",
1513                    e
1514                );
1515            }
1516        }
1517
1518        #[test]
1519        fn err_concurrent_open_same_file() {
1520            let (_dir, cfg) = tmp_path();
1521
1522            {
1523                let file = File::new(cfg.clone()).unwrap();
1524                assert_eq!(file.length(), BUFFER_SIZE * INIT_BUFFERS);
1525            }
1526
1527            let barrier = Arc::new(Barrier::new(2));
1528
1529            let t1 = {
1530                let cfg = cfg.clone();
1531                let b = barrier.clone();
1532                std::thread::spawn(move || {
1533                    b.wait();
1534                    File::open(cfg)
1535                })
1536            };
1537
1538            let t2 = {
1539                let b = barrier;
1540                std::thread::spawn(move || {
1541                    b.wait();
1542                    File::open(cfg)
1543                })
1544            };
1545
1546            let r1 = t1.join().unwrap();
1547            let r2 = t2.join().unwrap();
1548
1549            // NOTE:
1550            //
1551            // Exactly one must acquire exclusive flock and succeed, the other fails with LCK
1552            match (r1, r2) {
1553                (Ok(_), Err(e)) | (Err(e), Ok(_)) => {
1554                    assert_eq!(e.reason, err::LCK.reason, "error must be LCK: {:?}", e);
1555                }
1556                (Ok(_), Ok(_)) => {
1557                    panic!(
1558                        "Locking violation: both concurrent File::open acquired exclusive lock!"
1559                    );
1560                }
1561                (Err(e1), Err(e2)) => {
1562                    panic!("Unexpected failure of both threads: {:?}, {:?}", e1, e2);
1563                }
1564            }
1565        }
1566
1567        #[test]
1568        fn err_new_while_file_open() {
1569            let (_dir, cfg) = tmp_path();
1570            let file = File::new(cfg.clone()).unwrap();
1571
1572            let err = File::new(cfg).unwrap_err();
1573            assert_eq!(err.reason, err::EXS.reason);
1574
1575            drop(file);
1576        }
1577
1578        #[test]
1579        fn ok_reopen_after_drop() {
1580            let (_dir, cfg) = tmp_path();
1581            let file = File::new(cfg.clone()).unwrap();
1582            drop(file);
1583
1584            let reopened = File::open(cfg).unwrap();
1585            assert_eq!(reopened.length(), BUFFER_SIZE * INIT_BUFFERS);
1586        }
1587
1588        #[test]
1589        fn err_concurrent_open_during_and_after_delete() {
1590            let (_dir, cfg) = tmp_path();
1591            let file = File::new(cfg.clone()).unwrap();
1592            file.grow(2).unwrap();
1593
1594            let running = Arc::new(atomic::AtomicBool::new(true));
1595
1596            // Thread 2: constantly attempts File::open
1597            let opener = {
1598                let cfg = cfg.clone();
1599                let running = running.clone();
1600                std::thread::spawn(move || {
1601                    let mut attempts = 0;
1602                    while running.load(atomic::Ordering::Relaxed) || attempts < 0x32 {
1603                        attempts += 1;
1604                        let res = File::open(cfg.clone());
1605                        // While the file is open, open fails with LCK On Windows, opening a delete-pending file yields
1606                        // PRM Once the file is deleted, open fails with INV
1607                        //
1608                        // On POSIX, a descriptor opened right before unlink may momentarily succeed after deletion
1609                        // releases the writer's lock; if so, dropping it is completely safe
1610                        if let Err(err) = res {
1611                            assert!(
1612                                err.reason == err::LCK.reason
1613                                    || err.reason == err::INV.reason
1614                                    || err.reason == err::PRM.reason,
1615                                "Unexpected error reason: {:?}",
1616                                err
1617                            );
1618                        }
1619                        std::thread::yield_now();
1620                    }
1621                })
1622            };
1623
1624            // Thread 1: deletes the file
1625            std::thread::sleep(std::time::Duration::from_millis(5));
1626            file.delete().unwrap();
1627            running.store(false, atomic::Ordering::Relaxed);
1628
1629            opener.join().unwrap();
1630
1631            // After delete, opening must consistently fail with INV
1632            let err = File::open(cfg).unwrap_err();
1633            assert_eq!(err.reason, err::INV.reason);
1634        }
1635
1636        #[test]
1637        fn ok_concurrent_new_fails_before_delete_and_succeeds_after() {
1638            let (_dir, cfg) = tmp_path();
1639            let file = File::new(cfg.clone()).unwrap();
1640            file.grow(2).unwrap();
1641
1642            let deleted = Arc::new(atomic::AtomicBool::new(false));
1643
1644            let creator = {
1645                let cfg = cfg.clone();
1646                let deleted = deleted.clone();
1647                std::thread::spawn(move || {
1648                    let mut created = None;
1649                    // Before delete occurs, File::new must fail with EXS (or PRM on Windows while delete is pending)
1650                    // Once delete completes, File::new must eventually succeed
1651                    while created.is_none() {
1652                        match File::new(cfg.clone()) {
1653                            Ok(f) => {
1654                                assert!(
1655                                    deleted.load(atomic::Ordering::Acquire),
1656                                    "File::new succeeded before file.delete completed!"
1657                                );
1658                                created = Some(f);
1659                            }
1660                            Err(e) => {
1661                                assert!(
1662                                    e.reason == err::EXS.reason || e.reason == err::PRM.reason,
1663                                    "Unexpected error reason during race: {:?}",
1664                                    e
1665                                );
1666                                std::thread::yield_now();
1667                            }
1668                        }
1669                    }
1670                    created.unwrap()
1671                })
1672            };
1673
1674            std::thread::sleep(std::time::Duration::from_millis(0x0A));
1675            deleted.store(true, atomic::Ordering::Release);
1676            file.delete().unwrap();
1677
1678            let new_file = creator.join().unwrap();
1679            assert!(new_file.exists().unwrap());
1680            new_file.delete().unwrap();
1681        }
1682
1683        #[test]
1684        fn ok_concurrent_open_or_create_same_file() {
1685            let (_dir, cfg) = tmp_path();
1686            let num_threads = 8;
1687            let barrier = Arc::new(Barrier::new(num_threads));
1688            let mut handles = vec![];
1689
1690            for _ in 0..num_threads {
1691                let c = cfg.clone();
1692                let b = barrier.clone();
1693                handles.push(std::thread::spawn(move || {
1694                    b.wait();
1695                    File::open_or_create(c)
1696                }));
1697            }
1698
1699            let mut opened_files = vec![];
1700            let mut lock_failures = 0;
1701
1702            for h in handles {
1703                match h.join().unwrap() {
1704                    Ok(f) => {
1705                        assert_eq!(f.length(), BUFFER_SIZE * INIT_BUFFERS);
1706                        opened_files.push(f);
1707                    }
1708                    Err(e) => {
1709                        assert_eq!(e.reason, err::LCK.reason, "error must be LCK: {:?}", e);
1710                        lock_failures += 1;
1711                    }
1712                }
1713            }
1714
1715            // Exactly one must acquire the exclusive lock while other concurrent callers fail with LCK
1716            assert_eq!(opened_files.len(), 1);
1717            assert_eq!(lock_failures, num_threads - 1);
1718            assert!(cfg.path.exists());
1719        }
1720
1721        #[test]
1722        fn ok_file_send_and_sync() {
1723            fn assert_send_sync<T: Send + Sync>() {}
1724            assert_send_sync::<File>();
1725        }
1726    }
1727
1728    mod file_write_read {
1729        use super::*;
1730
1731        #[test]
1732        fn ok_single_buffer_write_read() {
1733            let (_dir, cfg) = tmp_path();
1734            let file = File::new(cfg).unwrap();
1735
1736            let data = [0xABu8; BUFFER_SIZE];
1737            file.write(&data, 2).unwrap();
1738            file.sync().unwrap();
1739
1740            let mut buf = [0u8; BUFFER_SIZE];
1741            file.read(&mut buf, 2).unwrap();
1742            assert_eq!(buf, data);
1743        }
1744
1745        #[test]
1746        fn ok_multi_buffer_write_read() {
1747            let (_dir, cfg) = tmp_path();
1748            let file = File::new(cfg).unwrap();
1749
1750            let multi_buf_data = [0x77u8; BUFFER_SIZE * 2];
1751            file.write(&multi_buf_data, 1).unwrap();
1752            file.sync().unwrap();
1753
1754            let mut read_buf = [0u8; BUFFER_SIZE * 2];
1755            file.read(&mut read_buf, 1).unwrap();
1756            assert_eq!(read_buf, multi_buf_data);
1757
1758            let mut chunk1 = [0u8; BUFFER_SIZE];
1759            let mut chunk2 = [0u8; BUFFER_SIZE];
1760            file.read(&mut chunk1, 1).unwrap();
1761            file.read(&mut chunk2, 2).unwrap();
1762            assert_eq!(chunk1, [0x77u8; BUFFER_SIZE]);
1763            assert_eq!(chunk2, [0x77u8; BUFFER_SIZE]);
1764        }
1765
1766        #[test]
1767        fn ok_empty_buffer_noop() {
1768            let (_dir, cfg) = tmp_path();
1769            let file = File::new(cfg).unwrap();
1770
1771            assert!(file.write(&[], 0).is_ok());
1772            assert!(file.read(&mut [], 0).is_ok());
1773        }
1774
1775        #[test]
1776        fn err_write_unaligned_buffer_size() {
1777            let (_dir, cfg) = tmp_path();
1778            let file = File::new(cfg).unwrap();
1779
1780            let data = [0u8; BUFFER_SIZE - 1];
1781            let err = file.write(&data, 0).unwrap_err();
1782            assert_eq!(err.reason, err::INV.reason);
1783
1784            let data_multi_unaligned = [0u8; (BUFFER_SIZE * 2) + 1];
1785            let err = file.write(&data_multi_unaligned, 0).unwrap_err();
1786            assert_eq!(err.reason, err::INV.reason);
1787        }
1788
1789        #[test]
1790        fn err_read_unaligned_buffer_size() {
1791            let (_dir, cfg) = tmp_path();
1792            let file = File::new(cfg).unwrap();
1793
1794            let mut buf = [0u8; BUFFER_SIZE - 1];
1795            let err = file.read(&mut buf, 0).unwrap_err();
1796            assert_eq!(err.reason, err::INV.reason);
1797
1798            let mut buf_multi_unaligned = [0u8; (BUFFER_SIZE * 2) + 3];
1799            let err = file.read(&mut buf_multi_unaligned, 0).unwrap_err();
1800            assert_eq!(err.reason, err::INV.reason);
1801        }
1802
1803        #[test]
1804        fn err_read_past_eof() {
1805            let (_dir, cfg) = tmp_path();
1806            let file = File::new(cfg).unwrap();
1807
1808            let mut buf = [0u8; BUFFER_SIZE];
1809            let err = file.read(&mut buf, INIT_BUFFERS).unwrap_err();
1810            assert_eq!(err.reason, err::HCF.reason);
1811
1812            let mut multi_buf = [0u8; BUFFER_SIZE * 2];
1813            let err = file.read(&mut multi_buf, INIT_BUFFERS - 1).unwrap_err();
1814            assert_eq!(err.reason, err::HCF.reason);
1815        }
1816
1817        #[test]
1818        fn err_write_past_eof() {
1819            let (_dir, cfg) = tmp_path();
1820            let file = File::new(cfg).unwrap();
1821
1822            let data = [1u8; BUFFER_SIZE];
1823            let err = file.write(&data, INIT_BUFFERS).unwrap_err();
1824            assert_eq!(err.reason, err::HCF.reason);
1825
1826            let multi_data = [1u8; BUFFER_SIZE * 2];
1827            let err = file.write(&multi_data, INIT_BUFFERS - 1).unwrap_err();
1828            assert_eq!(err.reason, err::HCF.reason);
1829        }
1830
1831        #[test]
1832        fn ok_concurrent_non_overlapping_writes() {
1833            let (_dir, mut cfg) = tmp_path();
1834            cfg.initial_available_buffers = 0x10;
1835            let file = Arc::new(File::new(cfg).unwrap());
1836
1837            let mut handles = vec![];
1838            for i in 0..4 {
1839                let f = file.clone();
1840                handles.push(std::thread::spawn(move || {
1841                    let data = [i as u8 + 0x0A; BUFFER_SIZE * 2];
1842                    f.write(&data, i * 2).unwrap();
1843                }));
1844            }
1845
1846            for h in handles {
1847                h.join().unwrap();
1848            }
1849
1850            file.sync().unwrap();
1851
1852            for i in 0..4 {
1853                let mut buf = [0u8; BUFFER_SIZE * 2];
1854                file.read(&mut buf, i * 2).unwrap();
1855                assert_eq!(buf, [i as u8 + 0x0A; BUFFER_SIZE * 2]);
1856            }
1857        }
1858
1859        #[test]
1860        fn ok_concurrent_readers() {
1861            let (_dir, cfg) = tmp_path();
1862            let file = Arc::new(File::new(cfg).unwrap());
1863
1864            let data = [0x42u8; BUFFER_SIZE * 2];
1865            file.write(&data, 0).unwrap();
1866            file.sync().unwrap();
1867
1868            let mut handles = vec![];
1869            for _ in 0..8 {
1870                let f = file.clone();
1871                let expected = data;
1872                handles.push(std::thread::spawn(move || {
1873                    let mut buf = [0u8; BUFFER_SIZE * 2];
1874                    f.read(&mut buf, 0).unwrap();
1875                    assert_eq!(buf, expected);
1876                }));
1877            }
1878
1879            for h in handles {
1880                h.join().unwrap();
1881            }
1882        }
1883    }
1884
1885    mod file_grow {
1886        use super::*;
1887
1888        #[test]
1889        fn ok_grow_updates_length() {
1890            let (_dir, cfg) = tmp_path();
1891            let file = File::new(cfg).unwrap();
1892            assert_eq!(file.length(), BUFFER_SIZE * INIT_BUFFERS);
1893            assert_eq!(file.total_buffers().unwrap(), INIT_BUFFERS);
1894
1895            file.grow(0x20).unwrap();
1896            assert_eq!(file.length(), BUFFER_SIZE * (INIT_BUFFERS + 0x20));
1897            assert_eq!(file.total_buffers().unwrap(), INIT_BUFFERS + 0x20);
1898        }
1899
1900        #[test]
1901        fn ok_grow_zero_count() {
1902            let (_dir, cfg) = tmp_path();
1903            let file = File::new(cfg).unwrap();
1904            file.grow(0).unwrap();
1905            assert_eq!(file.length(), BUFFER_SIZE * INIT_BUFFERS);
1906        }
1907
1908        #[test]
1909        fn ok_grow_and_multi_buffer_write_read() {
1910            let (_dir, cfg) = tmp_path();
1911            let file = File::new(cfg).unwrap();
1912
1913            file.grow(4).unwrap();
1914            let new_index = INIT_BUFFERS;
1915
1916            let data = [0x55u8; BUFFER_SIZE * 3];
1917            file.write(&data, new_index).unwrap();
1918            file.sync().unwrap();
1919
1920            let mut buf = [0u8; BUFFER_SIZE * 3];
1921            file.read(&mut buf, new_index).unwrap();
1922            assert_eq!(buf, data);
1923        }
1924
1925        #[test]
1926        fn ok_concurrent_grow_and_write() {
1927            let (_dir, cfg) = tmp_path();
1928            let file = Arc::new(File::new(cfg).unwrap());
1929
1930            let writer = {
1931                let f = file.clone();
1932                std::thread::spawn(move || {
1933                    for i in 0..INIT_BUFFERS {
1934                        let data = [i as u8; BUFFER_SIZE];
1935                        f.write(&data, i).unwrap();
1936                    }
1937                })
1938            };
1939
1940            let chunks_to_grow = 0x20;
1941            let grower = {
1942                let f = file.clone();
1943                std::thread::spawn(move || {
1944                    f.grow(chunks_to_grow).unwrap();
1945                })
1946            };
1947
1948            writer.join().unwrap();
1949            grower.join().unwrap();
1950
1951            file.sync().unwrap();
1952            assert_eq!(file.length(), BUFFER_SIZE * (INIT_BUFFERS + chunks_to_grow));
1953
1954            for i in 0..INIT_BUFFERS {
1955                let mut buf = [0u8; BUFFER_SIZE];
1956                file.read(&mut buf, i).unwrap();
1957                assert_eq!(buf, [i as u8; BUFFER_SIZE]);
1958            }
1959        }
1960
1961        #[test]
1962        #[cfg(target_os = "linux")]
1963        fn ok_sync_range() {
1964            let (_dir, cfg) = tmp_path();
1965            let file = File::new(cfg).unwrap();
1966
1967            let data = [0x33u8; BUFFER_SIZE * 2];
1968            file.write(&data, 0).unwrap();
1969            file.sync_range(0, 2).unwrap();
1970            file.sync().unwrap();
1971
1972            let mut buf = [0u8; BUFFER_SIZE * 2];
1973            file.read(&mut buf, 0).unwrap();
1974            assert_eq!(buf, data);
1975        }
1976
1977        #[test]
1978        #[cfg(target_os = "linux")]
1979        fn ok_sync_range_zero_count() {
1980            let (_dir, cfg) = tmp_path();
1981            let file = File::new(cfg).unwrap();
1982            assert!(file.sync_range(0, 0).is_ok());
1983            assert!(file.sync_range(5, 0).is_ok());
1984        }
1985
1986        #[test]
1987        #[cfg(target_os = "linux")]
1988        fn err_sync_range_index_overflow() {
1989            let (_dir, cfg) = tmp_path();
1990            let file = File::new(cfg).unwrap();
1991            let err1 = file.sync_range(usize::MAX, 1).unwrap_err();
1992            assert_eq!(err1.reason, err::INV.reason);
1993
1994            let err2 = file.sync_range(0, usize::MAX).unwrap_err();
1995            assert_eq!(err2.reason, err::INV.reason);
1996        }
1997
1998        #[test]
1999        fn ok_concurrent_multiple_growers() {
2000            let (_dir, cfg) = tmp_path();
2001            let file = Arc::new(File::new(cfg).unwrap());
2002            let num_threads = 4;
2003            let chunks_per_thread = 4;
2004            let barrier = Arc::new(Barrier::new(num_threads));
2005            let mut handles = vec![];
2006
2007            for _ in 0..num_threads {
2008                let f = file.clone();
2009                let b = barrier.clone();
2010                handles.push(std::thread::spawn(move || {
2011                    b.wait();
2012                    f.grow(chunks_per_thread).unwrap();
2013                }));
2014            }
2015
2016            for h in handles {
2017                h.join().unwrap();
2018            }
2019
2020            let expected_total_chunks = INIT_BUFFERS + (num_threads * chunks_per_thread);
2021            assert_eq!(file.length(), BUFFER_SIZE * expected_total_chunks);
2022            assert_eq!(file.total_buffers().unwrap(), expected_total_chunks);
2023            assert_eq!(file.current_epoch(), num_threads as TEpoch);
2024        }
2025
2026        #[test]
2027        fn ok_concurrent_readers_during_grow() {
2028            let (_dir, cfg) = tmp_path();
2029            let file = Arc::new(File::new(cfg).unwrap());
2030            let initial_data = [0x5Au8; BUFFER_SIZE];
2031            file.write(&initial_data, 0).unwrap();
2032            file.sync().unwrap();
2033
2034            let running = Arc::new(atomic::AtomicBool::new(true));
2035            let num_readers = 4;
2036            let mut readers = vec![];
2037
2038            for _ in 0..num_readers {
2039                let f = file.clone();
2040                let r = running.clone();
2041                let expected = initial_data;
2042                readers.push(std::thread::spawn(move || {
2043                    let mut buf = [0u8; BUFFER_SIZE];
2044                    while r.load(atomic::Ordering::Relaxed) {
2045                        f.read(&mut buf, 0).unwrap();
2046                        assert_eq!(buf, expected);
2047                        std::thread::yield_now();
2048                    }
2049                }));
2050            }
2051
2052            // Grower thread grows file in multiple increments
2053            for _ in 0..5 {
2054                file.grow(2).unwrap();
2055                std::thread::sleep(std::time::Duration::from_millis(5));
2056            }
2057
2058            running.store(false, atomic::Ordering::Relaxed);
2059            for r in readers {
2060                r.join().unwrap();
2061            }
2062
2063            assert_eq!(file.length(), BUFFER_SIZE * (INIT_BUFFERS + 0x0A));
2064            assert_eq!(file.total_buffers().unwrap(), INIT_BUFFERS + 0x0A);
2065        }
2066
2067        #[test]
2068        fn err_total_buffers_unaligned_file() {
2069            let (_dir, cfg) = tmp_path();
2070            let file = File::new(cfg).unwrap();
2071            assert_eq!(file.total_buffers().unwrap(), INIT_BUFFERS);
2072
2073            // Corrupt file length in-memory to simulate an unaligned layout
2074            file.inner.current_length.fetch_add(1, atomic::Ordering::SeqCst);
2075
2076            let err = file.total_buffers().unwrap_err();
2077            assert_eq!(err.reason, err::CPT.reason);
2078        }
2079    }
2080
2081    mod file_delete_exists {
2082        use super::*;
2083
2084        #[test]
2085        fn ok_exists_true_on_created_file() {
2086            let (_dir, cfg) = tmp_path();
2087            let file = File::new(cfg.clone()).unwrap();
2088
2089            assert!(file.exists().unwrap());
2090            assert!(cfg.path.exists());
2091        }
2092
2093        #[test]
2094        fn ok_delete_removes_file() {
2095            let (_dir, cfg) = tmp_path();
2096            let file = File::new(cfg.clone()).unwrap();
2097            let path = cfg.path.clone();
2098
2099            assert!(file.exists().unwrap());
2100            file.delete().unwrap();
2101
2102            assert!(!path.exists());
2103            assert!(File::new(cfg).is_ok());
2104        }
2105
2106        #[test]
2107        fn ok_open_and_delete() {
2108            let (_dir, cfg) = tmp_path();
2109            let file = File::new(cfg.clone()).unwrap();
2110            drop(file);
2111
2112            let opened = File::open(cfg.clone()).unwrap();
2113            assert!(opened.exists().unwrap());
2114            opened.delete().unwrap();
2115
2116            assert!(!cfg.path.exists());
2117        }
2118
2119        #[test]
2120        fn err_open_after_delete_grown_file() {
2121            let (_dir, cfg) = tmp_path();
2122            let file = File::new(cfg.clone()).unwrap();
2123
2124            file.grow(2).unwrap();
2125            assert_eq!(file.length(), (INIT_BUFFERS + 2) * BUFFER_SIZE);
2126
2127            file.delete().unwrap();
2128            assert!(!cfg.path.exists());
2129
2130            let err = File::open(cfg).unwrap_err();
2131            assert_eq!(err.reason, err::INV.reason);
2132        }
2133
2134        #[test]
2135        #[cfg(unix)]
2136        fn ok_exists_false_after_external_removal() {
2137            let (_dir, cfg) = tmp_path();
2138            let file = File::new(cfg.clone()).unwrap();
2139            assert!(file.exists().unwrap());
2140
2141            std::fs::remove_file(&cfg.path).unwrap();
2142            assert!(!file.exists().unwrap());
2143        }
2144
2145        #[test]
2146        #[cfg(windows)]
2147        fn ok_exists_false_after_external_removal() {
2148            let (_dir, cfg) = tmp_path();
2149            let file = File::new(cfg.clone()).unwrap();
2150            assert!(file.exists().unwrap());
2151            drop(file);
2152
2153            std::fs::remove_file(&cfg.path).unwrap();
2154            assert!(!PlatformFile::exists(&cfg.path).unwrap());
2155        }
2156
2157        #[test]
2158        #[cfg(unix)]
2159        fn err_delete_when_unlinked_externally() {
2160            let (_dir, cfg) = tmp_path();
2161            let file = File::new(cfg.clone()).unwrap();
2162
2163            std::fs::remove_file(&cfg.path).unwrap();
2164            let err = file.delete().unwrap_err();
2165            assert_eq!(err.reason, err::INV.reason);
2166        }
2167
2168        #[test]
2169        #[cfg(windows)]
2170        fn err_delete_when_unlinked_externally() {
2171            let (_dir, cfg) = tmp_path();
2172            let file = File::new(cfg.clone()).unwrap();
2173
2174            // NOTE:
2175            //
2176            // With `FILE_SHARE_DELETE` enabled, external deletion is permitted while open
2177            //
2178            // After external deletion, calling `file.delete()` fails with `err::INV` or `err::PRM` (delete-pending)
2179            std::fs::remove_file(&cfg.path).unwrap();
2180            let err = file.delete().unwrap_err();
2181            assert!(
2182                err.reason == err::INV.reason || err.reason == err::PRM.reason,
2183                "Unexpected error reason: {:?}",
2184                err
2185            );
2186        }
2187
2188        #[test]
2189        #[cfg(unix)]
2190        fn err_delete_permission_denied() {
2191            use std::os::unix::fs::PermissionsExt;
2192
2193            let dir = tempfile::tempdir().unwrap();
2194            let sub_dir = dir.path().join("readonly_dir");
2195            std::fs::create_dir(&sub_dir).unwrap();
2196
2197            let path = sub_dir.join("victim.db");
2198            let cfg = FileCfg {
2199                module_id: MID,
2200                path,
2201                buffer_size: BUFFER_SIZE,
2202                initial_available_buffers: INIT_BUFFERS,
2203                sync_interval: None,
2204            };
2205
2206            let file = File::new(cfg).unwrap();
2207            std::fs::set_permissions(&sub_dir, std::fs::Permissions::from_mode(0o555)).unwrap();
2208
2209            let err = file.delete().unwrap_err();
2210            std::fs::set_permissions(&sub_dir, std::fs::Permissions::from_mode(0o755)).unwrap();
2211
2212            assert_eq!(err.reason, err::PRM.reason);
2213        }
2214
2215        #[test]
2216        #[cfg(windows)]
2217        fn err_delete_permission_denied() {
2218            let (_dir, cfg) = tmp_path();
2219            let file = File::new(cfg.clone()).unwrap();
2220
2221            let orig_perms = std::fs::metadata(&cfg.path).unwrap().permissions();
2222            let mut read_only = orig_perms.clone();
2223            read_only.set_readonly(true);
2224            std::fs::set_permissions(&cfg.path, read_only).unwrap();
2225
2226            let err = file.delete().unwrap_err();
2227
2228            let _ = std::fs::set_permissions(&cfg.path, orig_perms);
2229
2230            assert_eq!(err.reason, err::PRM.reason);
2231        }
2232
2233        #[test]
2234        fn ok_delete_and_recreate_cycle() {
2235            let (_dir, cfg) = tmp_path();
2236
2237            for i in 0..3 {
2238                let file = File::new(cfg.clone()).unwrap();
2239                let data = [i as u8; BUFFER_SIZE];
2240                file.write(&data, 0).unwrap();
2241                assert!(file.exists().unwrap());
2242
2243                file.delete().unwrap();
2244                assert!(!cfg.path.exists());
2245            }
2246        }
2247    }
2248
2249    mod file_drop {
2250        use super::*;
2251
2252        #[test]
2253        fn ok_drop_persists_written_data() {
2254            let (_dir, cfg) = tmp_path();
2255            let data = [0x7Au8; BUFFER_SIZE * 2];
2256
2257            {
2258                let file = File::new(cfg.clone()).unwrap();
2259                file.write(&data, 0).unwrap();
2260                drop(file);
2261            }
2262
2263            {
2264                let opened = File::open(cfg).unwrap();
2265                let mut buf = [0u8; BUFFER_SIZE * 2];
2266                opened.read(&mut buf, 0).unwrap();
2267                assert_eq!(buf, data);
2268            }
2269        }
2270
2271        #[test]
2272        fn ok_drop_releases_exclusive_lock() {
2273            let (_dir, cfg) = tmp_path();
2274
2275            let file = File::new(cfg.clone()).unwrap();
2276            let err = File::open(cfg.clone()).unwrap_err();
2277            assert_eq!(err.reason, err::LCK.reason);
2278
2279            drop(file);
2280
2281            let opened = File::open(cfg);
2282            assert!(opened.is_ok());
2283        }
2284    }
2285
2286    mod file_sync_and_durability {
2287        use super::*;
2288        use std::time::{Duration, Instant};
2289
2290        #[test]
2291        fn ok_manual_sync_advances_epoch_and_tickets() {
2292            let (_dir, cfg) = tmp_path();
2293            let file = File::new(cfg).unwrap();
2294            let data = [0x55u8; BUFFER_SIZE];
2295
2296            assert_eq!(file.current_epoch(), 0);
2297            assert_eq!(file.durable_epoch(), 0);
2298
2299            let t1 = file.write(&data, 0).unwrap();
2300            assert_eq!(t1.epoch(), 1);
2301            assert!(!t1.is_durable());
2302            assert_eq!(file.current_epoch(), 1);
2303            assert_eq!(file.durable_epoch(), 0);
2304
2305            let t2 = file.write(&data, 1).unwrap();
2306            assert_eq!(t2.epoch(), 2);
2307            assert!(!t2.is_durable());
2308            assert_eq!(file.current_epoch(), 2);
2309            assert_eq!(file.durable_epoch(), 0);
2310
2311            file.sync().unwrap();
2312            assert!(t1.is_durable());
2313            assert!(t2.is_durable());
2314            assert_eq!(file.durable_epoch(), 2);
2315        }
2316
2317        #[test]
2318        fn ok_ticket_force_without_background_thread() {
2319            let (_dir, cfg) = tmp_path();
2320            let file = File::new(cfg).unwrap();
2321            let data = [0xAAu8; BUFFER_SIZE];
2322
2323            let ticket = file.write(&data, 0).unwrap();
2324            assert!(!ticket.is_durable());
2325
2326            let durable = ticket.force().unwrap();
2327            assert_eq!(durable, ticket.epoch());
2328            assert!(ticket.is_durable());
2329            assert_eq!(file.durable_epoch(), ticket.epoch());
2330        }
2331
2332        #[test]
2333        fn ok_background_sync_advances_epoch() {
2334            let (_dir, mut cfg) = tmp_path();
2335            cfg.sync_interval = Some(Duration::from_millis(0x19));
2336
2337            let file = File::new(cfg).unwrap();
2338            let data = [0x33u8; BUFFER_SIZE];
2339
2340            let ticket = file.write(&data, 0).unwrap();
2341            let epoch = ticket.wait().unwrap();
2342            assert_eq!(epoch, ticket.epoch());
2343            assert!(ticket.is_durable());
2344            assert!(file.durable_epoch() >= ticket.epoch());
2345        }
2346
2347        #[test]
2348        fn ok_ticket_force_with_background_thread() {
2349            let (_dir, mut cfg) = tmp_path();
2350            // Long interval to verify force wakes up immediately
2351            cfg.sync_interval = Some(Duration::from_secs(0x0A));
2352
2353            let file = File::new(cfg).unwrap();
2354            let data = [0x44u8; BUFFER_SIZE];
2355
2356            let ticket = file.write(&data, 0).unwrap();
2357            assert!(!ticket.is_durable());
2358
2359            let start = Instant::now();
2360            let epoch = ticket.force().unwrap();
2361            let elapsed = start.elapsed();
2362
2363            assert_eq!(epoch, ticket.epoch());
2364            assert!(ticket.is_durable());
2365            assert!(elapsed < Duration::from_secs(2));
2366        }
2367
2368        #[test]
2369        fn ok_drop_with_background_thread_persists_data() {
2370            let (_dir, mut cfg) = tmp_path();
2371            cfg.sync_interval = Some(Duration::from_secs(0x1E));
2372            let data = [0x88u8; BUFFER_SIZE];
2373
2374            {
2375                let file = File::new(cfg.clone()).unwrap();
2376                let _ticket = file.write(&data, 0).unwrap();
2377                // Drop without waiting for interval to expire
2378                drop(file);
2379            }
2380
2381            {
2382                let opened = File::open(cfg).unwrap();
2383                let mut buf = [0u8; BUFFER_SIZE];
2384                opened.read(&mut buf, 0).unwrap();
2385                assert_eq!(buf, data);
2386            }
2387        }
2388
2389        #[test]
2390        fn ok_delete_with_background_thread() {
2391            let (_dir, mut cfg) = tmp_path();
2392            cfg.sync_interval = Some(Duration::from_millis(0x14));
2393
2394            let file = File::new(cfg.clone()).unwrap();
2395            let data = [0x99u8; BUFFER_SIZE];
2396            let _ = file.write(&data, 0).unwrap();
2397
2398            assert!(file.exists().unwrap());
2399            file.delete().unwrap();
2400            assert!(!cfg.path.exists());
2401        }
2402
2403        #[test]
2404        fn ok_concurrent_writes_distinct_epochs() {
2405            let (_dir, cfg) = tmp_path();
2406            let file = Arc::new(File::new(cfg).unwrap());
2407            let num_threads = 4;
2408            let mut handles = Vec::new();
2409
2410            for i in 0..num_threads {
2411                let f = file.clone();
2412                handles.push(std::thread::spawn(move || {
2413                    let data = [i as u8; BUFFER_SIZE];
2414                    f.write(&data, i).unwrap()
2415                }));
2416            }
2417
2418            let mut epochs = Vec::new();
2419            for h in handles {
2420                let ticket = h.join().unwrap();
2421                epochs.push(ticket.epoch());
2422            }
2423
2424            epochs.sort();
2425            epochs.dedup();
2426            assert_eq!(epochs.len(), num_threads);
2427        }
2428
2429        #[test]
2430        fn err_ticket_wait_when_file_deleted() {
2431            let (_dir, cfg) = tmp_path();
2432            let file = File::new(cfg).unwrap();
2433            let data = [0x11u8; BUFFER_SIZE];
2434            let ticket = file.write(&data, 0).unwrap();
2435
2436            file.delete().unwrap();
2437
2438            // After delete, the weak SyncTrigger upgrade fails, returning an error on force
2439            let res = ticket.force();
2440            assert!(res.is_err());
2441        }
2442
2443        #[test]
2444        fn err_ticket_wait_surfaces_sync_failure() {
2445            let (_dir, cfg) = tmp_path();
2446            let file = File::new(cfg).unwrap();
2447            let data = [0x22u8; BUFFER_SIZE];
2448            let ticket = file.write(&data, 0).unwrap();
2449
2450            // Inject a synthetic durability error into the completion
2451            let sync_err = err::make_error(err::SYN);
2452            file.completion().set_err(sync_err.clone());
2453            file.completion().notify_all_listeners();
2454
2455            let err_wait = ticket.wait().unwrap_err();
2456            assert_eq!(err_wait.reason, err::SYN.reason);
2457
2458            let err_force = ticket.force().unwrap_err();
2459            assert_eq!(err_force.reason, err::SYN.reason);
2460        }
2461
2462        #[test]
2463        fn ok_concurrent_ticket_force_storm() {
2464            let (_dir, mut cfg) = tmp_path();
2465            cfg.initial_available_buffers = 0x20;
2466            cfg.sync_interval = Some(Duration::from_millis(0x0A));
2467            let file = Arc::new(File::new(cfg).unwrap());
2468            let num_threads = 0x10;
2469            let barrier = Arc::new(Barrier::new(num_threads));
2470            let mut handles = Vec::new();
2471
2472            for i in 0..num_threads {
2473                let f = file.clone();
2474                let b = barrier.clone();
2475                handles.push(std::thread::spawn(move || {
2476                    b.wait();
2477                    let data = [i as u8; BUFFER_SIZE];
2478                    let ticket = f.write(&data, i).unwrap();
2479                    let durable = ticket.force().unwrap();
2480                    assert_eq!(durable, ticket.epoch());
2481                    assert!(ticket.is_durable());
2482                }));
2483            }
2484
2485            for h in handles {
2486                h.join().unwrap();
2487            }
2488
2489            assert!(file.durable_epoch() >= num_threads as TEpoch);
2490        }
2491    }
2492
2493    mod module_binding {
2494        use super::*;
2495
2496        #[test]
2497        fn ok_multiple_files_distinct_module_ids() {
2498            let dir = tempfile::tempdir().unwrap();
2499            let cfg1 = FileCfg {
2500                module_id: 0x11,
2501                path: dir.path().join("file1.db"),
2502                buffer_size: BUFFER_SIZE,
2503                initial_available_buffers: INIT_BUFFERS,
2504                sync_interval: None,
2505            };
2506            let cfg2 = FileCfg {
2507                module_id: 0x22,
2508                path: dir.path().join("file2.db"),
2509                buffer_size: BUFFER_SIZE,
2510                initial_available_buffers: INIT_BUFFERS,
2511                sync_interval: None,
2512            };
2513
2514            let file1 = File::new(cfg1.clone()).unwrap();
2515            let file2 = File::new(cfg2.clone()).unwrap();
2516
2517            // Operations causing errors on file1 must carry module_id 0x11
2518            let mut invalid_buf = [0u8; BUFFER_SIZE - 1];
2519            let err1 = file1.read(&mut invalid_buf, 0).unwrap_err();
2520            assert_eq!(err1.module, 0x11);
2521            assert_eq!(err1.reason, err::INV.reason);
2522
2523            let err1_oob = file1.read(&mut [0u8; BUFFER_SIZE], 0x3E7).unwrap_err();
2524            assert_eq!(err1_oob.module, 0x11);
2525            assert_eq!(err1_oob.reason, err::HCF.reason);
2526
2527            // Operations causing errors on file2 must carry module_id 0x22
2528            let err2 = file2.read(&mut invalid_buf, 0).unwrap_err();
2529            assert_eq!(err2.module, 0x22);
2530            assert_eq!(err2.reason, err::INV.reason);
2531
2532            let err2_oob = file2.write(&[0u8; BUFFER_SIZE], 0x3E7).unwrap_err();
2533            assert_eq!(err2_oob.module, 0x22);
2534            assert_eq!(err2_oob.reason, err::HCF.reason);
2535
2536            // Existing file error on File::new carries respective module_id
2537            let err1_exs = File::new(cfg1).unwrap_err();
2538            assert_eq!(err1_exs.module, 0x11);
2539            assert_eq!(err1_exs.reason, err::EXS.reason);
2540
2541            let err2_exs = File::new(cfg2).unwrap_err();
2542            assert_eq!(err2_exs.module, 0x22);
2543            assert_eq!(err2_exs.reason, err::EXS.reason);
2544        }
2545    }
2546}