feoxdb/core/store/
persistence.rs

1use std::io::{self, Read, Seek};
2use std::sync::atomic::Ordering;
3use std::sync::Arc;
4
5use crate::constants::*;
6use crate::core::record::Record;
7use crate::error::{FeoxError, Result};
8use crate::storage::format::get_format;
9
10use super::FeoxStore;
11
12impl FeoxStore {
13    /// Force flush all pending writes to disk.
14    ///
15    /// In persistent mode, ensures all buffered writes are flushed to disk.
16    /// In memory-only mode, this is a no-op.
17    ///
18    /// # Example
19    ///
20    /// ```no_run
21    /// # use feoxdb::FeoxStore;
22    /// # fn main() -> feoxdb::Result<()> {
23    /// let store = FeoxStore::new(Some("/path/to/data.feox".to_string()))?;
24    /// store.insert(b"important", b"data")?;
25    /// store.flush_all()?;  // Ensure data is persisted
26    /// # Ok(())
27    /// # }
28    /// ```
29    pub fn flush_all(&self) -> Result<()> {
30        if !self.memory_only {
31            // First flush the write buffer to ensure all data is written
32            if let Some(ref wb) = self.write_buffer {
33                wb.force_flush()?;
34            }
35
36            if let Some(ref disk_io) = self.disk_io {
37                // Update metadata with current stats
38                let mut metadata = self._metadata.write();
39                metadata.total_records = self.stats.record_count.load(Ordering::Relaxed) as u64;
40                metadata.total_size = self.stats.disk_usage.load(Ordering::Relaxed);
41                metadata.fragmentation = self.free_space.read().get_fragmentation();
42                metadata.update();
43
44                // Write metadata
45                disk_io.write().write_metadata(metadata.as_bytes())?;
46                disk_io.write().flush()?;
47            }
48        }
49        Ok(())
50    }
51
52    pub(super) fn load_value_from_disk(&self, record: &Record) -> Result<Vec<u8>> {
53        let sector = record.sector.load(Ordering::Acquire);
54        if self.memory_only || sector == 0 {
55            return Err(FeoxError::InvalidRecord);
56        }
57
58        // Get the appropriate format handler
59        let metadata_version = self._metadata.read().version;
60        let format = get_format(metadata_version);
61
62        // Calculate how many sectors we need to read
63        let total_size = format.total_size(record.key.len(), record.value_len);
64        let sectors_needed = total_size.div_ceil(FEOX_BLOCK_SIZE);
65
66        // Read the sectors
67        let disk_io = self
68            .disk_io
69            .as_ref()
70            .ok_or_else(|| {
71                FeoxError::IoError(io::Error::new(
72                    io::ErrorKind::NotFound,
73                    "No disk IO available",
74                ))
75            })?
76            .read();
77
78        let data = disk_io.read_sectors_sync(sector, sectors_needed as u64)?;
79
80        // Use format to get the value offset
81        let offset = format.value_offset(record.key.len());
82        if offset + record.value_len > data.len() {
83            return Err(FeoxError::InvalidRecord);
84        }
85
86        Ok(data[offset..offset + record.value_len].to_vec())
87    }
88
89    pub(super) fn open_device(
90        &mut self,
91        device_path: &Option<String>,
92        file_size: Option<u64>,
93    ) -> Result<()> {
94        if let Some(path) = device_path {
95            // Open the device/file
96            use std::fs::OpenOptions;
97            #[cfg(target_os = "linux")]
98            use std::os::unix::fs::OpenOptionsExt;
99
100            #[cfg(unix)]
101            let (file, use_direct_io) = if std::path::Path::new("/.dockerenv").exists() {
102                let file = OpenOptions::new()
103                    .read(true)
104                    .write(true)
105                    .create(true)
106                    .truncate(false)
107                    .open(path)
108                    .map_err(FeoxError::IoError)?;
109                (file, false) // Don't use O_DIRECT in Docker
110            } else {
111                // Try with O_DIRECT on Linux, fall back without it on other Unix systems
112                #[cfg(target_os = "linux")]
113                {
114                    // Try to open with O_DIRECT first
115                    match OpenOptions::new()
116                        .read(true)
117                        .write(true)
118                        .create(true)
119                        .truncate(false)
120                        .custom_flags(libc::O_DIRECT)
121                        .open(path)
122                    {
123                        Ok(file) => (file, true), // Successfully opened with O_DIRECT
124                        Err(_) => {
125                            // Fallback to regular open
126                            let file = OpenOptions::new()
127                                .read(true)
128                                .write(true)
129                                .create(true)
130                                .truncate(false)
131                                .open(path)
132                                .map_err(FeoxError::IoError)?;
133                            (file, false)
134                        }
135                    }
136                }
137                #[cfg(not(target_os = "linux"))]
138                {
139                    let file = OpenOptions::new()
140                        .read(true)
141                        .write(true)
142                        .create(true)
143                        .truncate(false)
144                        .open(path)
145                        .map_err(FeoxError::IoError)?;
146                    (file, false) // O_DIRECT not supported on this platform
147                }
148            };
149
150            #[cfg(not(unix))]
151            let file = OpenOptions::new()
152                .read(true)
153                .write(true)
154                .create(true)
155                .truncate(false)
156                .open(path)
157                .map_err(FeoxError::IoError)?;
158
159            // Get file size
160            let metadata = file.metadata().map_err(FeoxError::IoError)?;
161            self.device_size = metadata.len();
162
163            // Track whether this is a newly created file
164            let was_newly_created = self.device_size == 0;
165
166            if was_newly_created {
167                // New empty file - set configured size or default and initialize free space
168                let target_size = file_size.unwrap_or(DEFAULT_DEVICE_SIZE);
169                file.set_len(target_size).map_err(FeoxError::IoError)?;
170                self.device_size = target_size;
171
172                // Initialize free space manager with all space free
173                self.free_space.write().initialize(self.device_size)?;
174
175                let mut metadata = self._metadata.write();
176                metadata.device_size = self.device_size;
177                metadata.update();
178            } else {
179                // Existing file - check if it's empty
180                // If empty, initialize free space; otherwise it will be rebuilt during scan
181                let is_empty_file = {
182                    let mut temp_file = file.try_clone().map_err(FeoxError::IoError)?;
183                    temp_file
184                        .metadata()
185                        .map(|m| {
186                            // Check if file is all zeros
187                            if m.len() > 0 {
188                                let mut buffer = vec![0u8; std::cmp::min(4096, m.len() as usize)];
189                                temp_file.seek(std::io::SeekFrom::Start(0)).ok();
190                                temp_file.read_exact(&mut buffer).ok();
191                                buffer.iter().all(|&b| b == 0)
192                            } else {
193                                false
194                            }
195                        })
196                        .unwrap_or(false)
197                };
198
199                if is_empty_file {
200                    // Empty pre-created file - initialize free space like a new file
201                    self.free_space.write().initialize(self.device_size)?;
202                } else {
203                    // Existing file with data - free space will be rebuilt during scan
204                    self.free_space.write().set_device_size(self.device_size);
205                }
206            }
207
208            #[cfg(unix)]
209            {
210                use std::os::unix::io::AsRawFd;
211                let file_arc = Arc::new(file);
212                let fd = file_arc.as_raw_fd();
213                self.device_fd = Some(fd);
214                // Store a clone of the file to keep it alive
215                self.device_file = Some(file_arc.as_ref().try_clone().map_err(FeoxError::IoError)?);
216                let disk_io = crate::storage::io::DiskIO::new(file_arc, use_direct_io)?;
217                self.disk_io = Some(Arc::new(parking_lot::RwLock::new(disk_io)));
218            }
219
220            #[cfg(not(unix))]
221            {
222                // Store a clone of the file to keep it alive
223                self.device_file = Some(file.try_clone().map_err(FeoxError::IoError)?);
224                let disk_io = crate::storage::io::DiskIO::new_from_file(file)?;
225                self.disk_io = Some(Arc::new(parking_lot::RwLock::new(disk_io)));
226            }
227
228            let disk_io = self.disk_io.as_ref().unwrap().read();
229
230            // Read metadata from existing files (not newly created ones)
231            if !was_newly_created {
232                if let Ok(metadata_bytes) = disk_io.read_metadata() {
233                    if let Some(loaded_metadata) =
234                        crate::storage::metadata::Metadata::from_bytes(&metadata_bytes)
235                    {
236                        // Initialize stats from metadata
237                        self.stats
238                            .disk_usage
239                            .store(loaded_metadata.total_size, Ordering::Relaxed);
240                        *self._metadata.write() = loaded_metadata;
241                    }
242                }
243            }
244        }
245        Ok(())
246    }
247}
248
249impl Drop for FeoxStore {
250    fn drop(&mut self) {
251        // Stop TTL sweeper if running
252        if let Some(mut sweeper) = self.ttl_sweeper.write().take() {
253            sweeper.stop();
254        }
255
256        // Signal shutdown to write buffer workers
257        if let Some(ref wb) = self.write_buffer {
258            wb.initiate_shutdown();
259        }
260
261        // Write metadata directly without using the write buffer
262        if !self.memory_only {
263            if let Some(ref disk_io) = self.disk_io {
264                // Update metadata with current stats
265                let mut metadata = self._metadata.write();
266                metadata.total_records = self.stats.record_count.load(Ordering::Relaxed) as u64;
267                metadata.total_size = self.stats.disk_usage.load(Ordering::Relaxed);
268                metadata.fragmentation = self.free_space.read().get_fragmentation();
269                metadata.update();
270
271                // Write metadata
272                let _ = disk_io.write().write_metadata(metadata.as_bytes());
273                let _ = disk_io.write().flush();
274            }
275        }
276
277        // Take ownership of write_buffer to properly shut it down
278        if let Some(wb_arc) = self.write_buffer.take() {
279            // Try to get mutable access if we're the only owner
280            if let Ok(wb) = Arc::try_unwrap(wb_arc) {
281                // We own it exclusively, can call complete_shutdown
282                let mut wb_mut = wb;
283                wb_mut.complete_shutdown();
284            }
285            // If we can't get exclusive access, workers are already shutting down via initiate_shutdown
286        }
287
288        // Now it's safe to shutdown disk I/O since workers have exited
289        if let Some(ref disk_io) = self.disk_io {
290            disk_io.write().shutdown();
291        }
292    }
293}
feoxdb/core/store/persistence.rs

feoxdb/core/store/
persistence.rs
