feoxdb/storage/
format.rs

1use crate::constants::*;
2use crate::core::record::Record;
3use crate::storage::seq_token::seq_token;
4use std::sync::atomic::Ordering;
5
6/// Trait for handling different record format versions
7pub trait RecordFormat: Send + Sync {
8    /// Calculate the size of a record on disk (excluding value)
9    fn record_header_size(&self, key_len: usize) -> usize;
10
11    /// Calculate total size including value
12    fn total_size(&self, key_len: usize, value_len: usize) -> usize;
13
14    /// Serialize a record to bytes for disk storage
15    fn serialize_record(&self, record: &Record, include_value: bool) -> Vec<u8>;
16
17    /// Append a serialized record without its sector header.
18    fn serialize_record_into(&self, record: &Record, include_value: bool, data: &mut Vec<u8>) {
19        data.extend_from_slice(&self.serialize_record(record, include_value));
20    }
21
22    /// Parse a record from disk bytes (returns key, value_len, timestamp, ttl_expiry)
23    fn parse_record(&self, data: &[u8]) -> Option<(Vec<u8>, usize, u64, u64)>;
24
25    /// Get the offset where value data starts in the serialized format
26    fn value_offset(&self, key_len: usize) -> usize;
27}
28
29fn serialization_buffer(
30    format: &dyn RecordFormat,
31    record: &Record,
32    include_value: bool,
33) -> Vec<u8> {
34    let value_len = if include_value { record.value_len } else { 0 };
35    Vec::with_capacity(format.total_size(record.key.len(), value_len) - SECTOR_HEADER_SIZE)
36}
37
38/// Version 1 format (no TTL support)
39pub struct FormatV1;
40static FORMAT_V1: FormatV1 = FormatV1;
41
42impl RecordFormat for FormatV1 {
43    fn record_header_size(&self, key_len: usize) -> usize {
44        SECTOR_HEADER_SIZE + 2 + key_len + 8 + 8 // header + key_len(2) + key + value_len(8) + timestamp(8)
45    }
46
47    fn total_size(&self, key_len: usize, value_len: usize) -> usize {
48        self.record_header_size(key_len) + value_len
49    }
50
51    fn serialize_record(&self, record: &Record, include_value: bool) -> Vec<u8> {
52        let mut data = serialization_buffer(self, record, include_value);
53        self.serialize_record_into(record, include_value, &mut data);
54        data
55    }
56
57    fn serialize_record_into(&self, record: &Record, include_value: bool, data: &mut Vec<u8>) {
58        data.extend_from_slice(&(record.key.len() as u16).to_le_bytes());
59        data.extend_from_slice(&record.key);
60        data.extend_from_slice(&(record.value_len as u64).to_le_bytes());
61        data.extend_from_slice(&record.timestamp.to_le_bytes());
62
63        if include_value {
64            if let Some(value) = record.value.read().as_ref() {
65                data.extend_from_slice(value);
66            }
67        }
68    }
69
70    fn parse_record(&self, data: &[u8]) -> Option<(Vec<u8>, usize, u64, u64)> {
71        if data.len() < SECTOR_HEADER_SIZE + 2 {
72            return None;
73        }
74
75        let mut offset = SECTOR_HEADER_SIZE + 2;
76        let key_len = u16::from_le_bytes(
77            data[SECTOR_HEADER_SIZE..SECTOR_HEADER_SIZE + 2]
78                .try_into()
79                .ok()?,
80        ) as usize;
81
82        if offset + key_len + 16 > data.len() {
83            return None;
84        }
85
86        let key = data[offset..offset + key_len].to_vec();
87        offset += key_len;
88
89        let value_len = u64::from_le_bytes(data[offset..offset + 8].try_into().ok()?) as usize;
90        offset += 8;
91
92        let timestamp = u64::from_le_bytes(data[offset..offset + 8].try_into().ok()?);
93
94        Some((key, value_len, timestamp, 0)) // No TTL in v1
95    }
96
97    fn value_offset(&self, key_len: usize) -> usize {
98        SECTOR_HEADER_SIZE + 2 + key_len + 8 + 8
99    }
100}
101
102/// Version 2 format (with TTL support)
103pub struct FormatV2;
104static FORMAT_V2: FormatV2 = FormatV2;
105
106impl RecordFormat for FormatV2 {
107    fn record_header_size(&self, key_len: usize) -> usize {
108        SECTOR_HEADER_SIZE + 2 + key_len + 8 + 8 + 8 // header + key_len(2) + key + value_len(8) + timestamp(8) + ttl(8)
109    }
110
111    fn total_size(&self, key_len: usize, value_len: usize) -> usize {
112        self.record_header_size(key_len) + value_len
113    }
114
115    fn serialize_record(&self, record: &Record, include_value: bool) -> Vec<u8> {
116        let mut data = serialization_buffer(self, record, include_value);
117        self.serialize_record_into(record, include_value, &mut data);
118        data
119    }
120
121    fn serialize_record_into(&self, record: &Record, include_value: bool, data: &mut Vec<u8>) {
122        data.extend_from_slice(&(record.key.len() as u16).to_le_bytes());
123        data.extend_from_slice(&record.key);
124        data.extend_from_slice(&(record.value_len as u64).to_le_bytes());
125        data.extend_from_slice(&record.timestamp.to_le_bytes());
126        data.extend_from_slice(&record.ttl_expiry.load(Ordering::Acquire).to_le_bytes());
127
128        if include_value {
129            if let Some(value) = record.value.read().as_ref() {
130                data.extend_from_slice(value);
131            }
132        }
133    }
134
135    fn parse_record(&self, data: &[u8]) -> Option<(Vec<u8>, usize, u64, u64)> {
136        if data.len() < SECTOR_HEADER_SIZE + 2 {
137            return None;
138        }
139
140        let mut offset = SECTOR_HEADER_SIZE + 2;
141        let key_len = u16::from_le_bytes(
142            data[SECTOR_HEADER_SIZE..SECTOR_HEADER_SIZE + 2]
143                .try_into()
144                .ok()?,
145        ) as usize;
146
147        if offset + key_len + 24 > data.len() {
148            // 24 = value_len(8) + timestamp(8) + ttl(8)
149            return None;
150        }
151
152        let key = data[offset..offset + key_len].to_vec();
153        offset += key_len;
154
155        let value_len = u64::from_le_bytes(data[offset..offset + 8].try_into().ok()?) as usize;
156        offset += 8;
157
158        let timestamp = u64::from_le_bytes(data[offset..offset + 8].try_into().ok()?);
159        offset += 8;
160
161        let ttl_expiry = u64::from_le_bytes(data[offset..offset + 8].try_into().ok()?);
162
163        Some((key, value_len, timestamp, ttl_expiry))
164    }
165
166    fn value_offset(&self, key_len: usize) -> usize {
167        SECTOR_HEADER_SIZE + 2 + key_len + 8 + 8 + 8
168    }
169}
170
171/// Check that a head sector still stores the record it was read for.
172///
173/// A retired extent can be released and handed to another record while a reader
174/// still holds the old `Record` and its stale sector number, so the bytes coming
175/// back from disk must be proven to belong to this record before they are used.
176/// Everything needed is already in the buffer the reader just filled, so this
177/// costs no extra I/O. The layout prefix is identical in every format version:
178/// marker(2) seq(2) key_len(2) key value_len(8) timestamp(8).
179pub fn sector_holds_record(data: &[u8], record: &Record) -> bool {
180    if data.len() < SECTOR_HEADER_SIZE + 2 {
181        return false;
182    }
183    if u16::from_le_bytes([data[0], data[1]]) != SECTOR_MARKER {
184        return false;
185    }
186    let key_len =
187        u16::from_le_bytes([data[SECTOR_HEADER_SIZE], data[SECTOR_HEADER_SIZE + 1]]) as usize;
188    if key_len != record.key.len() {
189        return false;
190    }
191    let key_at = SECTOR_HEADER_SIZE + 2;
192    let value_len_at = key_at + key_len;
193    if value_len_at + 16 > data.len() {
194        return false;
195    }
196    if data[key_at..value_len_at] != record.key[..] {
197        return false;
198    }
199    let Ok(value_len) = data[value_len_at..value_len_at + 8].try_into() else {
200        return false;
201    };
202    if u64::from_le_bytes(value_len) as usize != record.value_len {
203        return false;
204    }
205    let Ok(timestamp) = data[value_len_at + 8..value_len_at + 16].try_into() else {
206        return false;
207    };
208    u64::from_le_bytes(timestamp) == record.timestamp
209}
210
211#[cfg(test)]
212pub(crate) fn fill_retirement_extent(retired: &mut [u8], sector: u64, sectors: usize) {
213    debug_assert_eq!(retired.len(), sectors * FEOX_BLOCK_SIZE);
214    retired.fill(0);
215    fill_retirement_markers(retired, sector, sectors);
216}
217
218pub(crate) fn fill_retirement_markers(retired: &mut [u8], sector: u64, remaining: usize) {
219    debug_assert_eq!(retired.len() % FEOX_BLOCK_SIZE, 0);
220    let blocks = retired.len() / FEOX_BLOCK_SIZE;
221    debug_assert!(blocks <= remaining);
222    for offset in 0..blocks {
223        let start = offset * FEOX_BLOCK_SIZE;
224        fill_retirement_marker(
225            &mut retired[start..start + DELETION_MARKER_SIZE],
226            sector + offset as u64,
227            remaining - offset,
228        );
229    }
230}
231
232pub(crate) fn fill_retirement_marker(marker: &mut [u8], sector: u64, remaining: usize) {
233    write_retirement_marker(marker, sector, remaining, RETIREMENT_COMPLETE);
234}
235
236#[cfg(test)]
237pub(crate) fn retirement_block(sector: u64, remaining: usize) -> Vec<u8> {
238    let mut retired = vec![0u8; FEOX_BLOCK_SIZE];
239    fill_retirement_marker(&mut retired[..DELETION_MARKER_SIZE], sector, remaining);
240    retired
241}
242
243#[cfg(test)]
244pub(crate) fn pending_retirement_block(sector: u64, remaining: usize) -> Vec<u8> {
245    let mut retired = vec![0u8; FEOX_BLOCK_SIZE];
246    write_retirement_marker(
247        &mut retired[..DELETION_MARKER_SIZE],
248        sector,
249        remaining,
250        RETIREMENT_PENDING,
251    );
252    retired
253}
254
255pub(crate) fn retirement_marker_token(sector: u64, marker: &[u8]) -> u16 {
256    let mut protected = [0; 17];
257    protected[..16].copy_from_slice(&marker[..16]);
258    protected[16] = marker[18];
259    seq_token(sector, &protected)
260}
261
262fn write_retirement_marker(marker: &mut [u8], sector: u64, remaining: usize, state: u8) {
263    marker[..8].copy_from_slice(DELETION_MARKER);
264    marker[8..16].copy_from_slice(&(remaining as u64).to_le_bytes());
265    marker[18] = state;
266    let token = retirement_marker_token(sector, marker);
267    marker[16..18].copy_from_slice(&token.to_le_bytes());
268}
269
270pub fn get_format(version: u32) -> Box<dyn RecordFormat> {
271    match version {
272        1 => Box::new(FormatV1),
273        2 | 3 => Box::new(FormatV2),
274        _ => Box::new(FormatV2),
275    }
276}
277
278pub(crate) fn get_format_ref(version: u32) -> &'static dyn RecordFormat {
279    match version {
280        1 => &FORMAT_V1,
281        2 | 3 => &FORMAT_V2,
282        _ => &FORMAT_V2,
283    }
284}
feoxdb/storage/format.rs

feoxdb/storage/
format.rs
