[4/7][record][log][store] encode records with packed headers

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]

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::<RecordDataHeader>(), size_of::<u64>());
+        assert_eq!(align_of::<RecordDataHeader>(), align_of::<u64>());
+    }
+
+    #[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);
+    }
+}