use std::collections::HashMap;
use thiserror::Error;
#[inline]
pub fn fnv1a_64(data: &[u8]) -> u64 {
let mut h: u64 = 14_695_981_039_346_656_037;
for &b in data {
h ^= b as u64;
h = h.wrapping_mul(1_099_511_628_211);
}
h
}
pub const WAL_MAGIC: [u8; 4] = *b"WALX";
const MIN_ENTRY_BYTES: usize = 45;
#[derive(Debug, Error, PartialEq, Eq, Clone)]
pub enum WalError {
#[error("checksum mismatch for entry {entry_seq}: expected {expected:#018x}, got {got:#018x}")]
ChecksumMismatch {
entry_seq: u64,
expected: u64,
got: u64,
},
#[error("corrupted entry at seq {0}")]
CorruptedEntry(u64),
#[error("transaction {0} not found")]
TransactionNotFound(u64),
#[error("entry too large: {0} bytes")]
EntryTooLarge(usize),
#[error("segment full")]
SegmentFull,
#[error("invalid WAL magic bytes")]
InvalidMagic,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
#[repr(u8)]
pub enum WalOpType {
Put = 1,
Delete = 2,
Begin = 3,
Commit = 4,
Rollback = 5,
Checkpoint = 6,
}
impl WalOpType {
fn from_u8(v: u8) -> Option<Self> {
match v {
1 => Some(Self::Put),
2 => Some(Self::Delete),
3 => Some(Self::Begin),
4 => Some(Self::Commit),
5 => Some(Self::Rollback),
6 => Some(Self::Checkpoint),
_ => None,
}
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct WalEntry {
pub seq_num: u64,
pub tx_id: u64,
pub op_type: WalOpType,
pub key: Vec<u8>,
pub value: Vec<u8>,
pub timestamp: u64,
pub checksum: u64,
}
impl WalEntry {
pub fn encode(&self) -> Vec<u8> {
let total = 4 + 8 + 8 + 1 + 4 + self.key.len() + 4 + self.value.len() + 8 + 8;
let mut buf = Vec::with_capacity(total);
buf.extend_from_slice(&WAL_MAGIC);
buf.extend_from_slice(&self.seq_num.to_le_bytes());
buf.extend_from_slice(&self.tx_id.to_le_bytes());
buf.push(self.op_type as u8);
buf.extend_from_slice(&(self.key.len() as u32).to_le_bytes());
buf.extend_from_slice(&self.key);
buf.extend_from_slice(&(self.value.len() as u32).to_le_bytes());
buf.extend_from_slice(&self.value);
buf.extend_from_slice(&self.timestamp.to_le_bytes());
let checksum = fnv1a_64(&buf);
buf.extend_from_slice(&checksum.to_le_bytes());
buf
}
pub fn decode(data: &[u8], offset: usize) -> Result<(WalEntry, usize), WalError> {
let buf = &data[offset..];
if buf.len() < MIN_ENTRY_BYTES {
return Err(WalError::CorruptedEntry(0));
}
if buf[..4] != WAL_MAGIC {
return Err(WalError::InvalidMagic);
}
let seq_num = u64::from_le_bytes(
buf[4..12]
.try_into()
.map_err(|_| WalError::CorruptedEntry(0))?,
);
let tx_id = u64::from_le_bytes(
buf[12..20]
.try_into()
.map_err(|_| WalError::CorruptedEntry(seq_num))?,
);
let op_u8 = buf[20];
let op_type = WalOpType::from_u8(op_u8).ok_or(WalError::CorruptedEntry(seq_num))?;
let key_len = u32::from_le_bytes(
buf[21..25]
.try_into()
.map_err(|_| WalError::CorruptedEntry(seq_num))?,
) as usize;
let key_end = 25 + key_len;
if buf.len() < key_end + 4 {
return Err(WalError::CorruptedEntry(seq_num));
}
let key = buf[25..key_end].to_vec();
let value_len = u32::from_le_bytes(
buf[key_end..key_end + 4]
.try_into()
.map_err(|_| WalError::CorruptedEntry(seq_num))?,
) as usize;
let value_start = key_end + 4;
let value_end = value_start + value_len;
let ts_end = value_end + 8;
let cs_end = ts_end + 8;
if buf.len() < cs_end {
return Err(WalError::CorruptedEntry(seq_num));
}
let value = buf[value_start..value_end].to_vec();
let timestamp = u64::from_le_bytes(
buf[value_end..ts_end]
.try_into()
.map_err(|_| WalError::CorruptedEntry(seq_num))?,
);
let stored_checksum = u64::from_le_bytes(
buf[ts_end..cs_end]
.try_into()
.map_err(|_| WalError::CorruptedEntry(seq_num))?,
);
let computed = fnv1a_64(&buf[..ts_end]);
if computed != stored_checksum {
return Err(WalError::ChecksumMismatch {
entry_seq: seq_num,
expected: stored_checksum,
got: computed,
});
}
let entry = WalEntry {
seq_num,
tx_id,
op_type,
key,
value,
timestamp,
checksum: stored_checksum,
};
Ok((entry, cs_end))
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum TxState {
Active,
Committed,
RolledBack,
}
#[derive(Debug, Clone)]
pub struct Transaction {
pub tx_id: u64,
pub entries: Vec<WalEntry>,
pub state: TxState,
}
impl Transaction {
fn new(tx_id: u64) -> Self {
Self {
tx_id,
entries: Vec::new(),
state: TxState::Active,
}
}
}
#[derive(Debug, Clone)]
pub struct WalConfig {
pub max_segment_size: usize,
pub sync_on_write: bool,
pub max_entries_per_segment: usize,
pub checkpoint_interval: usize,
}
impl Default for WalConfig {
fn default() -> Self {
Self {
max_segment_size: 64 * 1024 * 1024, sync_on_write: false,
max_entries_per_segment: 100_000,
checkpoint_interval: 10_000,
}
}
}
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct RecoveryResult {
pub entries_recovered: usize,
pub transactions_recovered: usize,
pub entries_replayed: usize,
pub partial_tx_discarded: usize,
pub last_seq_num: u64,
}
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct WalStats {
pub total_entries: u64,
pub total_bytes: u64,
pub active_transactions: usize,
pub segments_count: usize,
pub last_checkpoint_seq: u64,
}
pub struct WalWriteAheadLog {
segment: Vec<u8>,
entries: Vec<WalEntry>,
transactions: HashMap<u64, Transaction>,
next_seq: u64,
next_tx_id: u64,
config: WalConfig,
stats: WalStats,
last_checkpoint_seq: u64,
writes_since_checkpoint: usize,
}
impl WalWriteAheadLog {
pub fn new(config: WalConfig) -> Self {
Self {
segment: Vec::new(),
entries: Vec::new(),
transactions: HashMap::new(),
next_seq: 1,
next_tx_id: 1,
config,
stats: WalStats {
segments_count: 1,
..WalStats::default()
},
last_checkpoint_seq: 0,
writes_since_checkpoint: 0,
}
}
fn now_ns() -> u64 {
use std::time::{SystemTime, UNIX_EPOCH};
SystemTime::now()
.duration_since(UNIX_EPOCH)
.unwrap_or_default()
.as_nanos() as u64
}
fn alloc_seq(&mut self) -> u64 {
let s = self.next_seq;
self.next_seq += 1;
s
}
fn alloc_tx_id(&mut self) -> u64 {
let id = self.next_tx_id;
self.next_tx_id += 1;
id
}
fn append_entry(&mut self, entry: WalEntry) -> Result<u64, WalError> {
let encoded = entry.encode();
let encoded_len = encoded.len();
if self.config.max_segment_size > 0
&& self.segment.len() + encoded_len > self.config.max_segment_size
{
return Err(WalError::SegmentFull);
}
if self.config.max_entries_per_segment > 0
&& self.entries.len() >= self.config.max_entries_per_segment
{
return Err(WalError::SegmentFull);
}
let seq = entry.seq_num;
self.segment.extend_from_slice(&encoded);
self.stats.total_entries += 1;
self.stats.total_bytes += encoded_len as u64;
self.entries.push(entry);
Ok(seq)
}
pub fn write(&mut self, key: Vec<u8>, value: Vec<u8>, op: WalOpType) -> Result<u64, WalError> {
let seq = self.alloc_seq();
let entry = WalEntry {
seq_num: seq,
tx_id: 0,
op_type: op,
key,
value,
timestamp: Self::now_ns(),
checksum: 0, };
let seq = self.append_entry(entry)?;
self.writes_since_checkpoint += 1;
if self.config.checkpoint_interval > 0
&& self.writes_since_checkpoint >= self.config.checkpoint_interval
{
self.checkpoint()?;
}
Ok(seq)
}
pub fn begin_transaction(&mut self) -> u64 {
let tx_id = self.alloc_tx_id();
let seq = self.alloc_seq();
let entry = WalEntry {
seq_num: seq,
tx_id,
op_type: WalOpType::Begin,
key: Vec::new(),
value: Vec::new(),
timestamp: Self::now_ns(),
checksum: 0,
};
let _ = self.append_entry(entry);
let tx = Transaction::new(tx_id);
self.transactions.insert(tx_id, tx);
self.stats.active_transactions = self
.transactions
.values()
.filter(|t| t.state == TxState::Active)
.count();
tx_id
}
pub fn write_tx(
&mut self,
tx_id: u64,
key: Vec<u8>,
value: Vec<u8>,
op: WalOpType,
) -> Result<u64, WalError> {
{
let tx = self
.transactions
.get(&tx_id)
.ok_or(WalError::TransactionNotFound(tx_id))?;
if tx.state != TxState::Active {
return Err(WalError::TransactionNotFound(tx_id));
}
}
let seq = self.alloc_seq();
let entry = WalEntry {
seq_num: seq,
tx_id,
op_type: op,
key,
value,
timestamp: Self::now_ns(),
checksum: 0,
};
let seq = self.append_entry(entry.clone())?;
if let Some(tx) = self.transactions.get_mut(&tx_id) {
tx.entries.push(entry);
}
Ok(seq)
}
pub fn commit_transaction(&mut self, tx_id: u64) -> Result<(), WalError> {
{
let tx = self
.transactions
.get(&tx_id)
.ok_or(WalError::TransactionNotFound(tx_id))?;
if tx.state != TxState::Active {
return Err(WalError::TransactionNotFound(tx_id));
}
}
let seq = self.alloc_seq();
let entry = WalEntry {
seq_num: seq,
tx_id,
op_type: WalOpType::Commit,
key: Vec::new(),
value: Vec::new(),
timestamp: Self::now_ns(),
checksum: 0,
};
self.append_entry(entry)?;
if let Some(tx) = self.transactions.get_mut(&tx_id) {
tx.state = TxState::Committed;
}
self.stats.active_transactions = self
.transactions
.values()
.filter(|t| t.state == TxState::Active)
.count();
Ok(())
}
pub fn rollback_transaction(&mut self, tx_id: u64) -> Result<(), WalError> {
{
let tx = self
.transactions
.get(&tx_id)
.ok_or(WalError::TransactionNotFound(tx_id))?;
if tx.state != TxState::Active {
return Err(WalError::TransactionNotFound(tx_id));
}
}
let seq = self.alloc_seq();
let entry = WalEntry {
seq_num: seq,
tx_id,
op_type: WalOpType::Rollback,
key: Vec::new(),
value: Vec::new(),
timestamp: Self::now_ns(),
checksum: 0,
};
self.append_entry(entry)?;
if let Some(tx) = self.transactions.get_mut(&tx_id) {
tx.state = TxState::RolledBack;
}
self.stats.active_transactions = self
.transactions
.values()
.filter(|t| t.state == TxState::Active)
.count();
Ok(())
}
pub fn checkpoint(&mut self) -> Result<u64, WalError> {
let seq = self.alloc_seq();
let entry = WalEntry {
seq_num: seq,
tx_id: 0,
op_type: WalOpType::Checkpoint,
key: Vec::new(),
value: Vec::new(),
timestamp: Self::now_ns(),
checksum: 0,
};
self.append_entry(entry)?;
let active_tx_ids: std::collections::HashSet<u64> = self
.transactions
.values()
.filter(|t| t.state == TxState::Active)
.map(|t| t.tx_id)
.collect();
let truncate_before = self.last_checkpoint_seq;
self.entries
.retain(|e| e.seq_num > truncate_before || active_tx_ids.contains(&e.tx_id));
let mut rebuilt = Vec::with_capacity(self.segment.len());
for e in &self.entries {
rebuilt.extend_from_slice(&e.encode());
}
self.segment = rebuilt;
self.last_checkpoint_seq = seq;
self.stats.last_checkpoint_seq = seq;
self.writes_since_checkpoint = 0;
Ok(seq)
}
pub fn recover(data: &[u8]) -> Result<RecoveryResult, WalError> {
let mut all_entries: Vec<WalEntry> = Vec::new();
let mut pos = 0usize;
while pos < data.len() {
if pos + 4 <= data.len() && data[pos..pos + 4] != WAL_MAGIC {
pos += 1;
continue;
}
match WalEntry::decode(data, pos) {
Ok((entry, consumed)) => {
pos += consumed;
all_entries.push(entry);
}
Err(WalError::ChecksumMismatch { .. }) => {
break;
}
Err(_) => {
break;
}
}
}
let mut committed_tx: std::collections::HashSet<u64> = std::collections::HashSet::new();
let mut rolled_back_tx: std::collections::HashSet<u64> = std::collections::HashSet::new();
let mut seen_tx: std::collections::HashSet<u64> = std::collections::HashSet::new();
for e in &all_entries {
if e.tx_id != 0 {
seen_tx.insert(e.tx_id);
match e.op_type {
WalOpType::Commit => {
committed_tx.insert(e.tx_id);
}
WalOpType::Rollback => {
rolled_back_tx.insert(e.tx_id);
}
_ => {}
}
}
}
let last_seq_num = all_entries.iter().map(|e| e.seq_num).max().unwrap_or(0);
let entries_recovered = all_entries.len();
let transactions_recovered = seen_tx.len();
let partial_tx_ids: std::collections::HashSet<u64> = seen_tx
.iter()
.filter(|id| !committed_tx.contains(id) && !rolled_back_tx.contains(id))
.copied()
.collect();
let partial_tx_discarded: usize = all_entries
.iter()
.filter(|e| e.tx_id != 0 && partial_tx_ids.contains(&e.tx_id))
.count();
let entries_replayed: usize = all_entries
.iter()
.filter(|e| {
let is_data_op = matches!(e.op_type, WalOpType::Put | WalOpType::Delete);
if !is_data_op {
return false;
}
if e.tx_id == 0 {
return true;
}
committed_tx.contains(&e.tx_id)
})
.count();
Ok(RecoveryResult {
entries_recovered,
transactions_recovered,
entries_replayed,
partial_tx_discarded,
last_seq_num,
})
}
pub fn replay(entries: &[WalEntry], handler: &mut dyn FnMut(&WalEntry)) -> usize {
let committed_tx: std::collections::HashSet<u64> = entries
.iter()
.filter(|e| e.op_type == WalOpType::Commit && e.tx_id != 0)
.map(|e| e.tx_id)
.collect();
let mut count = 0usize;
for entry in entries {
let is_data = matches!(entry.op_type, WalOpType::Put | WalOpType::Delete);
if !is_data {
continue;
}
let replayable = entry.tx_id == 0 || committed_tx.contains(&entry.tx_id);
if replayable {
handler(entry);
count += 1;
}
}
count
}
pub fn export_segment(&self) -> Vec<u8> {
self.segment.clone()
}
pub fn stats(&self) -> WalStats {
WalStats {
total_entries: self.stats.total_entries,
total_bytes: self.stats.total_bytes,
active_transactions: self.stats.active_transactions,
segments_count: 1,
last_checkpoint_seq: self.last_checkpoint_seq,
}
}
pub fn truncate_before(&mut self, seq_num: u64) -> usize {
let before = self.entries.len();
self.entries.retain(|e| e.seq_num >= seq_num);
let removed = before - self.entries.len();
if removed > 0 {
let mut rebuilt = Vec::with_capacity(self.segment.len());
for e in &self.entries {
rebuilt.extend_from_slice(&e.encode());
}
self.segment = rebuilt;
}
removed
}
pub fn entry_count(&self) -> usize {
self.entries.len()
}
}
#[cfg(test)]
mod tests {
use super::*;
struct Xorshift64 {
state: u64,
}
impl Xorshift64 {
fn new(seed: u64) -> Self {
Self {
state: if seed == 0 {
0xDEAD_BEEF_CAFE_1234
} else {
seed
},
}
}
fn next(&mut self) -> u64 {
self.state ^= self.state << 13;
self.state ^= self.state >> 7;
self.state ^= self.state << 17;
self.state
}
fn next_bytes(&mut self, len: usize) -> Vec<u8> {
let mut out = Vec::with_capacity(len);
while out.len() < len {
let v = self.next();
out.extend_from_slice(&v.to_le_bytes());
}
out.truncate(len);
out
}
}
fn default_wal() -> WalWriteAheadLog {
WalWriteAheadLog::new(WalConfig::default())
}
#[test]
fn test_fnv1a_64_empty() {
assert_eq!(fnv1a_64(&[]), 14_695_981_039_346_656_037u64);
}
#[test]
fn test_fnv1a_64_known_value() {
let h = fnv1a_64(b"foobar");
assert_ne!(h, 0);
assert_eq!(h, fnv1a_64(b"foobar"));
}
#[test]
fn test_fnv1a_64_differs_by_one_byte() {
let a = fnv1a_64(b"hello");
let b = fnv1a_64(b"hEllo");
assert_ne!(a, b);
}
#[test]
fn test_wal_op_type_from_u8_all_valid() {
for (byte, expected) in &[
(1u8, WalOpType::Put),
(2, WalOpType::Delete),
(3, WalOpType::Begin),
(4, WalOpType::Commit),
(5, WalOpType::Rollback),
(6, WalOpType::Checkpoint),
] {
assert_eq!(WalOpType::from_u8(*byte), Some(*expected));
}
}
#[test]
fn test_wal_op_type_from_u8_invalid() {
assert_eq!(WalOpType::from_u8(0), None);
assert_eq!(WalOpType::from_u8(7), None);
assert_eq!(WalOpType::from_u8(255), None);
}
#[test]
fn test_encode_decode_put_entry() {
let key = b"my-key".to_vec();
let value = b"my-value-data".to_vec();
let entry = WalEntry {
seq_num: 42,
tx_id: 0,
op_type: WalOpType::Put,
key: key.clone(),
value: value.clone(),
timestamp: 123_456_789,
checksum: 0,
};
let encoded = entry.encode();
let cs = u64::from_le_bytes(encoded[encoded.len() - 8..].try_into().unwrap());
let (decoded, consumed) = WalEntry::decode(&encoded, 0).unwrap();
assert_eq!(consumed, encoded.len());
assert_eq!(decoded.seq_num, 42);
assert_eq!(decoded.tx_id, 0);
assert_eq!(decoded.op_type, WalOpType::Put);
assert_eq!(decoded.key, key);
assert_eq!(decoded.value, value);
assert_eq!(decoded.timestamp, 123_456_789);
assert_eq!(decoded.checksum, cs);
}
#[test]
fn test_encode_decode_delete_entry() {
let entry = WalEntry {
seq_num: 1,
tx_id: 5,
op_type: WalOpType::Delete,
key: b"del-key".to_vec(),
value: Vec::new(),
timestamp: 0,
checksum: 0,
};
let enc = entry.encode();
let (dec, _) = WalEntry::decode(&enc, 0).unwrap();
assert_eq!(dec.op_type, WalOpType::Delete);
assert_eq!(dec.key, b"del-key");
assert!(dec.value.is_empty());
}
#[test]
fn test_encode_decode_empty_key_value() {
let entry = WalEntry {
seq_num: 99,
tx_id: 0,
op_type: WalOpType::Checkpoint,
key: Vec::new(),
value: Vec::new(),
timestamp: 0,
checksum: 0,
};
let enc = entry.encode();
let (dec, consumed) = WalEntry::decode(&enc, 0).unwrap();
assert_eq!(consumed, enc.len());
assert_eq!(dec.op_type, WalOpType::Checkpoint);
assert!(dec.key.is_empty());
assert!(dec.value.is_empty());
}
#[test]
fn test_encode_decode_with_offset() {
let padding = vec![0u8; 16];
let entry = WalEntry {
seq_num: 7,
tx_id: 3,
op_type: WalOpType::Put,
key: b"k".to_vec(),
value: b"v".to_vec(),
timestamp: 1,
checksum: 0,
};
let mut buf = padding.clone();
buf.extend_from_slice(&entry.encode());
let (dec, _) = WalEntry::decode(&buf, 16).unwrap();
assert_eq!(dec.seq_num, 7);
}
#[test]
fn test_encode_decode_large_payload() {
let mut rng = Xorshift64::new(0xABCDEF01);
let key = rng.next_bytes(512);
let value = rng.next_bytes(4096);
let entry = WalEntry {
seq_num: 1000,
tx_id: 0,
op_type: WalOpType::Put,
key: key.clone(),
value: value.clone(),
timestamp: 999,
checksum: 0,
};
let enc = entry.encode();
let (dec, _) = WalEntry::decode(&enc, 0).unwrap();
assert_eq!(dec.key, key);
assert_eq!(dec.value, value);
}
#[test]
fn test_checksum_mismatch_detected() {
let entry = WalEntry {
seq_num: 1,
tx_id: 0,
op_type: WalOpType::Put,
key: b"k".to_vec(),
value: b"v".to_vec(),
timestamp: 0,
checksum: 0,
};
let mut enc = entry.encode();
let flip_pos = enc.len() - 10; enc[flip_pos] ^= 0xFF;
let result = WalEntry::decode(&enc, 0);
assert!(matches!(result, Err(WalError::ChecksumMismatch { .. })));
}
#[test]
fn test_invalid_magic_detected() {
let entry = WalEntry {
seq_num: 2,
tx_id: 0,
op_type: WalOpType::Put,
key: b"k".to_vec(),
value: b"v".to_vec(),
timestamp: 0,
checksum: 0,
};
let mut enc = entry.encode();
enc[0] = b'Z';
let result = WalEntry::decode(&enc, 0);
assert!(matches!(result, Err(WalError::InvalidMagic)));
}
#[test]
fn test_corrupted_entry_too_short() {
let data = vec![b'W', b'A', b'L', b'X', 0x01]; let result = WalEntry::decode(&data, 0);
assert!(matches!(result, Err(WalError::CorruptedEntry(_))));
}
#[test]
fn test_corrupted_entry_truncated_key() {
let entry = WalEntry {
seq_num: 5,
tx_id: 0,
op_type: WalOpType::Put,
key: b"longkey12345678".to_vec(),
value: b"val".to_vec(),
timestamp: 0,
checksum: 0,
};
let mut enc = entry.encode();
enc.truncate(enc.len() / 2); let result = WalEntry::decode(&enc, 0);
assert!(result.is_err());
}
#[test]
fn test_write_returns_sequential_seq_nums() {
let mut wal = default_wal();
let s1 = wal
.write(b"k1".to_vec(), b"v1".to_vec(), WalOpType::Put)
.unwrap();
let s2 = wal
.write(b"k2".to_vec(), b"v2".to_vec(), WalOpType::Put)
.unwrap();
let s3 = wal
.write(b"k3".to_vec(), b"v3".to_vec(), WalOpType::Delete)
.unwrap();
assert!(s1 < s2);
assert!(s2 < s3);
}
#[test]
fn test_write_increases_entry_count() {
let mut wal = default_wal();
assert_eq!(wal.entry_count(), 0);
wal.write(b"a".to_vec(), b"b".to_vec(), WalOpType::Put)
.unwrap();
assert_eq!(wal.entry_count(), 1);
wal.write(b"c".to_vec(), b"d".to_vec(), WalOpType::Put)
.unwrap();
assert_eq!(wal.entry_count(), 2);
}
#[test]
fn test_write_segment_not_empty() {
let mut wal = default_wal();
wal.write(b"key".to_vec(), b"val".to_vec(), WalOpType::Put)
.unwrap();
assert!(!wal.export_segment().is_empty());
}
#[test]
fn test_stats_tracks_entries_and_bytes() {
let mut wal = default_wal();
wal.write(b"x".to_vec(), b"y".to_vec(), WalOpType::Put)
.unwrap();
wal.write(b"x".to_vec(), Vec::new(), WalOpType::Delete)
.unwrap();
let s = wal.stats();
assert_eq!(s.total_entries, 2);
assert!(s.total_bytes > 0);
}
#[test]
fn test_begin_transaction_returns_unique_ids() {
let mut wal = default_wal();
let t1 = wal.begin_transaction();
let t2 = wal.begin_transaction();
let t3 = wal.begin_transaction();
assert_ne!(t1, t2);
assert_ne!(t2, t3);
}
#[test]
fn test_write_tx_valid_transaction() {
let mut wal = default_wal();
let tx = wal.begin_transaction();
let seq = wal
.write_tx(tx, b"key".to_vec(), b"value".to_vec(), WalOpType::Put)
.unwrap();
assert!(seq > 0);
}
#[test]
fn test_write_tx_unknown_transaction_errors() {
let mut wal = default_wal();
let result = wal.write_tx(9999, b"k".to_vec(), b"v".to_vec(), WalOpType::Put);
assert!(matches!(result, Err(WalError::TransactionNotFound(9999))));
}
#[test]
fn test_commit_transaction_succeeds() {
let mut wal = default_wal();
let tx = wal.begin_transaction();
wal.write_tx(tx, b"k".to_vec(), b"v".to_vec(), WalOpType::Put)
.unwrap();
wal.commit_transaction(tx).unwrap();
}
#[test]
fn test_commit_unknown_transaction_errors() {
let mut wal = default_wal();
let result = wal.commit_transaction(42);
assert!(matches!(result, Err(WalError::TransactionNotFound(42))));
}
#[test]
fn test_rollback_transaction_succeeds() {
let mut wal = default_wal();
let tx = wal.begin_transaction();
wal.write_tx(tx, b"k".to_vec(), b"v".to_vec(), WalOpType::Put)
.unwrap();
wal.rollback_transaction(tx).unwrap();
}
#[test]
fn test_rollback_unknown_transaction_errors() {
let mut wal = default_wal();
let result = wal.rollback_transaction(77);
assert!(matches!(result, Err(WalError::TransactionNotFound(77))));
}
#[test]
fn test_double_commit_errors() {
let mut wal = default_wal();
let tx = wal.begin_transaction();
wal.commit_transaction(tx).unwrap();
let result = wal.commit_transaction(tx);
assert!(result.is_err());
}
#[test]
fn test_double_rollback_errors() {
let mut wal = default_wal();
let tx = wal.begin_transaction();
wal.rollback_transaction(tx).unwrap();
let result = wal.rollback_transaction(tx);
assert!(result.is_err());
}
#[test]
fn test_active_transactions_count_decreases_after_commit() {
let mut wal = default_wal();
let t1 = wal.begin_transaction();
let t2 = wal.begin_transaction();
assert_eq!(wal.stats().active_transactions, 2);
wal.commit_transaction(t1).unwrap();
assert_eq!(wal.stats().active_transactions, 1);
wal.rollback_transaction(t2).unwrap();
assert_eq!(wal.stats().active_transactions, 0);
}
#[test]
fn test_checkpoint_returns_seq_num() {
let mut wal = default_wal();
wal.write(b"a".to_vec(), b"b".to_vec(), WalOpType::Put)
.unwrap();
let cp_seq = wal.checkpoint().unwrap();
assert!(cp_seq > 0);
assert_eq!(wal.stats().last_checkpoint_seq, cp_seq);
}
#[test]
fn test_checkpoint_truncates_old_entries() {
let mut wal = default_wal();
for i in 0u8..50 {
wal.write(vec![i], vec![i], WalOpType::Put).unwrap();
}
let before = wal.entry_count();
wal.checkpoint().unwrap();
for i in 50u8..100 {
wal.write(vec![i], vec![i], WalOpType::Put).unwrap();
}
wal.checkpoint().unwrap();
let after = wal.entry_count();
assert!(after < before + 51); }
#[test]
fn test_checkpoint_preserves_active_tx_entries() {
let mut wal = default_wal();
let tx = wal.begin_transaction();
wal.write_tx(tx, b"k".to_vec(), b"v".to_vec(), WalOpType::Put)
.unwrap();
wal.checkpoint().unwrap();
assert!(wal.entries.iter().any(|e| e.tx_id == tx));
}
#[test]
fn test_auto_checkpoint_triggers() {
let config = WalConfig {
checkpoint_interval: 3,
..WalConfig::default()
};
let mut wal = WalWriteAheadLog::new(config);
for i in 0u8..3 {
wal.write(vec![i], vec![i], WalOpType::Put).unwrap();
}
assert_eq!(wal.writes_since_checkpoint, 0);
assert!(wal.last_checkpoint_seq > 0);
}
#[test]
fn test_export_segment_contains_all_entries() {
let mut wal = default_wal();
wal.write(b"k1".to_vec(), b"v1".to_vec(), WalOpType::Put)
.unwrap();
wal.write(b"k2".to_vec(), b"v2".to_vec(), WalOpType::Delete)
.unwrap();
let seg = wal.export_segment();
let mut pos = 0usize;
let mut count = 0usize;
while pos < seg.len() {
let (_, consumed) = WalEntry::decode(&seg, pos).unwrap();
pos += consumed;
count += 1;
}
assert_eq!(count, 2);
}
#[test]
fn test_export_segment_is_deterministic() {
let mut wal = default_wal();
wal.write(b"key".to_vec(), b"val".to_vec(), WalOpType::Put)
.unwrap();
let a = wal.export_segment();
let b = wal.export_segment();
assert_eq!(a, b);
}
#[test]
fn test_recover_clean_log() {
let mut wal = default_wal();
wal.write(b"a".to_vec(), b"1".to_vec(), WalOpType::Put)
.unwrap();
wal.write(b"b".to_vec(), b"2".to_vec(), WalOpType::Put)
.unwrap();
wal.write(b"c".to_vec(), Vec::new(), WalOpType::Delete)
.unwrap();
let seg = wal.export_segment();
let rr = WalWriteAheadLog::recover(&seg).unwrap();
assert_eq!(rr.entries_recovered, 3);
assert_eq!(rr.entries_replayed, 3); assert_eq!(rr.partial_tx_discarded, 0);
assert_eq!(rr.transactions_recovered, 0);
assert!(rr.last_seq_num >= 3);
}
#[test]
fn test_recover_committed_transaction() {
let mut wal = default_wal();
let tx = wal.begin_transaction();
wal.write_tx(tx, b"k".to_vec(), b"v".to_vec(), WalOpType::Put)
.unwrap();
wal.commit_transaction(tx).unwrap();
let seg = wal.export_segment();
let rr = WalWriteAheadLog::recover(&seg).unwrap();
assert_eq!(rr.entries_replayed, 1); assert_eq!(rr.partial_tx_discarded, 0);
assert_eq!(rr.transactions_recovered, 1);
}
#[test]
fn test_recover_partial_tx_discarded() {
let mut wal = default_wal();
let tx = wal.begin_transaction();
wal.write_tx(tx, b"k".to_vec(), b"v".to_vec(), WalOpType::Put)
.unwrap();
let seg = wal.export_segment();
let rr = WalWriteAheadLog::recover(&seg).unwrap();
assert_eq!(rr.entries_replayed, 0); assert!(rr.partial_tx_discarded > 0);
}
#[test]
fn test_recover_rolled_back_tx_not_replayed() {
let mut wal = default_wal();
let tx = wal.begin_transaction();
wal.write_tx(tx, b"k".to_vec(), b"v".to_vec(), WalOpType::Put)
.unwrap();
wal.rollback_transaction(tx).unwrap();
let seg = wal.export_segment();
let rr = WalWriteAheadLog::recover(&seg).unwrap();
assert_eq!(rr.entries_replayed, 0);
assert_eq!(rr.partial_tx_discarded, 0);
}
#[test]
fn test_recover_mixed_committed_and_partial() {
let mut wal = default_wal();
let t1 = wal.begin_transaction();
wal.write_tx(t1, b"k1".to_vec(), b"v1".to_vec(), WalOpType::Put)
.unwrap();
wal.commit_transaction(t1).unwrap();
let t2 = wal.begin_transaction();
wal.write_tx(t2, b"k2".to_vec(), b"v2".to_vec(), WalOpType::Put)
.unwrap();
wal.write_tx(t2, b"k3".to_vec(), b"v3".to_vec(), WalOpType::Put)
.unwrap();
let seg = wal.export_segment();
let rr = WalWriteAheadLog::recover(&seg).unwrap();
assert_eq!(rr.entries_replayed, 1); assert!(rr.partial_tx_discarded >= 2); assert_eq!(rr.transactions_recovered, 2);
}
#[test]
fn test_recover_empty_buffer() {
let rr = WalWriteAheadLog::recover(&[]).unwrap();
assert_eq!(rr.entries_recovered, 0);
assert_eq!(rr.last_seq_num, 0);
}
#[test]
fn test_recover_garbage_prefix_skipped() {
let mut wal = default_wal();
wal.write(b"k".to_vec(), b"v".to_vec(), WalOpType::Put)
.unwrap();
let mut seg = vec![0xDE, 0xAD, 0xBE, 0xEF, 0x00, 0x00];
seg.extend_from_slice(&wal.export_segment());
let rr = WalWriteAheadLog::recover(&seg).unwrap();
assert_eq!(rr.entries_recovered, 1);
}
#[test]
fn test_recover_last_seq_num_correct() {
let mut wal = default_wal();
for i in 0u8..10 {
wal.write(vec![i], vec![i], WalOpType::Put).unwrap();
}
let seg = wal.export_segment();
let rr = WalWriteAheadLog::recover(&seg).unwrap();
assert_eq!(rr.last_seq_num, 10);
}
#[test]
fn test_replay_standalone_entries() {
let mut wal = default_wal();
wal.write(b"k1".to_vec(), b"v1".to_vec(), WalOpType::Put)
.unwrap();
wal.write(b"k2".to_vec(), b"v2".to_vec(), WalOpType::Put)
.unwrap();
wal.write(b"k3".to_vec(), Vec::new(), WalOpType::Delete)
.unwrap();
let entries: Vec<WalEntry> = wal.entries.clone();
let mut replayed = Vec::new();
let count = WalWriteAheadLog::replay(&entries, &mut |e| {
replayed.push(e.clone());
});
assert_eq!(count, 3);
assert_eq!(replayed.len(), 3);
}
#[test]
fn test_replay_skips_control_entries() {
let mut wal = default_wal();
wal.write(b"k".to_vec(), b"v".to_vec(), WalOpType::Put)
.unwrap();
wal.checkpoint().unwrap();
let entries: Vec<WalEntry> = wal.entries.clone();
let mut count_called = 0usize;
WalWriteAheadLog::replay(&entries, &mut |_| {
count_called += 1;
});
assert_eq!(count_called, 1);
}
#[test]
fn test_replay_skips_uncommitted_tx() {
let mut wal = default_wal();
let tx = wal.begin_transaction();
wal.write_tx(tx, b"k".to_vec(), b"v".to_vec(), WalOpType::Put)
.unwrap();
let entries: Vec<WalEntry> = wal.entries.clone();
let mut count_called = 0usize;
WalWriteAheadLog::replay(&entries, &mut |_| {
count_called += 1;
});
assert_eq!(count_called, 0);
}
#[test]
fn test_replay_includes_committed_tx() {
let mut wal = default_wal();
let tx = wal.begin_transaction();
wal.write_tx(tx, b"k".to_vec(), b"v".to_vec(), WalOpType::Put)
.unwrap();
wal.write_tx(tx, b"k2".to_vec(), Vec::new(), WalOpType::Delete)
.unwrap();
wal.commit_transaction(tx).unwrap();
let entries: Vec<WalEntry> = wal.entries.clone();
let mut seen_keys: Vec<Vec<u8>> = Vec::new();
WalWriteAheadLog::replay(&entries, &mut |e| {
seen_keys.push(e.key.clone());
});
assert_eq!(seen_keys.len(), 2);
assert!(seen_keys.contains(&b"k".to_vec()));
assert!(seen_keys.contains(&b"k2".to_vec()));
}
#[test]
fn test_replay_order_preserved() {
let mut wal = default_wal();
let keys: &[&[u8]] = &[b"alpha", b"beta_", b"gamma", b"delta"];
for k in keys {
wal.write(k.to_vec(), b"v".to_vec(), WalOpType::Put)
.unwrap();
}
let entries: Vec<WalEntry> = wal.entries.clone();
let mut order: Vec<Vec<u8>> = Vec::new();
WalWriteAheadLog::replay(&entries, &mut |e| {
order.push(e.key.clone());
});
for (i, k) in keys.iter().enumerate() {
assert_eq!(order[i], k.to_vec());
}
}
#[test]
fn test_replay_empty_entries() {
let count = WalWriteAheadLog::replay(&[], &mut |_| {});
assert_eq!(count, 0);
}
#[test]
fn test_truncate_before_removes_entries() {
let mut wal = default_wal();
for i in 1u8..=10 {
wal.write(vec![i], vec![i], WalOpType::Put).unwrap();
}
assert_eq!(wal.entry_count(), 10);
let removed = wal.truncate_before(6);
assert_eq!(removed, 5); assert_eq!(wal.entry_count(), 5);
}
#[test]
fn test_truncate_before_zero_removes_nothing() {
let mut wal = default_wal();
wal.write(b"k".to_vec(), b"v".to_vec(), WalOpType::Put)
.unwrap();
let removed = wal.truncate_before(0);
assert_eq!(removed, 0);
assert_eq!(wal.entry_count(), 1);
}
#[test]
fn test_truncate_before_large_seq_removes_all() {
let mut wal = default_wal();
for i in 0u8..5 {
wal.write(vec![i], vec![i], WalOpType::Put).unwrap();
}
let removed = wal.truncate_before(u64::MAX);
assert_eq!(removed, 5);
assert_eq!(wal.entry_count(), 0);
assert!(wal.export_segment().is_empty());
}
#[test]
fn test_truncate_rebuilds_segment() {
let mut wal = default_wal();
for i in 1u8..=6 {
wal.write(vec![i], vec![i], WalOpType::Put).unwrap();
}
let seg_before = wal.export_segment().len();
wal.truncate_before(4);
let seg_after = wal.export_segment().len();
assert!(seg_after < seg_before);
}
#[test]
fn test_stats_segments_count() {
let wal = default_wal();
assert_eq!(wal.stats().segments_count, 1);
}
#[test]
fn test_stats_last_checkpoint_seq_initial_zero() {
let wal = default_wal();
assert_eq!(wal.stats().last_checkpoint_seq, 0);
}
#[test]
fn test_stats_after_checkpoint() {
let mut wal = default_wal();
wal.write(b"k".to_vec(), b"v".to_vec(), WalOpType::Put)
.unwrap();
let cp = wal.checkpoint().unwrap();
assert_eq!(wal.stats().last_checkpoint_seq, cp);
}
#[test]
fn test_stats_total_bytes_grows() {
let mut wal = default_wal();
let b0 = wal.stats().total_bytes;
wal.write(b"key-large".to_vec(), vec![0u8; 256], WalOpType::Put)
.unwrap();
let b1 = wal.stats().total_bytes;
assert!(b1 > b0);
}
#[test]
fn test_segment_full_on_size_limit() {
let config = WalConfig {
max_segment_size: 100, ..WalConfig::default()
};
let mut wal = WalWriteAheadLog::new(config);
let mut hit_full = false;
for i in 0u8..100 {
let result = wal.write(vec![i], vec![i; 10], WalOpType::Put);
if matches!(result, Err(WalError::SegmentFull)) {
hit_full = true;
break;
}
}
assert!(hit_full, "Expected SegmentFull error to be triggered");
}
#[test]
fn test_segment_full_on_entry_count_limit() {
let config = WalConfig {
max_entries_per_segment: 3,
max_segment_size: 0, checkpoint_interval: 0,
..WalConfig::default()
};
let mut wal = WalWriteAheadLog::new(config);
wal.write(b"a".to_vec(), b"1".to_vec(), WalOpType::Put)
.unwrap();
wal.write(b"b".to_vec(), b"2".to_vec(), WalOpType::Put)
.unwrap();
wal.write(b"c".to_vec(), b"3".to_vec(), WalOpType::Put)
.unwrap();
let result = wal.write(b"d".to_vec(), b"4".to_vec(), WalOpType::Put);
assert!(matches!(result, Err(WalError::SegmentFull)));
}
#[test]
fn test_multiple_concurrent_transactions() {
let mut wal = default_wal();
let t1 = wal.begin_transaction();
let t2 = wal.begin_transaction();
wal.write_tx(t1, b"t1k1".to_vec(), b"v1".to_vec(), WalOpType::Put)
.unwrap();
wal.write_tx(t2, b"t2k1".to_vec(), b"v2".to_vec(), WalOpType::Put)
.unwrap();
wal.write_tx(t1, b"t1k2".to_vec(), b"v3".to_vec(), WalOpType::Put)
.unwrap();
wal.commit_transaction(t1).unwrap();
wal.rollback_transaction(t2).unwrap();
let entries: Vec<WalEntry> = wal.entries.clone();
let mut replayed_keys: Vec<Vec<u8>> = Vec::new();
WalWriteAheadLog::replay(&entries, &mut |e| {
replayed_keys.push(e.key.clone());
});
assert!(replayed_keys.contains(&b"t1k1".to_vec()));
assert!(replayed_keys.contains(&b"t1k2".to_vec()));
assert!(!replayed_keys.contains(&b"t2k1".to_vec()));
}
#[test]
fn test_sequential_encode_decode_in_buffer() {
let mut buf = Vec::new();
let mut entries = Vec::new();
for i in 0u64..20 {
let e = WalEntry {
seq_num: i + 1,
tx_id: 0,
op_type: WalOpType::Put,
key: format!("key-{i}").into_bytes(),
value: format!("val-{i}").into_bytes(),
timestamp: i * 1000,
checksum: 0,
};
buf.extend_from_slice(&e.encode());
entries.push(e);
}
let mut pos = 0usize;
let mut decoded = Vec::new();
while pos < buf.len() {
let (e, consumed) = WalEntry::decode(&buf, pos).unwrap();
decoded.push(e);
pos += consumed;
}
assert_eq!(decoded.len(), 20);
for (orig, dec) in entries.iter().zip(decoded.iter()) {
assert_eq!(orig.seq_num, dec.seq_num);
assert_eq!(orig.key, dec.key);
assert_eq!(orig.value, dec.value);
}
}
#[test]
fn test_wal_config_defaults_reasonable() {
let cfg = WalConfig::default();
assert!(cfg.max_segment_size > 0);
assert!(cfg.max_entries_per_segment > 0);
assert!(cfg.checkpoint_interval > 0);
}
#[test]
fn test_stress_random_writes_and_recovery() {
let mut rng = Xorshift64::new(42);
let mut wal = default_wal();
for _ in 0..200 {
let key_len = (rng.next() % 64 + 1) as usize;
let val_len = (rng.next() % 256) as usize;
let key = rng.next_bytes(key_len);
let val = rng.next_bytes(val_len);
wal.write(key, val, WalOpType::Put).unwrap();
}
let seg = wal.export_segment();
let rr = WalWriteAheadLog::recover(&seg).unwrap();
assert_eq!(rr.entries_recovered, 200);
assert_eq!(rr.entries_replayed, 200);
}
#[test]
fn test_stress_mixed_transactions_and_standalone() {
let mut rng = Xorshift64::new(7777);
let mut wal = default_wal();
let mut committed_put_count = 0usize;
let mut standalone_count = 0usize;
for _ in 0..10 {
for j in 0u8..5 {
wal.write(vec![j], rng.next_bytes(8), WalOpType::Put)
.unwrap();
standalone_count += 1;
}
let tx = wal.begin_transaction();
for j in 0u8..3 {
wal.write_tx(tx, vec![j], rng.next_bytes(8), WalOpType::Put)
.unwrap();
committed_put_count += 1;
}
wal.commit_transaction(tx).unwrap();
let tx2 = wal.begin_transaction();
wal.write_tx(tx2, b"partial".to_vec(), b"v".to_vec(), WalOpType::Put)
.unwrap();
}
let seg = wal.export_segment();
let rr = WalWriteAheadLog::recover(&seg).unwrap();
assert_eq!(rr.entries_replayed, standalone_count + committed_put_count);
assert!(rr.partial_tx_discarded > 0);
}
}