feoxdb/core/store/
recovery.rs

1use std::sync::atomic::Ordering;
2use std::sync::Arc;
3
4use crate::constants::*;
5use crate::core::record::Record;
6use crate::error::{FeoxError, Result};
7use crate::storage::format::get_format;
8use crate::storage::metadata::Metadata;
9
10use super::FeoxStore;
11
12impl FeoxStore {
13    pub(super) fn load_indexes(&mut self) -> Result<()> {
14        if self.memory_only {
15            return Ok(());
16        }
17
18        // Try to read metadata from sector 0
19        if let Some(ref disk_io) = self.disk_io {
20            let metadata_data = disk_io.read().read_metadata()?;
21
22            // Check if metadata is valid (has our signature)
23            if metadata_data.len() >= FEOX_SIGNATURE_SIZE {
24                let signature = &metadata_data[..FEOX_SIGNATURE_SIZE];
25
26                if signature == FEOX_SIGNATURE {
27                    // Parse and store metadata
28                    if let Some(metadata) = Metadata::from_bytes(&metadata_data) {
29                        *self._metadata.write() = metadata;
30                    }
31
32                    // Valid metadata found, scan the disk to rebuild indexes
33                    self.scan_and_rebuild_indexes()?;
34                }
35            }
36        }
37
38        Ok(())
39    }
40
41    pub(super) fn scan_and_rebuild_indexes(&mut self) -> Result<()> {
42        if self.memory_only || self.device_size == 0 {
43            return Ok(());
44        }
45
46        let disk_io = self.disk_io.as_ref().ok_or(FeoxError::NoDevice)?;
47
48        // Get the appropriate format handler
49        let metadata_version = self._metadata.read().version;
50        let format = get_format(metadata_version);
51
52        let total_sectors = self.device_size / FEOX_BLOCK_SIZE as u64;
53        let occupied_len = usize::try_from(total_sectors).map_err(|_| FeoxError::InvalidDevice)?;
54        let mut sector: u64 = 1;
55        let mut occupied_sectors = vec![false; occupied_len];
56
57        while sector < total_sectors {
58            let data = match disk_io.read().read_sectors_sync(sector, 1) {
59                Ok(d) => d,
60                Err(_) => {
61                    sector += 1;
62                    continue;
63                }
64            };
65
66            if data.len() < SECTOR_HEADER_SIZE {
67                sector += 1;
68                continue;
69            }
70
71            // Check for deletion marker first
72            if data.len() >= 8 && &data[..8] == b"\0DELETED" {
73                // This sector has been deleted, skip it
74                sector += 1;
75                continue;
76            }
77
78            let marker = u16::from_le_bytes([data[0], data[1]]);
79            let seq_num = u16::from_le_bytes([data[2], data[3]]);
80
81            if marker != SECTOR_MARKER || seq_num != 0 {
82                sector += 1;
83                continue;
84            }
85
86            if data.len() < SECTOR_HEADER_SIZE + 2 {
87                sector += 1;
88                continue;
89            }
90
91            // Parse the record using format trait
92            let (key, value_len, timestamp, ttl_expiry) = match format.parse_record(&data) {
93                Some(parsed) => parsed,
94                None => {
95                    sector += 1;
96                    continue;
97                }
98            };
99
100            if key.is_empty() || key.len() > MAX_KEY_SIZE {
101                sector += 1;
102                continue;
103            }
104
105            // Calculate total size using format trait
106            let total_size = format.total_size(key.len(), value_len);
107            let sectors_needed = total_size.div_ceil(FEOX_BLOCK_SIZE);
108
109            let mut record = Record::new(key.clone(), Vec::new(), timestamp);
110            record.sector.store(sector, Ordering::Release);
111            record.value_len = value_len;
112            record.ttl_expiry.store(ttl_expiry, Ordering::Release);
113            record.clear_value();
114
115            // Skip expired records during load if TTL is enabled
116            if self.enable_ttl && ttl_expiry > 0 && self.get_timestamp() > ttl_expiry {
117                sector += sectors_needed as u64;
118                continue;
119            }
120
121            let record_arc = Arc::new(record);
122            let key_len = key.len();
123
124            let existing = self.hash_table.read(&key, |_, record| Arc::clone(record));
125            if existing
126                .as_ref()
127                .is_some_and(|record| record.timestamp > timestamp)
128            {
129                sector += sectors_needed as u64;
130                continue;
131            }
132
133            if let Some(existing) = existing {
134                let existing_sectors = format
135                    .total_size(existing.key.len(), existing.value_len)
136                    .div_ceil(FEOX_BLOCK_SIZE);
137                set_occupied(
138                    &mut occupied_sectors,
139                    existing.sector.load(Ordering::Acquire),
140                    existing_sectors,
141                    false,
142                )?;
143                self.stats.memory_usage.fetch_sub(
144                    self.calculate_record_size(existing.key.len(), existing.value_len),
145                    Ordering::AcqRel,
146                );
147                self.stats.disk_usage.fetch_sub(
148                    (existing_sectors * FEOX_BLOCK_SIZE) as u64,
149                    Ordering::AcqRel,
150                );
151            } else {
152                self.stats.record_count.fetch_add(1, Ordering::AcqRel);
153            }
154
155            self.hash_table.upsert(key.clone(), Arc::clone(&record_arc));
156            self.tree.insert(key, Arc::clone(&record_arc));
157
158            let record_size = self.calculate_record_size(key_len, value_len);
159            self.stats
160                .memory_usage
161                .fetch_add(record_size, Ordering::AcqRel);
162
163            // Track disk usage
164            self.stats
165                .disk_usage
166                .fetch_add((sectors_needed * FEOX_BLOCK_SIZE) as u64, Ordering::AcqRel);
167
168            set_occupied(&mut occupied_sectors, sector, sectors_needed, true)?;
169            sector += sectors_needed as u64;
170        }
171
172        let mut last_end = FEOX_DATA_START_BLOCK;
173
174        for occupied_start in FEOX_DATA_START_BLOCK..total_sectors {
175            if !occupied_sectors[occupied_start as usize] {
176                continue;
177            }
178            if occupied_start > last_end {
179                self.free_space
180                    .write()
181                    .release_sectors(last_end, occupied_start - last_end)?;
182            }
183            last_end = occupied_start + 1;
184        }
185
186        if last_end < total_sectors {
187            self.free_space
188                .write()
189                .release_sectors(last_end, total_sectors - last_end)?;
190        }
191
192        Ok(())
193    }
194}
195
196fn set_occupied(occupied: &mut [bool], sector: u64, sectors: usize, value: bool) -> Result<()> {
197    let start = usize::try_from(sector).map_err(|_| FeoxError::CorruptedRecord)?;
198    let end = start
199        .checked_add(sectors)
200        .filter(|end| *end <= occupied.len())
201        .ok_or(FeoxError::CorruptedRecord)?;
202    occupied[start..end].fill(value);
203    Ok(())
204}
feoxdb/core/store/recovery.rs

feoxdb/core/store/
recovery.rs
