use std::io::{Read, Write};
use crate::error::{HoronError, HoronResult};
use crate::format::*;
use crate::quant::SemLayout;
use crate::snapshot::NodeEntry;
#[derive(Debug, Clone)]
pub struct WalEntry {
pub seq: u32,
pub op: u8,
pub key: String,
pub payload: WalPayload,
}
#[derive(Debug, Clone)]
pub enum WalPayload {
Insert(NodeEntry),
Update {
data: Vec<u8>,
metadata: Vec<(String, String)>,
},
Delete,
SetMeta {
meta_key: String,
meta_value: String,
},
SetSemantic {
coords: Vec<u8>,
},
Epoch {
epoch_id: u64,
flags: u8,
},
}
impl WalEntry {
fn write_body<W: Write>(&self, w: &mut W, layout: &SemLayout) -> HoronResult<()> {
w.write_all(&self.seq.to_le_bytes())?;
w.write_all(&[self.op])?;
let key_bytes = self.key.as_bytes();
w.write_all(&(key_bytes.len() as u16).to_le_bytes())?;
w.write_all(key_bytes)?;
match &self.payload {
WalPayload::Insert(entry) => {
w.write_all(&(entry.data.len() as u32).to_le_bytes())?;
w.write_all(&entry.data)?;
w.write_all(&(entry.metadata.len() as u16).to_le_bytes())?;
for (mk, mv) in &entry.metadata {
let mk_b = mk.as_bytes();
let mv_b = mv.as_bytes();
w.write_all(&(mk_b.len() as u16).to_le_bytes())?;
w.write_all(mk_b)?;
w.write_all(&(mv_b.len() as u16).to_le_bytes())?;
w.write_all(mv_b)?;
}
if layout.quantized {
if layout.disk_bytes() > 0 {
w.write_all(&layout.encode_tail(&entry.semantic_coords)?)?;
}
} else if !entry.semantic_coords.is_empty() {
w.write_all(&entry.semantic_coords)?;
}
}
WalPayload::Update { data, metadata } => {
w.write_all(&(data.len() as u32).to_le_bytes())?;
w.write_all(data)?;
w.write_all(&(metadata.len() as u16).to_le_bytes())?;
for (mk, mv) in metadata {
let mk_b = mk.as_bytes();
let mv_b = mv.as_bytes();
w.write_all(&(mk_b.len() as u16).to_le_bytes())?;
w.write_all(mk_b)?;
w.write_all(&(mv_b.len() as u16).to_le_bytes())?;
w.write_all(mv_b)?;
}
}
WalPayload::Delete => {
}
WalPayload::SetMeta { meta_key, meta_value } => {
let mk_b = meta_key.as_bytes();
let mv_b = meta_value.as_bytes();
w.write_all(&(mk_b.len() as u16).to_le_bytes())?;
w.write_all(mk_b)?;
w.write_all(&(mv_b.len() as u16).to_le_bytes())?;
w.write_all(mv_b)?;
}
WalPayload::SetSemantic { coords } => {
if layout.quantized {
w.write_all(&layout.encode_tail(coords)?)?;
} else {
w.write_all(coords)?;
}
}
WalPayload::Epoch { epoch_id, flags } => {
w.write_all(&epoch_id.to_le_bytes())?;
w.write_all(&[*flags])?;
}
}
Ok(())
}
pub fn write_to<W: Write>(&self, w: &mut W, layout: &SemLayout) -> HoronResult<()> {
let mut body = Vec::new();
self.write_body(&mut body, layout)?;
let crc = crc32fast::hash(&body);
w.write_all(&body)?;
w.write_all(&crc.to_le_bytes())?;
Ok(())
}
pub fn read_from<R: Read>(
r: &mut R,
layout: &SemLayout,
) -> HoronResult<Option<Self>> {
let mut buf2 = [0u8; 2];
let mut buf4 = [0u8; 4];
if r.read_exact(&mut buf4).is_err() {
return Ok(None); }
let mut body = Vec::new();
body.extend_from_slice(&buf4);
let seq = u32::from_le_bytes(buf4);
let mut op_buf = [0u8; 1];
r.read_exact(&mut op_buf)?;
body.push(op_buf[0]);
let op = op_buf[0];
r.read_exact(&mut buf2)?;
body.extend_from_slice(&buf2);
let key_len = u16::from_le_bytes(buf2) as usize;
let mut key_buf = vec![0u8; key_len];
r.read_exact(&mut key_buf)?;
body.extend_from_slice(&key_buf);
let key = String::from_utf8(key_buf)
.map_err(|e| HoronError::InvalidFormat(format!("invalid UTF-8 key: {}", e)))?;
let payload = match op {
OP_INSERT => {
let entry = read_insert_body(r, &mut body, layout)?;
WalPayload::Insert(entry)
}
OP_UPDATE => {
let (data, metadata) = read_update_body(r, &mut body)?;
WalPayload::Update { data, metadata }
}
OP_DELETE => {
WalPayload::Delete
}
OP_SET_META => {
let (mk, mv) = read_meta_body(r, &mut body)?;
WalPayload::SetMeta { meta_key: mk, meta_value: mv }
}
OP_SET_SEMANTIC => {
let mut disk = vec![0u8; layout.disk_bytes()];
r.read_exact(&mut disk)?;
body.extend_from_slice(&disk);
let coords = if layout.quantized {
layout.decode_tail(&disk)?
} else {
disk
};
WalPayload::SetSemantic { coords }
}
OP_EPOCH => {
let mut buf8 = [0u8; 8];
r.read_exact(&mut buf8)?;
body.extend_from_slice(&buf8);
let epoch_id = u64::from_le_bytes(buf8);
let mut flag_buf = [0u8; 1];
r.read_exact(&mut flag_buf)?;
body.push(flag_buf[0]);
WalPayload::Epoch { epoch_id, flags: flag_buf[0] }
}
_ => {
return Err(HoronError::InvalidFormat(
format!("unknown WAL op code: 0x{:02X}", op)
));
}
};
r.read_exact(&mut buf4)?;
let stored_crc = u32::from_le_bytes(buf4);
let computed_crc = crc32fast::hash(&body);
if stored_crc != computed_crc {
return Ok(None);
}
Ok(Some(WalEntry { seq, op, key, payload }))
}
}
fn read_insert_body<R: Read>(
r: &mut R,
body: &mut Vec<u8>,
layout: &SemLayout,
) -> HoronResult<NodeEntry> {
let mut buf2 = [0u8; 2];
let mut buf4 = [0u8; 4];
r.read_exact(&mut buf4)?;
body.extend_from_slice(&buf4);
let data_len = u32::from_le_bytes(buf4) as usize;
if data_len > MAX_ENTRY_DATA {
return Err(HoronError::InvalidFormat(format!(
"WAL entry data length {} exceeds maximum {} — corrupt length field",
data_len, MAX_ENTRY_DATA
)));
}
let data = crate::format::read_bounded_vec(r, data_len, "WAL entry data")?;
body.extend_from_slice(&data);
r.read_exact(&mut buf2)?;
body.extend_from_slice(&buf2);
let meta_count = u16::from_le_bytes(buf2) as usize;
let mut metadata = Vec::with_capacity(meta_count);
for _ in 0..meta_count {
r.read_exact(&mut buf2)?;
body.extend_from_slice(&buf2);
let mk_len = u16::from_le_bytes(buf2) as usize;
let mut mk_buf = vec![0u8; mk_len];
r.read_exact(&mut mk_buf)?;
body.extend_from_slice(&mk_buf);
r.read_exact(&mut buf2)?;
body.extend_from_slice(&buf2);
let mv_len = u16::from_le_bytes(buf2) as usize;
let mut mv_buf = vec![0u8; mv_len];
r.read_exact(&mut mv_buf)?;
body.extend_from_slice(&mv_buf);
let mk = String::from_utf8(mk_buf)
.map_err(|e| HoronError::InvalidFormat(format!("invalid UTF-8: {}", e)))?;
let mv = String::from_utf8(mv_buf)
.map_err(|e| HoronError::InvalidFormat(format!("invalid UTF-8: {}", e)))?;
metadata.push((mk, mv));
}
let disk_bytes = layout.disk_bytes();
let mut semantic_coords = vec![0u8; disk_bytes];
if disk_bytes > 0 {
r.read_exact(&mut semantic_coords)?;
body.extend_from_slice(&semantic_coords);
if layout.quantized {
semantic_coords = layout.decode_tail(&semantic_coords)?;
}
}
Ok(NodeEntry {
key: String::new(), data,
metadata,
semantic_coords,
})
}
fn read_update_body<R: Read>(
r: &mut R,
body: &mut Vec<u8>,
) -> HoronResult<(Vec<u8>, Vec<(String, String)>)> {
let mut buf2 = [0u8; 2];
let mut buf4 = [0u8; 4];
r.read_exact(&mut buf4)?;
body.extend_from_slice(&buf4);
let data_len = u32::from_le_bytes(buf4) as usize;
if data_len > MAX_ENTRY_DATA {
return Err(HoronError::InvalidFormat(format!(
"WAL entry data length {} exceeds maximum {} — corrupt length field",
data_len, MAX_ENTRY_DATA
)));
}
let data = crate::format::read_bounded_vec(r, data_len, "WAL entry data")?;
body.extend_from_slice(&data);
r.read_exact(&mut buf2)?;
body.extend_from_slice(&buf2);
let meta_count = u16::from_le_bytes(buf2) as usize;
let mut metadata = Vec::with_capacity(meta_count);
for _ in 0..meta_count {
r.read_exact(&mut buf2)?;
body.extend_from_slice(&buf2);
let mk_len = u16::from_le_bytes(buf2) as usize;
let mut mk_buf = vec![0u8; mk_len];
r.read_exact(&mut mk_buf)?;
body.extend_from_slice(&mk_buf);
r.read_exact(&mut buf2)?;
body.extend_from_slice(&buf2);
let mv_len = u16::from_le_bytes(buf2) as usize;
let mut mv_buf = vec![0u8; mv_len];
r.read_exact(&mut mv_buf)?;
body.extend_from_slice(&mv_buf);
let mk = String::from_utf8(mk_buf)
.map_err(|e| HoronError::InvalidFormat(format!("invalid UTF-8: {}", e)))?;
let mv = String::from_utf8(mv_buf)
.map_err(|e| HoronError::InvalidFormat(format!("invalid UTF-8: {}", e)))?;
metadata.push((mk, mv));
}
Ok((data, metadata))
}
fn read_meta_body<R: Read>(
r: &mut R,
body: &mut Vec<u8>,
) -> HoronResult<(String, String)> {
let mut buf2 = [0u8; 2];
r.read_exact(&mut buf2)?;
body.extend_from_slice(&buf2);
let mk_len = u16::from_le_bytes(buf2) as usize;
let mut mk_buf = vec![0u8; mk_len];
r.read_exact(&mut mk_buf)?;
body.extend_from_slice(&mk_buf);
r.read_exact(&mut buf2)?;
body.extend_from_slice(&buf2);
let mv_len = u16::from_le_bytes(buf2) as usize;
let mut mv_buf = vec![0u8; mv_len];
r.read_exact(&mut mv_buf)?;
body.extend_from_slice(&mv_buf);
let mk = String::from_utf8(mk_buf)
.map_err(|e| HoronError::InvalidFormat(format!("invalid UTF-8: {}", e)))?;
let mv = String::from_utf8(mv_buf)
.map_err(|e| HoronError::InvalidFormat(format!("invalid UTF-8: {}", e)))?;
Ok((mk, mv))
}
const MAX_BLOCK_DECOMPRESSED: usize = 256 * 1024 * 1024;
pub fn write_wal_block<W: Write>(
w: &mut W,
entries_data: &[u8],
entry_count: u16,
algo: u8,
) -> HoronResult<()> {
let compressed = crate::compression::compress(entries_data, algo)?;
w.write_all(&entry_count.to_le_bytes())?;
w.write_all(&(compressed.len() as u32).to_le_bytes())?;
w.write_all(&compressed)?;
Ok(())
}
pub fn read_wal_block<R: Read>(
r: &mut R,
algo: u8,
) -> HoronResult<Option<(Vec<u8>, u16)>> {
let mut buf2 = [0u8; 2];
if r.read_exact(&mut buf2).is_err() {
return Ok(None); }
let entry_count = u16::from_le_bytes(buf2);
if entry_count == 0 || entry_count > WAL_BLOCK_SIZE as u16 {
return Ok(None); }
let mut buf4 = [0u8; 4];
if r.read_exact(&mut buf4).is_err() {
return Ok(None); }
let compressed_len = u32::from_le_bytes(buf4) as usize;
if compressed_len == 0 || compressed_len > MAX_BLOCK_DECOMPRESSED {
return Ok(None); }
let compressed = match crate::format::read_bounded_vec(r, compressed_len, "WAL block") {
Ok(c) => c,
Err(_) => return Ok(None), };
let decompressed = crate::compression::decompress(&compressed, algo, MAX_BLOCK_DECOMPRESSED)?;
Ok(Some((decompressed, entry_count)))
}
pub fn write_wal_header<W: Write>(w: &mut W, entry_count: u32, base_seq: u32) -> HoronResult<()> {
w.write_all(&entry_count.to_le_bytes())?;
w.write_all(&base_seq.to_le_bytes())?;
Ok(())
}
pub fn read_wal_header<R: Read>(r: &mut R) -> HoronResult<(u32, u32)> {
let mut buf4 = [0u8; 4];
r.read_exact(&mut buf4)?;
let entry_count = u32::from_le_bytes(buf4);
r.read_exact(&mut buf4)?;
let base_seq = u32::from_le_bytes(buf4);
Ok((entry_count, base_seq))
}
#[cfg(test)]
mod tests {
use super::*;
use std::io::Cursor;
fn l0() -> SemLayout {
SemLayout::plain(0)
}
fn make_insert_entry(seq: u32, key: &str, data: &[u8]) -> WalEntry {
WalEntry {
seq,
op: OP_INSERT,
key: key.to_string(),
payload: WalPayload::Insert(NodeEntry {
key: key.to_string(),
data: data.to_vec(),
metadata: vec![],
semantic_coords: vec![],
}),
}
}
#[test]
fn test_wal_entry_insert_roundtrip() {
let entry = make_insert_entry(1, "/test", b"hello");
let mut buf = Vec::new();
entry.write_to(&mut buf, &l0()).unwrap();
let mut cursor = Cursor::new(&buf);
let parsed = WalEntry::read_from(&mut cursor, &l0()).unwrap().unwrap();
assert_eq!(parsed.seq, 1);
assert_eq!(parsed.op, OP_INSERT);
assert_eq!(parsed.key, "/test");
match parsed.payload {
WalPayload::Insert(e) => assert_eq!(e.data, b"hello"),
_ => panic!("expected Insert"),
}
}
#[test]
fn test_wal_entry_delete_roundtrip() {
let entry = WalEntry {
seq: 5,
op: OP_DELETE,
key: "/gone".to_string(),
payload: WalPayload::Delete,
};
let mut buf = Vec::new();
entry.write_to(&mut buf, &l0()).unwrap();
let mut cursor = Cursor::new(&buf);
let parsed = WalEntry::read_from(&mut cursor, &l0()).unwrap().unwrap();
assert_eq!(parsed.seq, 5);
assert_eq!(parsed.op, OP_DELETE);
assert!(matches!(parsed.payload, WalPayload::Delete));
}
#[test]
fn test_wal_entry_set_meta_roundtrip() {
let entry = WalEntry {
seq: 10,
op: OP_SET_META,
key: "/doc".to_string(),
payload: WalPayload::SetMeta {
meta_key: "author".to_string(),
meta_value: "alice".to_string(),
},
};
let mut buf = Vec::new();
entry.write_to(&mut buf, &l0()).unwrap();
let mut cursor = Cursor::new(&buf);
let parsed = WalEntry::read_from(&mut cursor, &l0()).unwrap().unwrap();
assert_eq!(parsed.seq, 10);
match parsed.payload {
WalPayload::SetMeta { meta_key, meta_value } => {
assert_eq!(meta_key, "author");
assert_eq!(meta_value, "alice");
}
_ => panic!("expected SetMeta"),
}
}
#[test]
fn test_wal_corrupted_crc_returns_none() {
let entry = make_insert_entry(1, "/test", b"data");
let mut buf = Vec::new();
entry.write_to(&mut buf, &l0()).unwrap();
let len = buf.len();
buf[len - 1] ^= 0xFF;
let mut cursor = Cursor::new(&buf);
let result = WalEntry::read_from(&mut cursor, &l0()).unwrap();
assert!(result.is_none(), "corrupted CRC should return None");
}
#[test]
fn test_wal_multiple_entries() {
let entries = vec![
make_insert_entry(1, "/a", b"aaa"),
make_insert_entry(2, "/b", b"bbb"),
WalEntry {
seq: 3,
op: OP_DELETE,
key: "/a".to_string(),
payload: WalPayload::Delete,
},
];
let mut buf = Vec::new();
for e in &entries {
e.write_to(&mut buf, &l0()).unwrap();
}
let mut cursor = Cursor::new(&buf);
let e1 = WalEntry::read_from(&mut cursor, &l0()).unwrap().unwrap();
let e2 = WalEntry::read_from(&mut cursor, &l0()).unwrap().unwrap();
let e3 = WalEntry::read_from(&mut cursor, &l0()).unwrap().unwrap();
assert_eq!(e1.seq, 1);
assert_eq!(e2.seq, 2);
assert_eq!(e3.seq, 3);
assert_eq!(e3.op, OP_DELETE);
let e4 = WalEntry::read_from(&mut cursor, &l0()).unwrap();
assert!(e4.is_none());
}
#[test]
fn test_wal_header_roundtrip() {
let mut buf = Vec::new();
write_wal_header(&mut buf, 42, 100).unwrap();
let mut cursor = Cursor::new(&buf);
let (count, base) = read_wal_header(&mut cursor).unwrap();
assert_eq!(count, 42);
assert_eq!(base, 100);
}
#[test]
fn test_wal_block_roundtrip() {
let entries: Vec<WalEntry> = (1..=10)
.map(|i| make_insert_entry(i, &format!("/n{}", i), b"data"))
.collect();
let mut raw = Vec::new();
for e in &entries {
e.write_to(&mut raw, &l0()).unwrap();
}
let mut block_buf = Vec::new();
write_wal_block(&mut block_buf, &raw, entries.len() as u16, ALGO_ZSTD).unwrap();
let mut cursor = Cursor::new(&block_buf);
let (decompressed, count) = read_wal_block(&mut cursor, ALGO_ZSTD).unwrap().unwrap();
assert_eq!(count, 10);
assert_eq!(decompressed, raw);
let mut inner = Cursor::new(&decompressed);
for i in 1..=10u32 {
let parsed = WalEntry::read_from(&mut inner, &l0()).unwrap().unwrap();
assert_eq!(parsed.seq, i);
}
assert!(WalEntry::read_from(&mut inner, &l0()).unwrap().is_none());
}
#[test]
fn test_wal_block_single_entry() {
let entry = make_insert_entry(1, "/single", b"one");
let mut raw = Vec::new();
entry.write_to(&mut raw, &l0()).unwrap();
let mut block_buf = Vec::new();
write_wal_block(&mut block_buf, &raw, 1, ALGO_ZSTD).unwrap();
let mut cursor = Cursor::new(&block_buf);
let (decompressed, count) = read_wal_block(&mut cursor, ALGO_ZSTD).unwrap().unwrap();
assert_eq!(count, 1);
assert_eq!(decompressed, raw);
}
#[test]
fn test_wal_block_full_64() {
let entries: Vec<WalEntry> = (1..=64)
.map(|i| make_insert_entry(i, &format!("/node_{}", i), b"payload"))
.collect();
let mut raw = Vec::new();
for e in &entries {
e.write_to(&mut raw, &l0()).unwrap();
}
let mut block_buf = Vec::new();
write_wal_block(&mut block_buf, &raw, 64, ALGO_ZSTD).unwrap();
let mut cursor = Cursor::new(&block_buf);
let (decompressed, count) = read_wal_block(&mut cursor, ALGO_ZSTD).unwrap().unwrap();
assert_eq!(count, 64);
assert_eq!(decompressed, raw);
}
#[test]
fn test_wal_block_eof_returns_none() {
let cursor = Cursor::new(Vec::<u8>::new());
let result = read_wal_block(&mut cursor.clone(), ALGO_ZSTD).unwrap();
assert!(result.is_none());
}
#[test]
fn test_wal_block_truncated_header() {
let mut cursor = Cursor::new(vec![0x01]);
let result = read_wal_block(&mut cursor, ALGO_ZSTD).unwrap();
assert!(result.is_none());
}
#[test]
fn test_wal_block_multiple_blocks() {
let mut all_blocks = Vec::new();
let mut expected_entries = Vec::new();
for block_idx in 0..3u32 {
let count = if block_idx < 2 { 64 } else { 12 };
let entries: Vec<WalEntry> = (0..count)
.map(|i| {
let seq = block_idx * 64 + i + 1;
make_insert_entry(seq, &format!("/b{}/n{}", block_idx, i), b"x")
})
.collect();
let mut raw = Vec::new();
for e in &entries {
e.write_to(&mut raw, &l0()).unwrap();
}
write_wal_block(&mut all_blocks, &raw, count as u16, ALGO_ZSTD).unwrap();
expected_entries.extend(entries);
}
let mut cursor = Cursor::new(&all_blocks);
let mut read_count = 0u32;
for block_idx in 0..3 {
let (decompressed, count) = read_wal_block(&mut cursor, ALGO_ZSTD).unwrap().unwrap();
let expected_count: u16 = if block_idx < 2 { 64 } else { 12 };
assert_eq!(count, expected_count);
let mut inner = Cursor::new(&decompressed);
for _ in 0..count {
let parsed = WalEntry::read_from(&mut inner, &l0()).unwrap().unwrap();
assert_eq!(parsed.seq, read_count + 1);
read_count += 1;
}
}
assert_eq!(read_count, 140);
assert!(read_wal_block(&mut cursor, ALGO_ZSTD).unwrap().is_none());
}
}