krabka_records_legacy/message/
decode.rs

1//! Decoding one v0/v1 message: the CRC check over the frame, then the parse
2//! of the magic byte, the v1 timestamp, and the nullable key and value.
3//!
4//! The parse is its own module because nearly every branch in it is a
5//! rejection: a short buffer, an undersized frame, a bad CRC, an unknown
6//! magic byte, a negative length, or bytes left over after the value each
7//! map to a distinct [`LegacyRecordsError`] variant.
8
9use {
10    super::{Magic, Message},
11    crate::error::LegacyRecordsError,
12    bytes::{Buf, Bytes},
13};
14
15impl Message {
16    /// Decode a message from `buf`.
17    ///
18    /// `buf` must be positioned at the CRC and must contain at least
19    /// `frame_size` bytes. `frame_size` is the `message_size` from the
20    /// outer `MessageSet` frame.
21    /// # Errors
22    /// Returns `Truncated` if `buf` holds fewer than `frame_size` bytes.
23    /// Returns `Malformed` if `frame_size` is less than the 6-byte minimum.
24    /// Returns `CrcMismatch` if the computed CRC differs from the CRC field.
25    /// Returns `UnsupportedMagic` if the magic byte is not 0 or 1.
26    /// Returns `Truncated` if a v1 frame holds fewer than 8 timestamp bytes.
27    /// Returns `NegativeLength` if the `key` or `value` length is negative and is not -1.
28    /// Returns `Truncated` if the `key` or `value` bytes stop early.
29    /// Returns `Malformed` if bytes remain after the value.
30    pub fn decode_from<B: Buf>(buf: &mut B, frame_size: usize) -> Result<Self, LegacyRecordsError> {
31        if buf.remaining() < frame_size {
32            return Err(LegacyRecordsError::Truncated {
33                needed: frame_size - buf.remaining(),
34            });
35        }
36        if frame_size < 6 {
37            return Err(LegacyRecordsError::Malformed(format!(
38                "message frame {frame_size} bytes < 6 minimum"
39            )));
40        }
41        let mut frame = vec![0u8; frame_size];
42        buf.copy_to_slice(&mut frame);
43
44        let Some((&crc_bytes, body)) = frame.split_first_chunk::<4>() else {
45            return Err(LegacyRecordsError::Malformed(format!(
46                "message frame {frame_size} bytes < 6 minimum"
47            )));
48        };
49        let expected_crc = u32::from_be_bytes(crc_bytes);
50        let computed = crc32fast::hash(body);
51        if expected_crc != computed {
52            return Err(LegacyRecordsError::CrcMismatch {
53                expected: expected_crc,
54                computed,
55            });
56        }
57
58        let mut cur = body;
59        let magic = Magic::from_i8(cur.get_i8())?;
60        let attributes = cur.get_i8();
61        let timestamp = match magic {
62            Magic::V0 => None,
63            Magic::V1 => {
64                if cur.remaining() < 8 {
65                    return Err(LegacyRecordsError::Truncated {
66                        needed: 8 - cur.remaining(),
67                    });
68                }
69                Some(cur.get_i64())
70            }
71        };
72        let key = get_nullable_bytes(&mut cur, "key")?;
73        let value = get_nullable_bytes(&mut cur, "value")?;
74        if !cur.is_empty() {
75            return Err(LegacyRecordsError::Malformed(format!(
76                "trailing {} byte(s) inside message frame",
77                cur.len()
78            )));
79        }
80        Ok(Self {
81            magic,
82            attributes,
83            timestamp,
84            key,
85            value,
86        })
87    }
88}
89
90fn get_nullable_bytes(
91    buf: &mut &[u8],
92    label: &'static str,
93) -> Result<Option<Bytes>, LegacyRecordsError> {
94    if buf.remaining() < 4 {
95        return Err(LegacyRecordsError::Truncated {
96            needed: 4 - buf.remaining(),
97        });
98    }
99    let len = buf.get_i32();
100    let n = match usize::try_from(len) {
101        Ok(n) => n,
102        Err(_) if len == -1 => return Ok(None),
103        Err(_) => return Err(LegacyRecordsError::NegativeLength { label, len }),
104    };
105    let Some((data, rest)) = buf.split_at_checked(n) else {
106        return Err(LegacyRecordsError::Truncated {
107            needed: n - buf.remaining(),
108        });
109    };
110    let data = Bytes::copy_from_slice(data);
111    *buf = rest;
112    Ok(Some(data))
113}
114
115#[cfg(test)]
116mod tests {
117
118    use {super::*, crate::message::test_support::fixture_v1, bytes::BytesMut};
119
120    #[test]
121    fn rejects_bad_crc() {
122        let m = fixture_v1();
123        let mut buf = BytesMut::new();
124        m.encode_into(&mut buf);
125        buf[0] ^= 0xFF;
126        let mut cur: &[u8] = &buf[..];
127        assert2::assert!(matches!(
128            Message::decode_from(&mut cur, m.encoded_len()),
129            Err(LegacyRecordsError::CrcMismatch { .. })
130        ));
131    }
132
133    #[test]
134    fn rejects_unsupported_magic() {
135        let mut buf = BytesMut::new();
136        // Build a fake message with magic = 2.
137        let mut body = vec![2u8, 0u8]; // magic, attrs
138        body.extend_from_slice(&(-1i32).to_be_bytes()); // key=null
139        body.extend_from_slice(&(-1i32).to_be_bytes()); // value=null
140        let crc = crc32fast::hash(&body);
141        buf.extend_from_slice(&crc.to_be_bytes());
142        buf.extend_from_slice(&body);
143        let frame_size = buf.len();
144        let mut cur: &[u8] = &buf[..];
145        assert2::assert!(matches!(
146            Message::decode_from(&mut cur, frame_size),
147            Err(LegacyRecordsError::UnsupportedMagic { found: 2 })
148        ));
149    }
150
151    // Build a frame: crc(4) | magic | attrs | trailing, with a valid CRC.
152    fn frame_with_body(magic: u8, attrs: u8, trailing: &[u8]) -> Vec<u8> {
153        let mut body = vec![magic, attrs];
154        body.extend_from_slice(trailing);
155        let crc = crc32fast::hash(&body);
156        let mut frame = crc.to_be_bytes().to_vec();
157        frame.extend_from_slice(&body);
158        frame
159    }
160
161    #[test]
162    fn decode_buffer_shorter_than_frame_reports_needed() {
163        // 4 bytes available, frame claims 10: needed = 10 - 4 = 6.
164        let data = [0u8; 4];
165        let mut cur: &[u8] = &data;
166        let err = Message::decode_from(&mut cur, 10).unwrap_err();
167        assert2::assert!(matches!(err, LegacyRecordsError::Truncated { needed: 6 }));
168    }
169
170    #[test]
171    fn decode_rejects_frame_below_minimum() {
172        // frame_size 5 (< 6) is malformed; buffer has >= 5 bytes.
173        let data = [0u8; 5];
174        let mut cur: &[u8] = &data;
175        assert2::assert!(matches!(
176            Message::decode_from(&mut cur, 5),
177            Err(LegacyRecordsError::Malformed(_))
178        ));
179    }
180
181    #[test]
182    fn decode_min_size_frame_parses_past_guard() {
183        // A 6-byte frame clears the `< 6` guard (6 < 6 false) and then fails on
184        // the missing key-length field -> Truncated, not Malformed. This
185        // distinguishes `<` from `<=` at the boundary.
186        let frame = frame_with_body(0, 0, &[]);
187        assert2::assert!(frame.len() == 6);
188        let mut cur: &[u8] = &frame;
189        let err = Message::decode_from(&mut cur, 6).unwrap_err();
190        assert2::assert!(matches!(err, LegacyRecordsError::Truncated { .. }));
191    }
192
193    fn check_truncated_frame(magic: u8, trailing: &[u8], needed: usize) {
194        let frame = frame_with_body(magic, 0, trailing);
195        let err = Message::decode_from(&mut &frame[..], frame.len()).unwrap_err();
196        assert2::assert!(
197            matches!(err, LegacyRecordsError::Truncated { needed: actual } if actual == needed)
198        );
199    }
200
201    #[test]
202    fn decode_v1_truncated_timestamp_reports_needed() {
203        // magic+attrs then only 4 of the 8 timestamp bytes: needed = 8 - 4 = 4.
204        check_truncated_frame(1, &[0u8; 4], 4);
205    }
206
207    #[test]
208    fn decode_v1_timestamp_present_then_missing_kv() {
209        // Full 8 timestamp bytes but no key/value: clears the `< 8` timestamp
210        // guard (8 < 8 false) and fails on the missing key length (needed 4),
211        // not the guard's own needed (0). Distinguishes `<` from `<=`.
212        check_truncated_frame(1, &[0u8; 8], 4);
213    }
214
215    #[test]
216    fn decode_truncated_key_length_reports_needed() {
217        // V0 frame with only 1 byte where the 4-byte key length is expected:
218        // needed = 4 - 1 = 3.
219        check_truncated_frame(0, &[0xAA], 3);
220    }
221
222    #[test]
223    fn decode_truncated_key_body_reports_needed() {
224        // V0 frame: key length = 5 but only 2 body bytes present.
225        // needed = 5 - 2 = 3.
226        let mut trailing = 5i32.to_be_bytes().to_vec();
227        trailing.extend_from_slice(&[0xAA, 0xBB]);
228        check_truncated_frame(0, &trailing, 3);
229    }
230}
krabka_records_legacy/message/decode.rs

krabka_records_legacy/message/
decode.rs
