1use std::collections::BTreeMap;
19use std::ops::Range;
20
21use bytes::{BufMut, Bytes, BytesMut};
22use snafu::{OptionExt, ensure};
23use store_api::storage::RegionId;
24
25use crate::error::{CorruptedWalObjectSnafu, Result};
26
27const HEADER_MAGIC: &[u8; 8] = b"GTWALOBJ";
28const TRAILER_MAGIC: &[u8; 8] = b"GTWALTRL";
29const FORMAT_VERSION: u16 = 1;
30const NO_PREDECESSOR: u64 = u64::MAX;
33
34pub(crate) const HEADER_LEN: usize = 8 + 2 + 8 + 8 + 8 + 8 + 4;
37pub(crate) const TRAILER_LEN: usize = 8 + 8 + 4 + 4 + 8;
40const SEGMENT_HEADER_LEN: usize = 8 + 4;
41pub(crate) const FOOTER_ENTRY_LEN: usize = 8 + 8 + 8 + 4 + 8 + 8 + 4;
44#[cfg(test)]
46const FOOTER_COUNT_LEN: usize = 4;
47
48#[derive(Debug, Clone, PartialEq, Eq)]
50pub(crate) struct Header {
51 pub(crate) object_seq: u64,
52 pub(crate) epoch: u64,
56 pub(crate) prev: Option<ChainLink>,
58}
59
60#[derive(Debug, Clone, Copy, PartialEq, Eq)]
63pub(crate) struct ChainLink {
64 pub(crate) object_seq: u64,
65 pub(crate) epoch: u64,
66}
67
68#[derive(Debug, Clone, PartialEq, Eq)]
70pub(crate) struct Record {
71 pub(crate) region_id: RegionId,
72 pub(crate) entry_id: u64,
73 pub(crate) payload: Bytes,
74}
75
76#[derive(Debug, Clone, PartialEq, Eq)]
78pub(crate) struct FooterEntry {
79 pub(crate) region_id: RegionId,
80 pub(crate) min_entry_id: u64,
81 pub(crate) max_entry_id: u64,
82 pub(crate) entry_count: u32,
83 pub(crate) segment_offset: u64,
84 pub(crate) segment_len: u64,
85 pub(crate) segment_crc32: u32,
86}
87
88#[derive(Debug, Clone, Copy, PartialEq, Eq)]
90pub(crate) struct FixedTrailer {
91 pub(crate) footer_offset: u64,
92 pub(crate) footer_len: u64,
93 pub(crate) footer_crc32: u32,
94 pub(crate) object_crc32: u32,
95}
96
97#[derive(Debug, Clone, PartialEq, Eq)]
99pub(crate) struct EncodedObject {
100 pub(crate) bytes: Bytes,
101 pub(crate) footer: Vec<FooterEntry>,
102 pub(crate) trailer: FixedTrailer,
103}
104
105#[cfg(test)]
107#[derive(Debug, Clone, PartialEq, Eq)]
108pub(crate) struct DecodedObject {
109 pub(crate) header: Header,
110 pub(crate) records: Vec<Record>,
111 pub(crate) footer: Vec<FooterEntry>,
112}
113
114pub(crate) fn encode_object(header: Header, records: &[Record]) -> Result<EncodedObject> {
118 let mut grouped = BTreeMap::<RegionId, Vec<&Record>>::new();
119 for record in records {
120 grouped.entry(record.region_id).or_default().push(record);
121 }
122 for (region_id, records) in &mut grouped {
123 records.sort_unstable_by_key(|record| record.entry_id);
124 ensure!(
125 !records
126 .windows(2)
127 .any(|pair| pair[0].entry_id >= pair[1].entry_id),
128 CorruptedWalObjectSnafu {
129 reason: format!("non-monotonic entry id for region {region_id}"),
130 }
131 );
132 }
133
134 let mut output = BytesMut::new();
135 encode_header(&header, &mut output);
136
137 let mut directory = Vec::with_capacity(grouped.len());
138 for (region_id, records) in grouped {
139 let segment_offset = to_u64(output.len(), "segment offset")?;
140 let entry_count = to_u32(records.len(), "segment entry count")?;
141 let segment_start = output.len();
142
143 output.put_u64(region_id.as_u64());
144 output.put_u32(entry_count);
145 for record in &records {
146 output.put_u64(record.entry_id);
147 output.put_u32(to_u32(record.payload.len(), "record payload")?);
148 output.put_slice(&record.payload);
149 }
150
151 let segment = &output[segment_start..];
152 directory.push(FooterEntry {
153 region_id,
154 min_entry_id: records[0].entry_id,
155 max_entry_id: records[records.len() - 1].entry_id,
156 entry_count,
157 segment_offset,
158 segment_len: to_u64(segment.len(), "segment length")?,
159 segment_crc32: crc32fast::hash(segment),
160 });
161 }
162
163 let footer_offset = to_u64(output.len(), "footer offset")?;
164 let footer_start = output.len();
165 encode_footer(&directory, &mut output)?;
166 let footer = &output[footer_start..];
167 let trailer = FixedTrailer {
168 footer_offset,
169 footer_len: to_u64(footer.len(), "footer length")?,
170 footer_crc32: crc32fast::hash(footer),
171 object_crc32: 0,
172 };
173 let trailer = FixedTrailer {
174 object_crc32: object_crc32(&output, trailer),
175 ..trailer
176 };
177 encode_trailer(trailer, &mut output);
178
179 Ok(EncodedObject {
180 bytes: output.freeze(),
181 footer: directory,
182 trailer,
183 })
184}
185
186#[cfg(test)]
190pub(crate) fn object_len(footer: &[FooterEntry]) -> u64 {
191 let segments = footer.iter().map(|entry| entry.segment_len).sum::<u64>();
192 let framing = HEADER_LEN + FOOTER_COUNT_LEN + footer.len() * FOOTER_ENTRY_LEN + TRAILER_LEN;
193 segments + framing as u64
194}
195
196pub(crate) fn decode_header(bytes: &[u8]) -> Result<Header> {
199 ensure!(
200 bytes.len() >= HEADER_LEN,
201 CorruptedWalObjectSnafu {
202 reason: truncated("header", bytes.len(), HEADER_LEN),
203 }
204 );
205
206 let mut reader = Reader::new(&bytes[..HEADER_LEN]);
207 let magic = reader.take(HEADER_MAGIC.len(), "header")?;
208 ensure!(
209 magic == HEADER_MAGIC,
210 CorruptedWalObjectSnafu {
211 reason: invalid_magic("header", HEADER_MAGIC, magic),
212 }
213 );
214 let version = reader.u16("header")?;
215 ensure!(
216 version == FORMAT_VERSION,
217 CorruptedWalObjectSnafu {
218 reason: format!(
219 "unsupported format version {version}, expected version {FORMAT_VERSION}"
220 ),
221 }
222 );
223 let object_seq = reader.u64("header")?;
224 let epoch = reader.u64("header")?;
225 let prev_object_seq = reader.u64("header")?;
226 let prev_epoch = reader.u64("header")?;
227 let header_crc32 = reader.u32("header")?;
228 let checksum = crc32fast::hash(&bytes[..HEADER_LEN - 4]);
229 ensure!(
230 checksum == header_crc32,
231 CorruptedWalObjectSnafu {
232 reason: checksum_mismatch("header", header_crc32, checksum),
233 }
234 );
235 let prev = (prev_object_seq != NO_PREDECESSOR).then_some(ChainLink {
236 object_seq: prev_object_seq,
237 epoch: prev_epoch,
238 });
239 Ok(Header {
240 object_seq,
241 epoch,
242 prev,
243 })
244}
245
246pub(crate) fn decode_trailer(bytes: &[u8]) -> Result<FixedTrailer> {
248 ensure!(
249 bytes.len() >= TRAILER_LEN,
250 CorruptedWalObjectSnafu {
251 reason: truncated("trailer", bytes.len(), TRAILER_LEN),
252 }
253 );
254 ensure!(
255 bytes.len() <= TRAILER_LEN,
256 CorruptedWalObjectSnafu {
257 reason: trailing_bytes("trailer", bytes.len(), TRAILER_LEN),
258 }
259 );
260
261 let mut reader = Reader::new(bytes);
262 let trailer = FixedTrailer {
263 footer_offset: reader.u64("trailer")?,
264 footer_len: reader.u64("trailer")?,
265 footer_crc32: reader.u32("trailer")?,
266 object_crc32: reader.u32("trailer")?,
267 };
268 let magic = reader.take(TRAILER_MAGIC.len(), "trailer")?;
269 ensure!(
270 magic == TRAILER_MAGIC,
271 CorruptedWalObjectSnafu {
272 reason: invalid_magic("trailer", TRAILER_MAGIC, magic),
273 }
274 );
275 Ok(trailer)
276}
277
278pub(crate) fn decode_footer(bytes: &[u8], trailer: FixedTrailer) -> Result<Vec<FooterEntry>> {
280 let footer_len = to_usize(trailer.footer_len, "footer length")?;
281 ensure!(
282 bytes.len() >= footer_len,
283 CorruptedWalObjectSnafu {
284 reason: truncated("footer", bytes.len(), footer_len),
285 }
286 );
287 ensure!(
288 bytes.len() <= footer_len,
289 CorruptedWalObjectSnafu {
290 reason: trailing_bytes("footer", bytes.len(), footer_len),
291 }
292 );
293 let checksum = crc32fast::hash(bytes);
294 ensure!(
295 checksum == trailer.footer_crc32,
296 CorruptedWalObjectSnafu {
297 reason: checksum_mismatch("footer", trailer.footer_crc32, checksum),
298 }
299 );
300
301 let mut reader = Reader::new(bytes);
302 let count = reader.u32("footer")? as usize;
303 let expected_len = count
304 .checked_mul(FOOTER_ENTRY_LEN)
305 .and_then(|len| len.checked_add(4))
306 .with_context(|| CorruptedWalObjectSnafu {
307 reason: format!("footer declares {count} entries, which overflows its length"),
308 })?;
309 ensure!(
310 expected_len == bytes.len(),
311 CorruptedWalObjectSnafu {
312 reason: format!(
313 "footer declares {count} entries, expected length {expected_len}, actual {}",
314 bytes.len()
315 ),
316 }
317 );
318
319 let mut directory = Vec::with_capacity(count);
320 let mut previous_region = None;
321 for _ in 0..count {
322 let region_id = RegionId::from_u64(reader.u64("footer")?);
323 if let Some(previous) = previous_region {
324 ensure!(
325 previous < region_id,
326 CorruptedWalObjectSnafu {
327 reason: if previous == region_id {
328 format!("duplicate footer entry for region {region_id}")
329 } else {
330 format!(
331 "footer entries are not ordered by region, {previous} precedes {region_id}"
332 )
333 },
334 }
335 );
336 }
337 previous_region = Some(region_id);
338
339 let entry = FooterEntry {
340 region_id,
341 min_entry_id: reader.u64("footer")?,
342 max_entry_id: reader.u64("footer")?,
343 entry_count: reader.u32("footer")?,
344 segment_offset: reader.u64("footer")?,
345 segment_len: reader.u64("footer")?,
346 segment_crc32: reader.u32("footer")?,
347 };
348 ensure!(
349 entry.entry_count > 0
350 && entry.min_entry_id <= entry.max_entry_id
351 && entry.segment_len >= SEGMENT_HEADER_LEN as u64,
352 CorruptedWalObjectSnafu {
353 reason: format!(
354 "invalid footer entry for region {}, entry ids {}..={}, {} entries, segment length {}",
355 entry.region_id,
356 entry.min_entry_id,
357 entry.max_entry_id,
358 entry.entry_count,
359 entry.segment_len
360 ),
361 }
362 );
363 directory.push(entry);
364 }
365 Ok(directory)
366}
367
368pub(crate) fn decode_segment(bytes: &[u8], entry: &FooterEntry) -> Result<Vec<Record>> {
370 let segment_len = to_usize(entry.segment_len, "segment length")?;
371 ensure!(
372 bytes.len() >= segment_len,
373 CorruptedWalObjectSnafu {
374 reason: truncated("segment", bytes.len(), segment_len),
375 }
376 );
377 ensure!(
378 bytes.len() <= segment_len,
379 CorruptedWalObjectSnafu {
380 reason: trailing_bytes("segment", bytes.len(), segment_len),
381 }
382 );
383 let checksum = crc32fast::hash(bytes);
384 ensure!(
385 checksum == entry.segment_crc32,
386 CorruptedWalObjectSnafu {
387 reason: checksum_mismatch(
388 &format!("segment of region {}", entry.region_id),
389 entry.segment_crc32,
390 checksum
391 ),
392 }
393 );
394
395 let mut reader = Reader::new(bytes);
396 let region_id = RegionId::from_u64(reader.u64("segment")?);
397 let count = reader.u32("segment")?;
398 ensure!(
399 region_id == entry.region_id && count == entry.entry_count,
400 CorruptedWalObjectSnafu {
401 reason: format!(
402 "segment holds region {region_id} with {count} entries, footer expects region {} with {} entries",
403 entry.region_id, entry.entry_count
404 ),
405 }
406 );
407 let max_count = reader.remaining_len() / (8 + 4);
410 ensure!(
411 count as usize <= max_count,
412 CorruptedWalObjectSnafu {
413 reason: format!(
414 "segment declares {count} entries but only holds bytes for {max_count}"
415 ),
416 }
417 );
418
419 let mut records = Vec::with_capacity(count as usize);
420 let mut previous_id = None;
421 for _ in 0..count {
422 let entry_id = reader.u64("segment")?;
423 ensure!(
424 previous_id.is_none_or(|previous| previous < entry_id),
425 CorruptedWalObjectSnafu {
426 reason: format!("non-monotonic entry id for region {region_id}"),
427 }
428 );
429 previous_id = Some(entry_id);
430 let payload_len = reader.u32("segment")? as usize;
431 let payload = Bytes::copy_from_slice(reader.take(payload_len, "segment")?);
432 records.push(Record {
433 region_id,
434 entry_id,
435 payload,
436 });
437 }
438 ensure!(
439 reader.is_empty(),
440 CorruptedWalObjectSnafu {
441 reason: trailing_bytes("segment", bytes.len(), bytes.len() - reader.remaining_len()),
442 }
443 );
444 ensure!(
445 records.first().map(|record| record.entry_id) == Some(entry.min_entry_id)
446 && records.last().map(|record| record.entry_id) == Some(entry.max_entry_id),
447 CorruptedWalObjectSnafu {
448 reason: format!(
449 "segment of region {region_id} holds entry ids {:?}..={:?}, footer expects {}..={}",
450 records.first().map(|record| record.entry_id),
451 records.last().map(|record| record.entry_id),
452 entry.min_entry_id,
453 entry.max_entry_id
454 ),
455 }
456 );
457 Ok(records)
458}
459
460pub(crate) const MIN_OBJECT_LEN: usize = HEADER_LEN + 4 + TRAILER_LEN;
463
464pub(crate) fn footer_range(trailer: FixedTrailer, object_len: usize) -> Result<Range<usize>> {
467 ensure!(
468 object_len >= MIN_OBJECT_LEN,
469 CorruptedWalObjectSnafu {
470 reason: truncated("object", object_len, MIN_OBJECT_LEN),
471 }
472 );
473 let trailer_start = object_len - TRAILER_LEN;
474 let footer_start = to_usize(trailer.footer_offset, "footer offset")?;
475 let footer_len = to_usize(trailer.footer_len, "footer length")?;
476 let footer_end =
477 footer_start
478 .checked_add(footer_len)
479 .with_context(|| CorruptedWalObjectSnafu {
480 reason: format!("footer range {footer_start}..{footer_len} overflows the object"),
481 })?;
482 ensure!(
483 footer_start >= HEADER_LEN && footer_end == trailer_start,
484 CorruptedWalObjectSnafu {
485 reason: format!(
486 "invalid footer range {footer_start}..{footer_end}, expected {HEADER_LEN}..{trailer_start}"
487 ),
488 }
489 );
490 Ok(footer_start..footer_end)
491}
492
493pub(crate) fn verify_segment_ranges(footer: &[FooterEntry], footer_start: usize) -> Result<()> {
497 let mut expected_offset = HEADER_LEN;
498 for entry in footer {
499 let start = to_usize(entry.segment_offset, "segment offset")?;
500 let len = to_usize(entry.segment_len, "segment length")?;
501 let end = start
502 .checked_add(len)
503 .with_context(|| CorruptedWalObjectSnafu {
504 reason: format!(
505 "segment range {start}..{len} of region {} overflows the object",
506 entry.region_id
507 ),
508 })?;
509 ensure!(
510 start == expected_offset && end <= footer_start,
511 CorruptedWalObjectSnafu {
512 reason: format!(
513 "invalid segment range {start}..{end} of region {}, expected {expected_offset}..{footer_start}",
514 entry.region_id
515 ),
516 }
517 );
518 expected_offset = end;
519 }
520 ensure!(
521 expected_offset == footer_start,
522 CorruptedWalObjectSnafu {
523 reason: format!(
524 "invalid segment range, segments end at {expected_offset}, footer starts at {footer_start}"
525 ),
526 }
527 );
528 Ok(())
529}
530
531#[cfg(test)]
535pub(crate) fn decode_object(bytes: &[u8]) -> Result<DecodedObject> {
536 ensure!(
537 bytes.len() >= MIN_OBJECT_LEN,
538 CorruptedWalObjectSnafu {
539 reason: truncated("object", bytes.len(), MIN_OBJECT_LEN),
540 }
541 );
542
543 let header = decode_header(bytes)?;
544 let trailer_start = bytes.len() - TRAILER_LEN;
545 let trailer = decode_trailer(&bytes[trailer_start..])?;
546 let footer_range = footer_range(trailer, bytes.len())?;
547 let footer_start = footer_range.start;
548
549 let footer = decode_footer(&bytes[footer_range], trailer)?;
550 verify_segment_ranges(&footer, footer_start)?;
551
552 let mut records = Vec::new();
553 for entry in &footer {
554 let start = entry.segment_offset as usize;
555 let end = start + entry.segment_len as usize;
556 records.extend(decode_segment(&bytes[start..end], entry)?);
557 }
558 let checksum = object_crc32(&bytes[..trailer_start], trailer);
559 ensure!(
560 checksum == trailer.object_crc32,
561 CorruptedWalObjectSnafu {
562 reason: checksum_mismatch("object", trailer.object_crc32, checksum),
563 }
564 );
565
566 Ok(DecodedObject {
567 header,
568 records,
569 footer,
570 })
571}
572
573fn encode_header(header: &Header, output: &mut BytesMut) {
574 let start = output.len();
575 output.put_slice(HEADER_MAGIC);
576 output.put_u16(FORMAT_VERSION);
577 output.put_u64(header.object_seq);
578 output.put_u64(header.epoch);
579 let prev = header.prev.unwrap_or(ChainLink {
580 object_seq: NO_PREDECESSOR,
581 epoch: 0,
582 });
583 output.put_u64(prev.object_seq);
584 output.put_u64(prev.epoch);
585 output.put_u32(crc32fast::hash(&output[start..]));
586}
587
588fn encode_footer(directory: &[FooterEntry], output: &mut BytesMut) -> Result<()> {
589 output.put_u32(to_u32(directory.len(), "footer entry count")?);
590 for entry in directory {
591 output.put_u64(entry.region_id.as_u64());
592 output.put_u64(entry.min_entry_id);
593 output.put_u64(entry.max_entry_id);
594 output.put_u32(entry.entry_count);
595 output.put_u64(entry.segment_offset);
596 output.put_u64(entry.segment_len);
597 output.put_u32(entry.segment_crc32);
598 }
599 Ok(())
600}
601
602fn encode_trailer(trailer: FixedTrailer, output: &mut BytesMut) {
603 output.put_u64(trailer.footer_offset);
604 output.put_u64(trailer.footer_len);
605 output.put_u32(trailer.footer_crc32);
606 output.put_u32(trailer.object_crc32);
607 output.put_slice(TRAILER_MAGIC);
608}
609
610fn object_crc32(bytes_before_trailer: &[u8], trailer: FixedTrailer) -> u32 {
613 let mut checksum = crc32fast::Hasher::new();
614 checksum.update(bytes_before_trailer);
615 checksum.update(&trailer.footer_offset.to_be_bytes());
616 checksum.update(&trailer.footer_len.to_be_bytes());
617 checksum.update(&trailer.footer_crc32.to_be_bytes());
618 checksum.update(TRAILER_MAGIC);
619 checksum.finalize()
620}
621
622fn truncated(part: &str, actual: usize, expected: usize) -> String {
623 format!("truncated {part}, expected at least {expected} bytes, actual {actual}")
624}
625
626fn trailing_bytes(part: &str, actual: usize, expected: usize) -> String {
627 format!("trailing bytes in {part}, expected {expected} bytes, actual {actual}")
628}
629
630fn invalid_magic(part: &str, expected: &[u8], actual: &[u8]) -> String {
631 format!(
632 "invalid {part} magic, expected {:?}, actual {:?}",
633 String::from_utf8_lossy(expected),
634 String::from_utf8_lossy(actual)
635 )
636}
637
638fn checksum_mismatch(part: &str, expected: u32, actual: u32) -> String {
639 format!("{part} checksum mismatch, expected {expected:#010x}, actual {actual:#010x}")
640}
641
642fn to_u32(value: usize, name: &str) -> Result<u32> {
643 u32::try_from(value)
644 .ok()
645 .with_context(|| CorruptedWalObjectSnafu {
646 reason: format!("{name} {value} does not fit into u32"),
647 })
648}
649
650fn to_u64(value: usize, name: &str) -> Result<u64> {
651 u64::try_from(value)
652 .ok()
653 .with_context(|| CorruptedWalObjectSnafu {
654 reason: format!("{name} {value} does not fit into u64"),
655 })
656}
657
658fn to_usize(value: u64, name: &str) -> Result<usize> {
659 usize::try_from(value)
660 .ok()
661 .with_context(|| CorruptedWalObjectSnafu {
662 reason: format!("{name} {value} does not fit into usize"),
663 })
664}
665
666struct Reader<'a> {
667 bytes: &'a [u8],
668 offset: usize,
669}
670
671impl<'a> Reader<'a> {
672 fn new(bytes: &'a [u8]) -> Self {
673 Self { bytes, offset: 0 }
674 }
675
676 fn take(&mut self, len: usize, part: &'static str) -> Result<&'a [u8]> {
677 let end = self
678 .offset
679 .checked_add(len)
680 .with_context(|| CorruptedWalObjectSnafu {
681 reason: truncated(part, self.bytes.len(), len),
682 })?;
683 let value = self
684 .bytes
685 .get(self.offset..end)
686 .with_context(|| CorruptedWalObjectSnafu {
687 reason: truncated(part, self.bytes.len(), end),
688 })?;
689 self.offset = end;
690 Ok(value)
691 }
692
693 fn u16(&mut self, part: &'static str) -> Result<u16> {
694 Ok(u16::from_be_bytes(
695 self.take(2, part)?.try_into().expect("u16 has two bytes"),
696 ))
697 }
698
699 fn u32(&mut self, part: &'static str) -> Result<u32> {
700 Ok(u32::from_be_bytes(
701 self.take(4, part)?.try_into().expect("u32 has four bytes"),
702 ))
703 }
704
705 fn u64(&mut self, part: &'static str) -> Result<u64> {
706 Ok(u64::from_be_bytes(
707 self.take(8, part)?.try_into().expect("u64 has eight bytes"),
708 ))
709 }
710
711 fn is_empty(&self) -> bool {
712 self.offset == self.bytes.len()
713 }
714
715 fn remaining_len(&self) -> usize {
716 self.bytes.len() - self.offset
717 }
718}
719
720#[cfg(test)]
721mod tests {
722 use super::*;
723 use crate::error::Error;
724
725 fn header() -> Header {
726 Header {
727 object_seq: 42,
728 epoch: 5,
729 prev: Some(ChainLink {
730 object_seq: 40,
731 epoch: 4,
732 }),
733 }
734 }
735
736 fn records() -> Vec<Record> {
737 vec![
738 Record {
739 region_id: RegionId::new(2, 1),
740 entry_id: 7,
741 payload: Bytes::from_static(b"region-2"),
742 },
743 Record {
744 region_id: RegionId::new(1, 1),
745 entry_id: 2,
746 payload: Bytes::from_static(b"second"),
747 },
748 Record {
749 region_id: RegionId::new(1, 1),
750 entry_id: 1,
751 payload: Bytes::from_static(b"first"),
752 },
753 ]
754 }
755
756 #[test]
757 fn test_format_round_trip_with_multiple_regions_and_entries() {
758 let encoded = encode_object(header(), &records()).unwrap();
759 let decoded = decode_object(&encoded.bytes).unwrap();
760
761 assert_eq!(header(), decoded.header);
762 assert_eq!(2, decoded.footer.len());
763 assert_eq!(RegionId::new(1, 1), decoded.footer[0].region_id);
764 assert_eq!((1, 2, 2), footer_ids(&decoded.footer[0]));
765 assert_eq!(RegionId::new(2, 1), decoded.footer[1].region_id);
766 assert_eq!((7, 7, 1), footer_ids(&decoded.footer[1]));
767 assert_eq!(
768 vec![
769 (RegionId::new(1, 1), 1, Bytes::from_static(b"first")),
770 (RegionId::new(1, 1), 2, Bytes::from_static(b"second")),
771 (RegionId::new(2, 1), 7, Bytes::from_static(b"region-2")),
772 ],
773 decoded
774 .records
775 .into_iter()
776 .map(|record| (record.region_id, record.entry_id, record.payload))
777 .collect::<Vec<_>>()
778 );
779 }
780
781 #[test]
782 fn test_format_decodes_footer_trailer_and_segment_ranges() {
783 let encoded = encode_object(header(), &records()).unwrap();
784 let trailer_start = encoded.bytes.len() - TRAILER_LEN;
785 let trailer = decode_trailer(&encoded.bytes[trailer_start..]).unwrap();
786 assert_eq!(encoded.trailer, trailer);
787
788 let footer_start = trailer.footer_offset as usize;
789 let footer_end = footer_start + trailer.footer_len as usize;
790 let footer = decode_footer(&encoded.bytes[footer_start..footer_end], trailer).unwrap();
791 assert_eq!(encoded.footer, footer);
792
793 let entry = &footer[0];
794 let start = entry.segment_offset as usize;
795 let end = start + entry.segment_len as usize;
796 let segment = decode_segment(&encoded.bytes[start..end], entry).unwrap();
797 assert_eq!(
798 vec![1, 2],
799 segment
800 .iter()
801 .map(|entry| entry.entry_id)
802 .collect::<Vec<_>>()
803 );
804 }
805
806 #[test]
807 fn test_format_rejects_truncation() {
808 let encoded = encode_object(header(), &records()).unwrap();
809 assert_corrupted(
810 decode_object(&encoded.bytes[..HEADER_LEN - 1]),
811 "truncated object",
812 );
813 assert_corrupted(
814 decode_trailer(&encoded.bytes[encoded.bytes.len() - TRAILER_LEN + 1..]),
815 "truncated trailer",
816 );
817
818 let entry = &encoded.footer[0];
819 let start = entry.segment_offset as usize;
820 let end = start + entry.segment_len as usize - 1;
821 assert_corrupted(
822 decode_segment(&encoded.bytes[start..end], entry),
823 "truncated segment",
824 );
825 }
826
827 #[test]
828 fn test_format_rejects_header_magic_and_version() {
829 let encoded = encode_object(header(), &records()).unwrap();
830
831 let mut bad_magic = encoded.bytes.to_vec();
832 bad_magic[0] ^= 1;
833 assert_corrupted(decode_object(&bad_magic), "invalid header magic");
834
835 let mut bad_version = encoded.bytes.to_vec();
836 bad_version[8..10].copy_from_slice(&(FORMAT_VERSION + 1).to_be_bytes());
837 assert_corrupted(decode_object(&bad_version), "unsupported format version 2");
838
839 for offset in [10, 18, 26, 34, HEADER_LEN - 1] {
842 let mut bad_field = encoded.bytes.to_vec();
843 bad_field[offset] ^= 1;
844 assert_corrupted(decode_header(&bad_field), "header checksum mismatch");
845 }
846 }
847
848 #[test]
849 fn test_format_round_trips_the_first_object_of_a_chain() {
850 let first = Header {
851 prev: None,
852 ..header()
853 };
854 let encoded = encode_object(first.clone(), &records()).unwrap();
855 assert_eq!(first, decode_object(&encoded.bytes).unwrap().header);
856 }
857
858 #[test]
859 fn test_format_rejects_trailer_magic_and_invalid_footer_range() {
860 let encoded = encode_object(header(), &records()).unwrap();
861
862 let mut bad_magic = encoded.bytes.to_vec();
863 let last = bad_magic.len() - 1;
864 bad_magic[last] ^= 1;
865 assert_corrupted(decode_object(&bad_magic), "invalid trailer magic");
866
867 let mut bad_range = encoded.bytes.to_vec();
868 let trailer_start = bad_range.len() - TRAILER_LEN;
869 bad_range[trailer_start..trailer_start + 8].copy_from_slice(&u64::MAX.to_be_bytes());
870 assert_corrupted(decode_object(&bad_range), "footer");
871 }
872
873 #[test]
874 fn test_format_rejects_segment_and_footer_crc_errors() {
875 let encoded = encode_object(header(), &records()).unwrap();
876
877 let mut bad_segment = encoded.bytes.to_vec();
878 let segment = &encoded.footer[0];
879 bad_segment[segment.segment_offset as usize + SEGMENT_HEADER_LEN] ^= 1;
880 assert_corrupted(
881 decode_object(&bad_segment),
882 &format!("segment of region {} checksum mismatch", segment.region_id),
883 );
884
885 let mut bad_footer = encoded.bytes.to_vec();
886 bad_footer[encoded.trailer.footer_offset as usize] ^= 1;
887 assert_corrupted(decode_object(&bad_footer), "footer checksum mismatch");
888 }
889
890 #[test]
891 fn test_format_rejects_corrupted_trailer_checksums() {
892 let encoded = encode_object(header(), &records()).unwrap();
893 let trailer_start = encoded.bytes.len() - TRAILER_LEN;
894
895 let object_crc32_start = trailer_start + 8 + 8 + 4;
898 let mut bad_object_crc32 = encoded.bytes.to_vec();
899 bad_object_crc32[object_crc32_start] ^= 1;
900 assert_corrupted(decode_object(&bad_object_crc32), "object checksum mismatch");
901
902 let footer_crc32_start = trailer_start + 8 + 8;
903 let mut bad_footer_crc32 = encoded.bytes.to_vec();
904 bad_footer_crc32[footer_crc32_start] ^= 1;
905 assert_corrupted(decode_object(&bad_footer_crc32), "footer checksum mismatch");
906 }
907
908 #[test]
909 fn test_format_round_trips_an_object_without_records() {
910 let encoded = encode_object(header(), &[]).unwrap();
911 assert!(encoded.footer.is_empty());
912 assert_eq!(
913 MIN_OBJECT_LEN,
914 encoded.bytes.len(),
915 "an empty object is a header, an empty footer and the trailer"
916 );
917 assert_eq!(encoded.bytes.len() as u64, object_len(&encoded.footer));
918 let decoded = decode_object(&encoded.bytes).unwrap();
919 assert_eq!(header(), decoded.header);
920 assert!(decoded.records.is_empty());
921 assert!(decoded.footer.is_empty());
922 }
923
924 #[test]
925 fn test_format_rejects_duplicate_entry_ids() {
926 let region_id = RegionId::new(1, 1);
927 let records = vec![
928 Record {
929 region_id,
930 entry_id: 1,
931 payload: Bytes::from_static(b"first"),
932 },
933 Record {
934 region_id,
935 entry_id: 1,
936 payload: Bytes::from_static(b"duplicate"),
937 },
938 ];
939 assert_corrupted(
940 encode_object(header(), &records),
941 &format!("non-monotonic entry id for region {region_id}"),
942 );
943 }
944
945 #[test]
946 fn test_format_rejects_segment_count_larger_than_available_bytes() {
947 let region_id = RegionId::new(1, 1);
948 let mut segment = BytesMut::new();
949 segment.put_u64(region_id.as_u64());
950 segment.put_u32(u32::MAX);
951 let entry = FooterEntry {
952 region_id,
953 min_entry_id: 1,
954 max_entry_id: 1,
955 entry_count: u32::MAX,
956 segment_offset: HEADER_LEN as u64,
957 segment_len: segment.len() as u64,
958 segment_crc32: crc32fast::hash(&segment),
959 };
960
961 assert_corrupted(
962 decode_segment(&segment, &entry),
963 "segment declares 4294967295 entries but only holds bytes for 0",
964 );
965 }
966
967 const FIXTURE_V1_HEX: &str = concat!(
973 "475457414c4f424a",
976 "0001",
977 "0000000000000007",
978 "0000000000000003",
979 "0000000000000006",
980 "0000000000000002",
981 "1ca121b1",
982 "0000000100000001",
985 "00000003",
986 "0000000000000001",
987 "00000001",
988 "61",
989 "0000000000000002",
990 "00000002",
991 "6262",
992 "0000000000000003",
993 "00000003",
994 "636363",
995 "0000000200000001",
997 "00000002",
998 "000000000000000a",
999 "00000001",
1000 "78",
1001 "000000000000000b",
1002 "00000002",
1003 "7979",
1004 "00000002",
1007 "0000000100000001",
1008 "0000000000000001",
1009 "0000000000000003",
1010 "00000003",
1011 "000000000000002e",
1012 "0000000000000036",
1013 "25ac0486",
1014 "0000000200000001",
1015 "000000000000000a",
1016 "000000000000000b",
1017 "00000002",
1018 "0000000000000064",
1019 "0000000000000027",
1020 "65dc08ec",
1021 "000000000000008b",
1023 "0000000000000064",
1024 "6b97e3e4",
1025 "631bb7d1",
1026 "475457414c54524c",
1027 );
1028
1029 fn fixture_bytes() -> Vec<u8> {
1030 let hex = FIXTURE_V1_HEX.as_bytes();
1031 hex.chunks(2)
1032 .map(|pair| u8::from_str_radix(std::str::from_utf8(pair).unwrap(), 16).unwrap())
1033 .collect()
1034 }
1035
1036 fn fixture_records() -> Vec<Record> {
1037 [
1038 (RegionId::new(1, 1), 1, &b"a"[..]),
1039 (RegionId::new(1, 1), 2, b"bb"),
1040 (RegionId::new(1, 1), 3, b"ccc"),
1041 (RegionId::new(2, 1), 10, b"x"),
1042 (RegionId::new(2, 1), 11, b"yy"),
1043 ]
1044 .into_iter()
1045 .map(|(region_id, entry_id, payload)| Record {
1046 region_id,
1047 entry_id,
1048 payload: Bytes::copy_from_slice(payload),
1049 })
1050 .collect()
1051 }
1052
1053 #[test]
1054 fn test_format_matches_version_1_fixture() {
1055 let fixture = fixture_bytes();
1056 let header = Header {
1057 object_seq: 7,
1058 epoch: 3,
1059 prev: Some(ChainLink {
1060 object_seq: 6,
1061 epoch: 2,
1062 }),
1063 };
1064 let footer = vec![
1065 FooterEntry {
1066 region_id: RegionId::new(1, 1),
1067 min_entry_id: 1,
1068 max_entry_id: 3,
1069 entry_count: 3,
1070 segment_offset: 46,
1071 segment_len: 54,
1072 segment_crc32: 0x25ac0486,
1073 },
1074 FooterEntry {
1075 region_id: RegionId::new(2, 1),
1076 min_entry_id: 10,
1077 max_entry_id: 11,
1078 entry_count: 2,
1079 segment_offset: 100,
1080 segment_len: 39,
1081 segment_crc32: 0x65dc08ec,
1082 },
1083 ];
1084 assert_eq!(271, fixture.len());
1085
1086 let decoded = decode_object(&fixture).unwrap();
1087 assert_eq!(header, decoded.header);
1088 assert_eq!(footer, decoded.footer);
1089 assert_eq!(fixture_records(), decoded.records);
1090 assert_eq!(
1091 FixedTrailer {
1092 footer_offset: 139,
1093 footer_len: 100,
1094 footer_crc32: 0x6b97e3e4,
1095 object_crc32: 0x631bb7d1,
1096 },
1097 decode_trailer(&fixture[fixture.len() - TRAILER_LEN..]).unwrap()
1098 );
1099
1100 let mut reversed = fixture_records();
1102 reversed.reverse();
1103 for records in [fixture_records(), reversed] {
1104 let encoded = encode_object(header.clone(), &records).unwrap();
1105 assert_eq!(fixture, encoded.bytes.as_ref());
1106 assert_eq!(footer, encoded.footer);
1107 assert_eq!(fixture.len() as u64, object_len(&encoded.footer));
1108 }
1109 }
1110
1111 fn footer_ids(entry: &FooterEntry) -> (u64, u64, u32) {
1112 (entry.min_entry_id, entry.max_entry_id, entry.entry_count)
1113 }
1114
1115 fn assert_corrupted<T: std::fmt::Debug>(result: Result<T>, expected_reason: &str) {
1116 match result {
1117 Err(Error::CorruptedWalObject { reason, .. }) => assert!(
1118 reason.contains(expected_reason),
1119 "expected reason to contain {expected_reason:?}, actual {reason:?}"
1120 ),
1121 other => panic!("expected a corrupted object error, actual {other:?}"),
1122 }
1123 }
1124}