basalt/
wal.rs

1//! Append-only write-ahead log for committed state snapshots.
2//!
3//! A frame is considered committed only after its complete payload and CRC
4//! have reached the WAL file.  Recovery accepts valid complete frames and
5//! repairs and ignores an incomplete final frame, which is the normal result
6//! of a killed process during append.
7
8use std::fs::{self, File, OpenOptions};
9use std::io::{self, Read, Write};
10use std::path::Path;
11
12use crate::crc::crc32;
13use crate::db::{DbError, DbErrorKind};
14
15const MAGIC: &[u8; 4] = b"BSWL";
16const VERSION: u32 = 1;
17const HEADER: usize = 32;
18
19#[derive(Debug, Clone)]
20pub struct Frame {
21    pub generation: u64,
22    pub payload: Vec<u8>,
23}
24
25fn io_error(context: &str, e: io::Error) -> DbError {
26    DbError::new(
27        DbErrorKind::Io(format!("{context}: {e}")),
28        format!("{context}: {e}"),
29    )
30}
31
32pub fn append(path: &Path, generation: u64, payload: &[u8]) -> Result<(), DbError> {
33    if let Some(parent) = path
34        .parent()
35        .filter(|parent| !parent.as_os_str().is_empty())
36    {
37        fs::create_dir_all(parent).map_err(|e| io_error("create WAL directory", e))?;
38    }
39    let mut header = [0u8; HEADER];
40    header[..4].copy_from_slice(MAGIC);
41    header[4..8].copy_from_slice(&VERSION.to_le_bytes());
42    header[8..16].copy_from_slice(&generation.to_le_bytes());
43    header[16..24].copy_from_slice(&(payload.len() as u64).to_le_bytes());
44    header[24..28].copy_from_slice(&crc32(payload).to_le_bytes());
45    let mut file = OpenOptions::new()
46        .create(true)
47        .append(true)
48        .open(path)
49        .map_err(|e| io_error("open WAL", e))?;
50    file.write_all(&header)
51        .map_err(|e| io_error("write WAL header", e))?;
52    file.write_all(payload)
53        .map_err(|e| io_error("write WAL payload", e))?;
54    file.sync_all().map_err(|e| io_error("sync WAL", e))
55}
56
57/// Return the highest valid frame.  A partial/corrupt tail is ignored; an
58/// invalid frame before the tail is an error because it would hide later data.
59pub fn latest(path: &Path) -> Result<Option<Frame>, DbError> {
60    if !path.exists() {
61        return Ok(None);
62    }
63    let mut bytes = Vec::new();
64    File::open(path)
65        .map_err(|e| io_error("open WAL", e))?
66        .read_to_end(&mut bytes)
67        .map_err(|e| io_error("read WAL", e))?;
68    let mut offset = 0usize;
69    let mut latest = None;
70    while offset < bytes.len() {
71        if bytes.len() - offset < HEADER {
72            truncate_to(path, offset)?;
73            break;
74        }
75        if &bytes[offset..offset + 4] != MAGIC {
76            return Err(corrupt("invalid WAL magic"));
77        }
78        let version = u32_at(&bytes, offset + 4)?;
79        if version != VERSION {
80            return Err(corrupt("unsupported WAL version"));
81        }
82        let generation = u64_at(&bytes, offset + 8)?;
83        let declared_len = u64_at(&bytes, offset + 16)?;
84        let len = match usize::try_from(declared_len) {
85            Ok(len) => len,
86            Err(_) => {
87                truncate_to(path, offset)?;
88                break;
89            }
90        };
91        let checksum = u32_at(&bytes, offset + 24)?;
92        let end = match offset.checked_add(HEADER).and_then(|n| n.checked_add(len)) {
93            Some(end) if end <= bytes.len() => end,
94            _ => {
95                truncate_to(path, offset)?;
96                break;
97            }
98        };
99        let payload = &bytes[offset + HEADER..end];
100        if crc32(payload) != checksum {
101            return Err(corrupt("WAL frame checksum mismatch"));
102        }
103        if latest
104            .as_ref()
105            .map(|f: &Frame| generation > f.generation)
106            .unwrap_or(true)
107        {
108            latest = Some(Frame {
109                generation,
110                payload: payload.to_vec(),
111            });
112        }
113        offset = end;
114    }
115    Ok(latest)
116}
117
118pub fn truncate(path: &Path) -> Result<(), DbError> {
119    if !path.exists() {
120        return Ok(());
121    }
122    let file = OpenOptions::new()
123        .write(true)
124        .truncate(true)
125        .open(path)
126        .map_err(|e| io_error("truncate WAL", e))?;
127    file.sync_all()
128        .map_err(|e| io_error("sync truncated WAL", e))
129}
130
131fn truncate_to(path: &Path, length: usize) -> Result<(), DbError> {
132    let file = OpenOptions::new()
133        .write(true)
134        .open(path)
135        .map_err(|e| io_error("open WAL for tail repair", e))?;
136    file.set_len(length as u64)
137        .map_err(|e| io_error("truncate incomplete WAL frame", e))?;
138    file.sync_all()
139        .map_err(|e| io_error("sync repaired WAL", e))
140}
141
142fn corrupt(message: &str) -> DbError {
143    DbError::new(
144        DbErrorKind::Io(message.to_string()),
145        format!("corrupt WAL: {message}"),
146    )
147}
148
149fn u32_at(bytes: &[u8], offset: usize) -> Result<u32, DbError> {
150    let raw = bytes
151        .get(offset..offset + 4)
152        .ok_or_else(|| corrupt("WAL header is truncated"))?;
153    Ok(u32::from_le_bytes(raw.try_into().unwrap()))
154}
155
156fn u64_at(bytes: &[u8], offset: usize) -> Result<u64, DbError> {
157    let raw = bytes
158        .get(offset..offset + 8)
159        .ok_or_else(|| corrupt("WAL header is truncated"))?;
160    Ok(u64::from_le_bytes(raw.try_into().unwrap()))
161}
162
163#[cfg(test)]
164mod tests {
165    use super::*;
166
167    #[test]
168    fn ignores_torn_tail() {
169        let dir = std::env::temp_dir().join(format!("basalt-wal-{}", std::process::id()));
170        let _ = fs::remove_dir_all(&dir);
171        fs::create_dir_all(&dir).unwrap();
172        let path = dir.join("db.wal");
173        append(&path, 1, b"one").unwrap();
174        let mut file = OpenOptions::new().append(true).open(&path).unwrap();
175        file.write_all(b"BSWL").unwrap();
176        file.sync_all().unwrap();
177        assert_eq!(latest(&path).unwrap().unwrap().payload, b"one");
178        truncate(&path).unwrap();
179        assert!(latest(&path).unwrap().is_none());
180        let _ = fs::remove_dir_all(dir);
181    }
182
183    #[test]
184    fn rejects_a_complete_corrupt_frame() {
185        let dir = std::env::temp_dir().join(format!("basalt-wal-corrupt-{}", std::process::id()));
186        let _ = fs::remove_dir_all(&dir);
187        fs::create_dir_all(&dir).unwrap();
188        let path = dir.join("db.wal");
189        append(&path, 1, b"one").unwrap();
190        let mut bytes = fs::read(&path).unwrap();
191        *bytes.last_mut().unwrap() ^= 1;
192        fs::write(&path, bytes).unwrap();
193        assert!(latest(&path).is_err());
194        let _ = fs::remove_dir_all(dir);
195    }
196
197    #[test]
198    fn repairs_a_torn_tail_before_a_later_commit() {
199        let dir = std::env::temp_dir().join(format!("basalt-wal-tail-{}", std::process::id()));
200        let _ = fs::remove_dir_all(&dir);
201        fs::create_dir_all(&dir).unwrap();
202        let path = dir.join("db.wal");
203        append(&path, 1, b"one").unwrap();
204        OpenOptions::new()
205            .append(true)
206            .open(&path)
207            .unwrap()
208            .write_all(b"BSWL")
209            .unwrap();
210
211        assert_eq!(latest(&path).unwrap().unwrap().generation, 1);
212        append(&path, 2, b"two").unwrap();
213        let frame = latest(&path).unwrap().unwrap();
214        assert_eq!(frame.generation, 2);
215        assert_eq!(frame.payload, b"two");
216        let _ = fs::remove_dir_all(dir);
217    }
218}
basalt/wal.rs

basalt/
wal.rs
