krabka_records_legacy/message/
attributes.rs

1//! The legacy `attributes` byte: its bit layout, and the mapping between its
2//! low three bits and a [`CompressionType`].
3//!
4//! This is the one part of the v0/v1 header that is not a plain fixed-width
5//! field, so its two conversions live together here: the codes stop at 3, and
6//! a codec that v0/v1 cannot name has to fail rather than truncate.
7
8use {crate::error::LegacyRecordsError, krabka_compression::CompressionType};
9
10/// Bit layout of the legacy `attributes` byte.
11pub mod attrs {
12    pub const COMPRESSION_MASK: i8 = 0x07;
13    /// v1-only: 0 = `CreateTime`, 1 = `LogAppendTime`.
14    pub const TIMESTAMP_TYPE_BIT: i8 = 1 << 3;
15}
16
17/// Map a v0/v1 compression code to a [`CompressionType`].
18///
19/// The compression code is the low 3 bits of `attributes`.
20/// `0` => `None`, `1` => `Gzip`, `2` => `Snappy`, `3` => `Lz4`. v0/v1
21/// never carried Zstd on the wire. KIP-110 was v2-only.
22/// # Errors
23/// Returns `LegacyRecordsError::Malformed` if the compression code is not 0, 1, 2, or 3.
24pub fn compression_from_attrs(byte: i8) -> Result<CompressionType, LegacyRecordsError> {
25    match byte & attrs::COMPRESSION_MASK {
26        0 => Ok(CompressionType::None),
27        1 => Ok(CompressionType::Gzip),
28        2 => Ok(CompressionType::Snappy),
29        3 => Ok(CompressionType::Lz4),
30        other => Err(LegacyRecordsError::Malformed(format!(
31            "legacy compression code {other} not supported (v0/v1 carries 0..=3)"
32        ))),
33    }
34}
35
36#[must_use]
37/// # Panics
38/// Panics if `codec` is `CompressionType::Zstd`, or any other codec that v0/v1 cannot carry.
39pub fn attrs_with_compression(byte: i8, codec: CompressionType) -> i8 {
40    let code: i8 = match codec {
41        CompressionType::None => 0,
42        CompressionType::Gzip => 1,
43        CompressionType::Snappy => 2,
44        CompressionType::Lz4 => 3,
45        CompressionType::Zstd => panic!("legacy v0/v1 cannot carry zstd"),
46        _ => panic!("unrecognised compression codec {codec:?} for v0/v1"),
47    };
48    (byte & !attrs::COMPRESSION_MASK) | code
49}
50
51#[cfg(test)]
52mod tests {
53    use super::*;
54
55    #[yare::parameterized(
56        none = { CompressionType::None },
57        gzip = { CompressionType::Gzip },
58        snappy = { CompressionType::Snappy },
59        lz4 = { CompressionType::Lz4 },
60    )]
61    fn attrs_codec_roundtrip(compression: CompressionType) {
62        let bits = attrs_with_compression(0, compression);
63        assert2::assert!(compression_from_attrs(bits).unwrap() == compression);
64    }
65
66    // --- mutation-coverage tests --------------------------------------------
67    //
68    // The round-trip tests above pass through any arithmetic flip that cancels
69    // between encode and decode, and never exercise the malformed/boundary
70    // paths. The tests below pin exact bit layouts, the timestamp sentinel, and
71    // the precise `needed` byte counts / error variants on truncated input.
72    //
73    // A few mutants here are genuinely equivalent and intentionally left alone:
74    //   - `attrs_with_compression` line `(byte & !MASK) | code`: `| -> ^` is a
75    //     no-op because the two operands are bit-disjoint.
76    //   - `encoded_len() - 4` in `encode_into` only sizes a `Vec::with_capacity`
77    //     hint; the output bytes are identical regardless.
78
79    #[test]
80    fn attribute_bit_constants() {
81        // `1 << 3`; a `>>` flip would zero the timestamp-type bit.
82        assert2::assert!(attrs::TIMESTAMP_TYPE_BIT == 0b0000_1000);
83        assert2::assert!(attrs::COMPRESSION_MASK == 0b0000_0111);
84    }
85
86    #[yare::parameterized(
87        none = { CompressionType::None, 0 },
88        gzip = { CompressionType::Gzip, 1 },
89        snappy = { CompressionType::Snappy, 2 },
90        lz4 = { CompressionType::Lz4, 3 },
91    )]
92    fn attrs_with_compression_exact_codes(compression: CompressionType, expected: i8) {
93        assert2::assert!(attrs_with_compression(0, compression) == expected);
94    }
95
96    #[yare::parameterized(
97        replace_codec = { 3, CompressionType::Gzip, 1 },
98        preserve_timestamp_with_codec = {
99            attrs::TIMESTAMP_TYPE_BIT,
100            CompressionType::Gzip,
101            attrs::TIMESTAMP_TYPE_BIT | 1,
102        },
103        preserve_timestamp_without_codec = {
104            attrs::TIMESTAMP_TYPE_BIT,
105            CompressionType::None,
106            attrs::TIMESTAMP_TYPE_BIT,
107        },
108    )]
109    fn attrs_with_compression_replaces_low_bits_keeps_high(
110        initial: i8,
111        compression: CompressionType,
112        expected: i8,
113    ) {
114        assert2::assert!(attrs_with_compression(initial, compression) == expected);
115    }
116
117    #[test]
118    #[should_panic(expected = "cannot carry zstd")]
119    fn attrs_with_compression_panics_on_zstd() {
120        let _ = attrs_with_compression(0, CompressionType::Zstd);
121    }
122}
krabka_records_legacy/message/attributes.rs

krabka_records_legacy/message/
attributes.rs
