vecdb_core/storage/
wal.rs

1use std::fs::{File, OpenOptions};
2use std::io::{Read, Seek, SeekFrom, Write};
3use std::path::{Path, PathBuf};
4use std::time::{SystemTime, UNIX_EPOCH};
5
6use serde::{Deserialize, Serialize};
7
8use crate::errors::{Result, VecDbError};
9use crate::types::{VectorId, VectorRecord};
10
11fn now_unix() -> u64 {
12    SystemTime::now()
13        .duration_since(UNIX_EPOCH)
14        .unwrap_or_default()
15        .as_secs()
16}
17
18fn checksum(data: &[u8]) -> u32 {
19    xxhash_rust::xxh3::xxh3_64(data) as u32
20}
21
22#[derive(Debug, Clone, Serialize, Deserialize)]
23pub enum WalEntry {
24    Insert {
25        id: VectorId,
26        mmap_index: usize,
27        record: VectorRecord,
28    },
29    Delete {
30        id: VectorId,
31    },
32    Checkpoint {
33        entry_count: usize,
34        timestamp: u64,
35    },
36}
37
38pub struct WriteAheadLog {
39    pub path: PathBuf,
40    pub entry_count: usize,
41    file: File,
42}
43
44fn open_rw(path: &Path, create: bool) -> Result<File> {
45    OpenOptions::new()
46        .read(true)
47        .write(true)
48        .create(create)
49        .open(path)
50        .map_err(|e| VecDbError::StorageError(format!("wal open failed: {e}")))
51}
52
53impl WriteAheadLog {
54    pub fn open(path: &Path) -> Result<Self> {
55        let file = open_rw(path, true)?;
56
57        let mut wal = Self {
58            path: path.to_path_buf(),
59            entry_count: 0,
60            file,
61        };
62
63        // Count existing valid entries.
64        let entries = wal.replay()?;
65        wal.entry_count = entries.len();
66        Ok(wal)
67    }
68
69    pub fn append(&mut self, entry: &WalEntry) -> Result<()> {
70        let payload =
71            serde_json::to_vec(entry).map_err(|e| VecDbError::SerializationError(e.to_string()))?;
72
73        let length = payload.len() as u32;
74        let cksum = checksum(&payload);
75
76        // Seek to end before writing (replaces O_APPEND — safe for single-writer WAL).
77        self.file
78            .seek(SeekFrom::End(0))
79            .map_err(|e| VecDbError::StorageError(format!("wal seek failed: {e}")))?;
80
81        self.file
82            .write_all(&length.to_le_bytes())
83            .map_err(|e| VecDbError::StorageError(format!("wal write failed: {e}")))?;
84        self.file
85            .write_all(&payload)
86            .map_err(|e| VecDbError::StorageError(format!("wal write failed: {e}")))?;
87        self.file
88            .write_all(&cksum.to_le_bytes())
89            .map_err(|e| VecDbError::StorageError(format!("wal write failed: {e}")))?;
90        // sync_all = FlushFileBuffers on Windows: commits write buffer to page cache
91        // so that reads via any handle (including std::fs::read) see current data.
92        self.file
93            .sync_all()
94            .map_err(|e| VecDbError::StorageError(format!("wal flush failed: {e}")))?;
95
96        self.entry_count += 1;
97        tracing::debug!("WAL append: entry_count={}", self.entry_count);
98        Ok(())
99    }
100
101    /// Append many entries with a single `sync_all` at the end. Used by bulk
102    /// load, where per-entry fsync would dominate (one fsync per vector). The
103    /// batch is still crash-consistent: on a crash mid-batch, replay stops at
104    /// the last fully-written entry (length/checksum framing), and any partial
105    /// tail is ignored.
106    pub fn append_batch(&mut self, entries: &[WalEntry]) -> Result<()> {
107        if entries.is_empty() {
108            return Ok(());
109        }
110        self.file
111            .seek(SeekFrom::End(0))
112            .map_err(|e| VecDbError::StorageError(format!("wal seek failed: {e}")))?;
113
114        let mut buf = Vec::new();
115        for entry in entries {
116            let payload = serde_json::to_vec(entry)
117                .map_err(|e| VecDbError::SerializationError(e.to_string()))?;
118            let length = payload.len() as u32;
119            let cksum = checksum(&payload);
120            buf.extend_from_slice(&length.to_le_bytes());
121            buf.extend_from_slice(&payload);
122            buf.extend_from_slice(&cksum.to_le_bytes());
123        }
124        self.file
125            .write_all(&buf)
126            .map_err(|e| VecDbError::StorageError(format!("wal write failed: {e}")))?;
127        self.file
128            .sync_all()
129            .map_err(|e| VecDbError::StorageError(format!("wal flush failed: {e}")))?;
130
131        self.entry_count += entries.len();
132        Ok(())
133    }
134
135    pub fn replay(&mut self) -> Result<Vec<WalEntry>> {
136        // Read entire file via a fresh open/read/close cycle.
137        // This is the most reliable approach on all platforms — avoids any
138        // per-handle buffering or file-pointer aliasing issues.
139        let data = std::fs::read(&self.path)
140            .map_err(|e| VecDbError::StorageError(format!("wal read failed: {e}")))?;
141        let mut cursor = std::io::Cursor::new(data);
142
143        let mut entries = Vec::new();
144
145        loop {
146            // Read 4-byte length prefix.
147            let mut len_buf = [0u8; 4];
148            match cursor.read_exact(&mut len_buf) {
149                Ok(()) => {}
150                Err(e) if e.kind() == std::io::ErrorKind::UnexpectedEof => break,
151                Err(e) => {
152                    tracing::warn!("WAL replay: error reading length: {e}");
153                    break;
154                }
155            }
156            let length = u32::from_le_bytes(len_buf) as usize;
157
158            let mut payload = vec![0u8; length];
159            if let Err(e) = cursor.read_exact(&mut payload) {
160                tracing::warn!("WAL replay: truncated payload: {e}");
161                break;
162            }
163
164            let mut cksum_buf = [0u8; 4];
165            if let Err(e) = cursor.read_exact(&mut cksum_buf) {
166                tracing::warn!("WAL replay: truncated checksum: {e}");
167                break;
168            }
169            let stored = u32::from_le_bytes(cksum_buf);
170            let computed = checksum(&payload);
171            if computed != stored {
172                tracing::warn!(
173                    "WAL replay: checksum mismatch (computed={computed}, stored={stored}), stopping"
174                );
175                break;
176            }
177
178            match serde_json::from_slice::<WalEntry>(&payload) {
179                Ok(entry) => entries.push(entry),
180                Err(e) => {
181                    tracing::warn!("WAL replay: deserialization error: {e}, stopping");
182                    break;
183                }
184            }
185        }
186
187        tracing::info!("WAL replay: {} entries recovered", entries.len());
188        Ok(entries)
189    }
190
191    pub fn checkpoint(&mut self) -> Result<()> {
192        let entry = WalEntry::Checkpoint {
193            entry_count: self.entry_count,
194            timestamp: now_unix(),
195        };
196        self.append(&entry)
197    }
198
199    pub fn truncate_after_checkpoint(&mut self) -> Result<()> {
200        let entries = self.replay()?;
201
202        let Some(checkpoint_pos) = entries
203            .iter()
204            .rposition(|e| matches!(e, WalEntry::Checkpoint { .. }))
205        else {
206            return Ok(());
207        };
208
209        let remaining: Vec<_> = entries.into_iter().skip(checkpoint_pos + 1).collect();
210        let remaining_count = remaining.len();
211
212        // Truncate and rewrite via a separate handle so we can resize from 0.
213        let mut new_file = OpenOptions::new()
214            .write(true)
215            .truncate(true)
216            .open(&self.path)
217            .map_err(|e| VecDbError::StorageError(format!("wal truncate open failed: {e}")))?;
218
219        for entry in &remaining {
220            let payload = serde_json::to_vec(entry)
221                .map_err(|e| VecDbError::SerializationError(e.to_string()))?;
222            let length = payload.len() as u32;
223            let cksum = checksum(&payload);
224            new_file
225                .write_all(&length.to_le_bytes())
226                .map_err(|e| VecDbError::StorageError(format!("wal rewrite failed: {e}")))?;
227            new_file
228                .write_all(&payload)
229                .map_err(|e| VecDbError::StorageError(format!("wal rewrite failed: {e}")))?;
230            new_file
231                .write_all(&cksum.to_le_bytes())
232                .map_err(|e| VecDbError::StorageError(format!("wal rewrite failed: {e}")))?;
233        }
234        new_file
235            .flush()
236            .map_err(|e| VecDbError::StorageError(format!("wal rewrite flush failed: {e}")))?;
237        drop(new_file);
238
239        // Reopen own handle so it sees the truncated content.
240        self.file = open_rw(&self.path, false)?;
241        // Position at end ready for future appends.
242        self.file
243            .seek(SeekFrom::End(0))
244            .map_err(|e| VecDbError::StorageError(format!("wal seek failed: {e}")))?;
245
246        self.entry_count = remaining_count;
247        tracing::info!("WAL truncated: {} entries remaining", self.entry_count);
248        Ok(())
249    }
250
251    pub fn entry_count(&self) -> usize {
252        self.entry_count
253    }
254}
vecdb_core/storage/wal.rs

vecdb_core/storage/
wal.rs
