reth_ipc/
stream_codec.rs

1// Copyright (c) 2015-2017 Parity Technologies Limited
2//
3// Permission is hereby granted, free of charge, to any
4// person obtaining a copy of this software and associated
5// documentation files (the "Software"), to deal in the
6// Software without restriction, including without
7// limitation the rights to use, copy, modify, merge,
8// publish, distribute, sublicense, and/or sell copies of
9// the Software, and to permit persons to whom the Software
10// is furnished to do so, subject to the following
11// conditions:
12//
13// The above copyright notice and this permission notice
14// shall be included in all copies or substantial portions
15// of the Software.
16//
17// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF
18// ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED
19// TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A
20// PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT
21// SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY
22// CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION
23// OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR
24// IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
25// DEALINGS IN THE SOFTWARE.
26
27// This basis of this file has been taken from the deprecated jsonrpc codebase:
28// https://github.com/paritytech/jsonrpc
29
30use bytes::{Buf, BufMut, BytesMut};
31use std::{io, str};
32
33/// Separator for enveloping messages in streaming codecs
34#[derive(Debug, Clone)]
35pub enum Separator {
36    /// No envelope is expected between messages. Decoder will try to figure out
37    /// message boundaries by accumulating incoming bytes until valid JSON is formed.
38    /// Encoder will send messages without any boundaries between requests.
39    Empty,
40    /// Byte is used as a sentinel between messages
41    Byte(u8),
42}
43
44impl Default for Separator {
45    fn default() -> Self {
46        Self::Byte(b'\n')
47    }
48}
49
50/// Stream codec for streaming protocols (ipc, tcp)
51#[derive(Debug, Default)]
52pub struct StreamCodec {
53    incoming_separator: Separator,
54    outgoing_separator: Separator,
55    scan: ScanState,
56}
57
58impl StreamCodec {
59    /// Default codec with streaming input data. Input can be both enveloped and not.
60    pub fn stream_incoming() -> Self {
61        Self::new(Separator::Empty, Default::default())
62    }
63
64    /// New custom stream codec
65    pub const fn new(incoming_separator: Separator, outgoing_separator: Separator) -> Self {
66        Self { incoming_separator, outgoing_separator, scan: ScanState::new() }
67    }
68}
69
70/// Scan state of the [`Separator::Empty`] decoder, carried across `decode` calls so bytes
71/// scanned by a previous call are not scanned again.
72///
73/// Indices refer to the current read buffer and must be reset whenever bytes are consumed from
74/// it.
75#[derive(Debug)]
76struct ScanState {
77    depth: i32,
78    in_str: bool,
79    is_escaped: bool,
80    start_idx: usize,
81    whitespaces: usize,
82    cursor: usize,
83}
84
85impl ScanState {
86    const fn new() -> Self {
87        Self { depth: 0, in_str: false, is_escaped: false, start_idx: 0, whitespaces: 0, cursor: 0 }
88    }
89}
90
91impl Default for ScanState {
92    fn default() -> Self {
93        Self::new()
94    }
95}
96
97#[inline]
98const fn is_whitespace(byte: u8) -> bool {
99    matches!(byte, 0x0D | 0x0A | 0x20 | 0x09)
100}
101
102impl tokio_util::codec::Decoder for StreamCodec {
103    type Item = String;
104    type Error = io::Error;
105
106    fn decode(&mut self, buf: &mut BytesMut) -> io::Result<Option<Self::Item>> {
107        if let Separator::Byte(separator) = self.incoming_separator {
108            if let Some(i) = buf.as_ref().iter().position(|&b| b == separator) {
109                let line = buf.split_to(i);
110                let _ = buf.split_to(1);
111
112                match str::from_utf8(line.as_ref()) {
113                    Ok(s) => Ok(Some(s.to_string())),
114                    Err(_) => Err(io::Error::other("invalid UTF-8")),
115                }
116            } else {
117                Ok(None)
118            }
119        } else {
120            // resume scanning at the byte the previous call stopped at
121            while self.scan.cursor < buf.len() {
122                let idx = self.scan.cursor;
123                let byte = buf[idx];
124
125                if (byte == b'{' || byte == b'[') && !self.scan.in_str {
126                    if self.scan.depth == 0 {
127                        self.scan.start_idx = idx;
128                    }
129                    self.scan.depth += 1;
130                } else if (byte == b'}' || byte == b']') && !self.scan.in_str {
131                    self.scan.depth -= 1;
132                } else if byte == b'"' && !self.scan.is_escaped {
133                    self.scan.in_str = !self.scan.in_str;
134                } else if is_whitespace(byte) {
135                    self.scan.whitespaces += 1;
136                }
137                self.scan.is_escaped = byte == b'\\' && !self.scan.is_escaped && self.scan.in_str;
138
139                if self.scan.depth == 0 &&
140                    idx != self.scan.start_idx &&
141                    idx - self.scan.start_idx + 1 > self.scan.whitespaces
142                {
143                    let start = self.scan.start_idx;
144                    let end = idx + 1;
145                    // reset before advancing the buffer because the stored indices go stale
146                    self.scan = ScanState::new();
147                    if start > 0 {
148                        buf.advance(start);
149                    }
150                    let bts = buf.split_to(end - start);
151                    return Ok(String::from_utf8(bts.into()).ok())
152                }
153
154                self.scan.cursor += 1;
155            }
156            Ok(None)
157        }
158    }
159}
160
161impl tokio_util::codec::Encoder<String> for StreamCodec {
162    type Error = io::Error;
163
164    fn encode(&mut self, msg: String, buf: &mut BytesMut) -> io::Result<()> {
165        match self.outgoing_separator {
166            Separator::Byte(separator) => {
167                buf.reserve(msg.len() + 1);
168                buf.extend_from_slice(msg.as_bytes());
169                buf.put_u8(separator);
170            }
171            Separator::Empty => buf.extend_from_slice(msg.as_bytes()),
172        }
173        Ok(())
174    }
175}
176
177#[cfg(test)]
178mod tests {
179    use super::*;
180    use bytes::BufMut;
181    use rand::{rngs::StdRng, RngCore, SeedableRng};
182    use tokio_util::codec::Decoder;
183
184    #[test]
185    fn simple_encode() {
186        let mut buf = BytesMut::with_capacity(2048);
187        buf.put_slice(b"{ test: 1 }{ test: 2 }{ test: 3 }");
188
189        let mut codec = StreamCodec::stream_incoming();
190
191        let request = codec
192            .decode(&mut buf)
193            .expect("There should be no error in simple test")
194            .expect("There should be at least one request in simple test");
195
196        assert_eq!(request, "{ test: 1 }");
197    }
198
199    #[test]
200    fn escape() {
201        let mut buf = BytesMut::with_capacity(2048);
202        buf.put_slice(br#"{ test: "\"\\" }{ test: "\ " }{ test: "\}" }[ test: "\]" ]"#);
203
204        let mut codec = StreamCodec::stream_incoming();
205
206        let request = codec
207            .decode(&mut buf)
208            .expect("There should be no error in first escape test")
209            .expect("There should be a request in first escape test");
210
211        assert_eq!(request, r#"{ test: "\"\\" }"#);
212
213        let request2 = codec
214            .decode(&mut buf)
215            .expect("There should be no error in 2nd escape test")
216            .expect("There should be a request in 2nd escape test");
217        assert_eq!(request2, r#"{ test: "\ " }"#);
218
219        let request3 = codec
220            .decode(&mut buf)
221            .expect("There should be no error in 3rd escape test")
222            .expect("There should be a request in 3rd escape test");
223        assert_eq!(request3, r#"{ test: "\}" }"#);
224
225        let request4 = codec
226            .decode(&mut buf)
227            .expect("There should be no error in 4th escape test")
228            .expect("There should be a request in 4th escape test");
229        assert_eq!(request4, r#"[ test: "\]" ]"#);
230    }
231
232    #[test]
233    fn whitespace() {
234        let mut buf = BytesMut::with_capacity(2048);
235        buf.put_slice(b"{ test: 1 }\n\n\n\n{ test: 2 }\n\r{\n test: 3 }  ");
236
237        let mut codec = StreamCodec::stream_incoming();
238
239        let request = codec
240            .decode(&mut buf)
241            .expect("There should be no error in first whitespace test")
242            .expect("There should be a request in first whitespace test");
243
244        assert_eq!(request, "{ test: 1 }");
245
246        let request2 = codec
247            .decode(&mut buf)
248            .expect("There should be no error in first 2nd test")
249            .expect("There should be a request in 2nd whitespace test");
250        assert_eq!(request2, "{ test: 2 }");
251
252        let request3 = codec
253            .decode(&mut buf)
254            .expect("There should be no error in first 3rd test")
255            .expect("There should be a request in 3rd whitespace test");
256        assert_eq!(request3, "{\n test: 3 }");
257
258        let request4 = codec.decode(&mut buf).expect("There should be no error in first 4th test");
259        assert!(
260            request4.is_none(),
261            "There should be no 4th request because it contains only whitespaces"
262        );
263    }
264
265    #[test]
266    fn fragmented_encode() {
267        let mut buf = BytesMut::with_capacity(2048);
268        buf.put_slice(b"{ test: 1 }{ test: 2 }{ tes");
269
270        let mut codec = StreamCodec::stream_incoming();
271
272        let request = codec
273            .decode(&mut buf)
274            .expect("There should be no error in first fragmented test")
275            .expect("There should be at least one request in first fragmented test");
276        assert_eq!(request, "{ test: 1 }");
277        codec
278            .decode(&mut buf)
279            .expect("There should be no error in second fragmented test")
280            .expect("There should be at least one request in second fragmented test");
281        assert_eq!(String::from_utf8(buf.as_ref().to_vec()).unwrap(), "{ tes");
282
283        buf.put_slice(b"t: 3 }");
284        let request = codec
285            .decode(&mut buf)
286            .expect("There should be no error in third fragmented test")
287            .expect("There should be at least one request in third fragmented test");
288        assert_eq!(request, "{ test: 3 }");
289    }
290
291    #[test]
292    fn huge() {
293        let request = r#"{
294			"jsonrpc":"2.0",
295			"method":"say_hello",
296			"params": [
297				42,
298				0,
299				{
300					"from":"0xb60e8dd61c5d32be8058bb8eb970870f07233155",
301					"gas":"0x2dc6c0",
302					"data":"0x606060405260003411156010576002565b6001805433600160a060020a0319918216811790925560028054909116909117905561291f806100406000396000f3606060405236156100e55760e060020a600035046304029f2381146100ed5780630a1273621461015f57806317c1dd87146102335780631f9ea25d14610271578063266fa0e91461029357806349593f5314610429578063569aa0d8146104fc57806359a4669f14610673578063647a4d5f14610759578063656104f5146108095780636e9febfe1461082b57806370de8c6e1461090d57806371bde852146109ed5780638f30435d14610ab4578063916dbc1714610da35780639f5a7cd414610eef578063c91540f614610fe6578063eae99e1c146110b5578063fedc2a281461115a575b61122d610002565b61122d6004808035906020019082018035906020019191908080601f01602080910402602001604051908101604052809392919081815260200183838082843750949650509335935050604435915050606435600154600090600160a060020a03908116339091161461233357610002565b61122f6004808035906020019082018035906020019191908080601f016020809104026020016040519081016040528093929190818152602001838380828437509496505093359350506044359150506064355b60006000600060005086604051808280519060200190808383829060006004602084601f0104600f02600301f1509050019150509081526020016040518091039020600050905042816005016000508560ff1660028110156100025760040201835060010154604060020a90046001604060020a0316116115df576115d6565b6112416004355b604080516001604060020a038316408152606060020a33600160a060020a031602602082015290519081900360340190205b919050565b61122d600435600254600160a060020a0390811633909116146128e357610002565b61125e6004808035906020019082018035906020019191908080601f01602080910402602001604051908101604052809392919081815260200183838082843750949650509335935050505060006000600060006000600060005087604051808280519060200190808383829060006004602084601f0104600f02600301f1509050019150509081526020016040518091039020600050905080600001600050600087600160a060020a0316815260200190815260200160002060005060000160059054906101000a90046001604060020a03169450845080600001600050600087600160a060020a03168152602001908152602001600020600050600001600d9054906101000a90046001604060020a03169350835080600001600050600087600160a060020a0316815260200190815260200160002060005060000160009054906101000a900460ff169250825080600001600050600087600160a060020a0316815260200190815260200160002060005060000160019054906101000a900463ffffffff16915081505092959194509250565b61122d6004808035906020019082018035906020019191908080601f01602080910402602001604051908101604052809392919081815260200183838082843750949650509335935050604435915050606435608435600060006000600060005088604051808280519060200190808383829060006004602084601f0104600f02600301f15090500191505090815260200160405180910390206000509250346000141515611c0e5760405133600160a060020a0316908290349082818181858883f193505050501515611c1a57610002565b6112996004808035906020019082018035906020019191908080601f01602080910402602001604051908101604052809392919081815260200183838082843750949650509335935050604435915050600060006000600060006000600060006000508a604051808280519060200190808383829060006004602084601f0104600f02600301f15090500191505090815260200160405180910390206000509050806001016000508960ff16600281101561000257600160a060020a038a168452828101600101602052604084205463ffffffff1698506002811015610002576040842054606060020a90046001604060020a031697506002811015610002576040842054640100000000900463ffffffff169650600281101561000257604084206001015495506002811015610002576040842054604060020a900463ffffffff169450600281101561000257505060409091205495999498509296509094509260a060020a90046001604060020a0316919050565b61122d6004808035906020019082018035906020019191908080601f016020809104026020016040519081016040528093929190818152602001838380828437509496505050505050506000600060005082604051808280519060200190808383829060006004602084601f0104600f02600301f15090500191505090815260200160405180910390206000509050348160050160005082600d0160009054906101000a900460ff1660ff16600281101561000257600402830160070180546001608060020a0381169093016001608060020a03199390931692909217909155505b5050565b6112e26004808035906020019082018035906020019191908080601f01602080910003423423094734987103498712093847102938740192387401349857109487501938475"
303				}
304			]
305		}"#;
306
307        let mut buf = BytesMut::with_capacity(65536);
308        buf.put_slice(request.as_bytes());
309
310        let mut codec = StreamCodec::stream_incoming();
311
312        let parsed_request = codec
313            .decode(&mut buf)
314            .expect("There should be no error in huge test")
315            .expect("There should be at least one request huge test");
316        assert_eq!(request, parsed_request);
317    }
318
319    #[test]
320    fn simple_line_codec() {
321        let mut buf = BytesMut::with_capacity(2048);
322        buf.put_slice(b"{ test: 1 }\n{ test: 2 }\n{ test: 3 }");
323
324        let mut codec = StreamCodec::default();
325
326        let request = codec
327            .decode(&mut buf)
328            .expect("There should be no error in simple test")
329            .expect("There should be at least one request in simple test");
330        let request2 = codec
331            .decode(&mut buf)
332            .expect("There should be no error in simple test")
333            .expect("There should be at least one request in simple test");
334
335        assert_eq!(request, "{ test: 1 }");
336        assert_eq!(request2, "{ test: 2 }");
337    }
338
339    #[test]
340    fn serde_json_accepts_whitespace_wrapped_json() {
341        let json = "   { \"key\": \"value\" }   ";
342
343        #[derive(serde::Deserialize, Debug, PartialEq)]
344        struct Obj {
345            key: String,
346        }
347
348        let parsed: Result<Obj, _> = serde_json::from_str(json);
349        assert!(parsed.is_ok(), "serde_json should accept whitespace-wrapped JSON");
350        assert_eq!(parsed.unwrap(), Obj { key: "value".into() });
351    }
352
353    /// Multiple messages fed one byte at a time.
354    #[test]
355    fn pipelined_messages_byte_by_byte() {
356        let payload = br#"{"a":1}{"b":2}{"c":3}"#;
357        let mut buf = BytesMut::with_capacity(256);
358        let mut codec = StreamCodec::stream_incoming();
359        let mut decoded = Vec::new();
360
361        for byte in payload {
362            buf.put_u8(*byte);
363            while let Some(m) = codec.decode(&mut buf).unwrap() {
364                decoded.push(m);
365            }
366        }
367
368        assert_eq!(decoded, vec![r#"{"a":1}"#, r#"{"b":2}"#, r#"{"c":3}"#]);
369        assert!(buf.is_empty());
370    }
371
372    /// Escape sequence split across two `decode` calls, `is_escaped` must carry over.
373    #[test]
374    fn escape_split_across_chunk_boundary() {
375        let mut codec = StreamCodec::stream_incoming();
376        let mut buf = BytesMut::with_capacity(64);
377
378        buf.put_slice(br#"{"a":"\"#);
379        assert!(codec.decode(&mut buf).unwrap().is_none());
380
381        buf.put_slice(br#"\"}"#);
382        let msg = codec.decode(&mut buf).unwrap().unwrap();
383        assert_eq!(msg, r#"{"a":"\\"}"#);
384    }
385
386    /// Opening and closing bracket arriving in separate `decode` calls.
387    #[test]
388    fn depth_split_across_chunk_boundary() {
389        let mut codec = StreamCodec::stream_incoming();
390        let mut buf = BytesMut::with_capacity(64);
391
392        buf.put_u8(b'{');
393        assert!(codec.decode(&mut buf).unwrap().is_none());
394        buf.put_u8(b'}');
395        assert_eq!(codec.decode(&mut buf).unwrap().unwrap(), "{}");
396    }
397
398    /// Input that drives the depth negative must not wedge the codec.
399    #[test]
400    fn leading_close_bracket_does_not_poison() {
401        let mut codec = StreamCodec::stream_incoming();
402        let mut buf = BytesMut::with_capacity(8);
403        buf.put_slice(b"]{");
404        assert_eq!(codec.decode(&mut buf).unwrap(), Some("]{".to_string()));
405    }
406
407    /// `Separator::Byte` decoding fed one byte at a time.
408    #[test]
409    fn byte_separator_split_at_every_byte() {
410        let payload = b"first line\nsecond line\nthird line\n";
411        let mut codec = StreamCodec::default();
412        let mut buf = BytesMut::with_capacity(64);
413        let mut got = Vec::new();
414
415        for byte in payload {
416            buf.put_u8(*byte);
417            while let Some(m) = codec.decode(&mut buf).unwrap() {
418                got.push(m);
419            }
420        }
421
422        assert_eq!(got, vec!["first line", "second line", "third line"]);
423    }
424
425    /// The stateless `Separator::Empty` decode loop that `ScanState` replaced, kept as the
426    /// reference implementation for the differential tests below.
427    struct ReferenceDecoder;
428
429    impl Decoder for ReferenceDecoder {
430        type Item = String;
431        type Error = io::Error;
432
433        fn decode(&mut self, buf: &mut BytesMut) -> io::Result<Option<Self::Item>> {
434            let mut depth = 0i32;
435            let mut in_str = false;
436            let mut is_escaped = false;
437            let mut start_idx = 0usize;
438            let mut whitespaces = 0usize;
439
440            for idx in 0..buf.as_ref().len() {
441                let byte = buf.as_ref()[idx];
442
443                if (byte == b'{' || byte == b'[') && !in_str {
444                    if depth == 0 {
445                        start_idx = idx;
446                    }
447                    depth += 1;
448                } else if (byte == b'}' || byte == b']') && !in_str {
449                    depth -= 1;
450                } else if byte == b'"' && !is_escaped {
451                    in_str = !in_str;
452                } else if is_whitespace(byte) {
453                    whitespaces += 1;
454                }
455                is_escaped = byte == b'\\' && !is_escaped && in_str;
456
457                if depth == 0 && idx != start_idx && idx - start_idx + 1 > whitespaces {
458                    if start_idx > 0 {
459                        buf.advance(start_idx);
460                    }
461                    let bts = buf.split_to(idx + 1 - start_idx);
462                    return Ok(String::from_utf8(bts.into()).ok())
463                }
464            }
465            Ok(None)
466        }
467    }
468
469    /// Feeds the same bytes with the same chunk boundaries to the codec and the reference
470    /// decoder and returns the decode results of both. Asserts that both consume the same bytes.
471    fn drain_both(bytes: &[u8], chunks: &[usize]) -> (Vec<Option<String>>, Vec<Option<String>>) {
472        let mut ours = StreamCodec::stream_incoming();
473        let mut theirs = ReferenceDecoder;
474
475        let mut buf_ours = BytesMut::with_capacity(bytes.len() + 16);
476        let mut buf_theirs = BytesMut::with_capacity(bytes.len() + 16);
477        let mut out_ours = Vec::new();
478        let mut out_theirs = Vec::new();
479
480        let mut cur = 0usize;
481        for &c in chunks {
482            let end = cur.saturating_add(c.max(1)).min(bytes.len());
483            if cur == end {
484                break;
485            }
486            buf_ours.put_slice(&bytes[cur..end]);
487            buf_theirs.put_slice(&bytes[cur..end]);
488            loop {
489                let a = ours.decode(&mut buf_ours).unwrap();
490                let b = theirs.decode(&mut buf_theirs).unwrap();
491                out_ours.push(a.clone());
492                out_theirs.push(b.clone());
493                if a.is_none() && b.is_none() {
494                    break;
495                }
496            }
497            cur = end;
498        }
499        if cur < bytes.len() {
500            buf_ours.put_slice(&bytes[cur..]);
501            buf_theirs.put_slice(&bytes[cur..]);
502            loop {
503                let a = ours.decode(&mut buf_ours).unwrap();
504                let b = theirs.decode(&mut buf_theirs).unwrap();
505                out_ours.push(a.clone());
506                out_theirs.push(b.clone());
507                if a.is_none() && b.is_none() {
508                    break;
509                }
510            }
511        }
512
513        assert_eq!(buf_ours, buf_theirs, "residual buffers diverged");
514
515        (out_ours, out_theirs)
516    }
517
518    /// Hand-picked adversarial inputs under several chunk schedules.
519    #[test]
520    fn differential_curated_corpus() {
521        let inputs: &[&[u8]] = &[
522            b"",
523            b"{}",
524            b"[]",
525            b"{}{}",
526            br#"{"a":1}"#,
527            br#"{"a":1}{"b":2}"#,
528            b"   {}   ",
529            b"   { }   ",
530            b"   {\n}\n   ",
531            br#"{"a":"\""}"#,
532            br#"{"a":"\\"}"#,
533            br#"{"a":"\\\""}"#,
534            b"{\xff}",
535            b"{\xff}{\"a\":1}",
536            b"]{}",
537            b"}{}",
538            b"]{}{}",
539            b"}{}{}",
540            b"][{}",
541            b"abc",
542            b"a",
543            b"\n\n\n",
544            br#"{"\":1}"#,
545            br#"{"escaped":"line\nline"}"#,
546            b"{[]}",
547            b"[{}]",
548            b"[{},{},{}]",
549        ];
550        let chunk_variants: &[&[usize]] =
551            &[&[usize::MAX], &[1], &[2], &[3], &[5], &[7], &[1, 3, 2, 7, 1, 4]];
552
553        for input in inputs {
554            for chunks in chunk_variants {
555                let (ours, theirs) = drain_both(input, chunks);
556                assert_eq!(
557                    ours,
558                    theirs,
559                    "differential mismatch on {:?} with chunks {:?}",
560                    std::str::from_utf8(input).unwrap_or("<non-utf8>"),
561                    chunks
562                );
563            }
564        }
565    }
566
567    /// Random byte sequences over a JSON-heavy alphabet under random chunk schedules.
568    #[test]
569    fn differential_random_bytes() {
570        const SEQUENCES: usize = 200;
571        const SCHEDULES: usize = 5;
572        const MAX_LEN: usize = 128;
573
574        let mut rng = StdRng::seed_from_u64(0x00CC_2259_C0DE_C0DE);
575        let alphabet: &[u8] = b"{}[]\":,\\ \n \t a1{}[]\"";
576
577        for seq in 0..SEQUENCES {
578            let len = 1 + (rng.next_u32() as usize % MAX_LEN);
579            let mut bytes = Vec::with_capacity(len);
580            for _ in 0..len {
581                let b = alphabet[rng.next_u32() as usize % alphabet.len()];
582                bytes.push(b);
583            }
584            for sched in 0..SCHEDULES {
585                let mut chunks = Vec::new();
586                let mut remaining = len;
587                while remaining > 0 {
588                    let c = 1 + (rng.next_u32() as usize % remaining.min(16));
589                    chunks.push(c);
590                    remaining -= c;
591                }
592                let (ours, theirs) = drain_both(&bytes, &chunks);
593                assert_eq!(
594                    ours, theirs,
595                    "differential mismatch on seq={} sched={} bytes={:?}",
596                    seq, sched, bytes
597                );
598            }
599        }
600    }
601
602    /// Randomly chunked well-formed JSON-RPC envelopes.
603    #[test]
604    fn differential_random_wellformed_json_rpc() {
605        let mut rng = StdRng::seed_from_u64(0x00DE_C0DE_C002_2259);
606        for _ in 0..20 {
607            let str_len = 500 + (rng.next_u32() as usize % 4096);
608            let filler: String = (0..str_len)
609                .map(|_| {
610                    let b = rng.next_u32() as u8 & 0x7F;
611                    // printable ASCII without quotes or backslashes
612                    if b < 0x20 || b == b'"' || b == b'\\' {
613                        b'a'
614                    } else {
615                        b
616                    }
617                })
618                .map(|b| b as char)
619                .collect();
620            let msg = format!(
621                r#"{{"jsonrpc":"2.0","method":"engine_test","params":["{}"],"id":1}}"#,
622                filler
623            );
624            let bytes = msg.as_bytes();
625
626            let mut chunks = Vec::new();
627            let mut remaining = bytes.len();
628            while remaining > 0 {
629                let c = 1 + (rng.next_u32() as usize % remaining.min(64));
630                chunks.push(c);
631                remaining -= c;
632            }
633
634            let (ours, theirs) = drain_both(bytes, &chunks);
635            assert_eq!(ours, theirs, "differential mismatch on well-formed JSON-RPC");
636            assert!(
637                ours.iter().any(|o| o.is_some()),
638                "well-formed message should decode to at least one frame"
639            );
640        }
641    }
642}
reth_ipc/stream_codec.rs

reth_ipc/
stream_codec.rs
