From f75fba890ea3028510a07a8e4e79305cd0c3ae26 Mon Sep 17 00:00:00 2001 From: Ray Liu Date: Mon, 7 Sep 2026 19:49:50 -0400 Subject: [PATCH] [4/7][record][log][store] encode records with packed headers --- src/log.rs | 68 ++++++++++---- src/record/data_header.rs | 94 +++++++++++++++++++ src/record/mod.rs | 191 ++++++++++++++++++++++++++++---------- src/store.rs | 34 ++++--- 4 files changed, 308 insertions(+), 79 deletions(-) diff --git a/src/log.rs b/src/log.rs index bfeae50..168bd20 100644 --- a/src/log.rs +++ b/src/log.rs @@ -1,6 +1,6 @@ use crate::{ address::Address, - record::{self, RecordRef}, + record::{self, RecordKind, RecordRef}, }; pub(crate) struct Log { @@ -25,6 +25,33 @@ impl Log { } pub(crate) fn append(&mut self, previous: Address, key: &[u8], value: &[u8]) -> Address { + self.append_record(previous, RecordKind::Value, key, value) + } + + pub(crate) fn append_tombstone(&mut self, previous: Address, key: &[u8]) -> Address { + self.append_record(previous, RecordKind::Tombstone, key, &[]) + } + + pub(crate) fn read(&self, address: Address) -> RecordRef<'_> { + assert_ne!(address, Address::INVALID, "cannot read an invalid address"); + + let offset = address.as_offset(); + + assert!( + offset < self.buffer.len(), + "address extends beyond the log tail" + ); + + record::decode(&self.buffer, offset) + } + + fn append_record( + &mut self, + previous: Address, + kind: RecordKind, + key: &[u8], + value: &[u8], + ) -> Address { let address = self.tail(); assert!( @@ -36,28 +63,14 @@ impl Log { let start = self.buffer.len(); let end = start + length; - // validate that the resulting tail is representable Address::from_offset(end); self.buffer.resize(end, 0); - record::encode(&mut self.buffer[start..end], previous, key, value); + record::encode(&mut self.buffer[start..end], previous, kind, key, value); address } - - pub(crate) fn read(&self, address: Address) -> RecordRef<'_> { - assert_ne!(address, Address::INVALID, "cannot read an invalid address"); - - let offset = address.as_offset(); - - assert!( - offset < self.buffer.len(), - "address extends beyond the log tail" - ); - - record::decode(&self.buffer, offset) - } } #[cfg(test)] @@ -79,11 +92,26 @@ mod tests { let record = log.read(address); assert_eq!(address, Address::FIRST_VALID); - assert_eq!(record.previous, Address::INVALID); + assert_eq!(record.previous(), Address::INVALID); + assert!(!record.is_tombstone()); assert_eq!(record.key, b"foo"); assert_eq!(record.value, b"one"); } + #[test] + fn appends_and_reads_tombstone() { + let mut log = Log::new(); + let previous = log.append(Address::INVALID, b"foo", b"one"); + + let address = log.append_tombstone(previous, b"foo"); + let record = log.read(address); + + assert_eq!(record.previous(), previous); + assert!(record.is_tombstone()); + assert_eq!(record.key, b"foo"); + assert!(record.value.is_empty()); + } + #[test] fn appended_records_have_increasing_aligned_addresses() { let mut log = Log::new(); @@ -108,9 +136,9 @@ mod tests { let second = log.append(first, b"foo", b"two"); let third = log.append(second, b"foo", b"three"); - assert_eq!(log.read(third).previous, second); - assert_eq!(log.read(second).previous, first); - assert_eq!(log.read(first).previous, Address::INVALID); + assert_eq!(log.read(third).previous(), second); + assert_eq!(log.read(second).previous(), first); + assert_eq!(log.read(first).previous(), Address::INVALID); } #[test] diff --git a/src/record/data_header.rs b/src/record/data_header.rs index a4d3b10..6a55f07 100644 --- a/src/record/data_header.rs +++ b/src/record/data_header.rs @@ -73,3 +73,97 @@ impl RecordDataHeader { aligned + self.filler_words() * ALIGNMENT } } + +#[cfg(test)] +mod tests { + use super::*; + use std::mem::{align_of, size_of}; + + #[test] + fn data_header_is_one_word() { + assert_eq!(size_of::(), size_of::()); + assert_eq!(align_of::(), align_of::()); + } + + #[test] + fn default_header_has_empty_fields() { + let header = RecordDataHeader::default(); + + assert!(!header.key_is_inline()); + assert!(!header.value_is_inline()); + assert_eq!(header.key_len(), 0); + assert_eq!(header.value_len(), 0); + assert_eq!(header.filler_words(), 0); + } + + #[test] + fn inline_header_round_trips_lengths_and_flags() { + let header = RecordDataHeader::new_inline(123, 45_678); + + assert!(header.key_is_inline()); + assert!(header.value_is_inline()); + assert_eq!(header.key_len(), 123); + assert_eq!(header.value_len(), 45_678); + assert_eq!(header.filler_words(), 0); + } + + #[test] + fn inline_header_uses_expected_bit_positions() { + let header = RecordDataHeader::new_inline(5, 9); + let expected = RecordDataHeader::KEY_INLINE_BIT + | RecordDataHeader::VALUE_INLINE_BIT + | (5 << RecordDataHeader::KEY_LENGTH_SHIFT) + | (9 << RecordDataHeader::VALUE_LENGTH_SHIFT); + + assert_eq!(header.as_raw(), expected); + assert_eq!(RecordDataHeader::from_raw(expected), header); + } + + #[test] + fn maximum_inline_lengths_round_trip() { + let header = RecordDataHeader::new_inline( + RecordDataHeader::MAX_KEY_LENGTH, + RecordDataHeader::MAX_VALUE_LENGTH, + ); + + assert_eq!(header.key_len(), RecordDataHeader::MAX_KEY_LENGTH); + assert_eq!(header.value_len(), RecordDataHeader::MAX_VALUE_LENGTH); + } + + #[test] + #[should_panic(expected = "key length exceeds header field")] + fn rejects_oversized_inline_key() { + RecordDataHeader::new_inline(RecordDataHeader::MAX_KEY_LENGTH + 1, 0); + } + + #[test] + #[should_panic(expected = "value length exceeds header field")] + fn rejects_oversized_inline_value() { + RecordDataHeader::new_inline(0, RecordDataHeader::MAX_VALUE_LENGTH + 1); + } + + #[test] + fn allocated_length_is_aligned() { + for key_len in 0..16 { + for value_len in 0..16 { + let header = RecordDataHeader::new_inline(key_len, value_len); + let allocated_len = header.allocated_len(); + + assert!(allocated_len >= HEADER_SIZE + key_len + value_len); + assert_eq!(allocated_len % ALIGNMENT, 0); + } + } + } + + #[test] + fn filler_words_extend_allocated_length() { + let inline = RecordDataHeader::new_inline(1, 2); + let with_filler = RecordDataHeader::from_raw( + inline.as_raw() | (3 << RecordDataHeader::FILLER_WORDS_SHIFT), + ); + + assert_eq!(inline.allocated_len(), 24); + assert_eq!(with_filler.filler_words(), 3); + assert_eq!(with_filler.allocated_len(), 48); + } +} diff --git a/src/record/mod.rs b/src/record/mod.rs index 200590c..d8766d8 100644 --- a/src/record/mod.rs +++ b/src/record/mod.rs @@ -2,49 +2,69 @@ mod data_header; mod info; use crate::address::Address; +use data_header::RecordDataHeader; +use info::RecordInfo; pub const HEADER_SIZE: usize = 16; pub const ALIGNMENT: usize = 8; +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub(crate) enum RecordKind { + Value, + Tombstone, +} + #[derive(Clone, Copy, Debug, PartialEq, Eq)] pub struct RecordRef<'a> { - pub previous: Address, - pub key: &'a [u8], - pub value: &'a [u8], + info: RecordInfo, + data_header: RecordDataHeader, + pub(crate) key: &'a [u8], + pub(crate) value: &'a [u8], } -pub(crate) fn encoded_len(key_len: usize, value_len: usize) -> usize { - assert!( - key_len <= u32::MAX as usize, - "key length exceeds the record format" - ); - assert!( - value_len <= u32::MAX as usize, - "value length exceeds the record format" - ); +impl RecordRef<'_> { + pub(crate) const fn previous(self) -> Address { + self.info.previous() + } - (HEADER_SIZE + key_len + value_len).next_multiple_of(ALIGNMENT) + pub(crate) const fn is_tombstone(self) -> bool { + self.info.is_tombstone() + } } -pub(crate) fn encode(destination: &mut [u8], previous: Address, key: &[u8], value: &[u8]) { - let required_len = encoded_len(key.len(), value.len()); +pub(crate) const fn encoded_len(key_len: usize, value_len: usize) -> usize { + RecordDataHeader::new_inline(key_len, value_len).allocated_len() +} + +pub(crate) fn encode( + destination: &mut [u8], + previous: Address, + kind: RecordKind, + key: &[u8], + value: &[u8], +) { + let tombstone = matches!(kind, RecordKind::Tombstone); + + if tombstone { + assert!(value.is_empty(), "tombstone records cannot contain a value"); + } + + let info = RecordInfo::new_valid(previous, tombstone); + let data_header = RecordDataHeader::new_inline(key.len(), value.len()); assert_eq!( destination.len(), - required_len, - "record destination has the wrong length" + data_header.allocated_len(), + "record destination has wrong length" ); - let key_len = u32::try_from(key.len()).expect("key length exceeds the record format"); - let value_len = u32::try_from(value.len()).expect("value length exceeds the record format"); - let (header, payload) = destination.split_at_mut(HEADER_SIZE); - header[0..8].copy_from_slice(&previous.as_raw().to_le_bytes()); - header[8..12].copy_from_slice(&key_len.to_le_bytes()); - header[12..16].copy_from_slice(&value_len.to_le_bytes()); + header[0..8].copy_from_slice(&info.as_raw().to_le_bytes()); + header[8..16].copy_from_slice(&data_header.as_raw().to_le_bytes()); let (key_destination, remaining) = payload.split_at_mut(key.len()); + let (value_destination, padding) = remaining.split_at_mut(value.len()); key_destination.copy_from_slice(key); @@ -58,38 +78,34 @@ pub(crate) fn decode(buffer: &[u8], offset: usize) -> RecordRef<'_> { let record = buffer .get(offset..) .expect("record address extends past the log"); + let header = record .get(..HEADER_SIZE) .expect("record header extends past the log"); - let previous = Address::from_raw(u64::from_le_bytes( - header[0..8] - .try_into() - .expect("previous address has the wrong length"), - )); + let info = RecordInfo::from_raw(u64::from_le_bytes(header[0..8].try_into().unwrap())); - let key_len = u32::from_le_bytes( - header[8..12] - .try_into() - .expect("key length has the wrong length"), - ) as usize; + let data_header = + RecordDataHeader::from_raw(u64::from_le_bytes(header[8..16].try_into().unwrap())); - let value_len = u32::from_le_bytes( - header[12..16] - .try_into() - .expect("value length has the wrong length"), - ) as usize; + assert!(info.is_valid(), "record is invalid"); + assert!(data_header.key_is_inline(), "overflow keys not supported"); + assert!( + data_header.value_is_inline(), + "overflow values not supported" + ); - let record_len = encoded_len(key_len, value_len); + let record_len = data_header.allocated_len(); assert!(record_len <= record.len(), "record extends past the log"); let payload = &record[HEADER_SIZE..record_len]; - let (key, remaining) = payload.split_at(key_len); - let (value, _) = remaining.split_at(value_len); + let (key, remaining) = payload.split_at(data_header.key_len()); + let (value, _) = remaining.split_at(data_header.value_len()); RecordRef { - previous, + info, + data_header, key, value, } @@ -135,15 +151,67 @@ mod tests { buffer.resize(offset + length, 0); - encode(&mut buffer[offset..offset + length], previous, key, value); + encode( + &mut buffer[offset..offset + length], + previous, + RecordKind::Value, + key, + value, + ); let record = decode(&buffer, offset); - assert_eq!(record.previous, previous); + assert_eq!(record.previous(), previous); + assert!(!record.is_tombstone()); assert_eq!(record.key, key); assert_eq!(record.value, value); } + #[test] + fn tombstone_round_trips() { + let previous = Address::from_offset(128); + let key = b"hello"; + let length = encoded_len(key.len(), 0); + let offset = Address::FIRST_VALID.as_offset(); + let mut buffer = vec![0; offset + length]; + + encode( + &mut buffer[offset..offset + length], + previous, + RecordKind::Tombstone, + key, + &[], + ); + + let record = decode(&buffer, offset); + + assert_eq!(record.previous(), previous); + assert!(record.is_tombstone()); + assert_eq!(record.key, key); + assert!(record.value.is_empty()); + } + + #[test] + fn empty_value_is_not_a_tombstone() { + let key = b"hello"; + let length = encoded_len(key.len(), 0); + let offset = Address::FIRST_VALID.as_offset(); + let mut buffer = vec![0; offset + length]; + + encode( + &mut buffer[offset..offset + length], + Address::INVALID, + RecordKind::Value, + key, + &[], + ); + + let record = decode(&buffer, offset); + + assert!(!record.is_tombstone()); + assert!(record.value.is_empty()); + } + #[test] fn padding_is_zeroed() { let key = b"a"; @@ -153,7 +221,13 @@ mod tests { // fill with a nonzero value so the test proves encode clears padding let mut destination = vec![0xff; length]; - encode(&mut destination, Address::INVALID, key, value); + encode( + &mut destination, + Address::INVALID, + RecordKind::Value, + key, + value, + ); let data_end = HEADER_SIZE + key.len() + value.len(); @@ -174,19 +248,38 @@ mod tests { encode( &mut buffer[offset..offset + length], Address::INVALID, + RecordKind::Value, &key, &value, ); let record = decode(&buffer, offset); - assert_eq!(record.previous, Address::INVALID); + assert_eq!(record.previous(), Address::INVALID); + assert!(!record.is_tombstone()); assert_eq!(record.key, key); assert_eq!(record.value, value); } } } + #[test] + #[should_panic(expected = "tombstone records cannot contain a value")] + fn rejects_tombstone_with_value() { + let key = b"hello"; + let value = b"value"; + let length = encoded_len(key.len(), value.len()); + let mut destination = vec![0; length]; + + encode( + &mut destination, + Address::INVALID, + RecordKind::Tombstone, + key, + value, + ); + } + #[test] #[should_panic(expected = "record address is not aligned")] fn rejects_unaligned_offset() { @@ -203,9 +296,11 @@ mod tests { #[should_panic(expected = "record extends past the log")] fn rejects_truncated_body() { let mut buffer = vec![0; HEADER_SIZE]; + let info = RecordInfo::new_valid(Address::INVALID, false); + let data_header = RecordDataHeader::new_inline(10, 20); - buffer[8..12].copy_from_slice(&10_u32.to_le_bytes()); - buffer[12..16].copy_from_slice(&20_u32.to_le_bytes()); + buffer[0..8].copy_from_slice(&info.as_raw().to_le_bytes()); + buffer[8..16].copy_from_slice(&data_header.as_raw().to_le_bytes()); decode(&buffer, 0); } diff --git a/src/store.rs b/src/store.rs index 76008b2..5e206d0 100644 --- a/src/store.rs +++ b/src/store.rs @@ -14,19 +14,15 @@ impl Store { } pub fn get(&self, key: &[u8]) -> Option<&[u8]> { - let mut address = self.index.head(hash(key)); - - while address != Address::INVALID { - let record = self.log.read(address); - - if record.key == key { - return Some(record.value); - } + let hash = hash(key); + let address = self.find_address(hash, key)?; + let record = self.log.read(address); - address = record.previous; + if record.is_tombstone() { + None + } else { + Some(record.value) } - - None } pub fn upsert(&mut self, key: &[u8], value: &[u8]) { @@ -36,6 +32,22 @@ impl Store { self.index.set_head(hash, address); } + + fn find_address(&self, hash: u64, key: &[u8]) -> Option
{ + let mut address = self.index.head(hash); + + while address != Address::INVALID { + let record = self.log.read(address); + + if record.key == key { + return Some(address); + } + + address = record.previous(); + } + + None + } } #[cfg(test)]