1use crate::{align_up, get_u8, get_u16, get_u32, get_u64, put_u8, put_u16, put_u32, put_u64};
46use yo_common::{Code, Error, Result, crc32c};
47
48pub const HEADER_LEN: usize = 16;
50
51pub const HEADER_LEN_TTL: usize = 24;
53
54pub const TRAILER_LEN: usize = 4;
56
57pub const MAX_KEY_LEN: usize = u16::MAX as usize;
64
65pub mod record_flags {
67 pub const TIERED: u8 = 1 << 0;
69 pub const COMPRESSED: u8 = 1 << 1;
71 pub const HAS_TTL: u8 = 1 << 2;
73 pub const SHAPE_TAGGED: u8 = 1 << 3;
75 pub const CHECKSUMMED: u8 = 1 << 4;
82}
83
84#[derive(Debug, Clone, Copy, PartialEq, Eq)]
91#[repr(u8)]
92pub enum RecordKind {
93 String = 0,
95 CollectionChunk = 1,
97 Document = 2,
100 Vector = 3,
102 GraphNode = 4,
104 GraphAdj = 5,
106 Checkpoint = 6,
108 Tombstone = 7,
110 IndexDelta = 8,
112}
113
114impl RecordKind {
115 pub const ALL: [RecordKind; 9] = [
117 RecordKind::String,
118 RecordKind::CollectionChunk,
119 RecordKind::Document,
120 RecordKind::Vector,
121 RecordKind::GraphNode,
122 RecordKind::GraphAdj,
123 RecordKind::Checkpoint,
124 RecordKind::Tombstone,
125 RecordKind::IndexDelta,
126 ];
127
128 #[must_use]
130 pub const fn from_u8(b: u8) -> Option<RecordKind> {
131 match b {
132 0 => Some(RecordKind::String),
133 1 => Some(RecordKind::CollectionChunk),
134 2 => Some(RecordKind::Document),
135 3 => Some(RecordKind::Vector),
136 4 => Some(RecordKind::GraphNode),
137 5 => Some(RecordKind::GraphAdj),
138 6 => Some(RecordKind::Checkpoint),
139 7 => Some(RecordKind::Tombstone),
140 8 => Some(RecordKind::IndexDelta),
141 _ => None,
142 }
143 }
144
145 #[must_use]
147 pub const fn as_u8(self) -> u8 {
148 self as u8
149 }
150
151 #[must_use]
157 pub const fn carries_a_key(self) -> bool {
158 !matches!(self, RecordKind::CollectionChunk)
159 }
160}
161
162pub fn total_len(flags: u8, klen: usize, vlen: usize) -> Result<usize> {
169 if klen > MAX_KEY_LEN {
170 return Err(
171 Error::new(Code::Invalid, "the key is longer than 65535 bytes")
172 .with_detail(format!("klen={klen}")),
173 );
174 }
175 let n = header_len(flags) + klen + vlen + trailer_len(flags);
176 if n > u32::MAX as usize {
177 return Err(Error::new(
178 Code::Invalid,
179 "the record does not fit in a u32",
180 ));
181 }
182 Ok(n)
183}
184
185#[inline]
187#[must_use]
188pub const fn header_len(flags: u8) -> usize {
189 if flags & record_flags::HAS_TTL != 0 {
190 HEADER_LEN_TTL
191 } else {
192 HEADER_LEN
193 }
194}
195
196#[inline]
204#[must_use]
205pub const fn trailer_len(flags: u8) -> usize {
206 if flags & record_flags::CHECKSUMMED != 0 {
207 TRAILER_LEN
208 } else {
209 0
210 }
211}
212
213#[derive(Debug, Clone, Copy, PartialEq, Eq)]
215pub struct RecordHeader {
216 pub kind: u8,
220 pub flags: u8,
222 pub prev: u64,
224 pub ttl_ms: u64,
226}
227
228impl RecordHeader {
229 #[must_use]
231 pub const fn new(kind: RecordKind) -> RecordHeader {
232 RecordHeader {
233 kind: kind.as_u8(),
234 flags: record_flags::CHECKSUMMED,
235 prev: 0,
236 ttl_ms: 0,
237 }
238 }
239
240 #[must_use]
242 pub const fn with_ttl(mut self, unix_ms: u64) -> RecordHeader {
243 self.flags |= record_flags::HAS_TTL;
244 self.ttl_ms = unix_ms;
245 self
246 }
247
248 #[must_use]
250 pub const fn after(mut self, prev: u64) -> RecordHeader {
251 self.prev = prev;
252 self
253 }
254
255 pub fn fill(&self, buf: &mut [u8], key: &[u8], value: &[u8]) -> Result<usize> {
272 let flags = self.flags | record_flags::CHECKSUMMED;
277 let n = total_len(flags, key.len(), value.len())?;
278 if buf.len() < n {
279 return Err(
280 Error::new(Code::Full, "the record does not fit in the buffer")
281 .with_detail(format!("need={n} have={}", buf.len())),
282 );
283 }
284 let h = header_len(flags);
285 put_u8(buf, 4, self.kind);
286 put_u8(buf, 5, flags);
287 put_u16(buf, 6, key.len() as u16);
288 put_u64(buf, 8, self.prev);
289 if flags & record_flags::HAS_TTL != 0 {
290 put_u64(buf, 16, self.ttl_ms);
291 }
292 buf[h..h + key.len()].copy_from_slice(key);
293 let v = h + key.len();
294 buf[v..v + value.len()].copy_from_slice(value);
295
296 let c = crc32c(0, &(n as u32).to_le_bytes());
303 let c = crc32c(c, &buf[4..n - TRAILER_LEN]);
304 put_u32(buf, n - TRAILER_LEN, c);
305 Ok(n)
306 }
307}
308
309pub fn seal_len(buf: &mut [u8], len: usize) {
319 assert!(len != 0, "zero is the end of log sentinel, not a length");
320 put_u32(buf, 0, len as u32);
321}
322
323#[derive(Debug, Clone, Copy, PartialEq, Eq)]
325pub struct RecordRef<'a> {
326 pub len: u32,
328 pub kind: u8,
330 pub flags: u8,
332 pub prev: u64,
334 pub ttl_ms: Option<u64>,
336 pub key: &'a [u8],
338 pub value: &'a [u8],
340}
341
342impl<'a> RecordRef<'a> {
343 pub fn parse(bytes: &'a [u8]) -> Result<Option<RecordRef<'a>>> {
355 if bytes.len() < 4 {
356 return Ok(None);
357 }
358 let len = get_u32(bytes, 0) as usize;
359 if len == 0 {
360 return Ok(None);
361 }
362 let flags = get_u8(bytes, 5);
363 if flags & record_flags::CHECKSUMMED == 0 {
364 return Err(
369 Error::new(Code::Corrupt, "a record with its checksum flag clear")
370 .with_detail(format!("flags={flags:#04x}")),
371 );
372 }
373 let h = header_len(flags);
374 let t = trailer_len(flags);
375 let klen = get_u16(bytes, 6) as usize;
376
377 if len < h + klen + t {
378 return Err(
379 Error::new(Code::Corrupt, "the record is shorter than its own header")
380 .with_detail(format!("len={len} header={h} klen={klen} trailer={t}")),
381 );
382 }
383 if len > bytes.len() {
384 return Err(
388 Error::new(Code::Corrupt, "the record runs past the end of the page")
389 .with_detail(format!("len={len} available={}", bytes.len())),
390 );
391 }
392
393 if flags & record_flags::CHECKSUMMED != 0 {
394 let want = get_u32(bytes, len - TRAILER_LEN);
395 let got = crc32c(0, &bytes[..len - TRAILER_LEN]);
396 if want != got {
397 return Err(Error::new(Code::Corrupt, "record checksum mismatch")
398 .with_detail(format!("stored={want:#010x} computed={got:#010x}")));
399 }
400 }
401
402 let ttl_ms = if flags & record_flags::HAS_TTL != 0 {
403 Some(get_u64(bytes, 16))
404 } else {
405 None
406 };
407
408 Ok(Some(RecordRef {
409 len: len as u32,
410 kind: get_u8(bytes, 4),
411 flags,
412 prev: get_u64(bytes, 8),
413 ttl_ms,
414 key: &bytes[h..h + klen],
415 value: &bytes[h + klen..len - t],
416 }))
417 }
418
419 #[must_use]
424 pub fn stride(&self) -> usize {
425 align_up(self.len as usize)
426 }
427
428 #[must_use]
433 pub fn kind(&self) -> Option<RecordKind> {
434 RecordKind::from_u8(self.kind)
435 }
436
437 #[must_use]
439 pub fn is_tombstone(&self) -> bool {
440 self.kind == RecordKind::Tombstone.as_u8()
441 }
442
443 #[must_use]
445 pub fn is_tiered(&self) -> bool {
446 self.flags & record_flags::TIERED != 0
447 }
448}
449
450pub struct RecordIter<'a> {
456 bytes: &'a [u8],
457 at: usize,
458 done: bool,
459}
460
461impl<'a> RecordIter<'a> {
462 #[must_use]
464 pub const fn new(bytes: &'a [u8]) -> RecordIter<'a> {
465 RecordIter {
466 bytes,
467 at: 0,
468 done: false,
469 }
470 }
471
472 #[must_use]
474 pub const fn offset(&self) -> usize {
475 self.at
476 }
477}
478
479impl<'a> Iterator for RecordIter<'a> {
480 type Item = Result<RecordRef<'a>>;
481
482 fn next(&mut self) -> Option<Self::Item> {
483 if self.done || self.at >= self.bytes.len() {
484 return None;
485 }
486 match RecordRef::parse(&self.bytes[self.at..]) {
487 Ok(Some(r)) => {
488 self.at += r.stride();
489 Some(Ok(r))
490 }
491 Ok(None) => {
492 self.done = true;
493 None
494 }
495 Err(e) => {
496 self.done = true;
497 Some(Err(e))
498 }
499 }
500 }
501}
502
503#[cfg(test)]
504mod tests {
505 use super::*;
506 use crate::RECORD_ALIGN;
507
508 fn write(h: RecordHeader, key: &[u8], value: &[u8]) -> Vec<u8> {
509 let mut buf = vec![0u8; 4096];
510 let n = h.fill(&mut buf, key, value).unwrap();
511 seal_len(&mut buf, n);
512 buf.truncate(align_up(n));
513 buf
514 }
515
516 #[test]
517 fn a_record_round_trips() {
518 let h = RecordHeader::new(RecordKind::String).after(4096);
519 let buf = write(h, b"greeting", b"hello");
520 let r = RecordRef::parse(&buf).unwrap().unwrap();
521 assert_eq!(r.kind(), Some(RecordKind::String));
522 assert_eq!(r.key, b"greeting");
523 assert_eq!(r.value, b"hello");
524 assert_eq!(r.prev, 4096);
525 assert_eq!(r.ttl_ms, None);
526 assert!(!r.is_tombstone());
527 }
528
529 #[test]
530 fn every_field_lands_where_the_specification_says() {
531 let h = RecordHeader::new(RecordKind::Document)
532 .with_ttl(1_700_000_000_000)
533 .after(0x1122_3344_5566_7788);
534 let buf = write(h, b"k", b"v");
535 assert_eq!(get_u32(&buf, 0) as usize, HEADER_LEN_TTL + 1 + 1 + 4);
536 assert_eq!(get_u8(&buf, 4), 2, "document is kind 2");
537 assert_eq!(
538 get_u8(&buf, 5),
539 record_flags::CHECKSUMMED | record_flags::HAS_TTL
540 );
541 assert_eq!(get_u16(&buf, 6), 1);
542 assert_eq!(get_u64(&buf, 8), 0x1122_3344_5566_7788);
543 assert_eq!(get_u64(&buf, 16), 1_700_000_000_000);
544 assert_eq!(buf[24], b'k');
545 assert_eq!(buf[25], b'v');
546 }
547
548 #[test]
549 fn a_ttl_costs_eight_bytes_and_moves_the_key() {
550 let plain = write(RecordHeader::new(RecordKind::String), b"key", b"value");
551 let ttl = write(
552 RecordHeader::new(RecordKind::String).with_ttl(1),
553 b"key",
554 b"value",
555 );
556 assert_eq!(get_u32(&ttl, 0) - get_u32(&plain, 0), 8);
557 let r = RecordRef::parse(&ttl).unwrap().unwrap();
558 assert_eq!(r.ttl_ms, Some(1));
559 assert_eq!(r.key, b"key");
560 assert_eq!(r.value, b"value");
561 }
562
563 #[test]
564 fn a_zero_length_is_the_end_of_the_log_and_not_an_error() {
565 assert!(RecordRef::parse(&[0u8; 64]).unwrap().is_none());
566 assert!(RecordRef::parse(&[]).unwrap().is_none());
567 assert!(RecordRef::parse(&[1, 2, 3]).unwrap().is_none());
568 }
569
570 #[test]
571 fn len_is_exact_so_a_value_of_any_length_survives() {
572 for n in 0..=32usize {
576 let value: Vec<u8> = (0..n).map(|i| i as u8).collect();
577 let buf = write(RecordHeader::new(RecordKind::String), b"k", &value);
578 let r = RecordRef::parse(&buf).unwrap().unwrap();
579 assert_eq!(r.value, &value[..], "value of {n} bytes came back wrong");
580 assert_eq!(r.value.len(), n);
581 }
582 }
583
584 #[test]
585 fn the_stride_is_aligned_even_when_the_length_is_not() {
586 let buf = write(RecordHeader::new(RecordKind::String), b"k", b"abc");
587 let r = RecordRef::parse(&buf).unwrap().unwrap();
588 assert_eq!(r.len as usize, HEADER_LEN + 1 + 3 + TRAILER_LEN);
589 assert_eq!(r.len % RECORD_ALIGN as u32, 0);
590
591 let buf = write(RecordHeader::new(RecordKind::String), b"k", b"ab");
592 let r = RecordRef::parse(&buf).unwrap().unwrap();
593 assert_eq!(r.len as usize, 23);
594 assert_eq!(r.stride(), 24, "the padding is between records, not inside");
595 }
596
597 #[test]
598 fn walking_a_page_by_stride_stays_in_step() {
599 let mut page = vec![0u8; 4096];
602 let mut at = 0usize;
603 let mut written = Vec::new();
604 for i in 0..40usize {
605 let key = format!("key{i}");
606 let value = vec![b'v'; i];
607 let h = RecordHeader::new(RecordKind::String).after(at as u64);
608 let n = h.fill(&mut page[at..], key.as_bytes(), &value).unwrap();
609 seal_len(&mut page[at..], n);
610 written.push((key, value));
611 at += align_up(n);
612 }
613
614 let got: Vec<_> = RecordIter::new(&page).map(|r| r.unwrap()).collect();
615 assert_eq!(got.len(), 40);
616 for (r, (key, value)) in got.iter().zip(&written) {
617 assert_eq!(r.key, key.as_bytes());
618 assert_eq!(r.value, &value[..]);
619 }
620 }
621
622 #[test]
623 fn the_iterator_reports_where_it_stopped() {
624 let mut page = vec![0u8; 512];
625 let h = RecordHeader::new(RecordKind::String);
626 let n = h.fill(&mut page, b"a", b"bb").unwrap();
627 seal_len(&mut page, n);
628 let mut it = RecordIter::new(&page);
629 assert!(it.next().is_some());
630 assert!(it.next().is_none());
631 assert_eq!(it.offset(), align_up(n), "the tail is here");
632 }
633
634 #[test]
635 fn a_flipped_bit_anywhere_in_a_checksummed_record_is_caught() {
636 let good = write(
637 RecordHeader::new(RecordKind::String).with_ttl(99),
638 b"the key",
639 b"the value, which is long enough to be worth checking",
640 );
641 let len = get_u32(&good, 0) as usize;
642 for i in 0..len {
643 let mut bad = good.clone();
644 bad[i] ^= 0x20;
645 let r = RecordRef::parse(&bad);
646 match r {
651 Err(_) => {}
652 Ok(None) => {}
653 Ok(Some(rec)) => panic!("byte {i} was not caught, got {rec:?}"),
654 }
655 }
656 }
657
658 #[test]
663 fn a_header_written_without_the_checksum_flag_gets_one_anyway() {
664 let h = RecordHeader {
665 kind: RecordKind::String.as_u8(),
666 flags: 0,
667 prev: 0,
668 ttl_ms: 0,
669 };
670 let buf = write(h, b"k", b"v");
671 assert_eq!(get_u32(&buf, 0) as usize, HEADER_LEN + 2 + TRAILER_LEN);
672 assert_ne!(get_u8(&buf, 5) & record_flags::CHECKSUMMED, 0);
673 let r = RecordRef::parse(&buf).unwrap().unwrap();
674 assert_eq!(r.value, b"v");
675 }
676
677 #[test]
681 fn a_record_with_its_checksum_flag_cleared_is_corruption() {
682 let mut buf = write(RecordHeader::new(RecordKind::String), b"key", b"value");
683 buf[5] &= !record_flags::CHECKSUMMED;
684 let err = RecordRef::parse(&buf).unwrap_err();
685 assert_eq!(err.code(), Code::Corrupt);
686 }
687
688 #[test]
689 fn a_length_that_is_shorter_than_the_header_is_corruption() {
690 let mut buf = write(RecordHeader::new(RecordKind::String), b"key", b"value");
691 put_u32(&mut buf, 0, 12);
692 let err = RecordRef::parse(&buf).unwrap_err();
693 assert_eq!(err.code(), Code::Corrupt);
694 assert!(err.detail().unwrap().contains("len=12"));
695 }
696
697 #[test]
698 fn a_record_cut_off_by_a_torn_write_is_corruption() {
699 let buf = write(RecordHeader::new(RecordKind::String), b"key", b"value");
700 let err = RecordRef::parse(&buf[..8]).unwrap_err();
701 assert_eq!(err.code(), Code::Corrupt);
702 assert!(err.detail().unwrap().contains("available=8"));
703 }
704
705 #[test]
706 fn an_unknown_kind_is_skipped_rather_than_refused() {
707 let h = RecordHeader {
711 kind: 200,
712 flags: record_flags::CHECKSUMMED,
713 prev: 0,
714 ttl_ms: 0,
715 };
716 let mut page = vec![0u8; 512];
717 let n = h.fill(&mut page, b"future", b"stuff").unwrap();
718 seal_len(&mut page, n);
719 let after = align_up(n);
720 let m = RecordHeader::new(RecordKind::String)
721 .fill(&mut page[after..], b"k", b"v")
722 .unwrap();
723 seal_len(&mut page[after..], m);
724
725 let got: Vec<_> = RecordIter::new(&page).map(|r| r.unwrap()).collect();
726 assert_eq!(got.len(), 2);
727 assert_eq!(got[0].kind(), None, "not a kind this version knows");
728 assert_eq!(got[0].key, b"future");
729 assert_eq!(got[1].kind(), Some(RecordKind::String));
730 }
731
732 #[test]
733 fn a_key_larger_than_a_u16_is_refused_rather_than_truncated() {
734 let key = vec![b'k'; MAX_KEY_LEN + 1];
735 let mut buf = vec![0u8; MAX_KEY_LEN + 64];
736 let err = RecordHeader::new(RecordKind::String)
737 .fill(&mut buf, &key, b"v")
738 .unwrap_err();
739 assert_eq!(err.code(), Code::Invalid);
740 }
741
742 #[test]
743 fn a_buffer_with_no_room_says_how_much_it_needed() {
744 let mut buf = [0u8; 8];
745 let err = RecordHeader::new(RecordKind::String)
746 .fill(&mut buf, b"key", b"value")
747 .unwrap_err();
748 assert_eq!(err.code(), Code::Full);
749 assert!(err.detail().unwrap().contains("have=8"));
750 }
751
752 #[test]
753 fn kinds_round_trip_and_chunks_have_no_key() {
754 for k in RecordKind::ALL {
755 assert_eq!(RecordKind::from_u8(k.as_u8()), Some(k));
756 }
757 assert_eq!(RecordKind::from_u8(9), None);
758 assert!(!RecordKind::CollectionChunk.carries_a_key());
759 assert!(RecordKind::String.carries_a_key());
760 assert!(RecordKind::Tombstone.carries_a_key());
761 }
762
763 #[test]
764 fn a_tombstone_is_a_record_with_no_value() {
765 let buf = write(RecordHeader::new(RecordKind::Tombstone), b"gone", b"");
766 let r = RecordRef::parse(&buf).unwrap().unwrap();
767 assert!(r.is_tombstone());
768 assert_eq!(r.value, b"");
769 assert_eq!(r.key, b"gone");
770 }
771}