From cf17cfbdbd305fc81903fecd6029e3e7e292686e Mon Sep 17 00:00:00 2001 From: Windsor Date: Sat, 9 May 2026 14:02:56 -0700 Subject: [PATCH] feat(wal): add segmented write-ahead log --- Cargo.lock | 19 +++++ Cargo.toml | 5 +- cloud9-wal/Cargo.toml | 20 +++++ cloud9-wal/src/error.rs | 60 +++++++++++++++ cloud9-wal/src/format.rs | 141 +++++++++++++++++++++++++++++++++++ cloud9-wal/src/lib.rs | 25 +++++++ cloud9-wal/src/record.rs | 39 ++++++++++ cloud9-wal/src/segment.rs | 151 ++++++++++++++++++++++++++++++++++++++ cloud9-wal/src/tests.rs | 142 +++++++++++++++++++++++++++++++++++ cloud9-wal/src/wal.rs | 126 +++++++++++++++++++++++++++++++ 10 files changed, 727 insertions(+), 1 deletion(-) create mode 100644 cloud9-wal/Cargo.toml create mode 100644 cloud9-wal/src/error.rs create mode 100644 cloud9-wal/src/format.rs create mode 100644 cloud9-wal/src/lib.rs create mode 100644 cloud9-wal/src/record.rs create mode 100644 cloud9-wal/src/segment.rs create mode 100644 cloud9-wal/src/tests.rs create mode 100644 cloud9-wal/src/wal.rs diff --git a/Cargo.lock b/Cargo.lock index 5caa148..cb675d6 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -261,12 +261,31 @@ dependencies = [ "tracing", ] +[[package]] +name = "cloud9-wal" +version = "0.0.1" +dependencies = [ + "crc32fast", + "fs-err", + "tempfile", + "thiserror", +] + [[package]] name = "colorchoice" version = "1.0.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b05b61dc5112cbb17e4b6cd61790d9845d13888356391624cbe7e41efeac1e75" +[[package]] +name = "crc32fast" +version = "1.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9481c1c90cbf2ac953f07c8d4a58aa3945c425b7185c9154d67a65e4230da511" +dependencies = [ + "cfg-if", +] + [[package]] name = "deranged" version = "0.5.5" diff --git a/Cargo.toml b/Cargo.toml index dcd0abe..e96df29 100644 --- a/Cargo.toml +++ b/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" diff --git a/cloud9-wal/Cargo.toml b/cloud9-wal/Cargo.toml new file mode 100644 index 0000000..9ec18e9 --- /dev/null +++ b/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 } diff --git a/cloud9-wal/src/error.rs b/cloud9-wal/src/error.rs new file mode 100644 index 0000000..0e8d137 --- /dev/null +++ b/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 = std::result::Result; diff --git a/cloud9-wal/src/format.rs b/cloud9-wal/src/format.rs new file mode 100644 index 0000000..e9ca9bc --- /dev/null +++ b/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> { + 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 { + 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 { + 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 { + 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) +} diff --git a/cloud9-wal/src/lib.rs b/cloud9-wal/src/lib.rs new file mode 100644 index 0000000..6a159f1 --- /dev/null +++ b/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}; diff --git a/cloud9-wal/src/record.rs b/cloud9-wal/src/record.rs new file mode 100644 index 0000000..0da7fb4 --- /dev/null +++ b/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 { + 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, +} + +/// A record with its log position. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct StoredRecord { + pub lsn: Lsn, + pub record: Record, +} diff --git a/cloud9-wal/src/segment.rs b/cloud9-wal/src/segment.rs new file mode 100644 index 0000000..5fc0854 --- /dev/null +++ b/cloud9-wal/src/segment.rs @@ -0,0 +1,151 @@ +use std::path::{Path, PathBuf}; + +use fs_err::{self as fs, File, OpenOptions}; + +use crate::error::{Corruption, Result, WalError}; +use crate::format::{self, ReadOne}; +use crate::record::{Lsn, StoredRecord}; + +const SEGMENT_SUFFIX: &str = "wal"; + +pub(crate) fn recover(dir: &Path, segment_size: u64) -> Result<(u64, u64)> { + let mut ids = segment_ids(dir)?; + if ids.is_empty() { + let segment_id = 0; + File::create(segment_path(dir, segment_id)) + .map_err(|source| WalError::io(segment_path(dir, segment_id), source))?; + sync_dir(dir)?; + ids.push(segment_id); + } + + for (position, segment_id) in ids.iter().copied().enumerate() { + match valid_len(dir, segment_id, segment_size)? { + SegmentScan::Clean(len) => { + if position + 1 == ids.len() { + return Ok((segment_id, len)); + } + } + SegmentScan::IncompleteTail(len) => { + truncate_segment(dir, segment_id, len)?; + remove_segments_after(dir, &ids, position)?; + sync_dir(dir)?; + return Ok((segment_id, len)); + } + } + } + let last = ids.last().copied().ok_or(WalError::SegmentIdExhausted)?; + Ok((last, segment_len(dir, last)?)) +} + +pub(crate) fn read_segment( + dir: &Path, + segment_id: u64, + segment_size: u64, + records: &mut Vec, +) -> Result<()> { + let path = segment_path(dir, segment_id); + let mut file = File::open(&path).map_err(|source| WalError::io(path.clone(), source))?; + let mut offset = 0; + loop { + match format::read_one(&path, &mut file, Lsn { segment_id, offset }, segment_size)? { + ReadOne::Record { record, next_offset } => { + records.push(StoredRecord { lsn: Lsn { segment_id, offset }, record }); + offset = next_offset; + } + ReadOne::Eof => return Ok(()), + ReadOne::Incomplete => { + return Err(WalError::CorruptRecord { + lsn: Lsn { segment_id, offset }, + reason: Corruption::IncompleteRecord, + }); + } + } + } +} + +pub(crate) fn segment_ids(dir: &Path) -> Result> { + let mut ids = Vec::new(); + for entry in fs::read_dir(dir).map_err(|source| WalError::io(dir.to_path_buf(), source))? { + let entry = entry.map_err(|source| WalError::io(dir.to_path_buf(), source))?; + let path = entry.path(); + if path.extension().and_then(std::ffi::OsStr::to_str) == Some(SEGMENT_SUFFIX) { + ids.push(segment_id(&path)?); + } + } + ids.sort_unstable(); + for (expected, found) in ids.iter().copied().enumerate() { + let expected = u64::try_from(expected).map_err(|_| WalError::SegmentIdExhausted)?; + if found != expected { + return Err(WalError::MissingSegment { expected, found }); + } + } + Ok(ids) +} + +pub(crate) fn segment_path(dir: &Path, segment_id: u64) -> PathBuf { + dir.join(format!("{segment_id:020}.wal")) +} + +pub(crate) fn open_segment(dir: &Path, segment_id: u64) -> Result { + let path = segment_path(dir, segment_id); + OpenOptions::new() + .read(true) + .append(true) + .create(true) + .open(&path) + .map_err(|source| WalError::io(path, source)) +} + +pub(crate) fn sync_dir(dir: &Path) -> Result<()> { + File::open(dir) + .and_then(|file| file.sync_all()) + .map_err(|source| WalError::io(dir.to_path_buf(), source)) +} + +enum SegmentScan { + Clean(u64), + IncompleteTail(u64), +} + +fn valid_len(dir: &Path, segment_id: u64, segment_size: u64) -> Result { + let path = segment_path(dir, segment_id); + let mut file = File::open(&path).map_err(|source| WalError::io(path.clone(), source))?; + let mut offset = 0; + loop { + match format::read_one(&path, &mut file, Lsn { segment_id, offset }, segment_size)? { + ReadOne::Record { next_offset, .. } => offset = next_offset, + ReadOne::Eof => return Ok(SegmentScan::Clean(offset)), + ReadOne::Incomplete => return Ok(SegmentScan::IncompleteTail(offset)), + } + } +} + +fn segment_id(path: &Path) -> Result { + let Some(stem) = path.file_stem().and_then(std::ffi::OsStr::to_str) else { + return Err(WalError::BadSegmentName { path: path.to_path_buf() }); + }; + stem.parse::().map_err(|_| WalError::BadSegmentName { path: path.to_path_buf() }) +} + +fn segment_len(dir: &Path, segment_id: u64) -> Result { + let path = segment_path(dir, segment_id); + fs::metadata(&path).map_err(|source| WalError::io(path, source)).map(|meta| meta.len()) +} + +fn truncate_segment(dir: &Path, segment_id: u64, len: u64) -> Result<()> { + let path = segment_path(dir, segment_id); + let file = OpenOptions::new() + .write(true) + .open(&path) + .map_err(|source| WalError::io(path.clone(), source))?; + file.set_len(len).map_err(|source| WalError::io(path.clone(), source))?; + file.sync_all().map_err(|source| WalError::io(path, source)) +} + +fn remove_segments_after(dir: &Path, ids: &[u64], position: usize) -> Result<()> { + for segment_id in &ids[position + 1..] { + let path = segment_path(dir, *segment_id); + fs::remove_file(&path).map_err(|source| WalError::io(path, source))?; + } + Ok(()) +} diff --git a/cloud9-wal/src/tests.rs b/cloud9-wal/src/tests.rs new file mode 100644 index 0000000..4be7b59 --- /dev/null +++ b/cloud9-wal/src/tests.rs @@ -0,0 +1,142 @@ +use std::io::{Seek, SeekFrom, Write}; + +use fs_err::{self as fs, File, OpenOptions}; + +use crate::segment::segment_path; +use crate::*; + +fn kind(value: u16) -> RecordKind { + RecordKind::new(value).unwrap() +} + +fn payload(value: u8) -> Vec { + vec![value; 8] +} + +#[test] +fn append_and_recover_records() { + let dir = tempfile::tempdir().unwrap(); + let mut wal = Wal::open(dir.path(), WalOptions::default()).unwrap(); + + let first = wal.append(kind(1), payload(1)).unwrap(); + let second = wal.append(kind(2), payload(2)).unwrap(); + wal.sync().unwrap(); + drop(wal); + + let wal = Wal::open(dir.path(), WalOptions::default()).unwrap(); + let records = wal.records().unwrap(); + + assert_eq!(records.len(), 2); + assert_eq!(records[0].lsn, first); + assert_eq!(records[0].record.kind, kind(1)); + assert_eq!(records[1].lsn, second); + assert_eq!(records[1].record.payload, payload(2)); +} + +#[test] +fn rotates_segments() { + let dir = tempfile::tempdir().unwrap(); + let options = WalOptions { segment_size: 96, sync_on_append: false }; + let mut wal = Wal::open(dir.path(), options.clone()).unwrap(); + + let first = wal.append(kind(1), payload(1)).unwrap(); + let second = wal.append(kind(1), payload(2)).unwrap(); + let third = wal.append(kind(1), payload(3)).unwrap(); + wal.sync().unwrap(); + drop(wal); + + assert_eq!(first.segment_id, 0); + assert_eq!(second.segment_id, 0); + assert_eq!(third.segment_id, 1); + + let wal = Wal::open(dir.path(), options).unwrap(); + let records = wal.records().unwrap(); + assert_eq!(records.iter().map(|r| r.record.payload[0]).collect::>(), [1, 2, 3]); +} + +#[test] +fn recovery_truncates_incomplete_tail() { + let dir = tempfile::tempdir().unwrap(); + let mut wal = Wal::open(dir.path(), WalOptions::default()).unwrap(); + wal.append(kind(1), payload(1)).unwrap(); + wal.sync().unwrap(); + drop(wal); + + let segment = segment_path(dir.path(), 0); + let mut file = OpenOptions::new().append(true).open(&segment).unwrap(); + file.write_all(&[1, 2, 3]).unwrap(); + file.sync_all().unwrap(); + drop(file); + + let wal = Wal::open(dir.path(), WalOptions::default()).unwrap(); + let records = wal.records().unwrap(); + + assert_eq!(records.len(), 1); + assert_eq!(records[0].record.payload, payload(1)); + assert_eq!(fs::metadata(segment).unwrap().len(), crate::format::HEADER_LEN_U64 + 8); +} + +#[test] +fn recovery_rejects_checksum_corruption() { + let dir = tempfile::tempdir().unwrap(); + let mut wal = Wal::open(dir.path(), WalOptions::default()).unwrap(); + wal.append(kind(1), payload(1)).unwrap(); + wal.sync().unwrap(); + drop(wal); + + let segment = segment_path(dir.path(), 0); + let mut file = OpenOptions::new().read(true).write(true).open(segment).unwrap(); + file.seek(SeekFrom::Start(crate::format::HEADER_LEN_U64)).unwrap(); + file.write_all(&[9]).unwrap(); + file.sync_all().unwrap(); + drop(file); + + let err = Wal::open(dir.path(), WalOptions::default()).unwrap_err(); + assert!(matches!(err, WalError::CorruptRecord { reason: Corruption::PayloadChecksum, .. })); +} + +#[test] +fn recovery_rejects_missing_segment() { + let dir = tempfile::tempdir().unwrap(); + File::create(segment_path(dir.path(), 0)).unwrap(); + File::create(segment_path(dir.path(), 2)).unwrap(); + + let err = Wal::open(dir.path(), WalOptions::default()).unwrap_err(); + assert!(matches!(err, WalError::MissingSegment { expected: 1, found: 2 })); +} + +#[test] +fn recovery_rejects_record_larger_than_segment() { + let dir = tempfile::tempdir().unwrap(); + let mut wal = + Wal::open(dir.path(), WalOptions { segment_size: 40, sync_on_append: false }).unwrap(); + wal.append(kind(1), payload(1)).unwrap(); + wal.sync().unwrap(); + drop(wal); + + let err = + Wal::open(dir.path(), WalOptions { segment_size: 39, sync_on_append: false }).unwrap_err(); + assert!(matches!( + err, + WalError::CorruptRecord { + reason: Corruption::RecordTooLarge { len: 40, segment_size: 39 }, + .. + } + )); +} + +#[test] +fn rejects_reserved_kind() { + let err = RecordKind::new(0).unwrap_err(); + assert!(matches!(err, WalError::ReservedRecordKind)); +} + +#[test] +fn rejects_records_larger_than_segment() { + let dir = tempfile::tempdir().unwrap(); + let mut wal = + Wal::open(dir.path(), WalOptions { segment_size: 39, sync_on_append: false }).unwrap(); + + let err = wal.append(kind(1), payload(1)).unwrap_err(); + assert!(matches!(err, WalError::RecordTooLarge { .. })); +} diff --git a/cloud9-wal/src/wal.rs b/cloud9-wal/src/wal.rs new file mode 100644 index 0000000..78455a0 --- /dev/null +++ b/cloud9-wal/src/wal.rs @@ -0,0 +1,126 @@ +use std::io::{Seek, SeekFrom, Write}; +use std::path::{Path, PathBuf}; + +use fs_err::{self as fs, File}; + +use crate::error::{Result, WalError}; +use crate::format; +use crate::record::{Lsn, RecordKind, StoredRecord}; +use crate::segment; + +const DEFAULT_SEGMENT_SIZE: u64 = 64 * 1024 * 1024; + +/// WAL configuration. +#[derive(Debug, Clone)] +pub struct WalOptions { + pub segment_size: u64, + pub sync_on_append: bool, +} + +impl Default for WalOptions { + fn default() -> Self { + Self { segment_size: DEFAULT_SEGMENT_SIZE, sync_on_append: false } + } +} + +/// Append-only segmented WAL. +#[derive(Debug)] +pub struct Wal { + dir: PathBuf, + options: WalOptions, + active_id: u64, + active_len: u64, + active: File, +} + +impl Wal { + /// Open or create a WAL directory. + pub fn open(dir: impl AsRef, options: WalOptions) -> Result { + validate_options(&options)?; + let dir = dir.as_ref().to_path_buf(); + fs::create_dir_all(&dir).map_err(|source| WalError::io(dir.clone(), source))?; + let (active_id, active_len) = segment::recover(&dir, options.segment_size)?; + let active = segment::open_segment(&dir, active_id)?; + let mut wal = Self { dir, options, active_id, active_len, active }; + wal.active.seek(SeekFrom::Start(active_len)).map_err(|source| { + WalError::io(segment::segment_path(&wal.dir, wal.active_id), source) + })?; + Ok(wal) + } + + /// Append a record and return its LSN. + pub fn append(&mut self, kind: RecordKind, payload: impl AsRef<[u8]>) -> Result { + let encoded = format::encode_record(kind, payload.as_ref())?; + self.rotate_if_needed(encoded.len())?; + let lsn = Lsn { segment_id: self.active_id, offset: self.active_len }; + self.active.write_all(&encoded).map_err(|source| { + WalError::io(segment::segment_path(&self.dir, self.active_id), source) + })?; + self.active_len = checked_add_len(self.active_len, encoded.len())?; + if self.options.sync_on_append { + self.sync()?; + } + Ok(lsn) + } + + /// Force pending WAL bytes to stable storage. + pub fn sync(&mut self) -> Result<()> { + self.active.sync_all().map_err(|source| { + WalError::io(segment::segment_path(&self.dir, self.active_id), source) + }) + } + + /// Read every valid record in LSN order. + pub fn records(&self) -> Result> { + let mut records = Vec::new(); + for segment_id in segment::segment_ids(&self.dir)? { + segment::read_segment(&self.dir, segment_id, self.options.segment_size, &mut records)?; + } + Ok(records) + } + + fn rotate_if_needed(&mut self, record_len: usize) -> Result<()> { + let record_len = u64::try_from(record_len).map_err(|_| WalError::RecordTooLarge { + len: u64::MAX, + segment_size: self.options.segment_size, + })?; + if record_len > self.options.segment_size { + return Err(WalError::RecordTooLarge { + len: record_len, + segment_size: self.options.segment_size, + }); + } + let next_len = self.active_len.checked_add(record_len).ok_or(WalError::RecordTooLarge { + len: u64::MAX, + segment_size: self.options.segment_size, + })?; + if self.active_len != 0 && next_len > self.options.segment_size { + self.sync()?; + self.active_id = self.active_id.checked_add(1).ok_or(WalError::SegmentIdExhausted)?; + self.active = segment::open_segment(&self.dir, self.active_id)?; + self.active_len = 0; + segment::sync_dir(&self.dir)?; + } + Ok(()) + } +} + +fn validate_options(options: &WalOptions) -> Result<()> { + if options.segment_size < format::HEADER_LEN_U64 { + Err(WalError::SegmentTooSmall { + segment_size: options.segment_size, + header_len: format::HEADER_LEN, + }) + } else { + Ok(()) + } +} + +fn checked_add_len(offset: u64, len: usize) -> Result { + offset + .checked_add( + u64::try_from(len) + .map_err(|_| WalError::RecordTooLarge { len: u64::MAX, segment_size: u64::MAX })?, + ) + .ok_or(WalError::RecordTooLarge { len: u64::MAX, segment_size: u64::MAX }) +}