reddb_wire/redwire/
codec.rs

1//! Hand-rolled binary codec for v2 frames. No serde — the on-wire
2//! shape is fixed by ADR 0001, kept simple so a hex-dump is
3//! readable.
4
5use super::frame::{Flags, Frame, MessageKind, FRAME_HEADER_SIZE, MAX_FRAME_SIZE};
6
7/// Upper bound on the inflated size of a COMPRESSED payload. Four times
8/// the on-wire frame cap: enough headroom for compression to be useful,
9/// small enough that a hostile peer cannot exhaust memory with one frame.
10pub const MAX_DECOMPRESSED_PAYLOAD: usize = 4 * (MAX_FRAME_SIZE as usize);
11
12fn decompress_bounded(on_wire: &[u8], payload_len: usize) -> Result<Vec<u8>, FrameError> {
13    use std::io::Read;
14    let limit = MAX_DECOMPRESSED_PAYLOAD as u64;
15    let unparseable = |e: std::io::Error| FrameError::PayloadTruncated {
16        // Reuse PayloadTruncated for "decompression failed" rather than
17        // introduce a new variant — the wire-layer outcome is the same:
18        // the body is unparseable, drop the connection.
19        expected: payload_len as u32,
20        available: e.to_string().len() as u32,
21    };
22    let decoder = zstd::stream::Decoder::new(on_wire).map_err(unparseable)?;
23    let mut plain = Vec::new();
24    // Read one byte past the limit so an exactly-at-limit payload is
25    // accepted while anything larger is detected without buffering it.
26    decoder
27        .take(limit + 1)
28        .read_to_end(&mut plain)
29        .map_err(unparseable)?;
30    if plain.len() as u64 > limit {
31        return Err(FrameError::DecompressedTooLarge {
32            limit: MAX_DECOMPRESSED_PAYLOAD as u32,
33        });
34    }
35    Ok(plain)
36}
37
38#[derive(Debug, Clone, PartialEq, Eq)]
39pub enum FrameError {
40    Truncated,
41    InvalidLength(u32),
42    PayloadTruncated {
43        expected: u32,
44        available: u32,
45    },
46    UnknownKind(u8),
47    UnknownFlags(u8),
48    /// Catalog cross-check failed: the flag bits set on the frame are
49    /// not in `MessageKind::allowed_flags()` for this kind. The wire
50    /// catalog is the single source of truth for which bits a kind
51    /// may carry — see `frame.rs::MessageKind::allowed_flags`.
52    /// A COMPRESSED payload inflated past [`MAX_DECOMPRESSED_PAYLOAD`].
53    DecompressedTooLarge {
54        limit: u32,
55    },
56    FlagsNotAllowedForKind {
57        kind: u8,
58        flags: u8,
59    },
60}
61
62impl std::fmt::Display for FrameError {
63    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
64        match self {
65            Self::Truncated => write!(f, "frame header truncated (< 16 bytes)"),
66            Self::InvalidLength(n) => write!(f, "frame length field invalid: {n}"),
67            Self::PayloadTruncated {
68                expected,
69                available,
70            } => write!(
71                f,
72                "frame payload truncated: expected {expected} bytes, got {available}"
73            ),
74            Self::UnknownKind(byte) => write!(f, "unknown message kind 0x{byte:02x}"),
75            Self::UnknownFlags(byte) => write!(f, "unknown flag bits 0x{byte:02x}"),
76            Self::DecompressedTooLarge { limit } => {
77                write!(f, "compressed payload inflates past the {limit}-byte limit")
78            }
79            Self::FlagsNotAllowedForKind { kind, flags } => write!(
80                f,
81                "flag bits 0x{flags:02x} not allowed on kind 0x{kind:02x}"
82            ),
83        }
84    }
85}
86
87impl std::error::Error for FrameError {}
88
89pub fn frame_len_from_header(header: &[u8; FRAME_HEADER_SIZE]) -> Result<usize, FrameError> {
90    let length = u32::from_le_bytes([header[0], header[1], header[2], header[3]]);
91    if length < FRAME_HEADER_SIZE as u32 || length > MAX_FRAME_SIZE {
92        return Err(FrameError::InvalidLength(length));
93    }
94    Ok(length as usize)
95}
96
97pub fn decode_frame_parts(
98    header: &[u8; FRAME_HEADER_SIZE],
99    payload: &[u8],
100) -> Result<Frame, FrameError> {
101    let length = frame_len_from_header(header)?;
102    let expected_payload_len = length - FRAME_HEADER_SIZE;
103    if payload.len() < expected_payload_len {
104        return Err(FrameError::PayloadTruncated {
105            expected: expected_payload_len as u32,
106            available: payload.len() as u32,
107        });
108    }
109
110    let mut bytes = Vec::with_capacity(length);
111    bytes.extend_from_slice(header);
112    bytes.extend_from_slice(&payload[..expected_payload_len]);
113    decode_frame(&bytes).map(|(frame, _)| frame)
114}
115
116pub fn encode_frame(frame: &Frame) -> Vec<u8> {
117    // The frame's `payload` is always the plaintext form. If the
118    // COMPRESSED flag is set we compress on the wire and rewrite
119    // the length header to match the compressed size — the
120    // receiver inflates before delivering to the dispatch loop.
121    if frame.flags.contains(Flags::COMPRESSED) {
122        return encode_compressed(frame);
123    }
124    let total = frame.encoded_len() as usize;
125    let mut buf = Vec::with_capacity(total);
126    encode_uncompressed_frame_into(frame, &mut buf);
127    buf
128}
129
130/// Append an uncompressed frame to caller-owned capacity. Callers can reserve
131/// and admit the complete allocation before encoding. Compression has its own
132/// codec workspace and must use `encode_frame` instead.
133pub fn encode_uncompressed_frame_into(frame: &Frame, buf: &mut Vec<u8>) {
134    encode_uncompressed_payload_into(frame, &frame.payload, buf);
135}
136
137pub(crate) fn encode_uncompressed_payload_into(frame: &Frame, payload: &[u8], buf: &mut Vec<u8>) {
138    assert!(!frame.flags.contains(Flags::COMPRESSED));
139    let length = u32::try_from(FRAME_HEADER_SIZE + payload.len()).expect("validated frame size");
140    buf.extend_from_slice(&length.to_le_bytes());
141    buf.push(frame.kind as u8);
142    buf.push(frame.flags.bits());
143    buf.extend_from_slice(&frame.stream_id.to_le_bytes());
144    buf.extend_from_slice(&frame.correlation_id.to_le_bytes());
145    buf.extend_from_slice(payload);
146}
147
148fn encode_compressed(frame: &Frame) -> Vec<u8> {
149    // zstd level 1 — keeps CPU low while still cutting JSON +
150    // BulkInsertBinary by 60-80%. Operators that want max ratio
151    // can flip to level 3+ via `RED_REDWIRE_ZSTD_LEVEL` env.
152    let level = std::env::var("RED_REDWIRE_ZSTD_LEVEL")
153        .ok()
154        .and_then(|s| s.parse::<i32>().ok())
155        .unwrap_or(1);
156    let compressed = match zstd::stream::encode_all(frame.payload.as_slice(), level) {
157        Ok(buf) => buf,
158        Err(_) => {
159            // Fallback: drop the COMPRESSED flag and ship plaintext.
160            // Compression failures are rare (level 1 effectively
161            // never fails on bytes), but the fallback is safer
162            // than panicking inside the framing layer.
163            let mut clone = frame.clone();
164            clone.flags = Flags::from_bits(clone.flags.bits() & !Flags::COMPRESSED.bits());
165            return encode_frame(&clone);
166        }
167    };
168    let total = (FRAME_HEADER_SIZE + compressed.len()) as u32;
169    let mut buf = Vec::with_capacity(total as usize);
170    buf.extend_from_slice(&total.to_le_bytes());
171    buf.push(frame.kind as u8);
172    buf.push(frame.flags.bits());
173    buf.extend_from_slice(&frame.stream_id.to_le_bytes());
174    buf.extend_from_slice(&frame.correlation_id.to_le_bytes());
175    buf.extend_from_slice(&compressed);
176    buf
177}
178
179pub fn decode_frame(bytes: &[u8]) -> Result<(Frame, usize), FrameError> {
180    if bytes.len() < FRAME_HEADER_SIZE {
181        return Err(FrameError::Truncated);
182    }
183    let mut header = [0u8; FRAME_HEADER_SIZE];
184    header.copy_from_slice(&bytes[..FRAME_HEADER_SIZE]);
185    let length = frame_len_from_header(&header)? as u32;
186    if (bytes.len() as u32) < length {
187        return Err(FrameError::PayloadTruncated {
188            expected: length,
189            available: bytes.len() as u32,
190        });
191    }
192    let kind = MessageKind::from_u8(bytes[4]).ok_or(FrameError::UnknownKind(bytes[4]))?;
193    let flag_bits = bytes[5];
194    const KNOWN_FLAGS: u8 = 0b0000_0011;
195    if flag_bits & !KNOWN_FLAGS != 0 {
196        return Err(FrameError::UnknownFlags(flag_bits));
197    }
198    let flags = Flags::from_bits(flag_bits);
199    // Catalog cross-check: the kind's `allowed_flags()` is the single
200    // source of truth. Reject combinations the catalog forbids
201    // (e.g. COMPRESSED on tiny handshake payloads) so misframed
202    // frames fail at the boundary instead of reaching dispatch.
203    if !kind.permits_flags(flags) {
204        return Err(FrameError::FlagsNotAllowedForKind {
205            kind: bytes[4],
206            flags: flag_bits,
207        });
208    }
209    let stream_id = u16::from_le_bytes([bytes[6], bytes[7]]);
210    let correlation_id = u64::from_le_bytes([
211        bytes[8], bytes[9], bytes[10], bytes[11], bytes[12], bytes[13], bytes[14], bytes[15],
212    ]);
213    let payload_len = (length as usize) - FRAME_HEADER_SIZE;
214    let on_wire = &bytes[FRAME_HEADER_SIZE..FRAME_HEADER_SIZE + payload_len];
215    let payload = if flags.contains(Flags::COMPRESSED) {
216        // Decompress on read so the rest of the dispatch loop
217        // sees plaintext bytes regardless of how they arrived.
218        // `MAX_FRAME_SIZE` only bounds the *compressed* bytes; a
219        // streaming decoder capped at `MAX_DECOMPRESSED_PAYLOAD` keeps a
220        // 16 MiB zstd bomb from inflating into gigabytes of heap.
221        decompress_bounded(on_wire, payload_len)?
222    } else {
223        on_wire.to_vec()
224    };
225    Ok((
226        Frame {
227            kind,
228            // The flag stays on the decoded frame so dispatch can
229            // see it was compressed if it cares (audit, metrics).
230            flags,
231            stream_id,
232            correlation_id,
233            payload,
234        },
235        length as usize,
236    ))
237}
238
239#[cfg(test)]
240mod tests {
241    #[test]
242    fn caller_owned_encoding_matches_golden_header_and_preserves_prefix() {
243        use super::*;
244        let mut frame = Frame::new(MessageKind::Result, 0x0807060504030201, vec![0xaa, 0xbb]);
245        frame.stream_id = 0x0201;
246        let mut bytes = Vec::with_capacity(21);
247        bytes.extend_from_slice(b"pre");
248        encode_uncompressed_frame_into(&frame, &mut bytes);
249        assert_eq!(&bytes[..3], b"pre");
250        assert_eq!(
251            &bytes[3..],
252            &[
253                18,
254                0,
255                0,
256                0,
257                MessageKind::Result as u8,
258                0,
259                1,
260                2,
261                1,
262                2,
263                3,
264                4,
265                5,
266                6,
267                7,
268                8,
269                0xaa,
270                0xbb
271            ]
272        );
273        assert_eq!(bytes.capacity(), 21);
274        assert_eq!(&bytes[3..], encode_frame(&frame));
275        let (decoded, used) = decode_frame(&bytes[3..]).unwrap();
276        assert_eq!(used, 18);
277        assert_eq!(decoded.payload, frame.payload);
278        assert_eq!(decoded.stream_id, frame.stream_id);
279        assert_eq!(decoded.correlation_id, frame.correlation_id);
280    }
281
282    use super::*;
283
284    fn round_trip(frame: Frame) {
285        let bytes = encode_frame(&frame);
286        let (decoded, consumed) = decode_frame(&bytes).expect("decode");
287        assert_eq!(consumed, bytes.len());
288        assert_eq!(decoded, frame);
289    }
290
291    #[test]
292    fn round_trip_empty_payload() {
293        round_trip(Frame::new(MessageKind::Ping, 1, vec![]));
294    }
295
296    #[test]
297    fn frame_len_from_header_validates_bounds() {
298        let mut header = [0u8; FRAME_HEADER_SIZE];
299        header[..4].copy_from_slice(&(FRAME_HEADER_SIZE as u32).to_le_bytes());
300        assert_eq!(frame_len_from_header(&header).unwrap(), FRAME_HEADER_SIZE);
301
302        header[..4].copy_from_slice(&15u32.to_le_bytes());
303        assert_eq!(
304            frame_len_from_header(&header),
305            Err(FrameError::InvalidLength(15))
306        );
307
308        header[..4].copy_from_slice(&(MAX_FRAME_SIZE + 1).to_le_bytes());
309        assert_eq!(
310            frame_len_from_header(&header),
311            Err(FrameError::InvalidLength(MAX_FRAME_SIZE + 1))
312        );
313    }
314
315    #[test]
316    fn round_trip_with_payload() {
317        round_trip(Frame::new(MessageKind::Query, 42, b"SELECT 1".to_vec()));
318    }
319
320    #[test]
321    fn decode_frame_parts_matches_full_buffer_decode() {
322        let frame = Frame::new(MessageKind::Result, 42, br#"{"ok":true}"#.to_vec());
323        let bytes = encode_frame(&frame);
324        let mut header = [0u8; FRAME_HEADER_SIZE];
325        header.copy_from_slice(&bytes[..FRAME_HEADER_SIZE]);
326        let payload = &bytes[FRAME_HEADER_SIZE..];
327
328        let decoded = decode_frame_parts(&header, payload).expect("decode parts");
329        assert_eq!(decoded, frame);
330    }
331
332    #[test]
333    fn round_trip_with_stream_and_flags() {
334        let frame = Frame::new(MessageKind::Result, 999, vec![0xab; 256])
335            .with_stream(7)
336            .with_flags(Flags::COMPRESSED | Flags::MORE_FRAMES);
337        round_trip(frame);
338    }
339
340    #[test]
341    fn truncated_header_rejected() {
342        assert_eq!(decode_frame(&[]), Err(FrameError::Truncated));
343        assert_eq!(decode_frame(&[0; 15]), Err(FrameError::Truncated));
344    }
345
346    #[test]
347    fn length_below_header_rejected() {
348        let mut bytes = vec![0u8; 16];
349        bytes[..4].copy_from_slice(&15u32.to_le_bytes());
350        assert!(matches!(
351            decode_frame(&bytes),
352            Err(FrameError::InvalidLength(15))
353        ));
354    }
355
356    #[test]
357    fn unknown_kind_rejected() {
358        let mut bytes = vec![0u8; 16];
359        bytes[..4].copy_from_slice(&16u32.to_le_bytes());
360        bytes[4] = 0xff;
361        assert_eq!(decode_frame(&bytes), Err(FrameError::UnknownKind(0xff)));
362    }
363
364    #[test]
365    fn unknown_flag_bits_rejected() {
366        let mut bytes = vec![0u8; 16];
367        bytes[..4].copy_from_slice(&16u32.to_le_bytes());
368        bytes[4] = MessageKind::Ping as u8;
369        bytes[5] = 0b1000_0000;
370        assert!(matches!(
371            decode_frame(&bytes),
372            Err(FrameError::UnknownFlags(_))
373        ));
374    }
375
376    #[test]
377    fn flags_not_allowed_for_kind_rejected() {
378        // Ping is a handshake kind — the catalog forbids COMPRESSED
379        // on tiny handshake payloads. A frame with kind=Ping and the
380        // COMPRESSED bit set must be rejected at the boundary.
381        let mut bytes = vec![0u8; 16];
382        bytes[..4].copy_from_slice(&16u32.to_le_bytes());
383        bytes[4] = MessageKind::Ping as u8;
384        bytes[5] = Flags::COMPRESSED.bits();
385        match decode_frame(&bytes) {
386            Err(FrameError::FlagsNotAllowedForKind { kind, flags }) => {
387                assert_eq!(kind, MessageKind::Ping as u8);
388                assert_eq!(flags, Flags::COMPRESSED.bits());
389            }
390            other => panic!("expected FlagsNotAllowedForKind, got {other:?}"),
391        }
392    }
393
394    #[test]
395    fn streaming_decode_two_frames_back_to_back() {
396        let f1 = Frame::new(MessageKind::Query, 1, b"a".to_vec());
397        let f2 = Frame::new(MessageKind::Query, 2, b"b".to_vec());
398        let mut buf = encode_frame(&f1);
399        buf.extend(encode_frame(&f2));
400        let (got1, n1) = decode_frame(&buf).unwrap();
401        let (got2, _n2) = decode_frame(&buf[n1..]).unwrap();
402        assert_eq!(got1, f1);
403        assert_eq!(got2, f2);
404    }
405
406    #[test]
407    fn compressed_round_trip_recovers_plaintext() {
408        // A compressible payload — a kilobyte of repeating text.
409        let payload = b"abcabcabcabc".repeat(100);
410        let frame =
411            Frame::new(MessageKind::Result, 7, payload.clone()).with_flags(Flags::COMPRESSED);
412        let bytes = encode_frame(&frame);
413        // Wire form should be smaller than the plaintext frame.
414        assert!(
415            bytes.len() < FRAME_HEADER_SIZE + payload.len(),
416            "compressed frame ({}) must be smaller than plaintext payload ({})",
417            bytes.len(),
418            payload.len(),
419        );
420        let (decoded, _) = decode_frame(&bytes).expect("decode compressed");
421        assert_eq!(decoded.payload, payload);
422        assert!(decoded.flags.contains(Flags::COMPRESSED));
423    }
424
425    #[test]
426    fn output_stream_lifecycle_envelopes_round_trip() {
427        // Golden encode/decode for every new variant added in
428        // issue #762 / PRD #759 S3. Pins the exact byte values
429        // and confirms `stream_id` multiplex routing survives the
430        // round trip (it is what `StreamCancel`'s per-stream
431        // targeting relies on).
432        let open = Frame::new(
433            MessageKind::OpenStream,
434            10,
435            br#"{"sql":"SELECT 1","opts":{}}"#.to_vec(),
436        )
437        .with_stream(7);
438        round_trip(open.clone());
439        assert_eq!(encode_frame(&open)[4], 0x29);
440
441        let ack = Frame::new(
442            MessageKind::OpenAck,
443            10,
444            br#"{"lease_handle":"42","resumable":false,"snapshot_lsn":1234}"#.to_vec(),
445        )
446        .with_stream(7);
447        round_trip(ack.clone());
448        assert_eq!(encode_frame(&ack)[4], 0x2A);
449
450        let chunk = Frame::new(
451            MessageKind::StreamChunk,
452            10,
453            br#"{"seq":0,"rows":[{"a":1}],"terminal":false}"#.to_vec(),
454        )
455        .with_stream(7);
456        round_trip(chunk.clone());
457        assert_eq!(encode_frame(&chunk)[4], 0x2B);
458
459        let serr = Frame::new(
460            MessageKind::StreamError,
461            10,
462            br#"{"code":"unknown_stream","message":"x"}"#.to_vec(),
463        )
464        .with_stream(7);
465        round_trip(serr.clone());
466        assert_eq!(encode_frame(&serr)[4], 0x2C);
467
468        let end = Frame::new(
469            MessageKind::StreamEnd,
470            10,
471            br#"{"stats":{"row_count":1}}"#.to_vec(),
472        )
473        .with_stream(7);
474        round_trip(end.clone());
475        assert_eq!(encode_frame(&end)[4], 0x25);
476
477        let cancel = Frame::new(
478            MessageKind::StreamCancel,
479            10,
480            br#"{"reason":"client-abort"}"#.to_vec(),
481        )
482        .with_stream(7);
483        round_trip(cancel.clone());
484        assert_eq!(encode_frame(&cancel)[4], 0x2D);
485    }
486
487    #[test]
488    fn input_stream_envelopes_round_trip() {
489        // Golden encode/decode for the input-direction envelopes
490        // added in issue #764 / PRD #759 S5. The envelope *vocabulary*
491        // is reused from S3 — only the payload shapes and the
492        // direction of `StreamChunk` differ — so the byte values are
493        // pinned to the same kinds (no new MessageKind bytes).
494
495        // OpenStream with `direction:"in"` + target/columns instead of
496        // a `sql` field. Still kind 0x29, still multiplex via stream_id.
497        let open_in = Frame::new(
498            MessageKind::OpenStream,
499            20,
500            br#"{"direction":"in","target":"t","columns":["id","name"]}"#.to_vec(),
501        )
502        .with_stream(5);
503        round_trip(open_in.clone());
504        assert_eq!(encode_frame(&open_in)[4], 0x29);
505
506        // Client-originated chunk of rows on the input stream. Same
507        // 0x2B kind the server uses on output streams; the rows are
508        // JSON objects keyed by column.
509        let chunk_in = Frame::new(
510            MessageKind::StreamChunk,
511            20,
512            br#"{"seq":0,"rows":[{"id":1,"name":"a"}],"terminal":false}"#.to_vec(),
513        )
514        .with_stream(5);
515        round_trip(chunk_in.clone());
516        assert_eq!(encode_frame(&chunk_in)[4], 0x2B);
517
518        // Terminal chunk closes the input stream.
519        let chunk_terminal = Frame::new(
520            MessageKind::StreamChunk,
521            20,
522            br#"{"seq":2,"rows":[],"terminal":true}"#.to_vec(),
523        )
524        .with_stream(5);
525        round_trip(chunk_terminal.clone());
526        assert_eq!(encode_frame(&chunk_terminal)[4], 0x2B);
527
528        // Server StreamEnd carries the committed RID range + stats.
529        let end = Frame::new(
530            MessageKind::StreamEnd,
531            20,
532            br#"{"stats":{"row_count":3,"chunk_count":2,"committed_rid":42,"snapshot_lsn":40,"cancelled":false}}"#.to_vec(),
533        )
534        .with_stream(5);
535        round_trip(end.clone());
536        assert_eq!(encode_frame(&end)[4], 0x25);
537
538        // Server StreamError carries the recoverable_rid prefix.
539        let serr = Frame::new(
540            MessageKind::StreamError,
541            20,
542            br#"{"code":"invalid_row","message":"x","chunk_seq":1,"recoverable_rid":41}"#.to_vec(),
543        )
544        .with_stream(5);
545        round_trip(serr.clone());
546        assert_eq!(encode_frame(&serr)[4], 0x2C);
547    }
548
549    #[test]
550    fn queue_wait_envelopes_round_trip() {
551        // Golden encode/decode for the live queue-wait envelopes added
552        // in issue #917 / PRD #915. Pins the new byte values and
553        // confirms the request/push pair round-trips equal through the
554        // codec, multiplexed over `stream_id` like the other streamed
555        // envelopes.
556        let open = Frame::new(
557            MessageKind::QueueWaitOpen,
558            10,
559            br#"{"queue":"jobs","consumer":"w1","count":1,"wait_ms":5000}"#.to_vec(),
560        )
561        .with_stream(3);
562        round_trip(open.clone());
563        assert_eq!(encode_frame(&open)[4], 0x2E);
564
565        let push = Frame::new(
566            MessageKind::QueueEventPush,
567            10,
568            br#"{"message_id":"42","payload":{"hello":"world"},"consumer":"w1","delivery_count":1}"#
569                .to_vec(),
570        )
571        .with_stream(3);
572        round_trip(push.clone());
573        assert_eq!(encode_frame(&push)[4], 0x2F);
574    }
575
576    #[test]
577    fn uncompressed_frame_decodes_unchanged_when_flag_unset() {
578        let payload = b"hello world".to_vec();
579        let frame = Frame::new(MessageKind::Result, 1, payload.clone());
580        let bytes = encode_frame(&frame);
581        let (decoded, _) = decode_frame(&bytes).unwrap();
582        assert_eq!(decoded.payload, payload);
583        assert!(!decoded.flags.contains(Flags::COMPRESSED));
584    }
585
586    #[test]
587    fn compressed_payload_that_inflates_past_the_cap_is_refused() {
588        // MAX_FRAME_SIZE bounds only the *compressed* bytes. A highly
589        // compressible payload therefore lets a peer turn a small frame into
590        // a multi-gigabyte allocation unless the decoder is bounded.
591        let plain = vec![0u8; MAX_DECOMPRESSED_PAYLOAD + 1024];
592        let compressed = zstd::stream::encode_all(plain.as_slice(), 1).expect("compress");
593        assert!(
594            compressed.len() < MAX_FRAME_SIZE as usize,
595            "the bomb must fit in one legal frame to be a realistic attack"
596        );
597
598        let total = (FRAME_HEADER_SIZE + compressed.len()) as u32;
599        let mut bytes = Vec::with_capacity(total as usize);
600        bytes.extend_from_slice(&total.to_le_bytes());
601        bytes.push(MessageKind::Query as u8);
602        bytes.push(Flags::COMPRESSED.bits());
603        bytes.extend_from_slice(&0u16.to_le_bytes());
604        bytes.extend_from_slice(&0u64.to_le_bytes());
605        bytes.extend_from_slice(&compressed);
606
607        match decode_frame(&bytes) {
608            Err(FrameError::DecompressedTooLarge { limit }) => {
609                assert_eq!(limit as usize, MAX_DECOMPRESSED_PAYLOAD);
610            }
611            other => panic!("expected DecompressedTooLarge, got {other:?}"),
612        }
613    }
614}
reddb_wire/redwire/codec.rs

reddb_wire/redwire/
codec.rs
