Skip to main content

log_store/object_store_wal/
format.rs

1// Copyright 2023 Greptime Team
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15//! Encoding and decoding of a single WAL object. The byte layout is described
16//! in the [module documentation](super).
17
18use 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;
30/// Predecessor sequence recorded for an object that starts a chain. Object
31/// sequences stay below `OBJECT_SEQ_LIMIT`, so it never names an object.
32const NO_PREDECESSOR: u64 = u64::MAX;
33
34/// Length of the object header: magic, version, object sequence, epoch,
35/// predecessor sequence, predecessor epoch and the CRC32 of the header.
36pub(crate) const HEADER_LEN: usize = 8 + 2 + 8 + 8 + 8 + 8 + 4;
37/// Length of the fixed trailer: footer offset, footer length, footer CRC32,
38/// object CRC32 and magic.
39pub(crate) const TRAILER_LEN: usize = 8 + 8 + 4 + 4 + 8;
40const SEGMENT_HEADER_LEN: usize = 8 + 4;
41/// Length of one footer entry: region id, entry id range, entry count, segment
42/// offset, segment length and segment CRC32.
43pub(crate) const FOOTER_ENTRY_LEN: usize = 8 + 8 + 8 + 4 + 8 + 8 + 4;
44/// Length of the entry count the footer starts with.
45#[cfg(test)]
46const FOOTER_COUNT_LEN: usize = 4;
47
48/// Header of a WAL object.
49#[derive(Debug, Clone, PartialEq, Eq)]
50pub(crate) struct Header {
51    pub(crate) object_seq: u64,
52    /// Epoch of the store instance that wrote the object: one above the
53    /// sequence of the start object the instance created, so the epoch also
54    /// identifies the instance.
55    pub(crate) epoch: u64,
56    /// The object this object extends, `None` for the first object of a chain.
57    pub(crate) prev: Option<ChainLink>,
58}
59
60/// Names the object another object extends: its sequence and the epoch
61/// recorded in its header.
62#[derive(Debug, Clone, Copy, PartialEq, Eq)]
63pub(crate) struct ChainLink {
64    pub(crate) object_seq: u64,
65    pub(crate) epoch: u64,
66}
67
68/// A single WAL entry inside an object.
69#[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/// Footer entry describing the segment of one region.
77#[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/// Fixed-size trailer locating the footer of an object.
89#[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/// An encoded object together with the metadata a writer indexes it by.
98#[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/// A decoded object with its records ordered by region id and entry id.
106#[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
114/// Encodes `records` into one object. Records are grouped into a segment per
115/// region; entry ids within a region must be unique. An object without records
116/// only extends the chain.
117pub(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/// Returns the length of the object whose footer is `footer`, derived from the
187/// layout: the segments tile the body after the header, and the footer, which
188/// holds an entry count and one entry per segment, and the trailer follow them.
189#[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
196/// Decodes and verifies the header from the first [`HEADER_LEN`] bytes of an
197/// object.
198pub(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
246/// Decodes the trailer from the last [`TRAILER_LEN`] bytes of an object.
247pub(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
278/// Decodes the footer that `trailer` points at.
279pub(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
368/// Decodes the segment that `entry` describes.
369pub(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    // Every record needs at least an entry id and a payload length. Bound the
408    // allocation by bytes that are actually present before trusting `count`.
409    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
460/// Lower bound on the length of an object: header, a footer holding only its
461/// entry count and the trailer.
462pub(crate) const MIN_OBJECT_LEN: usize = HEADER_LEN + 4 + TRAILER_LEN;
463
464/// Locates the footer inside an object of `object_len` bytes from its trailer.
465/// The footer must follow the header and end where the trailer starts.
466pub(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
493/// Checks that the segments `footer` describes tile the object body: the first
494/// starts right after the header, each follows the previous one without a gap
495/// or overlap, and the last ends where the footer at `footer_start` begins.
496pub(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/// Decodes a whole object, verifying every checksum and byte range. Recovery
532/// reads only the header, trailer and footer, so this is the reference
533/// decoder that tests check the store against.
534#[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
610/// Checksum of everything but the object checksum itself, so the trailer can
611/// carry it.
612fn 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        // Every field after the version is covered by the header checksum,
840        // which recovery verifies without reading the whole object.
841        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        // The object checksum the trailer stores, at footer offset, footer length
896        // and footer checksum.
897        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    /// A version 1 object holding entries 1, 2 and 3 of region 1 and entries
968    /// 10 and 11 of region 2. The bytes were derived from the layout described in
969    /// the module documentation, independently of [`encode_object`], so a change
970    /// to field order, endianness or checksum coverage fails this test even when
971    /// the encoder and the decoder change together.
972    const FIXTURE_V1_HEX: &str = concat!(
973        // Header: magic, version 1, object sequence 7, epoch 3, predecessor
974        // sequence 6, predecessor epoch 2, header CRC32.
975        "475457414c4f424a",
976        "0001",
977        "0000000000000007",
978        "0000000000000003",
979        "0000000000000006",
980        "0000000000000002",
981        "1ca121b1",
982        // Segment of region 1 at offset 46: region id, entry count, then
983        // (entry id, payload length, payload) per entry.
984        "0000000100000001",
985        "00000003",
986        "0000000000000001",
987        "00000001",
988        "61",
989        "0000000000000002",
990        "00000002",
991        "6262",
992        "0000000000000003",
993        "00000003",
994        "636363",
995        // Segment of region 2 at offset 100.
996        "0000000200000001",
997        "00000002",
998        "000000000000000a",
999        "00000001",
1000        "78",
1001        "000000000000000b",
1002        "00000002",
1003        "7979",
1004        // Footer at offset 139: entry count, then per segment region id, min and
1005        // max entry id, entry count, segment offset, length and CRC32.
1006        "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        // Trailer: footer offset, footer length, footer CRC32, object CRC32, magic.
1022        "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        // The encoding does not depend on the order records are passed in.
1101        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}