use super::buffer::{Mutation, WalError};
pub(crate) const TAG_PUT: u8 = 0x01;
pub(crate) const TAG_DELETE: u8 = 0x02;
const U32_SIZE: usize = 4;
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum OperationType {
Put,
Delete,
}
impl OperationType {
pub(crate) const fn tag(self) -> u8 {
match self {
Self::Put => TAG_PUT,
Self::Delete => TAG_DELETE,
}
}
const fn from_tag(tag: u8) -> Result<Self, WalError> {
match tag {
TAG_PUT => Ok(Self::Put),
TAG_DELETE => Ok(Self::Delete),
found => Err(WalError::InvalidTag { found }),
}
}
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub enum WalEntry {
Put {
operation_type: OperationType,
key: Vec<u8>,
value: Vec<u8>,
crc32: u32,
},
Delete {
operation_type: OperationType,
key: Vec<u8>,
crc32: u32,
},
}
impl WalEntry {
pub fn put<K, V>(key: K, value: V) -> Self
where
K: Into<Vec<u8>>,
V: Into<Vec<u8>>,
{
let key = key.into();
let value = value.into();
let operation_type = OperationType::Put;
let crc32 = checksum_for(operation_type, &key, Some(&value));
Self::Put {
operation_type,
key,
value,
crc32,
}
}
pub fn delete<K>(key: K) -> Self
where
K: Into<Vec<u8>>,
{
let key = key.into();
let operation_type = OperationType::Delete;
let crc32 = checksum_for(operation_type, &key, None);
Self::Delete {
operation_type,
key,
crc32,
}
}
#[must_use]
pub fn serialise(&self) -> Vec<u8> {
let mut bytes = self.payload_bytes();
bytes.extend_from_slice(&self.crc32().to_le_bytes());
bytes
}
pub fn deserialise(bytes: &[u8]) -> Result<Self, WalError> {
let payload_len = bytes
.len()
.checked_sub(U32_SIZE)
.ok_or(WalError::Truncated)?;
let (payload, checksum_bytes) = bytes.split_at(payload_len);
let expected = read_u32_le(checksum_bytes)?;
let actual = crc32fast::hash(payload);
if expected != actual {
return Err(WalError::ChecksumMismatch { expected, actual });
}
let mut cursor = ByteCursor::new(payload);
let operation_type = OperationType::from_tag(cursor.read_u8()?)?;
match operation_type {
OperationType::Put => {
let key = cursor.read_len_prefixed_bytes()?.to_vec();
let value = cursor.read_len_prefixed_bytes()?.to_vec();
cursor.finish()?;
Ok(Self::Put {
operation_type,
key,
value,
crc32: expected,
})
}
OperationType::Delete => {
let key = cursor.read_len_prefixed_bytes()?.to_vec();
cursor.finish()?;
Ok(Self::Delete {
operation_type,
key,
crc32: expected,
})
}
}
}
#[must_use]
pub const fn operation_type(&self) -> OperationType {
match self {
Self::Put { operation_type, .. } | Self::Delete { operation_type, .. } => {
*operation_type
}
}
}
#[must_use]
pub fn key(&self) -> &[u8] {
match self {
Self::Put { key, .. } | Self::Delete { key, .. } => key,
}
}
#[must_use]
pub fn value(&self) -> Option<&[u8]> {
match self {
Self::Put { value, .. } => Some(value),
Self::Delete { .. } => None,
}
}
#[must_use]
pub const fn crc32(&self) -> u32 {
match self {
Self::Put { crc32, .. } | Self::Delete { crc32, .. } => *crc32,
}
}
#[must_use]
pub fn computed_crc32(&self) -> u32 {
crc32fast::hash(&self.payload_bytes())
}
fn payload_bytes(&self) -> Vec<u8> {
match self {
Self::Put {
operation_type,
key,
value,
..
} => payload_bytes(*operation_type, key, Some(value)),
Self::Delete {
operation_type,
key,
..
} => payload_bytes(*operation_type, key, None),
}
}
}
impl From<&Mutation> for WalEntry {
fn from(mutation: &Mutation) -> Self {
match mutation {
Mutation::Put { key, value } => Self::put(key.clone(), value.clone()),
Mutation::Delete { key } => Self::delete(key.clone()),
}
}
}
impl From<Mutation> for WalEntry {
fn from(mutation: Mutation) -> Self {
match mutation {
Mutation::Put { key, value } => Self::put(key, value),
Mutation::Delete { key } => Self::delete(key),
}
}
}
fn checksum_for(operation_type: OperationType, key: &[u8], value: Option<&[u8]>) -> u32 {
crc32fast::hash(&payload_bytes(operation_type, key, value))
}
fn payload_bytes(operation_type: OperationType, key: &[u8], value: Option<&[u8]>) -> Vec<u8> {
let value_len = value.map_or(0, <[u8]>::len);
let mut bytes = Vec::with_capacity(1 + U32_SIZE + key.len() + U32_SIZE + value_len);
bytes.push(operation_type.tag());
append_len(&mut bytes, key.len());
bytes.extend_from_slice(key);
if let Some(value) = value {
append_len(&mut bytes, value.len());
bytes.extend_from_slice(value);
}
bytes
}
fn append_len(bytes: &mut Vec<u8>, len: usize) {
bytes.extend_from_slice(&(len as u32).to_le_bytes());
}
fn read_u32_le(bytes: &[u8]) -> Result<u32, WalError> {
if bytes.len() != U32_SIZE {
return Err(WalError::Truncated);
}
Ok(u32::from_le_bytes([bytes[0], bytes[1], bytes[2], bytes[3]]))
}
#[derive(Debug)]
struct ByteCursor<'a> {
bytes: &'a [u8],
offset: usize,
}
impl<'a> ByteCursor<'a> {
const fn new(bytes: &'a [u8]) -> Self {
Self { bytes, offset: 0 }
}
fn read_u8(&mut self) -> Result<u8, WalError> {
let byte = *self.bytes.get(self.offset).ok_or(WalError::Truncated)?;
self.offset += 1;
Ok(byte)
}
fn read_u32(&mut self) -> Result<u32, WalError> {
read_u32_le(self.read_bytes(U32_SIZE)?)
}
fn read_len_prefixed_bytes(&mut self) -> Result<&'a [u8], WalError> {
let len = usize::try_from(self.read_u32()?).map_err(|_| WalError::LengthOverflow)?;
self.read_bytes(len)
}
fn read_bytes(&mut self, len: usize) -> Result<&'a [u8], WalError> {
let end = self
.offset
.checked_add(len)
.ok_or(WalError::LengthOverflow)?;
let slice = self
.bytes
.get(self.offset..end)
.ok_or(WalError::Truncated)?;
self.offset = end;
Ok(slice)
}
const fn finish(&self) -> Result<(), WalError> {
let trailing = self.bytes.len().saturating_sub(self.offset);
if trailing == 0 {
Ok(())
} else {
Err(WalError::TrailingBytes { trailing })
}
}
}
#[cfg(test)]
mod tests {
use super::{OperationType, TAG_DELETE, TAG_PUT, WalEntry, payload_bytes};
use crate::wal::buffer::{Mutation, WalError};
fn entry_bytes(payload: &[u8]) -> Vec<u8> {
let mut bytes = Vec::from(payload);
bytes.extend_from_slice(&crc32fast::hash(payload).to_le_bytes());
bytes
}
#[test]
fn put_entry_contains_operation_key_value_and_crc() {
let entry = WalEntry::put(b"key".to_vec(), b"value".to_vec());
let payload = payload_bytes(OperationType::Put, b"key", Some(b"value"));
if let WalEntry::Put {
operation_type,
key,
value,
crc32,
} = entry
{
assert_eq!(operation_type, OperationType::Put);
assert_eq!(key, b"key".to_vec());
assert_eq!(value, b"value".to_vec());
assert_eq!(crc32, crc32fast::hash(&payload));
} else {
unreachable_put_entry();
}
}
#[test]
fn delete_entry_contains_operation_key_and_crc() {
let entry = WalEntry::delete(b"key".to_vec());
let payload = payload_bytes(OperationType::Delete, b"key", None);
if let WalEntry::Delete {
operation_type,
key,
crc32,
} = entry
{
assert_eq!(operation_type, OperationType::Delete);
assert_eq!(key, b"key".to_vec());
assert_eq!(crc32, crc32fast::hash(&payload));
} else {
unreachable_delete_entry();
}
}
#[test]
fn crc_covers_operation_type_key_and_value_payload() {
let entry = WalEntry::put(b"ab".to_vec(), b"xyz".to_vec());
let mut payload = vec![TAG_PUT];
payload.extend_from_slice(&2u32.to_le_bytes());
payload.extend_from_slice(b"ab");
payload.extend_from_slice(&3u32.to_le_bytes());
payload.extend_from_slice(b"xyz");
assert_eq!(entry.computed_crc32(), crc32fast::hash(&payload));
assert_eq!(entry.crc32(), entry.computed_crc32());
}
#[test]
fn put_serialise_deserialise_round_trips() -> Result<(), WalError> {
let entry = WalEntry::put(b"alpha".to_vec(), b"beta".to_vec());
let decoded = WalEntry::deserialise(&entry.serialise())?;
assert_eq!(decoded, entry);
Ok(())
}
#[test]
fn delete_serialise_deserialise_round_trips() -> Result<(), WalError> {
let entry = WalEntry::delete(b"alpha".to_vec());
let decoded = WalEntry::deserialise(&entry.serialise())?;
assert_eq!(decoded, entry);
Ok(())
}
#[test]
fn serialised_put_is_tagged_and_length_prefixed() {
let bytes = WalEntry::put(b"ab".to_vec(), b"xyz".to_vec()).serialise();
let mut expected = vec![TAG_PUT];
expected.extend_from_slice(&2u32.to_le_bytes());
expected.extend_from_slice(b"ab");
expected.extend_from_slice(&3u32.to_le_bytes());
expected.extend_from_slice(b"xyz");
expected.extend_from_slice(&crc32fast::hash(&expected).to_le_bytes());
assert_eq!(bytes, expected);
}
#[test]
fn serialised_delete_is_tagged_and_length_prefixed() {
let bytes = WalEntry::delete(b"ab".to_vec()).serialise();
let mut expected = vec![TAG_DELETE];
expected.extend_from_slice(&2u32.to_le_bytes());
expected.extend_from_slice(b"ab");
expected.extend_from_slice(&crc32fast::hash(&expected).to_le_bytes());
assert_eq!(bytes, expected);
}
#[test]
fn deserialise_detects_checksum_mismatch() {
let mut bytes = WalEntry::put(b"k".to_vec(), b"v".to_vec()).serialise();
if let Some(byte) = bytes.last_mut() {
*byte ^= 0xff;
}
assert!(matches!(
WalEntry::deserialise(&bytes),
Err(WalError::ChecksumMismatch { .. })
));
}
#[test]
fn deserialise_rejects_invalid_operation_tag() {
let payload = [0xff, 0, 0, 0, 0];
let bytes = entry_bytes(&payload);
assert!(matches!(
WalEntry::deserialise(&bytes),
Err(WalError::InvalidTag { found: 0xff })
));
}
#[test]
fn deserialise_rejects_truncated_input() {
assert!(matches!(
WalEntry::deserialise(&[TAG_PUT]),
Err(WalError::Truncated)
));
}
#[test]
fn deserialise_rejects_trailing_bytes() {
let payload = [TAG_DELETE, 0, 0, 0, 0, 0xaa];
let bytes = entry_bytes(&payload);
assert!(matches!(
WalEntry::deserialise(&bytes),
Err(WalError::TrailingBytes { trailing: 1 })
));
}
#[test]
fn wal_entry_is_debug() {
let entry = WalEntry::delete(b"debug".to_vec());
assert!(format!("{entry:?}").contains("Delete"));
}
#[test]
fn mutation_converts_to_wal_entry() {
let mutation = Mutation::Put {
key: b"k".to_vec(),
value: b"v".to_vec(),
};
let entry = WalEntry::from(&mutation);
assert_eq!(entry, WalEntry::put(b"k".to_vec(), b"v".to_vec()));
}
fn unreachable_put_entry() {
unreachable!("WalEntry::put must construct the Put variant");
}
fn unreachable_delete_entry() {
unreachable!("WalEntry::delete must construct the Delete variant");
}
}