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::{MINIMUM_WIRE_VERSION, 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 Some(frame_end) = offset
91                .checked_add(5)
92                .and_then(|v| v.checked_add(frame_length))
93            else {
94                return Err(DecodeError::invalid("frame length overflow"));
95            };
96            if self.buffer.len() < frame_end {
97                self.offset = Some(offset);
98                return Ok(None);
99            }
100            offset = frame_end;
101            if frame_type == frame_type::END {
102                self.complete = true;
103                self.offset = Some(offset);
104                return Ok(Some(ScannedMessage {
105                    message: self.buffer[..frame_end].to_vec(),
106                    excess: self.buffer.len() - frame_end,
107                }));
108            }
109        }
110    }
111
112    fn check_header(&self) -> Result<()> {
113        let b = &self.buffer;
114        if b[0..4] != SSP2_MAGIC[..] {
115            return Err(DecodeError::invalid("bad envelope magic (expected SSP2)"));
116        }
117        let wire_version = u16::from_le_bytes([b[4], b[5]]);
118        if !(MINIMUM_WIRE_VERSION..=WIRE_VERSION).contains(&wire_version) {
119            return Err(DecodeError::invalid(format!(
120                "unsupported wireVersion {wire_version}"
121            )));
122        }
123        if b[6] != 0x01 && b[6] != 0x02 {
124            return Err(DecodeError::invalid(format!(
125                "unknown msgKind byte 0x{:02x}",
126                b[6]
127            )));
128        }
129        if b[7] != 0x00 {
130            return Err(DecodeError::invalid(format!(
131                "envelope flags must be 0x00, got 0x{:02x}",
132                b[7]
133            )));
134        }
135        Ok(())
136    }
137}
138
139#[cfg(test)]
140mod tests {
141    use super::*;
142    use crate::encode::encode_message;
143    use crate::model::{Frame, Message, MsgKind};
144
145    /// A small but real request envelope (REQ_HEADER + PULL_HEADER + END) —
146    /// the same shape the TS scanner test uses.
147    fn request_bytes() -> Vec<u8> {
148        let message = Message {
149            wire_version: WIRE_VERSION,
150            msg_kind: MsgKind::Request,
151            frames: vec![
152                Frame::ReqHeader {
153                    client_id: "c1".to_owned(),
154                    schema_version: 1,
155                    log_epoch: None,
156                },
157                Frame::PullHeader {
158                    limit_commits: 0,
159                    limit_snapshot_rows: 0,
160                    max_snapshot_pages: 0,
161                    accept: 0b0011,
162                },
163            ],
164        };
165        encode_message(&message)
166    }
167
168    #[test]
169    fn whole_message_in_one_chunk_completes_with_zero_excess() {
170        let request = request_bytes();
171        let mut scanner = MessageStreamScanner::new();
172        let result = scanner.push(&request).unwrap().expect("complete");
173        assert_eq!(result.excess, 0);
174        assert_eq!(result.message, request);
175    }
176
177    #[test]
178    fn every_split_point_reassembles_byte_exactly() {
179        let request = request_bytes();
180        for split in 1..request.len() {
181            let mut scanner = MessageStreamScanner::new();
182            assert!(
183                scanner.push(&request[..split]).unwrap().is_none(),
184                "split {split}: first half must be incomplete"
185            );
186            let result = scanner
187                .push(&request[split..])
188                .unwrap()
189                .unwrap_or_else(|| panic!("split {split}: second half must complete"));
190            assert_eq!(result.excess, 0, "split {split}");
191            assert_eq!(result.message, request, "split {split}");
192        }
193    }
194
195    #[test]
196    fn one_byte_at_a_time_trickle_completes() {
197        let request = request_bytes();
198        let mut scanner = MessageStreamScanner::new();
199        let mut result = None;
200        for byte in &request {
201            result = scanner.push(&[*byte]).unwrap();
202        }
203        let result = result.expect("complete after last byte");
204        assert_eq!(result.excess, 0);
205        assert_eq!(result.message, request);
206    }
207
208    #[test]
209    fn bytes_past_end_are_reported_as_excess() {
210        let request = request_bytes();
211        let mut with_excess = request.clone();
212        with_excess.extend_from_slice(&[0xaa, 0xbb, 0xcc]);
213        let mut scanner = MessageStreamScanner::new();
214        let result = scanner.push(&with_excess).unwrap().expect("complete");
215        assert_eq!(result.excess, 3);
216        assert_eq!(result.message, request);
217    }
218
219    #[test]
220    fn two_message_stream_every_split_reassembles_first_exactly() {
221        // The exhaustive split-point pattern over a *two*-message stream:
222        // the scanner must find the first message's END regardless of where
223        // the chunk boundary falls, reassembling it byte-exactly. Excess is
224        // whatever second-message bytes were already buffered at the moment
225        // the first message's END was consumed — the §8.7 pipelining signal.
226        let first = request_bytes();
227        let second = request_bytes();
228        let mut stream = first.clone();
229        stream.extend_from_slice(&second);
230        for split in 1..stream.len() {
231            let mut scanner = MessageStreamScanner::new();
232            // Feed the first chunk; if it already completes (split at or past
233            // the first message's END), the excess is what that chunk carried
234            // past END. Otherwise feed the rest and complete then.
235            let (result, buffered_at_completion) = match scanner.push(&stream[..split]).unwrap() {
236                Some(done) => (done, split),
237                None => {
238                    let done = scanner
239                        .push(&stream[split..])
240                        .unwrap()
241                        .unwrap_or_else(|| panic!("split {split}: first message must complete"));
242                    (done, stream.len())
243                }
244            };
245            assert_eq!(result.message, first, "split {split}: first message bytes");
246            assert_eq!(
247                result.excess,
248                buffered_at_completion - first.len(),
249                "split {split}: excess equals second-message bytes buffered at END"
250            );
251        }
252    }
253
254    #[test]
255    fn bad_magic_is_a_decode_error_once_header_arrived() {
256        let mut request = request_bytes();
257        request[0] = 0x58;
258        let mut scanner = MessageStreamScanner::new();
259        assert!(scanner.push(&request[..4]).unwrap().is_none());
260        assert!(scanner.push(&request[4..]).is_err());
261    }
262
263    #[test]
264    fn non_zero_flags_and_unknown_msg_kind_are_decode_errors() {
265        for (index, value) in [(6usize, 0x03u8), (7usize, 0x01u8)] {
266            let mut request = request_bytes();
267            request[index] = value;
268            let mut scanner = MessageStreamScanner::new();
269            assert!(scanner.push(&request).is_err(), "index {index}");
270        }
271    }
272
273    #[test]
274    #[should_panic(expected = "already complete")]
275    fn push_after_completion_panics() {
276        let request = request_bytes();
277        let mut scanner = MessageStreamScanner::new();
278        scanner.push(&request).unwrap();
279        let _ = scanner.push(&[0]);
280    }
281}
ssp2/stream.rs

ssp2/
stream.rs
