1use crate::{
16 DATA_START, FORMAT_VERSION, MAGIC, MIN_READER_VERSION, SUPERBLOCK_LEN, checksum_skipping,
17 get_u16, get_u32, get_u64, is_legal_page_size, put_u16, put_u32, put_u64,
18};
19use yo_common::{Code, Error, Result, SLOT_COUNT};
20
21pub const CRC_OFFSET: usize = 16380;
23
24pub const DEFAULT_SHARD_TABLE_OFF: u32 = 160;
26
27pub const CHECKPOINT_ENTRY_LEN: usize = 64;
29
30pub mod superblock_flags {
32 pub const CLEAN_SHUTDOWN: u32 = 1 << 0;
37 pub const ENCRYPTED: u32 = 1 << 1;
39 pub const HAS_ARCHIVAL: u32 = 1 << 2;
41 pub const TIERING_ENGAGED: u32 = 1 << 3;
43}
44
45#[derive(Debug, Clone, PartialEq, Eq)]
52pub struct Superblock {
53 pub format_version: u32,
55 pub min_reader_version: u32,
57 pub page_size: u32,
59 pub shard_count: u32,
61 pub seq: u64,
63 pub file_size: u64,
65 pub file_uuid: [u8; 16],
67 pub created_unix_ms: u64,
69 pub checkpoint_unix_ms: u64,
71 pub db_count: u32,
73 pub flags: u32,
75 pub replid_hi: u64,
77 pub replid_lo: u64,
79 pub replid_ext: u32,
81 pub shard_table_off: u32,
83 pub shard_table_len: u16,
85 pub catalog_addr: u64,
87 pub free_list_addr: u64,
89 pub archival_root: u64,
91}
92
93impl Default for Superblock {
94 fn default() -> Self {
95 Superblock {
96 format_version: FORMAT_VERSION,
97 min_reader_version: MIN_READER_VERSION,
98 page_size: crate::DEFAULT_PAGE_SIZE,
99 shard_count: 1,
100 seq: 0,
101 file_size: DATA_START,
102 file_uuid: [0; 16],
103 created_unix_ms: 0,
104 checkpoint_unix_ms: 0,
105 db_count: 16,
106 flags: 0,
107 replid_hi: 0,
108 replid_lo: 0,
109 replid_ext: 0,
110 shard_table_off: DEFAULT_SHARD_TABLE_OFF,
111 shard_table_len: 0,
112 catalog_addr: 0,
113 free_list_addr: 0,
114 archival_root: 0,
115 }
116 }
117}
118
119impl Superblock {
120 pub fn encode(&self, slot: &mut [u8]) {
130 assert_eq!(slot.len(), SUPERBLOCK_LEN, "a superblock slot is 16 KiB");
131 slot[..16].copy_from_slice(&MAGIC);
132 put_u32(slot, 16, self.format_version);
133 put_u32(slot, 20, self.min_reader_version);
134 put_u32(slot, 24, self.page_size);
135 put_u32(slot, 28, self.shard_count);
136 put_u64(slot, 32, self.seq);
137 put_u64(slot, 40, self.file_size);
138 slot[48..64].copy_from_slice(&self.file_uuid);
139 put_u64(slot, 64, self.created_unix_ms);
140 put_u64(slot, 72, self.checkpoint_unix_ms);
141 put_u32(slot, 80, self.db_count);
142 put_u32(slot, 84, self.flags);
143 put_u64(slot, 88, self.replid_hi);
144 put_u64(slot, 96, self.replid_lo);
145 put_u32(slot, 104, self.replid_ext);
146 put_u32(slot, 108, self.shard_table_off);
147 put_u64(slot, 112, self.catalog_addr);
148 put_u64(slot, 120, self.free_list_addr);
149 put_u64(slot, 128, self.archival_root);
150 slot[136..156].fill(0);
155 put_u16(slot, 156, self.shard_table_len);
156 put_u16(slot, 158, 0);
157 }
158
159 pub fn decode(slot: &[u8]) -> Result<Superblock> {
166 if slot.len() != SUPERBLOCK_LEN {
167 return Err(Error::new(
168 Code::Invalid,
169 "a superblock slot is 16384 bytes",
170 ));
171 }
172 if slot[..16] != MAGIC {
173 return Err(Error::new(
174 Code::Invalid,
175 "not a .yo file: the first sixteen bytes are not the magic",
176 ));
177 }
178 let want = get_u32(slot, CRC_OFFSET);
179 let got = checksum_skipping(slot, CRC_OFFSET);
180 if want != got {
181 return Err(Error::new(Code::Corrupt, "superblock checksum mismatch")
182 .with_detail(format!("stored={want:#010x} computed={got:#010x}")));
183 }
184
185 let min_reader_version = get_u32(slot, 20);
186 if min_reader_version > FORMAT_VERSION {
187 return Err(
188 Error::new(Code::VersionTooNew, "this file needs a newer yo to read it")
189 .with_detail(format!(
190 "file_min_reader={min_reader_version} this_reader={FORMAT_VERSION}"
191 )),
192 );
193 }
194
195 let page_size = get_u32(slot, 24);
196 if !is_legal_page_size(page_size) {
197 return Err(
198 Error::new(Code::Corrupt, "the segment size is not a legal value")
199 .with_detail(format!("page_size={page_size}")),
200 );
201 }
202 let shard_count = get_u32(slot, 28);
203 if shard_count == 0 {
204 return Err(Error::new(Code::Corrupt, "a file with no shards"));
205 }
206
207 let mut file_uuid = [0u8; 16];
208 file_uuid.copy_from_slice(&slot[48..64]);
209
210 Ok(Superblock {
211 format_version: get_u32(slot, 16),
212 min_reader_version,
213 page_size,
214 shard_count,
215 seq: get_u64(slot, 32),
216 file_size: get_u64(slot, 40),
217 file_uuid,
218 created_unix_ms: get_u64(slot, 64),
219 checkpoint_unix_ms: get_u64(slot, 72),
220 db_count: get_u32(slot, 80),
221 flags: get_u32(slot, 84),
222 replid_hi: get_u64(slot, 88),
223 replid_lo: get_u64(slot, 96),
224 replid_ext: get_u32(slot, 104),
225 shard_table_off: get_u32(slot, 108),
226 shard_table_len: get_u16(slot, 156),
227 catalog_addr: get_u64(slot, 112),
228 free_list_addr: get_u64(slot, 120),
229 archival_root: get_u64(slot, 128),
230 })
231 }
232
233 #[must_use]
235 pub const fn checkpoint_off(&self) -> usize {
236 self.shard_table_off as usize + self.shard_table_len as usize
237 }
238
239 #[must_use]
245 pub fn checkpoints_fit(&self) -> bool {
246 let end = self
247 .checkpoint_off()
248 .saturating_add(self.shard_count as usize * CHECKPOINT_ENTRY_LEN);
249 self.shard_table_off as usize >= 160 && end <= CRC_OFFSET
250 }
251}
252
253pub fn seal(slot: &mut [u8]) -> u32 {
259 assert_eq!(slot.len(), SUPERBLOCK_LEN, "a superblock slot is 16 KiB");
260 let crc = checksum_skipping(slot, CRC_OFFSET);
261 put_u32(slot, CRC_OFFSET, crc);
262 crc
263}
264
265pub fn pick(a: &[u8], b: &[u8]) -> Result<(usize, Superblock)> {
278 match (Superblock::decode(a), Superblock::decode(b)) {
279 (Ok(sa), Ok(sb)) => {
280 if sb.seq > sa.seq {
285 Ok((1, sb))
286 } else {
287 Ok((0, sa))
288 }
289 }
290 (Ok(sa), Err(_)) => Ok((0, sa)),
291 (Err(_), Ok(sb)) => Ok((1, sb)),
292 (Err(ea), Err(eb)) => Err(ea.with_detail(format!("slot B: {eb}"))),
293 }
294}
295
296#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
301pub struct CheckpointEntry {
302 pub log_begin: u64,
304 pub log_head: u64,
306 pub log_read_only: u64,
308 pub log_tail: u64,
310 pub index_image_addr: u64,
312 pub index_image_len: u64,
314 pub key_count: u64,
317 pub epoch: u32,
319}
320
321impl CheckpointEntry {
322 pub fn encode(&self, buf: &mut [u8]) {
328 assert_eq!(buf.len(), CHECKPOINT_ENTRY_LEN, "an entry is 64 bytes");
329 put_u64(buf, 0, self.log_begin);
330 put_u64(buf, 8, self.log_head);
331 put_u64(buf, 16, self.log_read_only);
332 put_u64(buf, 24, self.log_tail);
333 put_u64(buf, 32, self.index_image_addr);
334 put_u64(buf, 40, self.index_image_len);
335 put_u64(buf, 48, self.key_count);
336 put_u32(buf, 56, self.epoch);
337 let crc = checksum_skipping(buf, 60);
338 put_u32(buf, 60, crc);
339 }
340
341 pub fn decode(buf: &[u8]) -> Result<CheckpointEntry> {
347 if buf.len() != CHECKPOINT_ENTRY_LEN {
348 return Err(Error::new(Code::Invalid, "a checkpoint entry is 64 bytes"));
349 }
350 let want = get_u32(buf, 60);
351 let got = checksum_skipping(buf, 60);
352 if want != got {
353 return Err(Error::new(
354 Code::Corrupt,
355 "checkpoint entry checksum mismatch",
356 ));
357 }
358 let e = CheckpointEntry {
359 log_begin: get_u64(buf, 0),
360 log_head: get_u64(buf, 8),
361 log_read_only: get_u64(buf, 16),
362 log_tail: get_u64(buf, 24),
363 index_image_addr: get_u64(buf, 32),
364 index_image_len: get_u64(buf, 40),
365 key_count: get_u64(buf, 48),
366 epoch: get_u32(buf, 56),
367 };
368 if !e.addresses_are_ordered() {
369 return Err(
370 Error::new(Code::Corrupt, "the four log addresses are not in order").with_detail(
371 format!(
372 "begin={} head={} read_only={} tail={}",
373 e.log_begin, e.log_head, e.log_read_only, e.log_tail
374 ),
375 ),
376 );
377 }
378 Ok(e)
379 }
380
381 #[must_use]
383 pub const fn addresses_are_ordered(&self) -> bool {
384 self.log_begin <= self.log_head
385 && self.log_head <= self.log_read_only
386 && self.log_read_only <= self.log_tail
387 }
388}
389
390pub fn encode_shard_table(slot_shard: &[u16], out: &mut [u8]) -> Result<usize> {
404 if slot_shard.len() != SLOT_COUNT as usize {
405 return Err(Error::new(
406 Code::Invalid,
407 "the shard table has one entry per Redis slot",
408 )
409 .with_detail(format!("got={} want={SLOT_COUNT}", slot_shard.len())));
410 }
411 let mut n = 0usize;
412 let mut i = 0usize;
413 while i < slot_shard.len() {
414 let shard = slot_shard[i];
415 let mut run = 1usize;
416 while i + run < slot_shard.len() && slot_shard[i + run] == shard {
417 run += 1;
418 }
419 if n + 4 > out.len() {
422 return Err(Error::new(
423 Code::Full,
424 "the shard table does not fit in the superblock",
425 ));
426 }
427 put_u16(out, n, run as u16);
428 put_u16(out, n + 2, shard);
429 n += 4;
430 i += run;
431 }
432 Ok(n)
433}
434
435pub fn decode_shard_table(bytes: &[u8], shard_count: u32, out: &mut [u16]) -> Result<()> {
444 if out.len() != SLOT_COUNT as usize {
445 return Err(Error::new(
446 Code::Invalid,
447 "the output needs one entry per Redis slot",
448 ));
449 }
450 if !bytes.len().is_multiple_of(4) {
451 return Err(Error::new(
452 Code::Corrupt,
453 "the shard table is not a whole number of runs",
454 ));
455 }
456 let mut filled = 0usize;
457 for pair in bytes.as_chunks::<4>().0 {
458 let run = u16::from_le_bytes([pair[0], pair[1]]) as usize;
459 let shard = u16::from_le_bytes([pair[2], pair[3]]);
460 if run == 0 {
461 return Err(Error::new(Code::Corrupt, "a zero length run"));
462 }
463 if u32::from(shard) >= shard_count {
464 return Err(
465 Error::new(Code::Corrupt, "a slot points at a shard that is not there")
466 .with_detail(format!("shard={shard} shard_count={shard_count}")),
467 );
468 }
469 if filled + run > out.len() {
470 return Err(Error::new(
471 Code::Corrupt,
472 "the runs cover more than 16384 slots",
473 ));
474 }
475 out[filled..filled + run].fill(shard);
476 filled += run;
477 }
478 if filled != out.len() {
479 return Err(
480 Error::new(Code::Corrupt, "the runs do not cover every slot")
481 .with_detail(format!("covered={filled} want={SLOT_COUNT}")),
482 );
483 }
484 Ok(())
485}
486
487#[cfg(test)]
488mod tests {
489 use super::*;
490 use crate::DEFAULT_PAGE_SIZE;
491
492 #[cfg(miri)]
496 const STEP: usize = 2039;
497 #[cfg(not(miri))]
498 const STEP: usize = 1;
499
500 #[cfg(miri)]
504 const BY_HAND: [usize; 9] = [
505 0,
506 15,
507 16,
508 20,
509 24,
510 CRC_OFFSET,
511 CRC_OFFSET + 3,
512 16382,
513 SUPERBLOCK_LEN - 1,
514 ];
515 #[cfg(not(miri))]
516 const BY_HAND: [usize; 0] = [];
517
518 fn a_slot() -> ([u8; SUPERBLOCK_LEN], Superblock) {
519 let sb = Superblock {
520 page_size: DEFAULT_PAGE_SIZE,
521 shard_count: 8,
522 seq: 42,
523 file_size: DATA_START + 2048 * u64::from(DEFAULT_PAGE_SIZE),
524 file_uuid: [0xab; 16],
525 created_unix_ms: 1_700_000_000_000,
526 checkpoint_unix_ms: 1_700_000_001_000,
527 flags: superblock_flags::CLEAN_SHUTDOWN,
528 replid_hi: 0x0123_4567_89ab_cdef,
529 replid_lo: 0xfedc_ba98_7654_3210,
530 replid_ext: 0xdead_beef,
531 shard_table_len: 4,
532 catalog_addr: 32768,
533 free_list_addr: 65536,
534 ..Superblock::default()
535 };
536 let mut buf = [0u8; SUPERBLOCK_LEN];
537 sb.encode(&mut buf);
538 seal(&mut buf);
539 (buf, sb)
540 }
541
542 #[test]
543 fn a_superblock_round_trips_field_for_field() {
544 let (buf, sb) = a_slot();
545 assert_eq!(Superblock::decode(&buf).unwrap(), sb);
546 }
547
548 #[test]
549 fn every_field_lands_at_the_offset_the_specification_names() {
550 let (buf, sb) = a_slot();
551 assert_eq!(&buf[..16], &MAGIC);
552 assert_eq!(get_u32(&buf, 16), sb.format_version);
553 assert_eq!(get_u32(&buf, 20), sb.min_reader_version);
554 assert_eq!(get_u32(&buf, 24), sb.page_size);
555 assert_eq!(get_u32(&buf, 28), sb.shard_count);
556 assert_eq!(get_u64(&buf, 32), sb.seq);
557 assert_eq!(get_u64(&buf, 40), sb.file_size);
558 assert_eq!(&buf[48..64], &sb.file_uuid);
559 assert_eq!(get_u64(&buf, 64), sb.created_unix_ms);
560 assert_eq!(get_u64(&buf, 72), sb.checkpoint_unix_ms);
561 assert_eq!(get_u32(&buf, 80), sb.db_count);
562 assert_eq!(get_u32(&buf, 84), sb.flags);
563 assert_eq!(get_u64(&buf, 88), sb.replid_hi);
564 assert_eq!(get_u64(&buf, 96), sb.replid_lo);
565 assert_eq!(get_u32(&buf, 104), sb.replid_ext);
566 assert_eq!(get_u32(&buf, 108), sb.shard_table_off);
567 assert_eq!(get_u64(&buf, 112), sb.catalog_addr);
568 assert_eq!(get_u64(&buf, 120), sb.free_list_addr);
569 assert_eq!(get_u64(&buf, 128), sb.archival_root);
570 assert!(buf[136..156].iter().all(|&b| b == 0), "reserved is zero");
571 assert_eq!(get_u16(&buf, 156), sb.shard_table_len);
572 assert_eq!(
573 get_u32(&buf, CRC_OFFSET),
574 checksum_skipping(&buf, CRC_OFFSET)
575 );
576 }
577
578 #[test]
579 fn a_flipped_bit_anywhere_in_the_slot_is_caught() {
580 let (good, _) = a_slot();
581 for i in (0..SUPERBLOCK_LEN).step_by(STEP).chain(BY_HAND) {
593 let mut bad = good;
594 bad[i] ^= 0x40;
595 let err = Superblock::decode(&bad).unwrap_err();
596 assert!(
599 matches!(err.code(), Code::Corrupt | Code::Invalid),
600 "byte {i} was not caught: {err}"
601 );
602 }
603 }
604
605 #[test]
606 fn a_file_from_the_future_is_refused_by_name() {
607 let (mut buf, _) = a_slot();
608 put_u32(&mut buf, 20, FORMAT_VERSION + 1);
609 seal(&mut buf);
610 let err = Superblock::decode(&buf).unwrap_err();
611 assert_eq!(err.code(), Code::VersionTooNew);
612 assert!(err.detail().unwrap().contains("file_min_reader=2"));
613 }
614
615 #[test]
616 fn corruption_is_reported_before_the_version_is_believed() {
617 let (mut buf, _) = a_slot();
621 put_u32(&mut buf, 20, 0x4000_0000);
622 let err = Superblock::decode(&buf).unwrap_err();
623 assert_eq!(err.code(), Code::Corrupt);
624 }
625
626 #[test]
627 fn a_nonsense_page_size_is_refused_even_with_a_good_checksum() {
628 let (mut buf, _) = a_slot();
629 put_u32(&mut buf, 24, 12288);
630 seal(&mut buf);
631 let err = Superblock::decode(&buf).unwrap_err();
632 assert_eq!(err.code(), Code::Corrupt);
633 assert!(err.detail().unwrap().contains("12288"));
634 }
635
636 #[test]
637 fn the_higher_sequence_number_wins_and_a_bad_slot_does_not_vote() {
638 let (mut a, _) = a_slot();
639 let mut b = a;
640 put_u64(&mut b, 32, 43);
641 seal(&mut b);
642 assert_eq!(pick(&a, &b).unwrap().0, 1, "B is newer");
643 assert_eq!(pick(&b, &a).unwrap().0, 0, "B is still newer");
644
645 b[9000] ^= 0xff;
648 assert_eq!(pick(&a, &b).unwrap().0, 0);
649
650 a[9000] ^= 0xff;
652 let err = pick(&a, &b).unwrap_err();
653 assert_eq!(err.code(), Code::Corrupt);
654 assert!(err.detail().unwrap().contains("slot B"));
655 }
656
657 #[test]
658 fn equal_sequence_numbers_pick_a_deterministically() {
659 let (a, _) = a_slot();
660 let b = a;
661 assert_eq!(pick(&a, &b).unwrap().0, 0);
662 }
663
664 #[test]
665 fn a_checkpoint_entry_round_trips_and_fills_one_cache_line() {
666 assert_eq!(CHECKPOINT_ENTRY_LEN, yo_common::CACHE_LINE);
667 let e = CheckpointEntry {
668 log_begin: 0,
669 log_head: 1 << 20,
670 log_read_only: 2 << 20,
671 log_tail: 3 << 20,
672 index_image_addr: 1 << 30,
673 index_image_len: 4096,
674 key_count: 1_000_000,
675 epoch: 17,
676 };
677 let mut buf = [0u8; CHECKPOINT_ENTRY_LEN];
678 e.encode(&mut buf);
679 assert_eq!(CheckpointEntry::decode(&buf).unwrap(), e);
680 }
681
682 #[test]
683 fn out_of_order_log_addresses_are_corruption_not_an_empty_log() {
684 let e = CheckpointEntry {
685 log_begin: 100,
686 log_head: 50,
687 log_read_only: 200,
688 log_tail: 300,
689 ..CheckpointEntry::default()
690 };
691 let mut buf = [0u8; CHECKPOINT_ENTRY_LEN];
692 e.encode(&mut buf);
693 let err = CheckpointEntry::decode(&buf).unwrap_err();
694 assert_eq!(err.code(), Code::Corrupt);
695 assert!(err.detail().unwrap().contains("head=50"));
696 }
697
698 #[test]
699 fn a_flipped_bit_in_a_checkpoint_entry_is_caught() {
700 let e = CheckpointEntry {
701 log_tail: 1 << 40,
702 ..CheckpointEntry::default()
703 };
704 let mut good = [0u8; CHECKPOINT_ENTRY_LEN];
705 e.encode(&mut good);
706 for i in 0..CHECKPOINT_ENTRY_LEN {
707 let mut bad = good;
708 bad[i] ^= 1;
709 assert!(
710 CheckpointEntry::decode(&bad).is_err(),
711 "byte {i} was not caught"
712 );
713 }
714 }
715
716 #[test]
717 fn a_contiguous_shard_table_costs_four_bytes_per_shard() {
718 let shards = 64u16;
719 let per = SLOT_COUNT as usize / shards as usize;
720 let table: Vec<u16> = (0..SLOT_COUNT as usize).map(|s| (s / per) as u16).collect();
721
722 let mut out = [0u8; 1024];
723 let n = encode_shard_table(&table, &mut out).unwrap();
724 assert_eq!(n, 4 * shards as usize);
725
726 let mut back = vec![0u16; SLOT_COUNT as usize];
727 decode_shard_table(&out[..n], u32::from(shards), &mut back).unwrap();
728 assert_eq!(back, table);
729 }
730
731 #[test]
732 fn one_shard_is_one_run() {
733 let table = vec![0u16; SLOT_COUNT as usize];
734 let mut out = [0u8; 64];
735 let n = encode_shard_table(&table, &mut out).unwrap();
736 assert_eq!(n, 4);
737 assert_eq!(get_u16(&out, 0), SLOT_COUNT, "one run of every slot");
738 }
739
740 #[test]
741 fn the_worst_case_table_is_refused_rather_than_truncated() {
742 let table: Vec<u16> = (0..SLOT_COUNT).map(|s| s % 2).collect();
745 let mut out = [0u8; 1024];
746 let err = encode_shard_table(&table, &mut out).unwrap_err();
747 assert_eq!(err.code(), Code::Full);
748 }
749
750 #[test]
751 fn a_table_that_does_not_cover_every_slot_is_corruption() {
752 let bytes = [0x00u8, 0x40, 0x00, 0x00]; let mut out = vec![0u16; SLOT_COUNT as usize];
754 decode_shard_table(&bytes, 1, &mut out).unwrap();
755
756 let short = [0x10u8, 0x00, 0x00, 0x00]; let err = decode_shard_table(&short, 1, &mut out).unwrap_err();
758 assert_eq!(err.code(), Code::Corrupt);
759 assert!(err.detail().unwrap().contains("covered=16"));
760 }
761
762 #[test]
763 fn a_slot_pointing_at_a_shard_that_is_not_there_is_corruption() {
764 let bytes = [0x00u8, 0x40, 0x09, 0x00]; let mut out = vec![0u16; SLOT_COUNT as usize];
766 let err = decode_shard_table(&bytes, 8, &mut out).unwrap_err();
767 assert_eq!(err.code(), Code::Corrupt);
768 assert!(err.detail().unwrap().contains("shard=9"));
769 }
770
771 #[test]
772 fn a_zero_length_run_is_corruption_and_not_an_infinite_loop() {
773 let bytes = [0x00u8, 0x00, 0x00, 0x00];
774 let mut out = vec![0u16; SLOT_COUNT as usize];
775 assert_eq!(
776 decode_shard_table(&bytes, 1, &mut out).unwrap_err().code(),
777 Code::Corrupt
778 );
779 }
780
781 #[test]
782 fn the_checkpoint_entries_have_to_fit_where_the_header_says_they_do() {
783 let sb = Superblock {
784 shard_count: 8,
785 shard_table_len: 32,
786 ..Superblock::default()
787 };
788 assert_eq!(sb.checkpoint_off(), 192);
789 assert!(sb.checkpoints_fit());
790
791 let too_many = Superblock {
794 shard_count: 253,
795 ..sb.clone()
796 };
797 assert!(!too_many.checkpoints_fit());
798
799 let overlapping = Superblock {
800 shard_table_off: 8,
801 ..sb
802 };
803 assert!(!overlapping.checkpoints_fit(), "the table is in the header");
804 }
805}