velesdb_core/storage/
mmap_capacity.rs

1//! `MmapStorage` capacity management and compaction.
2//!
3//! Extracted from `mmap.rs` to reduce NLOC below the 500 threshold.
4
5use super::compaction::CompactionContext;
6use super::mmap::MmapStorage;
7
8use memmap2::MmapMut;
9use std::fs::OpenOptions;
10use std::io;
11use std::time::Instant;
12
13impl MmapStorage {
14    /// Ensures the memory map is large enough to hold data at `offset`.
15    ///
16    /// # P2 Optimization
17    ///
18    /// Uses aggressive pre-allocation to minimize blocking:
19    /// - Exponential growth (2x) for amortized O(1)
20    /// - 64MB minimum growth to reduce resize frequency
21    pub(crate) fn ensure_capacity(&mut self, required_len: usize) -> io::Result<()> {
22        let start = Instant::now();
23        let mut did_resize = false;
24        let mut bytes_resized = 0u64;
25
26        let mut mmap = self.mmap.write();
27        if mmap.len() < required_len {
28            mmap.flush()?;
29
30            let current_len = mmap.len() as u64;
31            let required_u64 = required_len as u64;
32
33            let doubled = current_len.saturating_mul(Self::GROWTH_FACTOR);
34            let with_headroom = required_u64.saturating_add(Self::MIN_GROWTH);
35            let min_growth = current_len.saturating_add(Self::MIN_GROWTH);
36
37            let new_len = doubled.max(with_headroom).max(min_growth).max(required_u64);
38
39            self.data_file.set_len(new_len)?;
40
41            // SAFETY: data_file has been resized with set_len(new_len) above,
42            // ensuring the new mapping range is fully allocated.
43            // - Condition 1: File was resized to new_len before remapping.
44            // - Condition 2: Old mmap is dropped when we assign the new one.
45            // - Condition 3: File remains open with read+write permissions.
46            // SAFETY: Memory mapping requires unsafe; resizing ensures mapping doesn't exceed file bounds.
47            *mmap = unsafe { MmapMut::map_mut(&self.data_file)? };
48            self.remap_epoch
49                .fetch_add(1, std::sync::atomic::Ordering::Release);
50
51            did_resize = true;
52            bytes_resized = new_len.saturating_sub(current_len);
53        }
54
55        self.metrics
56            .record_ensure_capacity(start.elapsed(), did_resize, bytes_resized);
57
58        Ok(())
59    }
60
61    /// Pre-allocates storage capacity for a known number of vectors.
62    ///
63    /// # Errors
64    ///
65    /// Returns an error if file operations fail.
66    pub fn reserve_capacity(&mut self, vector_count: usize) -> io::Result<()> {
67        let vector_size = self.dimension * std::mem::size_of::<f32>();
68        let required_len = vector_count.saturating_mul(vector_size);
69        let with_headroom = required_len.saturating_add(required_len / 10);
70        self.ensure_capacity(with_headroom)
71    }
72
73    /// Compacts the storage by rewriting only active vectors.
74    ///
75    /// # Returns
76    ///
77    /// The number of bytes reclaimed.
78    ///
79    /// # Errors
80    ///
81    /// Returns an error if file operations fail.
82    pub fn compact(&mut self) -> io::Result<usize> {
83        let ctx = CompactionContext {
84            path: &self.path,
85            dimension: self.dimension,
86            index: &self.index,
87            mmap: &self.mmap,
88            next_offset: &self.next_offset,
89            wal: &self.wal,
90            initial_size: Self::INITIAL_SIZE,
91        };
92
93        let bytes_reclaimed = ctx.compact()?;
94
95        if bytes_reclaimed > 0 {
96            let data_path = self.path.join("vectors.dat");
97            self.data_file = OpenOptions::new().read(true).write(true).open(&data_path)?;
98            // No flush_full() here: `commit_compaction` already synced the
99            // compacted data file before the swap, promoted an identical
100            // fsynced vectors.idx and truncated the WAL. Rewriting
101            // vectors.idx at this point is redundant (`&mut self` excludes
102            // concurrent mutation) and used to reopen the exact crash window
103            // the staged commit closed — a torn index next to an
104            // already-empty WAL is unrecoverable (audit 2026-06, finding 3).
105        }
106
107        Ok(bytes_reclaimed)
108    }
109
110    /// Returns the fragmentation ratio (0.0 = no fragmentation, 1.0 = 100% fragmented).
111    #[must_use]
112    pub fn fragmentation_ratio(&self) -> f64 {
113        let ctx = CompactionContext {
114            path: &self.path,
115            dimension: self.dimension,
116            index: &self.index,
117            mmap: &self.mmap,
118            next_offset: &self.next_offset,
119            wal: &self.wal,
120            initial_size: Self::INITIAL_SIZE,
121        };
122
123        ctx.fragmentation_ratio()
124    }
125}
velesdb_core/storage/mmap_capacity.rs

velesdb_core/storage/
mmap_capacity.rs
