feat(wal): add segmented write-ahead log

Cargo.toml

@@ -1,5 +1,5 @@
 [workspace]
-members = ["cloud9", "cloud9-core", "cloud9-node", "cloud9-proto", "cloud9-storage", "consensus/cloud9-raft", "consensus/cloud9-raft-io"]
+members = ["cloud9", "cloud9-core", "cloud9-node", "cloud9-proto", "cloud9-storage", "cloud9-wal", "consensus/cloud9-raft", "consensus/cloud9-raft-io"]
 resolver = "2"
 
 [workspace.package]
@@ -45,6 +45,7 @@ unimplemented = "warn"
 # Internal crates
 cloud9-core = { path = "cloud9-core" }
 cloud9-storage = { path = "cloud9-storage" }
+cloud9-wal = { path = "cloud9-wal" }
 cloud9-raft = { path = "consensus/cloud9-raft" }
 cloud9-raft-io = { path = "consensus/cloud9-raft-io" }
 cloud9-proto = { path = "cloud9-proto" }
@@ -64,6 +65,7 @@ arcstr = { version = "1.2", features = ["serde"] }
 fs-err = { version = "3", features = ["tokio"] }
 miette = { version = "7.2", features = ["fancy-no-backtrace"] }
 textwrap = "0.16"
+crc32fast = "1.5"
 
 # Testing - concurrency
 loom = "0.7.2"
@@ -74,3 +76,4 @@ proptest-state-machine = "0.6"
 
 # Testing - concurrency (randomized)
 shuttle = "0.8"
+tempfile = "3.20"

cloud9-wal/Cargo.toml

@@ -0,0 +1,20 @@
+[package]
+name = "cloud9-wal"
+version = { workspace = true }
+edition = { workspace = true }
+rust-version = { workspace = true }
+license = { workspace = true }
+authors = { workspace = true }
+repository = { workspace = true }
+homepage = { workspace = true }
+
+[lints]
+workspace = true
+
+[dependencies]
+crc32fast = { workspace = true }
+fs-err = { workspace = true }
+thiserror = { workspace = true }
+
+[dev-dependencies]
+tempfile = { workspace = true }

cloud9-wal/src/error.rs

@@ -0,0 +1,60 @@
+use std::io;
+use std::path::PathBuf;
+
+use thiserror::Error;
+
+use crate::record::Lsn;
+
+/// WAL failures.
+#[derive(Debug, Error)]
+pub enum WalError {
+    #[error("I/O error at `{path}`")]
+    Io {
+        path: PathBuf,
+        #[source]
+        source: io::Error,
+    },
+    #[error("record kind zero is reserved")]
+    ReservedRecordKind,
+    #[error("segment size {segment_size} is smaller than record header {header_len}")]
+    SegmentTooSmall { segment_size: u64, header_len: usize },
+    #[error("record length {len} exceeds segment size {segment_size}")]
+    RecordTooLarge { len: u64, segment_size: u64 },
+    #[error("payload length {len} exceeds u32::MAX")]
+    PayloadTooLarge { len: usize },
+    #[error("segment id exhausted")]
+    SegmentIdExhausted,
+    #[error("corrupt WAL record at segment {lsn:?}: {reason}")]
+    CorruptRecord { lsn: Lsn, reason: Corruption },
+    #[error("malformed WAL segment filename `{path}`")]
+    BadSegmentName { path: PathBuf },
+    #[error("missing WAL segment {expected:020}.wal before {found:020}.wal")]
+    MissingSegment { expected: u64, found: u64 },
+}
+
+impl WalError {
+    pub(crate) fn io(path: PathBuf, source: io::Error) -> Self {
+        Self::Io { path, source }
+    }
+}
+
+/// Specific corruption detected while scanning records.
+#[derive(Debug, Error)]
+pub enum Corruption {
+    #[error("bad magic {found:#x}")]
+    BadMagic { found: u32 },
+    #[error("unsupported version {found}")]
+    UnsupportedVersion { found: u16 },
+    #[error("reserved record kind")]
+    ReservedRecordKind,
+    #[error("header checksum mismatch")]
+    HeaderChecksum,
+    #[error("payload checksum mismatch")]
+    PayloadChecksum,
+    #[error("record length {len} exceeds segment size {segment_size}")]
+    RecordTooLarge { len: u64, segment_size: u64 },
+    #[error("incomplete record")]
+    IncompleteRecord,
+}
+
+pub type Result<T> = std::result::Result<T, WalError>;

cloud9-wal/src/format.rs

@@ -0,0 +1,141 @@
+use std::io::{ErrorKind, Read};
+use std::path::Path;
+
+use crc32fast::Hasher;
+
+use crate::error::{Corruption, Result, WalError};
+use crate::record::{Lsn, Record, RecordKind};
+
+pub(crate) const MAGIC: u32 = 0x4339_574c; // C9WL
+pub(crate) const VERSION: u16 = 1;
+pub(crate) const HEADER_LEN: usize = 32;
+pub(crate) const HEADER_LEN_U64: u64 = 32;
+const HEADER_CRC_END: usize = 28;
+
+pub(crate) enum ReadOne {
+    Record { record: Record, next_offset: u64 },
+    Eof,
+    Incomplete,
+}
+
+pub(crate) fn encode_record(kind: RecordKind, payload: &[u8]) -> Result<Vec<u8>> {
+    let payload_len = u32::try_from(payload.len())
+        .map_err(|_| WalError::PayloadTooLarge { len: payload.len() })?;
+    let mut header = [0_u8; HEADER_LEN];
+    header[0..4].copy_from_slice(&MAGIC.to_le_bytes());
+    header[4..6].copy_from_slice(&VERSION.to_le_bytes());
+    header[6..8].copy_from_slice(&kind.get().to_le_bytes());
+    header[8..12].copy_from_slice(&payload_len.to_le_bytes());
+    header[12..16].copy_from_slice(&crc32(payload).to_le_bytes());
+    let header_crc = crc32(&header[..HEADER_CRC_END]);
+    header[28..32].copy_from_slice(&header_crc.to_le_bytes());
+
+    let mut encoded = Vec::with_capacity(HEADER_LEN + payload.len());
+    encoded.extend_from_slice(&header);
+    encoded.extend_from_slice(payload);
+    Ok(encoded)
+}
+
+pub(crate) fn read_one(
+    path: &Path,
+    reader: &mut impl Read,
+    lsn: Lsn,
+    segment_size: u64,
+) -> Result<ReadOne> {
+    let mut header = [0_u8; HEADER_LEN];
+    match read_exact_or_tail(path, reader, &mut header)? {
+        ReadExact::Eof => return Ok(ReadOne::Eof),
+        ReadExact::Incomplete => return Ok(ReadOne::Incomplete),
+        ReadExact::Complete => {}
+    }
+
+    let kind = decode_header(&header, lsn)?;
+    let payload_len = u64::from(read_u32(&header, 8));
+    let record_len = HEADER_LEN_U64.checked_add(payload_len).ok_or(WalError::CorruptRecord {
+        lsn,
+        reason: Corruption::RecordTooLarge { len: u64::MAX, segment_size },
+    })?;
+    if record_len > segment_size {
+        return Err(WalError::CorruptRecord {
+            lsn,
+            reason: Corruption::RecordTooLarge { len: record_len, segment_size },
+        });
+    }
+    let payload_len = usize::try_from(payload_len).map_err(|_| WalError::CorruptRecord {
+        lsn,
+        reason: Corruption::RecordTooLarge { len: record_len, segment_size },
+    })?;
+
+    let mut payload = vec![0; payload_len];
+    if matches!(
+        read_exact_or_tail(path, reader, &mut payload)?,
+        ReadExact::Incomplete | ReadExact::Eof
+    ) {
+        return Ok(ReadOne::Incomplete);
+    }
+    if crc32(&payload) != read_u32(&header, 12) {
+        return Err(WalError::CorruptRecord { lsn, reason: Corruption::PayloadChecksum });
+    }
+    let next_offset = lsn.offset.checked_add(record_len).ok_or(WalError::CorruptRecord {
+        lsn,
+        reason: Corruption::RecordTooLarge { len: u64::MAX, segment_size },
+    })?;
+    Ok(ReadOne::Record { record: Record { kind, payload }, next_offset })
+}
+
+enum ReadExact {
+    Complete,
+    Eof,
+    Incomplete,
+}
+
+fn read_exact_or_tail(path: &Path, reader: &mut impl Read, buf: &mut [u8]) -> Result<ReadExact> {
+    let mut filled = 0;
+    while filled < buf.len() {
+        match reader.read(&mut buf[filled..]) {
+            Ok(0) if filled == 0 => return Ok(ReadExact::Eof),
+            Ok(0) => return Ok(ReadExact::Incomplete),
+            Ok(n) => filled += n,
+            Err(source) if source.kind() == ErrorKind::Interrupted => {}
+            Err(source) => return Err(WalError::io(path.to_path_buf(), source)),
+        }
+    }
+    Ok(ReadExact::Complete)
+}
+
+fn decode_header(header: &[u8; HEADER_LEN], lsn: Lsn) -> Result<RecordKind> {
+    let magic = read_u32(header, 0);
+    if magic != MAGIC {
+        return Err(WalError::CorruptRecord { lsn, reason: Corruption::BadMagic { found: magic } });
+    }
+    let version = read_u16(header, 4);
+    if version != VERSION {
+        return Err(WalError::CorruptRecord {
+            lsn,
+            reason: Corruption::UnsupportedVersion { found: version },
+        });
+    }
+    if crc32(&header[..HEADER_CRC_END]) != read_u32(header, 28) {
+        return Err(WalError::CorruptRecord { lsn, reason: Corruption::HeaderChecksum });
+    }
+    RecordKind::new(read_u16(header, 6))
+        .map_err(|_| WalError::CorruptRecord { lsn, reason: Corruption::ReservedRecordKind })
+}
+
+fn crc32(bytes: &[u8]) -> u32 {
+    let mut hasher = Hasher::new();
+    hasher.update(bytes);
+    hasher.finalize()
+}
+
+fn read_u16(bytes: &[u8], start: usize) -> u16 {
+    let mut out = [0; 2];
+    out.copy_from_slice(&bytes[start..start + 2]);
+    u16::from_le_bytes(out)
+}
+
+fn read_u32(bytes: &[u8], start: usize) -> u32 {
+    let mut out = [0; 4];
+    out.copy_from_slice(&bytes[start..start + 4]);
+    u32::from_le_bytes(out)
+}

cloud9-wal/src/lib.rs

@@ -0,0 +1,25 @@
+#![forbid(unsafe_code)]
+#![deny(clippy::unwrap_used, clippy::expect_used)]
+#![cfg_attr(test, allow(clippy::unwrap_used, clippy::expect_used, clippy::panic))]
+
+//! Tiny segmented write-ahead log.
+//!
+//! The WAL deliberately owns only the byte-durability problem:
+//! 1. append one typed byte record to the active segment,
+//! 2. sync the active segment when the caller asks for durability,
+//! 3. recover by scanning segment files and truncating only an incomplete tail.
+//!
+//! Higher layers own meaning. Raft entries, hard state, logical truncation, and
+//! snapshots are just payloads encoded by the caller.
+
+mod error;
+mod format;
+mod record;
+mod segment;
+#[cfg(test)]
+mod tests;
+mod wal;
+
+pub use error::{Corruption, Result, WalError};
+pub use record::{Lsn, Record, RecordKind, StoredRecord};
+pub use wal::{Wal, WalOptions};

cloud9-wal/src/record.rs

@@ -0,0 +1,39 @@
+use crate::{Result, WalError};
+
+/// Position of a WAL record.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)]
+pub struct Lsn {
+    pub segment_id: u64,
+    pub offset: u64,
+}
+
+/// Caller-owned record kind. Kind zero is reserved for invalid headers.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub struct RecordKind(u16);
+
+impl RecordKind {
+    /// Create a non-zero record kind.
+    pub fn new(value: u16) -> Result<Self> {
+        if value == 0 { Err(WalError::ReservedRecordKind) } else { Ok(Self(value)) }
+    }
+
+    /// Return the on-disk kind value.
+    #[must_use]
+    pub const fn get(self) -> u16 {
+        self.0
+    }
+}
+
+/// One logical WAL record.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct Record {
+    pub kind: RecordKind,
+    pub payload: Vec<u8>,
+}
+
+/// A record with its log position.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct StoredRecord {
+    pub lsn: Lsn,
+    pub record: Record,
+}