Crash-Safe Write-Ahead Log With CRC-Checked Replay in Rust

use std::fs::{File, OpenOptions};
use std::io::{self, BufWriter, Write};
use std::path::{Path, PathBuf};

use serde::{Deserialize, Serialize};

use crate::record::Record;

#[derive(Debug, Serialize, Deserialize)]
pub enum Entry {
    Set { key: String, value: Vec<u8> },
    Delete { key: String },
}

pub struct Wal {
    path: PathBuf,
    writer: BufWriter<File>,
}

impl Wal {
    pub fn append(&mut self, entry: &Entry) -> io::Result<()> {
        let payload = bincode::serialize(entry)
            .map_err(|e| io::Error::new(io::ErrorKind::InvalidData, e))?;
        Record::encode(&mut self.writer, &payload)?;
        self.writer.flush()?;
        // Force bytes to stable storage before returning to the caller.
        self.writer.get_ref().sync_data()?;
        Ok(())
    }

    pub fn path(&self) -> &Path {
        &self.path
    }

    fn from_file(path: PathBuf, file: File) -> Self {
        Wal {
            path,
            writer: BufWriter::new(file),
        }
    }

    pub(crate) fn open_append(path: &Path) -> io::Result<File> {
        OpenOptions::new().create(true).append(true).open(path)
    }
}

use std::fs::{File, OpenOptions};
use std::io::{self, BufReader, Seek, SeekFrom};
use std::path::PathBuf;

use crate::record::{DecodeError, Record};
use crate::wal::{Entry, Wal};

impl Wal {
    pub fn open<F>(path: PathBuf, mut apply: F) -> io::Result<Wal>
    where
        F: FnMut(Entry),
    {
        let read_file = OpenOptions::new().read(true).open(&path).or_else(|e| {
            if e.kind() == io::ErrorKind::NotFound {
                File::create(&path)
            } else {
                Err(e)
            }
        })?;

        let mut reader = BufReader::new(read_file);
        let mut good_offset: u64 = 0;

        loop {
            match Record::decode(&mut reader) {
                Ok(payload) => match bincode::deserialize::<Entry>(&payload) {
                    Ok(entry) => {
                        apply(entry);
                        good_offset = reader.stream_position()?;
                    }
                    Err(_) => break, // undecodable record: stop and drop tail
                },
                Err(DecodeError::UnexpectedEof) | Err(DecodeError::ChecksumMismatch) => break,
                Err(DecodeError::Io(e)) => return Err(e),
            }
        }

        let append_file = Wal::open_append(&path)?;
        // Discard any torn trailing bytes left by a crash mid-append.
        append_file.set_len(good_offset)?;
        append_file.seek(SeekFrom::End(0)).ok();

        Ok(Wal::from_file(path, append_file))
    }
}