ursula_runtime/
journal.rs

1//! Append-only framed journal.
2//!
3//! Persistence is kept orthogonal to serialization. The journal moves opaque
4//! versioned, checksummed frames to and from a file and handles the durability
5//! concerns — append, `fsync`, bounded recovery, and recovery of a torn trailing
6//! frame after a crash.
7//! How a record turns into a payload is entirely the [`FrameCodec`]'s business, so
8//! the Raft log store can frame protobuf while the WAL engine frames JSON over the
9//! exact same code.
10
11use std::fs;
12use std::fs::File;
13use std::fs::OpenOptions;
14use std::io;
15use std::io::Read;
16use std::io::Seek;
17use std::io::Write;
18use std::marker::PhantomData;
19use std::path::Path;
20use std::path::PathBuf;
21
22const JOURNAL_MAGIC: [u8; 8] = *b"URSJWAL\0";
23const JOURNAL_VERSION: u16 = 1;
24const JOURNAL_HEADER_LEN: usize = 16;
25const FRAME_HEADER_LEN: usize = 8;
26
27/// Maximum encoded payload accepted from disk or written as one journal frame.
28///
29/// This is intentionally above Ursula's 256 MiB Raft RPC limit while still
30/// preventing a corrupted length field from requesting an unbounded allocation.
31pub const MAX_FRAME_PAYLOAD_BYTES: usize = 512 * 1024 * 1024;
32
33fn journal_header() -> [u8; JOURNAL_HEADER_LEN] {
34    let mut header = [0_u8; JOURNAL_HEADER_LEN];
35    header[..JOURNAL_MAGIC.len()].copy_from_slice(&JOURNAL_MAGIC);
36    header[8..10].copy_from_slice(&JOURNAL_VERSION.to_le_bytes());
37    header[10..12].copy_from_slice(
38        &u16::try_from(JOURNAL_HEADER_LEN)
39            .expect("journal header length fits u16")
40            .to_le_bytes(),
41    );
42    header
43}
44
45/// Serialization seam: how one record becomes a frame payload and back.
46///
47/// `encode` is infallible because the codecs we use (protobuf, JSON over plain
48/// owned types) cannot fail in practice; a codec with fallible encoding should
49/// surface that as an `io::Error` from a panic-documented invariant instead.
50pub trait FrameCodec {
51    /// The record type carried in each frame.
52    type Record;
53
54    /// Serialize a record into a frame payload.
55    fn encode(record: &Self::Record) -> Vec<u8>;
56
57    /// Deserialize a frame payload back into a record.
58    fn decode(payload: &[u8]) -> io::Result<Self::Record>;
59}
60
61/// JSON frame codec for any owned, serde-serializable record.
62pub struct JsonCodec<T>(PhantomData<T>);
63
64impl<T> FrameCodec for JsonCodec<T>
65where T: serde::Serialize + serde::de::DeserializeOwned
66{
67    type Record = T;
68
69    fn encode(record: &T) -> Vec<u8> {
70        serde_json::to_vec(record).expect("journal record serializes to JSON")
71    }
72
73    fn decode(payload: &[u8]) -> io::Result<T> {
74        serde_json::from_slice(payload)
75            .map_err(|err| io::Error::new(io::ErrorKind::InvalidData, err))
76    }
77}
78
79/// An append handle over a single journal file.
80///
81/// The file is opened lazily on first append. The parent directory is `fsync`ed
82/// once on the first [`JournalWriter::sync`] when the file may have been freshly
83/// created, so the file's existence survives a crash.
84#[derive(Debug)]
85pub struct JournalWriter {
86    file: Option<File>,
87    parent_unsynced: bool,
88}
89
90impl JournalWriter {
91    /// Create a writer. Set `needs_parent_sync` when the file may not exist yet, so
92    /// the parent directory is `fsync`ed once the file is created.
93    pub fn new(needs_parent_sync: bool) -> Self {
94        Self {
95            file: None,
96            parent_unsynced: needs_parent_sync,
97        }
98    }
99
100    /// Create and initialize the journal file even when there are no records.
101    pub fn ensure_created(&mut self, path: &Path) -> io::Result<()> {
102        let _ = self.file_mut(path)?;
103        Ok(())
104    }
105
106    /// Append one record as a framed payload. Does not durably flush; pair with
107    /// [`JournalWriter::sync`] once per batch.
108    pub fn append<C: FrameCodec>(&mut self, path: &Path, record: &C::Record) -> io::Result<()> {
109        let payload = C::encode(record);
110        if payload.len() > MAX_FRAME_PAYLOAD_BYTES {
111            return Err(io::Error::new(
112                io::ErrorKind::InvalidData,
113                format!(
114                    "journal record is {} bytes, exceeding the {} byte limit",
115                    payload.len(),
116                    MAX_FRAME_PAYLOAD_BYTES
117                ),
118            ));
119        }
120        let len = u32::try_from(payload.len())
121            .map_err(|_| io::Error::new(io::ErrorKind::InvalidData, "journal record too large"))?;
122        let checksum = crc32fast::hash(&payload);
123        let file = self.file_mut(path)?;
124        file.write_all(&len.to_le_bytes())?;
125        file.write_all(&checksum.to_le_bytes())?;
126        file.write_all(&payload)
127    }
128
129    /// `fsync` the file data, plus the parent directory once if it was freshly created.
130    pub fn sync(&mut self, path: &Path) -> io::Result<()> {
131        let file = self.file.as_mut().expect("file opened before sync");
132        file.sync_data()?;
133        if self.parent_unsynced
134            && let Some(parent) = path.parent()
135            && let Ok(dir) = File::open(parent)
136        {
137            dir.sync_all()?;
138            self.parent_unsynced = false;
139        }
140        Ok(())
141    }
142
143    fn file_mut(&mut self, path: &Path) -> io::Result<&mut File> {
144        if self.file.is_none() {
145            if let Some(parent) = path.parent() {
146                fs::create_dir_all(parent)?;
147            }
148            let mut file = OpenOptions::new()
149                .create(true)
150                .read(true)
151                .append(true)
152                .open(path)?;
153            let file_len = file.metadata()?.len();
154            if file_len == 0 {
155                file.write_all(&journal_header())?;
156            } else {
157                validate_file_header(&mut file, path, file_len)?;
158            }
159            self.file = Some(file);
160        }
161        Ok(self.file.as_mut().expect("file opened above"))
162    }
163}
164
165/// Read every record from `path`, decoding with `C`. A torn trailing frame left by a
166/// crash mid-write is truncated away and ignored, leaving the file at its last clean
167/// record boundary.
168pub fn replay<C: FrameCodec>(path: &Path) -> io::Result<Vec<C::Record>> {
169    let mut records = Vec::new();
170    replay_each::<C>(path, |record| {
171        records.push(record);
172        Ok(())
173    })?;
174    Ok(records)
175}
176
177/// Stream every valid record from `path` through `visit` without retaining the
178/// entire journal in memory. A torn trailing frame is truncated with the same
179/// recovery semantics as [`replay`].
180pub fn replay_each<C: FrameCodec>(
181    path: &Path,
182    mut visit: impl FnMut(C::Record) -> io::Result<()>,
183) -> io::Result<()> {
184    if !path.exists() {
185        return Ok(());
186    }
187
188    let mut file = File::open(path)?;
189    let file_len = file.metadata()?.len();
190    if file_len == 0 {
191        return Ok(());
192    }
193    if file_len < u64::try_from(JOURNAL_HEADER_LEN).expect("header length fits u64") {
194        return Err(io::Error::new(
195            io::ErrorKind::InvalidData,
196            format!(
197                "journal '{}' has no complete Ursula WAL header; legacy unversioned journals require an explicit reset or migration",
198                path.display()
199            ),
200        ));
201    }
202    validate_file_header(&mut file, path, file_len)?;
203    let mut valid_len = u64::try_from(JOURNAL_HEADER_LEN).expect("header length fits u64");
204    let mut frame_index = 0_u64;
205    while valid_len < file_len {
206        let remaining = file_len.saturating_sub(valid_len);
207        if remaining < u64::try_from(FRAME_HEADER_LEN).expect("frame header length fits u64") {
208            break;
209        }
210
211        let mut len_bytes = [0_u8; 4];
212        file.read_exact(&mut len_bytes)?;
213        let payload_len = u64::from(u32::from_le_bytes(len_bytes));
214        let mut checksum_bytes = [0_u8; 4];
215        file.read_exact(&mut checksum_bytes)?;
216        let expected_checksum = u32::from_le_bytes(checksum_bytes);
217        if payload_len > u64::try_from(MAX_FRAME_PAYLOAD_BYTES).expect("frame limit fits u64") {
218            return Err(io::Error::new(
219                io::ErrorKind::InvalidData,
220                format!(
221                    "journal '{}' frame {} declares {} bytes, exceeding the {} byte limit",
222                    path.display(),
223                    frame_index + 1,
224                    payload_len,
225                    MAX_FRAME_PAYLOAD_BYTES
226                ),
227            ));
228        }
229        if remaining
230            .saturating_sub(u64::try_from(FRAME_HEADER_LEN).expect("frame header length fits u64"))
231            < payload_len
232        {
233            break;
234        }
235
236        let payload_len = usize::try_from(payload_len).expect("u32 fits usize");
237        let mut payload = vec![0_u8; payload_len];
238        file.read_exact(&mut payload)?;
239        let actual_checksum = crc32fast::hash(&payload);
240        if actual_checksum != expected_checksum {
241            return Err(io::Error::new(
242                io::ErrorKind::InvalidData,
243                format!(
244                    "journal '{}' frame {} checksum mismatch: expected {expected_checksum:#010x}, got {actual_checksum:#010x}",
245                    path.display(),
246                    frame_index + 1
247                ),
248            ));
249        }
250        let record = C::decode(&payload).map_err(|err| {
251            io::Error::new(
252                err.kind(),
253                format!(
254                    "journal '{}' frame {} decode failed: {err}",
255                    path.display(),
256                    frame_index + 1
257                ),
258            )
259        })?;
260        visit(record)?;
261        valid_len = valid_len
262            .checked_add(
263                u64::try_from(FRAME_HEADER_LEN)
264                    .expect("frame header length fits u64")
265                    .saturating_add(u64::try_from(payload_len).expect("usize fits u64")),
266            )
267            .ok_or_else(|| io::Error::new(io::ErrorKind::InvalidData, "journal offset overflow"))?;
268        frame_index = frame_index.saturating_add(1);
269    }
270
271    if valid_len < file_len {
272        truncate_to(
273            path,
274            usize::try_from(valid_len).map_err(|_| {
275                io::Error::new(io::ErrorKind::InvalidData, "journal offset exceeds usize")
276            })?,
277        )?;
278    }
279    Ok(())
280}
281
282/// Migrate the legacy `[length][payload]` journal format to the current
283/// checksummed format, retaining a hard-linked `.v0.bak` rollback copy.
284///
285/// A current-format or empty journal is left untouched. Unsupported future
286/// versions are also left for [`replay`] to reject explicitly.
287pub fn migrate_legacy<C: FrameCodec>(path: &Path) -> io::Result<bool> {
288    if !path.exists() {
289        return Ok(false);
290    }
291    let mut source = File::open(path)?;
292    let file_len = source.metadata()?.len();
293    if file_len == 0 {
294        return Ok(false);
295    }
296    let mut prefix = [0_u8; JOURNAL_MAGIC.len()];
297    let prefix_len = source.read(&mut prefix)?;
298    source.rewind()?;
299    if prefix_len == JOURNAL_MAGIC.len() && prefix == JOURNAL_MAGIC {
300        return Ok(false);
301    }
302
303    let migrate_path = suffixed_path(path, ".migrate-v1");
304    let backup_path = suffixed_path(path, ".v0.bak");
305    if migrate_path.exists() {
306        fs::remove_file(&migrate_path)?;
307    }
308    let mut writer = JournalWriter::new(true);
309    let mut valid_len = 0_u64;
310    let mut frame_index = 0_u64;
311    while valid_len < file_len {
312        let remaining = file_len.saturating_sub(valid_len);
313        if remaining < 4 {
314            break;
315        }
316        let mut len_bytes = [0_u8; 4];
317        source.read_exact(&mut len_bytes)?;
318        let payload_len = u64::from(u32::from_le_bytes(len_bytes));
319        if payload_len > u64::try_from(MAX_FRAME_PAYLOAD_BYTES).expect("frame limit fits u64") {
320            return Err(io::Error::new(
321                io::ErrorKind::InvalidData,
322                format!(
323                    "legacy journal '{}' frame {} declares {} bytes, exceeding the {} byte limit",
324                    path.display(),
325                    frame_index + 1,
326                    payload_len,
327                    MAX_FRAME_PAYLOAD_BYTES
328                ),
329            ));
330        }
331        if remaining.saturating_sub(4) < payload_len {
332            break;
333        }
334        let payload_len = usize::try_from(payload_len).expect("u32 fits usize");
335        let mut payload = vec![0_u8; payload_len];
336        source.read_exact(&mut payload)?;
337        let record = C::decode(&payload).map_err(|err| {
338            io::Error::new(
339                err.kind(),
340                format!(
341                    "legacy journal '{}' frame {} decode failed: {err}",
342                    path.display(),
343                    frame_index + 1
344                ),
345            )
346        })?;
347        writer.append::<C>(&migrate_path, &record)?;
348        valid_len = valid_len
349            .checked_add(4_u64.saturating_add(u64::try_from(payload_len).expect("usize fits u64")))
350            .ok_or_else(|| io::Error::new(io::ErrorKind::InvalidData, "journal offset overflow"))?;
351        frame_index = frame_index.saturating_add(1);
352    }
353    if frame_index == 0 {
354        return Err(io::Error::new(
355            io::ErrorKind::InvalidData,
356            format!(
357                "journal '{}' has neither the Ursula WAL header nor a complete legacy frame",
358                path.display()
359            ),
360        ));
361    }
362    if valid_len < file_len {
363        tracing::warn!(
364            path = %path.display(),
365            valid_bytes = valid_len,
366            discarded_torn_bytes = file_len.saturating_sub(valid_len),
367            "discarding a torn legacy WAL tail during format migration"
368        );
369    }
370    writer.sync(&migrate_path)?;
371    drop(writer);
372
373    if backup_path.exists() {
374        fs::remove_file(&backup_path)?;
375    }
376    fs::hard_link(path, &backup_path)?;
377    sync_parent(path)?;
378    fs::rename(&migrate_path, path)?;
379    sync_parent(path)?;
380    Ok(true)
381}
382
383fn suffixed_path(path: &Path, suffix: &str) -> PathBuf {
384    let mut name = path.as_os_str().to_owned();
385    name.push(suffix);
386    PathBuf::from(name)
387}
388
389fn sync_parent(path: &Path) -> io::Result<()> {
390    if let Some(parent) = path.parent() {
391        File::open(parent)?.sync_all()?;
392    }
393    Ok(())
394}
395
396/// Decode framed records from an in-memory buffer, returning the records and the byte
397/// length of the valid (fully-written) prefix. A torn trailing frame ends the scan.
398pub fn decode_frames<C: FrameCodec>(bytes: &[u8]) -> io::Result<(Vec<C::Record>, usize)> {
399    let mut records = Vec::new();
400    if bytes.is_empty() {
401        return Ok((records, 0));
402    }
403    if bytes.len() < JOURNAL_HEADER_LEN {
404        return Err(io::Error::new(
405            io::ErrorKind::InvalidData,
406            "in-memory journal has no complete Ursula WAL header",
407        ));
408    }
409    validate_header_bytes(&bytes[..JOURNAL_HEADER_LEN], "in-memory journal")?;
410    let mut offset = JOURNAL_HEADER_LEN;
411    let mut frame_index = 0_usize;
412    while offset < bytes.len() {
413        let Some(frame_header) = bytes.get(offset..offset.saturating_add(FRAME_HEADER_LEN)) else {
414            return Ok((records, offset)); // torn length prefix
415        };
416        let len = usize::try_from(u32::from_le_bytes(
417            frame_header[..4]
418                .try_into()
419                .expect("slice is exactly four bytes"),
420        ))
421        .expect("u32 fits usize");
422        if len > MAX_FRAME_PAYLOAD_BYTES {
423            return Err(io::Error::new(
424                io::ErrorKind::InvalidData,
425                format!(
426                    "in-memory journal frame {} declares {len} bytes, exceeding the {MAX_FRAME_PAYLOAD_BYTES} byte limit",
427                    frame_index + 1
428                ),
429            ));
430        }
431        let expected_checksum = u32::from_le_bytes(
432            frame_header[4..8]
433                .try_into()
434                .expect("slice is exactly four bytes"),
435        );
436        let start = offset.saturating_add(FRAME_HEADER_LEN);
437        let end = start.checked_add(len).ok_or_else(|| {
438            io::Error::new(io::ErrorKind::InvalidData, "journal frame length overflow")
439        })?;
440        let Some(payload) = bytes.get(start..end) else {
441            return Ok((records, offset)); // torn payload
442        };
443        let actual_checksum = crc32fast::hash(payload);
444        if actual_checksum != expected_checksum {
445            return Err(io::Error::new(
446                io::ErrorKind::InvalidData,
447                format!(
448                    "in-memory journal frame {} checksum mismatch: expected {expected_checksum:#010x}, got {actual_checksum:#010x}",
449                    frame_index + 1
450                ),
451            ));
452        }
453        records.push(C::decode(payload)?);
454        offset = end;
455        frame_index = frame_index.saturating_add(1);
456    }
457    Ok((records, bytes.len()))
458}
459
460fn validate_file_header(file: &mut File, path: &Path, file_len: u64) -> io::Result<()> {
461    if file_len < u64::try_from(JOURNAL_HEADER_LEN).expect("header length fits u64") {
462        return Err(io::Error::new(
463            io::ErrorKind::InvalidData,
464            format!("journal '{}' has a torn file header", path.display()),
465        ));
466    }
467    file.rewind()?;
468    let mut header = [0_u8; JOURNAL_HEADER_LEN];
469    file.read_exact(&mut header)?;
470    validate_header_bytes(&header, &format!("journal '{}'", path.display()))
471}
472
473fn validate_header_bytes(header: &[u8], description: &str) -> io::Result<()> {
474    if header.get(..JOURNAL_MAGIC.len()) != Some(JOURNAL_MAGIC.as_slice()) {
475        return Err(io::Error::new(
476            io::ErrorKind::InvalidData,
477            format!(
478                "{description} has no Ursula WAL magic; legacy unversioned journals require an explicit reset or migration"
479            ),
480        ));
481    }
482    let version = u16::from_le_bytes(
483        header[8..10]
484            .try_into()
485            .expect("validated journal header has version bytes"),
486    );
487    if version != JOURNAL_VERSION {
488        return Err(io::Error::new(
489            io::ErrorKind::InvalidData,
490            format!(
491                "{description} uses unsupported Ursula WAL version {version}; this binary supports version {JOURNAL_VERSION}"
492            ),
493        ));
494    }
495    let header_len = usize::from(u16::from_le_bytes(
496        header[10..12]
497            .try_into()
498            .expect("validated journal header has length bytes"),
499    ));
500    if header_len != JOURNAL_HEADER_LEN {
501        return Err(io::Error::new(
502            io::ErrorKind::InvalidData,
503            format!(
504                "{description} declares unsupported header length {header_len}; expected {JOURNAL_HEADER_LEN}"
505            ),
506        ));
507    }
508    if header[12..].iter().any(|byte| *byte != 0) {
509        return Err(io::Error::new(
510            io::ErrorKind::InvalidData,
511            format!("{description} has non-zero reserved header bytes"),
512        ));
513    }
514    Ok(())
515}
516
517/// Truncate `path` to `valid_len` bytes, dropping a torn trailing frame, then `fsync`.
518pub fn truncate_to(path: &Path, valid_len: usize) -> io::Result<()> {
519    let file = OpenOptions::new().write(true).open(path)?;
520    file.set_len(u64::try_from(valid_len).expect("valid frame offset fits u64"))?;
521    file.sync_data()
522}
523
524#[cfg(test)]
525mod tests {
526    use std::io::SeekFrom;
527
528    use super::*;
529
530    fn write_all(path: &Path, records: &[String]) {
531        let mut writer = JournalWriter::new(true);
532        for record in records {
533            writer
534                .append::<JsonCodec<String>>(path, record)
535                .expect("append record");
536        }
537        writer.sync(path).expect("sync journal");
538    }
539
540    #[test]
541    fn replays_appended_records_in_order() {
542        let dir = tempfile::tempdir().expect("temp dir");
543        let path = dir.path().join("journal");
544        let records = vec!["a".to_owned(), "bb".to_owned(), "ccc".to_owned()];
545        write_all(&path, &records);
546
547        let replayed = replay::<JsonCodec<String>>(&path).expect("replay");
548        assert_eq!(replayed, records);
549    }
550
551    #[test]
552    fn replay_of_missing_file_is_empty() {
553        let dir = tempfile::tempdir().expect("temp dir");
554        let path = dir.path().join("absent");
555        let replayed = replay::<JsonCodec<String>>(&path).expect("replay");
556        assert!(replayed.is_empty());
557    }
558
559    #[test]
560    fn append_reopens_and_extends_existing_journal() {
561        let dir = tempfile::tempdir().expect("temp dir");
562        let path = dir.path().join("journal");
563        write_all(&path, &["first".to_owned()]);
564        write_all(&path, &["second".to_owned()]);
565
566        let replayed = replay::<JsonCodec<String>>(&path).expect("replay");
567        assert_eq!(replayed, vec!["first".to_owned(), "second".to_owned()]);
568    }
569
570    #[test]
571    fn replay_truncates_a_torn_trailing_frame() {
572        let dir = tempfile::tempdir().expect("temp dir");
573        let path = dir.path().join("journal");
574        write_all(&path, &["clean".to_owned()]);
575
576        // Append a frame whose length header promises more bytes than follow.
577        let mut file = OpenOptions::new().append(true).open(&path).expect("reopen");
578        file.write_all(&64_u32.to_le_bytes()).expect("torn length");
579        file.write_all(b"torn").expect("torn payload");
580        file.sync_data().expect("sync torn tail");
581        let torn_len = fs::metadata(&path).expect("metadata").len();
582
583        let replayed = replay::<JsonCodec<String>>(&path).expect("replay");
584        assert_eq!(replayed, vec!["clean".to_owned()]);
585
586        // The torn tail was truncated away, so a re-read is clean and shorter.
587        let healed_len = fs::metadata(&path).expect("metadata").len();
588        assert!(healed_len < torn_len);
589        let reread = replay::<JsonCodec<String>>(&path).expect("re-replay");
590        assert_eq!(reread, vec!["clean".to_owned()]);
591    }
592
593    #[test]
594    fn replay_each_visits_records_without_collecting_them() {
595        let dir = tempfile::tempdir().expect("temp dir");
596        let path = dir.path().join("journal");
597        write_all(&path, &[
598            "first".to_owned(),
599            "second".to_owned(),
600            "third".to_owned(),
601        ]);
602
603        let mut replayed = Vec::new();
604        replay_each::<JsonCodec<String>>(&path, |record| {
605            replayed.push(record);
606            Ok(())
607        })
608        .expect("stream replay");
609
610        assert_eq!(replayed, vec!["first", "second", "third"]);
611    }
612
613    #[test]
614    fn replay_rejects_checksum_corruption() {
615        let dir = tempfile::tempdir().expect("temp dir");
616        let path = dir.path().join("journal");
617        write_all(&path, &["clean".to_owned()]);
618
619        let mut file = OpenOptions::new()
620            .read(true)
621            .write(true)
622            .open(&path)
623            .expect("open journal");
624        file.seek(SeekFrom::End(-1))
625            .expect("seek final payload byte");
626        file.write_all(b"x").expect("corrupt payload");
627        file.sync_data().expect("sync corruption");
628
629        let err = replay::<JsonCodec<String>>(&path).expect_err("checksum must fail closed");
630        assert_eq!(err.kind(), io::ErrorKind::InvalidData);
631        assert!(err.to_string().contains("frame 1 checksum mismatch"));
632    }
633
634    #[test]
635    fn replay_rejects_legacy_unversioned_journal() {
636        let dir = tempfile::tempdir().expect("temp dir");
637        let path = dir.path().join("journal");
638        let payload = serde_json::to_vec("legacy").expect("encode legacy payload");
639        let mut file = File::create(&path).expect("create legacy journal");
640        file.write_all(
641            &u32::try_from(payload.len())
642                .expect("payload length fits u32")
643                .to_le_bytes(),
644        )
645        .expect("write legacy length");
646        file.write_all(&payload).expect("write legacy payload");
647        file.sync_data().expect("sync legacy journal");
648
649        let err = replay::<JsonCodec<String>>(&path).expect_err("legacy format must be refused");
650        assert_eq!(err.kind(), io::ErrorKind::InvalidData);
651        assert!(err.to_string().contains("explicit reset or migration"));
652    }
653
654    #[test]
655    fn legacy_migration_preserves_records_and_a_rollback_copy() {
656        let dir = tempfile::tempdir().expect("temp dir");
657        let path = dir.path().join("journal");
658        let records = ["first".to_owned(), "second".to_owned()];
659        let mut legacy = File::create(&path).expect("create legacy journal");
660        for record in &records {
661            let payload = serde_json::to_vec(record).expect("encode legacy payload");
662            legacy
663                .write_all(
664                    &u32::try_from(payload.len())
665                        .expect("payload length fits u32")
666                        .to_le_bytes(),
667                )
668                .expect("write legacy length");
669            legacy.write_all(&payload).expect("write legacy payload");
670        }
671        legacy.sync_data().expect("sync legacy journal");
672        drop(legacy);
673
674        assert!(migrate_legacy::<JsonCodec<String>>(&path).expect("migrate legacy journal"));
675        assert_eq!(
676            replay::<JsonCodec<String>>(&path).expect("replay migrated journal"),
677            records
678        );
679        assert!(suffixed_path(&path, ".v0.bak").exists());
680        assert!(!migrate_legacy::<JsonCodec<String>>(&path).expect("migration is idempotent"));
681    }
682
683    #[test]
684    fn replay_rejects_unsupported_version() {
685        let dir = tempfile::tempdir().expect("temp dir");
686        let path = dir.path().join("journal");
687        write_all(&path, &["clean".to_owned()]);
688
689        let mut file = OpenOptions::new()
690            .write(true)
691            .open(&path)
692            .expect("open journal");
693        file.seek(SeekFrom::Start(8)).expect("seek version");
694        file.write_all(&2_u16.to_le_bytes())
695            .expect("write unsupported version");
696        file.sync_data().expect("sync unsupported version");
697
698        let err = replay::<JsonCodec<String>>(&path).expect_err("version must fail closed");
699        assert!(err.to_string().contains("unsupported Ursula WAL version 2"));
700    }
701
702    #[test]
703    fn replay_rejects_oversized_frame_before_allocating() {
704        let dir = tempfile::tempdir().expect("temp dir");
705        let path = dir.path().join("journal");
706        write_all(&path, &["clean".to_owned()]);
707
708        let mut file = OpenOptions::new()
709            .append(true)
710            .open(&path)
711            .expect("open journal");
712        let oversized =
713            u32::try_from(MAX_FRAME_PAYLOAD_BYTES + 1).expect("configured frame limit fits u32");
714        file.write_all(&oversized.to_le_bytes())
715            .expect("write oversized length");
716        file.write_all(&0_u32.to_le_bytes())
717            .expect("write placeholder checksum");
718        file.sync_data().expect("sync oversized frame");
719
720        let err = replay::<JsonCodec<String>>(&path).expect_err("oversized frame must fail closed");
721        assert!(
722            err.to_string()
723                .contains("exceeding the 536870912 byte limit")
724        );
725    }
726}
ursula_runtime/journal.rs

ursula_runtime/
journal.rs
