grove
grove/src/log.rs
at @13
1//! The append-only log: every fact grove knows, in order.
2//!
3//! File layout: an 8-byte header `grovelg1`, then records. Each record is
4//!
5//! ```text
6//! length: u32 little-endian (payload bytes)
7//! length check: 4 bytes (first 4 bytes of BLAKE3(length))
8//! payload check: 8 bytes (first 8 bytes of BLAKE3(payload))
9//! payload
10//! ```
11//!
12//! An append is acknowledged only once flushed to disk. A crash mid-append can leave a torn
13//! last record: a prefix of it, or (on some filesystems) zeros where its bytes should be.
14//! Opening the log recognises exactly those cases and cuts the file back to the last whole
15//! record. Anything else that fails a check is damage, and opening refuses rather than guess;
16//! the separate length check means a damaged length can't pass for a torn tail and take the
17//! records after it with it. While open, the log holds an exclusive lock on its file.
18
19use crate::blake3;
20use std::fs::{File, OpenOptions, TryLockError};
21use std::io::{self, Read, Seek, SeekFrom, Write};
22use std::path::Path;
23
24const HEADER: &[u8; 8] = b"grovelg1";
25const FRAME: usize = 4 + 4 + 8;
26/// Largest record grove will write or read: 64 MiB.
27pub const MAX_RECORD: usize = 64 << 20;
28
29pub struct Log {
30 file: File,
31 len: u64,
32 count: u64,
33}
34
35#[derive(Debug)]
36pub enum OpenError {
37 Io(io::Error),
38 Locked,
39 NotALog,
40 Damaged { offset: u64, reason: &'static str },
41}
42
43impl std::fmt::Display for OpenError {
44 fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
45 match self {
46 OpenError::Io(e) => write!(f, "{e}"),
47 OpenError::Locked => write!(f, "another grove process has this store open"),
48 OpenError::NotALog => write!(f, "not a grove log"),
49 OpenError::Damaged { offset, reason } => write!(f, "log damaged at byte {offset}: {reason}"),
50 }
51 }
52}
53
54impl From<io::Error> for OpenError {
55 fn from(e: io::Error) -> Self {
56 OpenError::Io(e)
57 }
58}
59
60/// What opening found, besides the records.
61#[derive(Debug, Default, PartialEq)]
62pub struct Recovery {
63 /// Bytes cut from the end because the last record was torn.
64 pub truncated: u64,
65}
66
67fn check(payload: &[u8]) -> [u8; 8] {
68 let h = blake3::hash(payload);
69 let mut c = [0; 8];
70 c.copy_from_slice(&h[..8]);
71 c
72}
73
74fn len_check(len: [u8; 4]) -> [u8; 4] {
75 let h = blake3::hash(&len);
76 [h[0], h[1], h[2], h[3]]
77}
78
79/// True if a crash could have left these bytes at the end of the file: nothing but zeros.
80fn zeros(bytes: &[u8]) -> bool {
81 bytes.iter().all(|&b| b == 0)
82}
83
84impl Log {
85 /// Open (creating if needed) and read every record.
86 pub fn open(path: &Path) -> Result<(Log, Vec<Vec<u8>>, Recovery), OpenError> {
87 let mut file = OpenOptions::new().read(true).write(true).create(true).truncate(false).open(path)?;
88 match file.try_lock() {
89 Ok(()) => {}
90 Err(TryLockError::WouldBlock) => return Err(OpenError::Locked),
91 Err(TryLockError::Error(e)) => return Err(e.into()),
92 }
93
94 let mut bytes = Vec::new();
95 file.read_to_end(&mut bytes)?;
96 if bytes.is_empty() {
97 file.write_all(HEADER)?;
98 file.sync_all()?;
99 sync_parent(path)?;
100 return Ok((Log { file, len: HEADER.len() as u64, count: 0 }, Vec::new(), Recovery::default()));
101 }
102 if bytes.len() < HEADER.len() || &bytes[..HEADER.len()] != HEADER {
103 return Err(OpenError::NotALog);
104 }
105
106 let mut records = Vec::new();
107 let mut at = HEADER.len();
108 let mut recovery = Recovery::default();
109 while at < bytes.len() {
110 let rest = &bytes[at..];
111 let damaged = |reason| Err(OpenError::Damaged { offset: at as u64, reason });
112 if rest.len() < FRAME {
113 // A frame cut short: the crash came while writing it.
114 recovery.truncated = rest.len() as u64;
115 break;
116 }
117 let len_bytes: [u8; 4] = rest[..4].try_into().unwrap();
118 if len_check(len_bytes) != rest[4..8] {
119 if zeros(rest) {
120 recovery.truncated = rest.len() as u64;
121 break;
122 }
123 return damaged("record header fails its check");
124 }
125 let len = u32::from_le_bytes(len_bytes) as usize;
126 if len > MAX_RECORD {
127 return damaged("record length out of range");
128 }
129 if rest.len() < FRAME + len {
130 // A sound header with the payload cut short.
131 recovery.truncated = rest.len() as u64;
132 break;
133 }
134 let payload = &rest[FRAME..FRAME + len];
135 if check(payload) != rest[8..FRAME] {
136 // Torn only if this is the last record; anything after it means real damage.
137 if FRAME + len == rest.len() || zeros(&rest[FRAME + len..]) {
138 recovery.truncated = rest.len() as u64;
139 break;
140 }
141 return damaged("record fails its check");
142 }
143 records.push(payload.to_vec());
144 at += FRAME + len;
145 }
146
147 if recovery.truncated > 0 {
148 file.set_len(at as u64)?;
149 file.sync_all()?;
150 }
151 file.seek(SeekFrom::Start(at as u64))?;
152 let count = records.len() as u64;
153 Ok((Log { file, len: at as u64, count }, records, recovery))
154 }
155
156 /// Append one record and flush it to disk. Returns the record's index.
157 pub fn append(&mut self, payload: &[u8]) -> io::Result<u64> {
158 if payload.len() > MAX_RECORD {
159 return Err(io::Error::new(io::ErrorKind::InvalidInput, "record too large"));
160 }
161 let mut frame = Vec::with_capacity(FRAME + payload.len());
162 let len = (payload.len() as u32).to_le_bytes();
163 frame.extend_from_slice(&len);
164 frame.extend_from_slice(&len_check(len));
165 frame.extend_from_slice(&check(payload));
166 frame.extend_from_slice(payload);
167 let result = self.file.write_all(&frame).and_then(|_| self.file.sync_data());
168 if let Err(e) = result {
169 // Leave the file as it was before this append, so the next one starts clean.
170 let _ = self.file.set_len(self.len);
171 let _ = self.file.seek(SeekFrom::Start(self.len));
172 return Err(e);
173 }
174 self.len += frame.len() as u64;
175 self.count += 1;
176 Ok(self.count - 1)
177 }
178
179 pub fn len(&self) -> u64 {
180 self.len
181 }
182
183 pub fn count(&self) -> u64 {
184 self.count
185 }
186
187 pub fn is_empty(&self) -> bool {
188 self.count == 0
189 }
190}
191
192/// Flush a directory entry (a newly created or renamed file) to disk.
193pub fn sync_parent(path: &Path) -> io::Result<()> {
194 match path.parent() {
195 Some(dir) if !dir.as_os_str().is_empty() => File::open(dir)?.sync_all(),
196 _ => Ok(()),
197 }
198}
199
200#[cfg(test)]
201mod tests {
202 use super::*;
203 use crate::testdir::TestDir;
204
205 fn reopen(path: &Path) -> (Log, Vec<Vec<u8>>, Recovery) {
206 Log::open(path).unwrap()
207 }
208
209 #[test]
210 fn append_and_replay() {
211 let dir = TestDir::new("log-replay");
212 let path = dir.path().join("log");
213 {
214 let (mut log, recs, _) = reopen(&path);
215 assert!(recs.is_empty());
216 assert_eq!(log.append(b"one").unwrap(), 0);
217 assert_eq!(log.append(b"").unwrap(), 1);
218 assert_eq!(log.append(&vec![7u8; 100_000]).unwrap(), 2);
219 }
220 let (log, recs, rec) = reopen(&path);
221 assert_eq!(recs, vec![b"one".to_vec(), Vec::new(), vec![7u8; 100_000]]);
222 assert_eq!(log.count(), 3);
223 assert_eq!(rec, Recovery::default());
224 }
225
226 #[test]
227 fn second_open_is_locked_out() {
228 let dir = TestDir::new("log-lock");
229 let path = dir.path().join("log");
230 let _held = reopen(&path);
231 assert!(matches!(Log::open(&path), Err(OpenError::Locked)));
232 }
233
234 #[test]
235 fn torn_tail_is_cut_at_every_byte() {
236 // Simulate a crash at every possible point while writing the third record.
237 let dir = TestDir::new("log-torn");
238 let full = dir.path().join("full");
239 {
240 let (mut log, _, _) = reopen(&full);
241 log.append(b"first").unwrap();
242 log.append(b"second").unwrap();
243 log.append(b"third record").unwrap();
244 }
245 let bytes = std::fs::read(&full).unwrap();
246 let two_records = HEADER.len() + 2 * FRAME + b"first".len() + b"second".len();
247 for cut in two_records..bytes.len() {
248 let path = dir.path().join(format!("cut{cut}"));
249 std::fs::write(&path, &bytes[..cut]).unwrap();
250 let (mut log, recs, rec) = reopen(&path);
251 assert_eq!(recs, vec![b"first".to_vec(), b"second".to_vec()], "cut at {cut}");
252 assert_eq!(rec.truncated, (cut - two_records) as u64);
253 // The log is usable again afterwards.
254 log.append(b"after").unwrap();
255 drop(log);
256 assert_eq!(reopen(&path).1.len(), 3);
257 }
258 }
259
260 #[test]
261 fn garbage_in_last_record_is_cut() {
262 let dir = TestDir::new("log-garbage-tail");
263 let path = dir.path().join("log");
264 {
265 let (mut log, _, _) = reopen(&path);
266 log.append(b"keep").unwrap();
267 log.append(b"lose").unwrap();
268 }
269 let mut bytes = std::fs::read(&path).unwrap();
270 let n = bytes.len();
271 bytes[n - 1] ^= 0xff; // the tail record's payload no longer matches its check
272 std::fs::write(&path, &bytes).unwrap();
273 let (_, recs, rec) = reopen(&path);
274 assert_eq!(recs, vec![b"keep".to_vec()]);
275 assert!(rec.truncated > 0);
276 }
277
278 #[test]
279 fn damage_before_the_tail_is_refused() {
280 let dir = TestDir::new("log-damage");
281 let path = dir.path().join("log");
282 {
283 let (mut log, _, _) = reopen(&path);
284 log.append(b"first").unwrap();
285 log.append(b"second").unwrap();
286 }
287 let mut bytes = std::fs::read(&path).unwrap();
288 bytes[HEADER.len() + FRAME] ^= 0xff; // inside the first payload
289 std::fs::write(&path, &bytes).unwrap();
290 assert!(matches!(Log::open(&path), Err(OpenError::Damaged { .. })));
291 // And nothing was cut.
292 assert_eq!(std::fs::read(&path).unwrap(), bytes);
293 }
294
295 #[test]
296 fn damaged_length_does_not_eat_later_records() {
297 let dir = TestDir::new("log-bad-length");
298 let path = dir.path().join("log");
299 {
300 let (mut log, _, _) = reopen(&path);
301 log.append(b"first").unwrap();
302 log.append(b"second").unwrap();
303 log.append(b"third").unwrap();
304 }
305 let mut bytes = std::fs::read(&path).unwrap();
306 bytes[HEADER.len() + 1] = 0x40; // first record now claims to be 16 MiB long
307 std::fs::write(&path, &bytes).unwrap();
308 assert!(matches!(Log::open(&path), Err(OpenError::Damaged { .. })));
309 assert_eq!(std::fs::read(&path).unwrap(), bytes, "nothing cut");
310 }
311
312 #[test]
313 fn zero_filled_tail_is_cut() {
314 // Some filesystems extend the file before the data lands; a crash leaves zeros.
315 let dir = TestDir::new("log-zeros");
316 let path = dir.path().join("log");
317 {
318 let (mut log, _, _) = reopen(&path);
319 log.append(b"kept").unwrap();
320 }
321 let mut bytes = std::fs::read(&path).unwrap();
322 bytes.extend_from_slice(&[0u8; 300]);
323 std::fs::write(&path, &bytes).unwrap();
324 let (_, recs, rec) = reopen(&path);
325 assert_eq!(recs, vec![b"kept".to_vec()]);
326 assert_eq!(rec.truncated, 300);
327 }
328
329 #[test]
330 fn not_a_log() {
331 let dir = TestDir::new("log-foreign");
332 let path = dir.path().join("log");
333 std::fs::write(&path, b"hello world").unwrap();
334 assert!(matches!(Log::open(&path), Err(OpenError::NotALog)));
335 }
336}