ssp2/
stream.rs

1//! Incremental SSP2 message-stream scanner (SPEC.md §1.4, §8.7).
2//!
3//! The envelope grammar is self-delimiting — an 8-byte header, then
4//! length-prefixed frames until `END` — so a byte stream split across
5//! arbitrary chunk boundaries (WebSocket messages, §8.7) needs no
6//! reassembly protocol beyond concatenation plus this scanner: feed
7//! chunks, learn exactly where one complete envelope ends. Used by the
8//! native transport for round-response reassembly (the Rust mirror of
9//! `packages/core/src/stream.ts`'s `MessageStreamScanner`, the TS
10//! reference the conformance host uses).
11//!
12//! The scanner validates only the 8-byte header (a stream whose header is
13//! not a valid SSP2 envelope has no findable end, §8.7 connection-fatal
14//! rule) and walks frame length prefixes; full decoding stays with
15//! [`crate::decode_message`].
16
17use crate::decode::{SSP2_MAGIC, WIRE_VERSION};
18use crate::error::{DecodeError, Result};
19use crate::model::frame_type;
20
21/// A completed scan: the exact envelope bytes plus any surplus buffered
22/// past the `END` frame.
23#[derive(Debug, Clone, PartialEq, Eq)]
24pub struct ScannedMessage {
25    /// The complete envelope bytes (header through the `END` frame).
26    pub message: Vec<u8>,
27    /// Byte count buffered PAST the `END` frame — a §8.7 stream MUST end
28    /// exactly at the `END` frame's last byte, so any excess is a protocol
29    /// violation (pipelining) for the caller to act on.
30    pub excess: usize,
31}
32
33/// Feed bytes, learn exactly where one complete SSP2 envelope ends.
34///
35/// Mirrors `MessageStreamScanner` in `packages/core/src/stream.ts` frame
36/// for frame: header validation, frame-length walking, exact-`END`
37/// detection, and excess reporting. Once complete, [`push`](Self::push)
38/// must not be called again.
39#[derive(Default)]
40pub struct MessageStreamScanner {
41    buffer: Vec<u8>,
42    /// Parse cursor: start of the next frame header; `None` while the
43    /// 8-byte envelope header is still incomplete.
44    offset: Option<usize>,
45    complete: bool,
46}
47
48impl MessageStreamScanner {
49    pub fn new() -> Self {
50        Self::default()
51    }
52
53    /// True once any bytes were fed.
54    pub fn started(&self) -> bool {
55        !self.buffer.is_empty()
56    }
57
58    /// Feed one chunk. Returns `Ok(Some(..))` once the `END` frame is fully
59    /// buffered, `Ok(None)` while more bytes are needed. Returns a
60    /// `DecodeError` on an invalid envelope header (connection-fatal per
61    /// §8.7). Panics on use after completion (a caller bug, mirroring the
62    /// TS scanner's thrown error).
63    pub fn push(&mut self, chunk: &[u8]) -> Result<Option<ScannedMessage>> {
64        assert!(
65            !self.complete,
66            "MessageStreamScanner: message already complete"
67        );
68        self.buffer.extend_from_slice(chunk);
69        if self.offset.is_none() {
70            if self.buffer.len() < 8 {
71                return Ok(None);
72            }
73            self.check_header()?;
74            self.offset = Some(8);
75        }
76        let mut offset = self.offset.expect("offset set once header parsed");
77        loop {
78            // Every frame is a 1-byte type + 4-byte little-endian length.
79            if self.buffer.len() - offset < 5 {
80                self.offset = Some(offset);
81                return Ok(None);
82            }
83            let frame_type = self.buffer[offset];
84            let frame_length = u32::from_le_bytes([
85                self.buffer[offset + 1],
86                self.buffer[offset + 2],
87                self.buffer[offset + 3],
88                self.buffer[offset + 4],
89            ]) as usize;
90            let frame_end = offset + 5 + frame_length;
91            if self.buffer.len() < frame_end {
92                self.offset = Some(offset);
93                return Ok(None);
94            }
95            offset = frame_end;
96            if frame_type == frame_type::END {
97                self.complete = true;
98                self.offset = Some(offset);
99                return Ok(Some(ScannedMessage {
100                    message: self.buffer[..frame_end].to_vec(),
101                    excess: self.buffer.len() - frame_end,
102                }));
103            }
104        }
105    }
106
107    fn check_header(&self) -> Result<()> {
108        let b = &self.buffer;
109        if b[0..4] != SSP2_MAGIC[..] {
110            return Err(DecodeError::invalid("bad envelope magic (expected SSP2)"));
111        }
112        let wire_version = u16::from_le_bytes([b[4], b[5]]);
113        if wire_version != WIRE_VERSION {
114            return Err(DecodeError::invalid(format!(
115                "unsupported wireVersion {wire_version}"
116            )));
117        }
118        if b[6] != 0x01 && b[6] != 0x02 {
119            return Err(DecodeError::invalid(format!(
120                "unknown msgKind byte 0x{:02x}",
121                b[6]
122            )));
123        }
124        if b[7] != 0x00 {
125            return Err(DecodeError::invalid(format!(
126                "envelope flags must be 0x00, got 0x{:02x}",
127                b[7]
128            )));
129        }
130        Ok(())
131    }
132}
133
134#[cfg(test)]
135mod tests {
136    use super::*;
137    use crate::encode::encode_message;
138    use crate::model::{Frame, Message, MsgKind};
139
140    /// A small but real request envelope (REQ_HEADER + PULL_HEADER + END) —
141    /// the same shape the TS scanner test uses.
142    fn request_bytes() -> Vec<u8> {
143        let message = Message {
144            msg_kind: MsgKind::Request,
145            frames: vec![
146                Frame::ReqHeader {
147                    client_id: "c1".to_owned(),
148                    schema_version: 1,
149                },
150                Frame::PullHeader {
151                    limit_commits: 0,
152                    limit_snapshot_rows: 0,
153                    max_snapshot_pages: 0,
154                    accept: 0b0011,
155                },
156            ],
157        };
158        encode_message(&message)
159    }
160
161    #[test]
162    fn whole_message_in_one_chunk_completes_with_zero_excess() {
163        let request = request_bytes();
164        let mut scanner = MessageStreamScanner::new();
165        let result = scanner.push(&request).unwrap().expect("complete");
166        assert_eq!(result.excess, 0);
167        assert_eq!(result.message, request);
168    }
169
170    #[test]
171    fn every_split_point_reassembles_byte_exactly() {
172        let request = request_bytes();
173        for split in 1..request.len() {
174            let mut scanner = MessageStreamScanner::new();
175            assert!(
176                scanner.push(&request[..split]).unwrap().is_none(),
177                "split {split}: first half must be incomplete"
178            );
179            let result = scanner
180                .push(&request[split..])
181                .unwrap()
182                .unwrap_or_else(|| panic!("split {split}: second half must complete"));
183            assert_eq!(result.excess, 0, "split {split}");
184            assert_eq!(result.message, request, "split {split}");
185        }
186    }
187
188    #[test]
189    fn one_byte_at_a_time_trickle_completes() {
190        let request = request_bytes();
191        let mut scanner = MessageStreamScanner::new();
192        let mut result = None;
193        for byte in &request {
194            result = scanner.push(&[*byte]).unwrap();
195        }
196        let result = result.expect("complete after last byte");
197        assert_eq!(result.excess, 0);
198        assert_eq!(result.message, request);
199    }
200
201    #[test]
202    fn bytes_past_end_are_reported_as_excess() {
203        let request = request_bytes();
204        let mut with_excess = request.clone();
205        with_excess.extend_from_slice(&[0xaa, 0xbb, 0xcc]);
206        let mut scanner = MessageStreamScanner::new();
207        let result = scanner.push(&with_excess).unwrap().expect("complete");
208        assert_eq!(result.excess, 3);
209        assert_eq!(result.message, request);
210    }
211
212    #[test]
213    fn two_message_stream_every_split_reassembles_first_exactly() {
214        // The exhaustive split-point pattern over a *two*-message stream:
215        // the scanner must find the first message's END regardless of where
216        // the chunk boundary falls, reassembling it byte-exactly. Excess is
217        // whatever second-message bytes were already buffered at the moment
218        // the first message's END was consumed — the §8.7 pipelining signal.
219        let first = request_bytes();
220        let second = request_bytes();
221        let mut stream = first.clone();
222        stream.extend_from_slice(&second);
223        for split in 1..stream.len() {
224            let mut scanner = MessageStreamScanner::new();
225            // Feed the first chunk; if it already completes (split at or past
226            // the first message's END), the excess is what that chunk carried
227            // past END. Otherwise feed the rest and complete then.
228            let (result, buffered_at_completion) = match scanner.push(&stream[..split]).unwrap() {
229                Some(done) => (done, split),
230                None => {
231                    let done = scanner
232                        .push(&stream[split..])
233                        .unwrap()
234                        .unwrap_or_else(|| panic!("split {split}: first message must complete"));
235                    (done, stream.len())
236                }
237            };
238            assert_eq!(result.message, first, "split {split}: first message bytes");
239            assert_eq!(
240                result.excess,
241                buffered_at_completion - first.len(),
242                "split {split}: excess equals second-message bytes buffered at END"
243            );
244        }
245    }
246
247    #[test]
248    fn bad_magic_is_a_decode_error_once_header_arrived() {
249        let mut request = request_bytes();
250        request[0] = 0x58;
251        let mut scanner = MessageStreamScanner::new();
252        assert!(scanner.push(&request[..4]).unwrap().is_none());
253        assert!(scanner.push(&request[4..]).is_err());
254    }
255
256    #[test]
257    fn non_zero_flags_and_unknown_msg_kind_are_decode_errors() {
258        for (index, value) in [(6usize, 0x03u8), (7usize, 0x01u8)] {
259            let mut request = request_bytes();
260            request[index] = value;
261            let mut scanner = MessageStreamScanner::new();
262            assert!(scanner.push(&request).is_err(), "index {index}");
263        }
264    }
265
266    #[test]
267    #[should_panic(expected = "already complete")]
268    fn push_after_completion_panics() {
269        let request = request_bytes();
270        let mut scanner = MessageStreamScanner::new();
271        scanner.push(&request).unwrap();
272        let _ = scanner.push(&[0]);
273    }
274}
ssp2/stream.rs

ssp2/
stream.rs
