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