From d45f5d6eaf598d9a2f90405adfc3d17c91dd67bc Mon Sep 17 00:00:00 2001 From: jeremyhi Date: Thu, 17 Sep 2026 09:14:22 +0000 Subject: [PATCH] feat(log-store): add the object store WAL object format (#9200) * feat(log-store): add the object store WAL object format Add the byte format of a single object store WAL object: a header with the GTWALOBJ magic, format version 1, the object sequence and the writer instance id; one segment per region ordered by region id with entries ordered by entry id; a footer that records each segment's region id, entry id range, entry count, byte range and CRC32; and a fixed trailer with the GTWALTRL magic, the footer location, the footer CRC32 and the whole-object CRC32. The module encodes objects deterministically and decodes the header, trailer, footer and segments separately, with structural checks on footer ranges and segment tiling. It has no callers yet; the store that writes and reads objects follows in later changes. Signed-off-by: jeremyhi * test(log-store): pin the object store WAL format with a byte fixture Add a fixed version 1 object with two regions and five entries as a hex literal. The test decodes it and checks the exact header, trailer, footer entries and records, and checks that encoding the same records, in either input order, reproduces the fixture byte for byte. Round-trip tests alone pass when a refactor changes field order, endianness or checksum coverage in both the encoder and the decoder. The fixture bytes were derived from the documented layout rather than from the encoder, so such a change now fails. Signed-off-by: jeremyhi * fix(log-store): reject an empty footer when decoding a WAL object The encoder never writes an object without records, but decode_footer accepted a footer that declares zero segments, and verify_segment_ranges accepts an empty footer too. Only the test-only decode_object rejected it, so a checksum-valid empty object would pass the header, trailer and footer checks that recovery runs. Reject a zero entry count in decode_footer and drop the now unreachable check in decode_object. Add a test that builds a checksum-valid object with an empty footer and checks that decode_footer and decode_object reject it. Signed-off-by: jeremyhi * refactor(log-store): use pub(crate) for the WAL object format API Other log-store modules use pub(crate) for items shared across module boundaries. Switch the format module from pub(super) to pub(crate) to follow that convention. No behavior change. Signed-off-by: jeremyhi --------- Signed-off-by: jeremyhi --- Cargo.lock | 1 + Cargo.toml | 1 + src/log-store/Cargo.toml | 1 + src/log-store/src/error.rs | 9 + src/log-store/src/lib.rs | 1 + src/log-store/src/object_store_wal.rs | 35 + src/log-store/src/object_store_wal/format.rs | 1096 ++++++++++++++++++ 7 files changed, 1144 insertions(+) create mode 100644 src/log-store/src/object_store_wal.rs create mode 100644 src/log-store/src/object_store_wal/format.rs diff --git a/Cargo.lock b/Cargo.lock index db47e98e355..dba349c74e5 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -8047,6 +8047,7 @@ dependencies = [ "common-test-util", "common-time", "common-wal", + "crc32fast", "dashmap", "delta-encoding", "derive_builder 0.20.2", diff --git a/Cargo.toml b/Cargo.toml index 730901925a9..886c1bb9104 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -126,6 +126,7 @@ chrono-tz = { version = "0.10", features = ["case-insensitive"] } clap = { version = "4.4", features = ["derive"] } config = "0.13.0" const_format = "0.2" +crc32fast = "1" criterion = "0.7" crossbeam-utils = "0.8" dashmap = "6.1" diff --git a/src/log-store/Cargo.toml b/src/log-store/Cargo.toml index c8372c578bd..ee2c6a12dc0 100644 --- a/src/log-store/Cargo.toml +++ b/src/log-store/Cargo.toml @@ -25,6 +25,7 @@ common-runtime.workspace = true common-telemetry.workspace = true common-time.workspace = true common-wal.workspace = true +crc32fast.workspace = true dashmap.workspace = true delta-encoding = "0.4" derive_builder.workspace = true diff --git a/src/log-store/src/error.rs b/src/log-store/src/error.rs index a83944d5174..dec7b42c3a5 100644 --- a/src/log-store/src/error.rs +++ b/src/log-store/src/error.rs @@ -308,6 +308,13 @@ pub enum Error { #[snafu(implicit)] location: Location, }, + + #[snafu(display("Corrupted WAL object, {}", reason))] + CorruptedWalObject { + reason: String, + #[snafu(implicit)] + location: Location, + }, } pub type Result = std::result::Result; @@ -353,6 +360,8 @@ impl ErrorExt for Error { | WaitDumpIndex { .. } | MetaLengthExceededLimit { .. } => StatusCode::Internal, + CorruptedWalObject { .. } => StatusCode::Unexpected, + // Object store related errors CreateWriter { .. } | WriteIndex { .. } | ReadIndex { .. } | Io { .. } => { StatusCode::StorageUnavailable diff --git a/src/log-store/src/lib.rs b/src/log-store/src/lib.rs index c054fbc1d80..5773efa7aaa 100644 --- a/src/log-store/src/lib.rs +++ b/src/log-store/src/lib.rs @@ -16,5 +16,6 @@ pub mod error; pub mod kafka; pub mod metrics; pub mod noop; +pub mod object_store_wal; pub mod raft_engine; pub mod test_util; diff --git a/src/log-store/src/object_store_wal.rs b/src/log-store/src/object_store_wal.rs new file mode 100644 index 00000000000..f4cd7e081c7 --- /dev/null +++ b/src/log-store/src/object_store_wal.rs @@ -0,0 +1,35 @@ +// Copyright 2023 Greptime Team +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +//! Durable primitives of a WAL that stores entries as immutable object store +//! objects. +//! +//! Entries of many regions are batched into a single object, written once and +//! never mutated afterwards. An object is laid out as +//! +//! ```text +//! header | segment (region 1) | ... | segment (region N) | footer | trailer +//! ``` +//! +//! The header carries the magic `GTWALOBJ`, the format version, the object +//! sequence and the instance that wrote the object. Each segment holds the +//! entries of exactly one region, ordered by entry id, and segments are ordered +//! by region id. The footer indexes every segment with its region id, entry id +//! range, byte range and CRC32. The fixed-size trailer points at the footer and +//! carries the CRC32 of the footer and of the whole object, so a reader locates +//! the footer by reading the fixed-length trailer at the end of the object. + +// The format has no callers until the store that writes and reads objects lands. +#[allow(dead_code)] +mod format; diff --git a/src/log-store/src/object_store_wal/format.rs b/src/log-store/src/object_store_wal/format.rs new file mode 100644 index 00000000000..62fe290aa82 --- /dev/null +++ b/src/log-store/src/object_store_wal/format.rs @@ -0,0 +1,1096 @@ +// Copyright 2023 Greptime Team +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +//! Encoding and decoding of a single WAL object. The byte layout is described +//! in the [module documentation](super). + +use std::collections::BTreeMap; +use std::ops::Range; + +use bytes::{BufMut, Bytes, BytesMut}; +use snafu::{OptionExt, ensure}; +use store_api::storage::RegionId; + +use crate::error::{CorruptedWalObjectSnafu, Result}; + +const HEADER_MAGIC: &[u8; 8] = b"GTWALOBJ"; +const TRAILER_MAGIC: &[u8; 8] = b"GTWALTRL"; +const FORMAT_VERSION: u16 = 1; + +/// Length of the object header: magic, version, object sequence, writer instance. +pub(crate) const HEADER_LEN: usize = 8 + 2 + 8 + 16; +/// Length of the fixed trailer: footer offset, footer length, footer CRC32, +/// object CRC32 and magic. +pub(crate) const TRAILER_LEN: usize = 8 + 8 + 4 + 4 + 8; +const SEGMENT_HEADER_LEN: usize = 8 + 4; +/// Length of one footer entry: region id, entry id range, entry count, segment +/// offset, segment length and segment CRC32. +pub(crate) const FOOTER_ENTRY_LEN: usize = 8 + 8 + 8 + 4 + 8 + 8 + 4; +/// Length of the entry count the footer starts with. +const FOOTER_COUNT_LEN: usize = 4; + +/// Header of a WAL object. +#[derive(Debug, Clone, PartialEq, Eq)] +pub(crate) struct Header { + pub(crate) object_seq: u64, + pub(crate) writer_instance: [u8; 16], +} + +/// A single WAL entry inside an object. +#[derive(Debug, Clone, PartialEq, Eq)] +pub(crate) struct Record { + pub(crate) region_id: RegionId, + pub(crate) entry_id: u64, + pub(crate) payload: Bytes, +} + +/// Footer entry describing the segment of one region. +#[derive(Debug, Clone, PartialEq, Eq)] +pub(crate) struct FooterEntry { + pub(crate) region_id: RegionId, + pub(crate) min_entry_id: u64, + pub(crate) max_entry_id: u64, + pub(crate) entry_count: u32, + pub(crate) segment_offset: u64, + pub(crate) segment_len: u64, + pub(crate) segment_crc32: u32, +} + +/// Fixed-size trailer locating the footer of an object. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) struct FixedTrailer { + pub(crate) footer_offset: u64, + pub(crate) footer_len: u64, + pub(crate) footer_crc32: u32, + pub(crate) object_crc32: u32, +} + +/// An encoded object together with the metadata a writer indexes it by. +#[derive(Debug, Clone, PartialEq, Eq)] +pub(crate) struct EncodedObject { + pub(crate) bytes: Bytes, + pub(crate) footer: Vec, + pub(crate) trailer: FixedTrailer, +} + +/// A decoded object with its records ordered by region id and entry id. +#[cfg(test)] +#[derive(Debug, Clone, PartialEq, Eq)] +pub(crate) struct DecodedObject { + pub(crate) header: Header, + pub(crate) records: Vec, + pub(crate) footer: Vec, +} + +/// Encodes `records` into one object. Records are grouped into a segment per +/// region; entry ids within a region must be unique. +pub(crate) fn encode_object(header: Header, records: &[Record]) -> Result { + ensure!( + !records.is_empty(), + CorruptedWalObjectSnafu { + reason: "object has no records", + } + ); + + let mut grouped = BTreeMap::>::new(); + for record in records { + grouped.entry(record.region_id).or_default().push(record); + } + for (region_id, records) in &mut grouped { + records.sort_unstable_by_key(|record| record.entry_id); + ensure!( + !records + .windows(2) + .any(|pair| pair[0].entry_id >= pair[1].entry_id), + CorruptedWalObjectSnafu { + reason: format!("non-monotonic entry id for region {region_id}"), + } + ); + } + + let mut output = BytesMut::new(); + encode_header(&header, &mut output); + + let mut directory = Vec::with_capacity(grouped.len()); + for (region_id, records) in grouped { + let segment_offset = to_u64(output.len(), "segment offset")?; + let entry_count = to_u32(records.len(), "segment entry count")?; + let segment_start = output.len(); + + output.put_u64(region_id.as_u64()); + output.put_u32(entry_count); + for record in &records { + output.put_u64(record.entry_id); + output.put_u32(to_u32(record.payload.len(), "record payload")?); + output.put_slice(&record.payload); + } + + let segment = &output[segment_start..]; + directory.push(FooterEntry { + region_id, + min_entry_id: records[0].entry_id, + max_entry_id: records[records.len() - 1].entry_id, + entry_count, + segment_offset, + segment_len: to_u64(segment.len(), "segment length")?, + segment_crc32: crc32fast::hash(segment), + }); + } + + let footer_offset = to_u64(output.len(), "footer offset")?; + let footer_start = output.len(); + encode_footer(&directory, &mut output)?; + let footer = &output[footer_start..]; + let trailer = FixedTrailer { + footer_offset, + footer_len: to_u64(footer.len(), "footer length")?, + footer_crc32: crc32fast::hash(footer), + object_crc32: 0, + }; + let trailer = FixedTrailer { + object_crc32: object_crc32(&output, trailer), + ..trailer + }; + encode_trailer(trailer, &mut output); + + Ok(EncodedObject { + bytes: output.freeze(), + footer: directory, + trailer, + }) +} + +/// Returns the length of the object whose footer is `footer`, derived from the +/// layout: the segments tile the body after the header, and the footer, which +/// holds an entry count and one entry per segment, and the trailer follow them. +pub(crate) fn object_len(footer: &[FooterEntry]) -> u64 { + let segments = footer.iter().map(|entry| entry.segment_len).sum::(); + let framing = HEADER_LEN + FOOTER_COUNT_LEN + footer.len() * FOOTER_ENTRY_LEN + TRAILER_LEN; + segments + framing as u64 +} + +/// Decodes the header from the first [`HEADER_LEN`] bytes of an object. +pub(crate) fn decode_header(bytes: &[u8]) -> Result
{ + ensure!( + bytes.len() >= HEADER_LEN, + CorruptedWalObjectSnafu { + reason: truncated("header", bytes.len(), HEADER_LEN), + } + ); + + let mut reader = Reader::new(&bytes[..HEADER_LEN]); + let magic = reader.take(HEADER_MAGIC.len(), "header")?; + ensure!( + magic == HEADER_MAGIC, + CorruptedWalObjectSnafu { + reason: invalid_magic("header", HEADER_MAGIC, magic), + } + ); + let version = reader.u16("header")?; + ensure!( + version == FORMAT_VERSION, + CorruptedWalObjectSnafu { + reason: format!( + "unsupported format version {version}, expected version {FORMAT_VERSION}" + ), + } + ); + let object_seq = reader.u64("header")?; + let writer_instance = reader + .take(16, "header")? + .try_into() + .expect("writer instance has a fixed length"); + Ok(Header { + object_seq, + writer_instance, + }) +} + +/// Decodes the trailer from the last [`TRAILER_LEN`] bytes of an object. +pub(crate) fn decode_trailer(bytes: &[u8]) -> Result { + ensure!( + bytes.len() >= TRAILER_LEN, + CorruptedWalObjectSnafu { + reason: truncated("trailer", bytes.len(), TRAILER_LEN), + } + ); + ensure!( + bytes.len() <= TRAILER_LEN, + CorruptedWalObjectSnafu { + reason: trailing_bytes("trailer", bytes.len(), TRAILER_LEN), + } + ); + + let mut reader = Reader::new(bytes); + let trailer = FixedTrailer { + footer_offset: reader.u64("trailer")?, + footer_len: reader.u64("trailer")?, + footer_crc32: reader.u32("trailer")?, + object_crc32: reader.u32("trailer")?, + }; + let magic = reader.take(TRAILER_MAGIC.len(), "trailer")?; + ensure!( + magic == TRAILER_MAGIC, + CorruptedWalObjectSnafu { + reason: invalid_magic("trailer", TRAILER_MAGIC, magic), + } + ); + Ok(trailer) +} + +/// Decodes the footer that `trailer` points at. +pub(crate) fn decode_footer(bytes: &[u8], trailer: FixedTrailer) -> Result> { + let footer_len = to_usize(trailer.footer_len, "footer length")?; + ensure!( + bytes.len() >= footer_len, + CorruptedWalObjectSnafu { + reason: truncated("footer", bytes.len(), footer_len), + } + ); + ensure!( + bytes.len() <= footer_len, + CorruptedWalObjectSnafu { + reason: trailing_bytes("footer", bytes.len(), footer_len), + } + ); + let checksum = crc32fast::hash(bytes); + ensure!( + checksum == trailer.footer_crc32, + CorruptedWalObjectSnafu { + reason: checksum_mismatch("footer", trailer.footer_crc32, checksum), + } + ); + + let mut reader = Reader::new(bytes); + let count = reader.u32("footer")? as usize; + ensure!( + count > 0, + CorruptedWalObjectSnafu { + reason: "object has no records", + } + ); + let expected_len = count + .checked_mul(FOOTER_ENTRY_LEN) + .and_then(|len| len.checked_add(4)) + .with_context(|| CorruptedWalObjectSnafu { + reason: format!("footer declares {count} entries, which overflows its length"), + })?; + ensure!( + expected_len == bytes.len(), + CorruptedWalObjectSnafu { + reason: format!( + "footer declares {count} entries, expected length {expected_len}, actual {}", + bytes.len() + ), + } + ); + + let mut directory = Vec::with_capacity(count); + let mut previous_region = None; + for _ in 0..count { + let region_id = RegionId::from_u64(reader.u64("footer")?); + if let Some(previous) = previous_region { + ensure!( + previous < region_id, + CorruptedWalObjectSnafu { + reason: if previous == region_id { + format!("duplicate footer entry for region {region_id}") + } else { + format!( + "footer entries are not ordered by region, {previous} precedes {region_id}" + ) + }, + } + ); + } + previous_region = Some(region_id); + + let entry = FooterEntry { + region_id, + min_entry_id: reader.u64("footer")?, + max_entry_id: reader.u64("footer")?, + entry_count: reader.u32("footer")?, + segment_offset: reader.u64("footer")?, + segment_len: reader.u64("footer")?, + segment_crc32: reader.u32("footer")?, + }; + ensure!( + entry.entry_count > 0 + && entry.min_entry_id <= entry.max_entry_id + && entry.segment_len >= SEGMENT_HEADER_LEN as u64, + CorruptedWalObjectSnafu { + reason: format!( + "invalid footer entry for region {}, entry ids {}..={}, {} entries, segment length {}", + entry.region_id, + entry.min_entry_id, + entry.max_entry_id, + entry.entry_count, + entry.segment_len + ), + } + ); + directory.push(entry); + } + Ok(directory) +} + +/// Decodes the segment that `entry` describes. +pub(crate) fn decode_segment(bytes: &[u8], entry: &FooterEntry) -> Result> { + let segment_len = to_usize(entry.segment_len, "segment length")?; + ensure!( + bytes.len() >= segment_len, + CorruptedWalObjectSnafu { + reason: truncated("segment", bytes.len(), segment_len), + } + ); + ensure!( + bytes.len() <= segment_len, + CorruptedWalObjectSnafu { + reason: trailing_bytes("segment", bytes.len(), segment_len), + } + ); + let checksum = crc32fast::hash(bytes); + ensure!( + checksum == entry.segment_crc32, + CorruptedWalObjectSnafu { + reason: checksum_mismatch( + &format!("segment of region {}", entry.region_id), + entry.segment_crc32, + checksum + ), + } + ); + + let mut reader = Reader::new(bytes); + let region_id = RegionId::from_u64(reader.u64("segment")?); + let count = reader.u32("segment")?; + ensure!( + region_id == entry.region_id && count == entry.entry_count, + CorruptedWalObjectSnafu { + reason: format!( + "segment holds region {region_id} with {count} entries, footer expects region {} with {} entries", + entry.region_id, entry.entry_count + ), + } + ); + // Every record needs at least an entry id and a payload length. Bound the + // allocation by bytes that are actually present before trusting `count`. + let max_count = reader.remaining_len() / (8 + 4); + ensure!( + count as usize <= max_count, + CorruptedWalObjectSnafu { + reason: format!( + "segment declares {count} entries but only holds bytes for {max_count}" + ), + } + ); + + let mut records = Vec::with_capacity(count as usize); + let mut previous_id = None; + for _ in 0..count { + let entry_id = reader.u64("segment")?; + ensure!( + previous_id.is_none_or(|previous| previous < entry_id), + CorruptedWalObjectSnafu { + reason: format!("non-monotonic entry id for region {region_id}"), + } + ); + previous_id = Some(entry_id); + let payload_len = reader.u32("segment")? as usize; + let payload = Bytes::copy_from_slice(reader.take(payload_len, "segment")?); + records.push(Record { + region_id, + entry_id, + payload, + }); + } + ensure!( + reader.is_empty(), + CorruptedWalObjectSnafu { + reason: trailing_bytes("segment", bytes.len(), bytes.len() - reader.remaining_len()), + } + ); + ensure!( + records.first().map(|record| record.entry_id) == Some(entry.min_entry_id) + && records.last().map(|record| record.entry_id) == Some(entry.max_entry_id), + CorruptedWalObjectSnafu { + reason: format!( + "segment of region {region_id} holds entry ids {:?}..={:?}, footer expects {}..={}", + records.first().map(|record| record.entry_id), + records.last().map(|record| record.entry_id), + entry.min_entry_id, + entry.max_entry_id + ), + } + ); + Ok(records) +} + +/// Lower bound on the length of an object: header, a footer holding only its +/// entry count and the trailer. +pub(crate) const MIN_OBJECT_LEN: usize = HEADER_LEN + 4 + TRAILER_LEN; + +/// Locates the footer inside an object of `object_len` bytes from its trailer. +/// The footer must follow the header and end where the trailer starts. +pub(crate) fn footer_range(trailer: FixedTrailer, object_len: usize) -> Result> { + ensure!( + object_len >= MIN_OBJECT_LEN, + CorruptedWalObjectSnafu { + reason: truncated("object", object_len, MIN_OBJECT_LEN), + } + ); + let trailer_start = object_len - TRAILER_LEN; + let footer_start = to_usize(trailer.footer_offset, "footer offset")?; + let footer_len = to_usize(trailer.footer_len, "footer length")?; + let footer_end = + footer_start + .checked_add(footer_len) + .with_context(|| CorruptedWalObjectSnafu { + reason: format!("footer range {footer_start}..{footer_len} overflows the object"), + })?; + ensure!( + footer_start >= HEADER_LEN && footer_end == trailer_start, + CorruptedWalObjectSnafu { + reason: format!( + "invalid footer range {footer_start}..{footer_end}, expected {HEADER_LEN}..{trailer_start}" + ), + } + ); + Ok(footer_start..footer_end) +} + +/// Checks that the segments `footer` describes tile the object body: the first +/// starts right after the header, each follows the previous one without a gap +/// or overlap, and the last ends where the footer at `footer_start` begins. +pub(crate) fn verify_segment_ranges(footer: &[FooterEntry], footer_start: usize) -> Result<()> { + let mut expected_offset = HEADER_LEN; + for entry in footer { + let start = to_usize(entry.segment_offset, "segment offset")?; + let len = to_usize(entry.segment_len, "segment length")?; + let end = start + .checked_add(len) + .with_context(|| CorruptedWalObjectSnafu { + reason: format!( + "segment range {start}..{len} of region {} overflows the object", + entry.region_id + ), + })?; + ensure!( + start == expected_offset && end <= footer_start, + CorruptedWalObjectSnafu { + reason: format!( + "invalid segment range {start}..{end} of region {}, expected {expected_offset}..{footer_start}", + entry.region_id + ), + } + ); + expected_offset = end; + } + ensure!( + expected_offset == footer_start, + CorruptedWalObjectSnafu { + reason: format!( + "invalid segment range, segments end at {expected_offset}, footer starts at {footer_start}" + ), + } + ); + Ok(()) +} + +/// Decodes a whole object, verifying every checksum and byte range. Recovery +/// reads only the header, trailer and footer, so this is the reference +/// decoder that tests check the store against. +#[cfg(test)] +pub(crate) fn decode_object(bytes: &[u8]) -> Result { + ensure!( + bytes.len() >= MIN_OBJECT_LEN, + CorruptedWalObjectSnafu { + reason: truncated("object", bytes.len(), MIN_OBJECT_LEN), + } + ); + + let header = decode_header(bytes)?; + let trailer_start = bytes.len() - TRAILER_LEN; + let trailer = decode_trailer(&bytes[trailer_start..])?; + let footer_range = footer_range(trailer, bytes.len())?; + let footer_start = footer_range.start; + + let footer = decode_footer(&bytes[footer_range], trailer)?; + verify_segment_ranges(&footer, footer_start)?; + + let mut records = Vec::new(); + for entry in &footer { + let start = entry.segment_offset as usize; + let end = start + entry.segment_len as usize; + records.extend(decode_segment(&bytes[start..end], entry)?); + } + let checksum = object_crc32(&bytes[..trailer_start], trailer); + ensure!( + checksum == trailer.object_crc32, + CorruptedWalObjectSnafu { + reason: checksum_mismatch("object", trailer.object_crc32, checksum), + } + ); + + Ok(DecodedObject { + header, + records, + footer, + }) +} + +fn encode_header(header: &Header, output: &mut BytesMut) { + output.put_slice(HEADER_MAGIC); + output.put_u16(FORMAT_VERSION); + output.put_u64(header.object_seq); + output.put_slice(&header.writer_instance); +} + +fn encode_footer(directory: &[FooterEntry], output: &mut BytesMut) -> Result<()> { + output.put_u32(to_u32(directory.len(), "footer entry count")?); + for entry in directory { + output.put_u64(entry.region_id.as_u64()); + output.put_u64(entry.min_entry_id); + output.put_u64(entry.max_entry_id); + output.put_u32(entry.entry_count); + output.put_u64(entry.segment_offset); + output.put_u64(entry.segment_len); + output.put_u32(entry.segment_crc32); + } + Ok(()) +} + +fn encode_trailer(trailer: FixedTrailer, output: &mut BytesMut) { + output.put_u64(trailer.footer_offset); + output.put_u64(trailer.footer_len); + output.put_u32(trailer.footer_crc32); + output.put_u32(trailer.object_crc32); + output.put_slice(TRAILER_MAGIC); +} + +/// Checksum of everything but the object checksum itself, so the trailer can +/// carry it. +fn object_crc32(bytes_before_trailer: &[u8], trailer: FixedTrailer) -> u32 { + let mut checksum = crc32fast::Hasher::new(); + checksum.update(bytes_before_trailer); + checksum.update(&trailer.footer_offset.to_be_bytes()); + checksum.update(&trailer.footer_len.to_be_bytes()); + checksum.update(&trailer.footer_crc32.to_be_bytes()); + checksum.update(TRAILER_MAGIC); + checksum.finalize() +} + +fn truncated(part: &str, actual: usize, expected: usize) -> String { + format!("truncated {part}, expected at least {expected} bytes, actual {actual}") +} + +fn trailing_bytes(part: &str, actual: usize, expected: usize) -> String { + format!("trailing bytes in {part}, expected {expected} bytes, actual {actual}") +} + +fn invalid_magic(part: &str, expected: &[u8], actual: &[u8]) -> String { + format!( + "invalid {part} magic, expected {:?}, actual {:?}", + String::from_utf8_lossy(expected), + String::from_utf8_lossy(actual) + ) +} + +fn checksum_mismatch(part: &str, expected: u32, actual: u32) -> String { + format!("{part} checksum mismatch, expected {expected:#010x}, actual {actual:#010x}") +} + +fn to_u32(value: usize, name: &str) -> Result { + u32::try_from(value) + .ok() + .with_context(|| CorruptedWalObjectSnafu { + reason: format!("{name} {value} does not fit into u32"), + }) +} + +fn to_u64(value: usize, name: &str) -> Result { + u64::try_from(value) + .ok() + .with_context(|| CorruptedWalObjectSnafu { + reason: format!("{name} {value} does not fit into u64"), + }) +} + +fn to_usize(value: u64, name: &str) -> Result { + usize::try_from(value) + .ok() + .with_context(|| CorruptedWalObjectSnafu { + reason: format!("{name} {value} does not fit into usize"), + }) +} + +struct Reader<'a> { + bytes: &'a [u8], + offset: usize, +} + +impl<'a> Reader<'a> { + fn new(bytes: &'a [u8]) -> Self { + Self { bytes, offset: 0 } + } + + fn take(&mut self, len: usize, part: &'static str) -> Result<&'a [u8]> { + let end = self + .offset + .checked_add(len) + .with_context(|| CorruptedWalObjectSnafu { + reason: truncated(part, self.bytes.len(), len), + })?; + let value = self + .bytes + .get(self.offset..end) + .with_context(|| CorruptedWalObjectSnafu { + reason: truncated(part, self.bytes.len(), end), + })?; + self.offset = end; + Ok(value) + } + + fn u16(&mut self, part: &'static str) -> Result { + Ok(u16::from_be_bytes( + self.take(2, part)?.try_into().expect("u16 has two bytes"), + )) + } + + fn u32(&mut self, part: &'static str) -> Result { + Ok(u32::from_be_bytes( + self.take(4, part)?.try_into().expect("u32 has four bytes"), + )) + } + + fn u64(&mut self, part: &'static str) -> Result { + Ok(u64::from_be_bytes( + self.take(8, part)?.try_into().expect("u64 has eight bytes"), + )) + } + + fn is_empty(&self) -> bool { + self.offset == self.bytes.len() + } + + fn remaining_len(&self) -> usize { + self.bytes.len() - self.offset + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::error::Error; + + fn header() -> Header { + Header { + object_seq: 42, + writer_instance: *b"writer-instance!", + } + } + + fn records() -> Vec { + vec![ + Record { + region_id: RegionId::new(2, 1), + entry_id: 7, + payload: Bytes::from_static(b"region-2"), + }, + Record { + region_id: RegionId::new(1, 1), + entry_id: 2, + payload: Bytes::from_static(b"second"), + }, + Record { + region_id: RegionId::new(1, 1), + entry_id: 1, + payload: Bytes::from_static(b"first"), + }, + ] + } + + #[test] + fn test_format_round_trip_with_multiple_regions_and_entries() { + let encoded = encode_object(header(), &records()).unwrap(); + let decoded = decode_object(&encoded.bytes).unwrap(); + + assert_eq!(header(), decoded.header); + assert_eq!(2, decoded.footer.len()); + assert_eq!(RegionId::new(1, 1), decoded.footer[0].region_id); + assert_eq!((1, 2, 2), footer_ids(&decoded.footer[0])); + assert_eq!(RegionId::new(2, 1), decoded.footer[1].region_id); + assert_eq!((7, 7, 1), footer_ids(&decoded.footer[1])); + assert_eq!( + vec![ + (RegionId::new(1, 1), 1, Bytes::from_static(b"first")), + (RegionId::new(1, 1), 2, Bytes::from_static(b"second")), + (RegionId::new(2, 1), 7, Bytes::from_static(b"region-2")), + ], + decoded + .records + .into_iter() + .map(|record| (record.region_id, record.entry_id, record.payload)) + .collect::>() + ); + } + + #[test] + fn test_format_decodes_footer_trailer_and_segment_ranges() { + let encoded = encode_object(header(), &records()).unwrap(); + let trailer_start = encoded.bytes.len() - TRAILER_LEN; + let trailer = decode_trailer(&encoded.bytes[trailer_start..]).unwrap(); + assert_eq!(encoded.trailer, trailer); + + let footer_start = trailer.footer_offset as usize; + let footer_end = footer_start + trailer.footer_len as usize; + let footer = decode_footer(&encoded.bytes[footer_start..footer_end], trailer).unwrap(); + assert_eq!(encoded.footer, footer); + + let entry = &footer[0]; + let start = entry.segment_offset as usize; + let end = start + entry.segment_len as usize; + let segment = decode_segment(&encoded.bytes[start..end], entry).unwrap(); + assert_eq!( + vec![1, 2], + segment + .iter() + .map(|entry| entry.entry_id) + .collect::>() + ); + } + + #[test] + fn test_format_rejects_truncation() { + let encoded = encode_object(header(), &records()).unwrap(); + assert_corrupted( + decode_object(&encoded.bytes[..HEADER_LEN - 1]), + "truncated object", + ); + assert_corrupted( + decode_trailer(&encoded.bytes[encoded.bytes.len() - TRAILER_LEN + 1..]), + "truncated trailer", + ); + + let entry = &encoded.footer[0]; + let start = entry.segment_offset as usize; + let end = start + entry.segment_len as usize - 1; + assert_corrupted( + decode_segment(&encoded.bytes[start..end], entry), + "truncated segment", + ); + } + + #[test] + fn test_format_rejects_header_magic_and_version() { + let encoded = encode_object(header(), &records()).unwrap(); + + let mut bad_magic = encoded.bytes.to_vec(); + bad_magic[0] ^= 1; + assert_corrupted(decode_object(&bad_magic), "invalid header magic"); + + let mut bad_version = encoded.bytes.to_vec(); + bad_version[8..10].copy_from_slice(&(FORMAT_VERSION + 1).to_be_bytes()); + assert_corrupted(decode_object(&bad_version), "unsupported format version 2"); + + let mut bad_object_seq = encoded.bytes.to_vec(); + bad_object_seq[10] ^= 1; + assert_corrupted(decode_object(&bad_object_seq), "object checksum mismatch"); + + let mut bad_writer = encoded.bytes.to_vec(); + bad_writer[18] ^= 1; + assert_corrupted(decode_object(&bad_writer), "object checksum mismatch"); + } + + #[test] + fn test_format_rejects_trailer_magic_and_invalid_footer_range() { + let encoded = encode_object(header(), &records()).unwrap(); + + let mut bad_magic = encoded.bytes.to_vec(); + let last = bad_magic.len() - 1; + bad_magic[last] ^= 1; + assert_corrupted(decode_object(&bad_magic), "invalid trailer magic"); + + let mut bad_range = encoded.bytes.to_vec(); + let trailer_start = bad_range.len() - TRAILER_LEN; + bad_range[trailer_start..trailer_start + 8].copy_from_slice(&u64::MAX.to_be_bytes()); + assert_corrupted(decode_object(&bad_range), "footer"); + } + + #[test] + fn test_format_rejects_segment_and_footer_crc_errors() { + let encoded = encode_object(header(), &records()).unwrap(); + + let mut bad_segment = encoded.bytes.to_vec(); + let segment = &encoded.footer[0]; + bad_segment[segment.segment_offset as usize + SEGMENT_HEADER_LEN] ^= 1; + assert_corrupted( + decode_object(&bad_segment), + &format!("segment of region {} checksum mismatch", segment.region_id), + ); + + let mut bad_footer = encoded.bytes.to_vec(); + bad_footer[encoded.trailer.footer_offset as usize] ^= 1; + assert_corrupted(decode_object(&bad_footer), "footer checksum mismatch"); + } + + #[test] + fn test_format_rejects_corrupted_trailer_checksums() { + let encoded = encode_object(header(), &records()).unwrap(); + let trailer_start = encoded.bytes.len() - TRAILER_LEN; + + // The object checksum the trailer stores, at footer offset, footer length + // and footer checksum. + let object_crc32_start = trailer_start + 8 + 8 + 4; + let mut bad_object_crc32 = encoded.bytes.to_vec(); + bad_object_crc32[object_crc32_start] ^= 1; + assert_corrupted(decode_object(&bad_object_crc32), "object checksum mismatch"); + + let footer_crc32_start = trailer_start + 8 + 8; + let mut bad_footer_crc32 = encoded.bytes.to_vec(); + bad_footer_crc32[footer_crc32_start] ^= 1; + assert_corrupted(decode_object(&bad_footer_crc32), "footer checksum mismatch"); + } + + #[test] + fn test_format_rejects_empty_object() { + assert_corrupted(encode_object(header(), &[]), "object has no records"); + } + + #[test] + fn test_format_rejects_checksum_valid_empty_footer() { + let mut object = BytesMut::new(); + encode_header(&header(), &mut object); + let footer_offset = object.len(); + encode_footer(&[], &mut object).unwrap(); + let footer = object[footer_offset..].to_vec(); + let trailer = FixedTrailer { + footer_offset: footer_offset as u64, + footer_len: footer.len() as u64, + footer_crc32: crc32fast::hash(&footer), + object_crc32: 0, + }; + let trailer = FixedTrailer { + object_crc32: object_crc32(&object, trailer), + ..trailer + }; + encode_trailer(trailer, &mut object); + + // Every structural check that recovery runs before the footer passes. + assert_eq!(header(), decode_header(&object).unwrap()); + let trailer_start = object.len() - TRAILER_LEN; + assert_eq!(trailer, decode_trailer(&object[trailer_start..]).unwrap()); + assert_eq!( + footer_offset..trailer_start, + footer_range(trailer, object.len()).unwrap() + ); + verify_segment_ranges(&[], footer_offset).unwrap(); + + assert_corrupted(decode_footer(&footer, trailer), "object has no records"); + assert_corrupted(decode_object(&object), "object has no records"); + } + + #[test] + fn test_format_rejects_duplicate_entry_ids() { + let region_id = RegionId::new(1, 1); + let records = vec![ + Record { + region_id, + entry_id: 1, + payload: Bytes::from_static(b"first"), + }, + Record { + region_id, + entry_id: 1, + payload: Bytes::from_static(b"duplicate"), + }, + ]; + assert_corrupted( + encode_object(header(), &records), + &format!("non-monotonic entry id for region {region_id}"), + ); + } + + #[test] + fn test_format_rejects_segment_count_larger_than_available_bytes() { + let region_id = RegionId::new(1, 1); + let mut segment = BytesMut::new(); + segment.put_u64(region_id.as_u64()); + segment.put_u32(u32::MAX); + let entry = FooterEntry { + region_id, + min_entry_id: 1, + max_entry_id: 1, + entry_count: u32::MAX, + segment_offset: HEADER_LEN as u64, + segment_len: segment.len() as u64, + segment_crc32: crc32fast::hash(&segment), + }; + + assert_corrupted( + decode_segment(&segment, &entry), + "segment declares 4294967295 entries but only holds bytes for 0", + ); + } + + /// A version 1 object holding entries 1, 2 and 3 of region 1 and entries + /// 10 and 11 of region 2. The bytes were derived from the layout described in + /// the module documentation, independently of [`encode_object`], so a change + /// to field order, endianness or checksum coverage fails this test even when + /// the encoder and the decoder change together. + const FIXTURE_V1_HEX: &str = concat!( + // Header: magic, version 1, object sequence 7, writer instance. + "475457414c4f424a", + "0001", + "0000000000000007", + "7772697465722d666978747572652d31", + // Segment of region 1 at offset 34: region id, entry count, then + // (entry id, payload length, payload) per entry. + "0000000100000001", + "00000003", + "0000000000000001", + "00000001", + "61", + "0000000000000002", + "00000002", + "6262", + "0000000000000003", + "00000003", + "636363", + // Segment of region 2 at offset 88. + "0000000200000001", + "00000002", + "000000000000000a", + "00000001", + "78", + "000000000000000b", + "00000002", + "7979", + // Footer at offset 127: entry count, then per segment region id, min and + // max entry id, entry count, segment offset, length and CRC32. + "00000002", + "0000000100000001", + "0000000000000001", + "0000000000000003", + "00000003", + "0000000000000022", + "0000000000000036", + "25ac0486", + "0000000200000001", + "000000000000000a", + "000000000000000b", + "00000002", + "0000000000000058", + "0000000000000027", + "65dc08ec", + // Trailer: footer offset, footer length, footer CRC32, object CRC32, magic. + "000000000000007f", + "0000000000000064", + "8298e0f7", + "7a009db1", + "475457414c54524c", + ); + + fn fixture_bytes() -> Vec { + let hex = FIXTURE_V1_HEX.as_bytes(); + hex.chunks(2) + .map(|pair| u8::from_str_radix(std::str::from_utf8(pair).unwrap(), 16).unwrap()) + .collect() + } + + fn fixture_records() -> Vec { + [ + (RegionId::new(1, 1), 1, &b"a"[..]), + (RegionId::new(1, 1), 2, b"bb"), + (RegionId::new(1, 1), 3, b"ccc"), + (RegionId::new(2, 1), 10, b"x"), + (RegionId::new(2, 1), 11, b"yy"), + ] + .into_iter() + .map(|(region_id, entry_id, payload)| Record { + region_id, + entry_id, + payload: Bytes::copy_from_slice(payload), + }) + .collect() + } + + #[test] + fn test_format_matches_version_1_fixture() { + let fixture = fixture_bytes(); + let header = Header { + object_seq: 7, + writer_instance: *b"writer-fixture-1", + }; + let footer = vec![ + FooterEntry { + region_id: RegionId::new(1, 1), + min_entry_id: 1, + max_entry_id: 3, + entry_count: 3, + segment_offset: 34, + segment_len: 54, + segment_crc32: 0x25ac0486, + }, + FooterEntry { + region_id: RegionId::new(2, 1), + min_entry_id: 10, + max_entry_id: 11, + entry_count: 2, + segment_offset: 88, + segment_len: 39, + segment_crc32: 0x65dc08ec, + }, + ]; + assert_eq!(259, fixture.len()); + + let decoded = decode_object(&fixture).unwrap(); + assert_eq!(header, decoded.header); + assert_eq!(footer, decoded.footer); + assert_eq!(fixture_records(), decoded.records); + assert_eq!( + FixedTrailer { + footer_offset: 127, + footer_len: 100, + footer_crc32: 0x8298e0f7, + object_crc32: 0x7a009db1, + }, + decode_trailer(&fixture[fixture.len() - TRAILER_LEN..]).unwrap() + ); + + // The encoding does not depend on the order records are passed in. + let mut reversed = fixture_records(); + reversed.reverse(); + for records in [fixture_records(), reversed] { + let encoded = encode_object(header.clone(), &records).unwrap(); + assert_eq!(fixture, encoded.bytes.as_ref()); + assert_eq!(footer, encoded.footer); + assert_eq!(fixture.len() as u64, object_len(&encoded.footer)); + } + } + + fn footer_ids(entry: &FooterEntry) -> (u64, u64, u32) { + (entry.min_entry_id, entry.max_entry_id, entry.entry_count) + } + + fn assert_corrupted(result: Result, expected_reason: &str) { + match result { + Err(Error::CorruptedWalObject { reason, .. }) => assert!( + reason.contains(expected_reason), + "expected reason to contain {expected_reason:?}, actual {reason:?}" + ), + other => panic!("expected a corrupted object error, actual {other:?}"), + } + } +}