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 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 wire_version != 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                    outbox_id: None,
157                },
158                Frame::PullHeader {
159                    limit_commits: 0,
160                    limit_snapshot_rows: 0,
161                    max_snapshot_pages: 0,
162                    accept: 0b0011,
163                    cursor: -1,
164                    registration_digest: [0; 32],
165                    echo_digest: None,
166                },
167            ],
168        };
169        encode_message(&message)
170    }
171
172    #[test]
173    fn whole_message_in_one_chunk_completes_with_zero_excess() {
174        let request = request_bytes();
175        let mut scanner = MessageStreamScanner::new();
176        let result = scanner.push(&request).unwrap().expect("complete");
177        assert_eq!(result.excess, 0);
178        assert_eq!(result.message, request);
179    }
180
181    #[test]
182    fn every_split_point_reassembles_byte_exactly() {
183        let request = request_bytes();
184        for split in 1..request.len() {
185            let mut scanner = MessageStreamScanner::new();
186            assert!(
187                scanner.push(&request[..split]).unwrap().is_none(),
188                "split {split}: first half must be incomplete"
189            );
190            let result = scanner
191                .push(&request[split..])
192                .unwrap()
193                .unwrap_or_else(|| panic!("split {split}: second half must complete"));
194            assert_eq!(result.excess, 0, "split {split}");
195            assert_eq!(result.message, request, "split {split}");
196        }
197    }
198
199    #[test]
200    fn one_byte_at_a_time_trickle_completes() {
201        let request = request_bytes();
202        let mut scanner = MessageStreamScanner::new();
203        let mut result = None;
204        for byte in &request {
205            result = scanner.push(&[*byte]).unwrap();
206        }
207        let result = result.expect("complete after last byte");
208        assert_eq!(result.excess, 0);
209        assert_eq!(result.message, request);
210    }
211
212    #[test]
213    fn bytes_past_end_are_reported_as_excess() {
214        let request = request_bytes();
215        let mut with_excess = request.clone();
216        with_excess.extend_from_slice(&[0xaa, 0xbb, 0xcc]);
217        let mut scanner = MessageStreamScanner::new();
218        let result = scanner.push(&with_excess).unwrap().expect("complete");
219        assert_eq!(result.excess, 3);
220        assert_eq!(result.message, request);
221    }
222
223    #[test]
224    fn two_message_stream_every_split_reassembles_first_exactly() {
225        // The exhaustive split-point pattern over a *two*-message stream:
226        // the scanner must find the first message's END regardless of where
227        // the chunk boundary falls, reassembling it byte-exactly. Excess is
228        // whatever second-message bytes were already buffered at the moment
229        // the first message's END was consumed — the §8.7 pipelining signal.
230        let first = request_bytes();
231        let second = request_bytes();
232        let mut stream = first.clone();
233        stream.extend_from_slice(&second);
234        for split in 1..stream.len() {
235            let mut scanner = MessageStreamScanner::new();
236            // Feed the first chunk; if it already completes (split at or past
237            // the first message's END), the excess is what that chunk carried
238            // past END. Otherwise feed the rest and complete then.
239            let (result, buffered_at_completion) = match scanner.push(&stream[..split]).unwrap() {
240                Some(done) => (done, split),
241                None => {
242                    let done = scanner
243                        .push(&stream[split..])
244                        .unwrap()
245                        .unwrap_or_else(|| panic!("split {split}: first message must complete"));
246                    (done, stream.len())
247                }
248            };
249            assert_eq!(result.message, first, "split {split}: first message bytes");
250            assert_eq!(
251                result.excess,
252                buffered_at_completion - first.len(),
253                "split {split}: excess equals second-message bytes buffered at END"
254            );
255        }
256    }
257
258    #[test]
259    fn bad_magic_is_a_decode_error_once_header_arrived() {
260        let mut request = request_bytes();
261        request[0] = 0x58;
262        let mut scanner = MessageStreamScanner::new();
263        assert!(scanner.push(&request[..4]).unwrap().is_none());
264        assert!(scanner.push(&request[4..]).is_err());
265    }
266
267    #[test]
268    fn non_zero_flags_and_unknown_msg_kind_are_decode_errors() {
269        for (index, value) in [(6usize, 0x03u8), (7usize, 0x01u8)] {
270            let mut request = request_bytes();
271            request[index] = value;
272            let mut scanner = MessageStreamScanner::new();
273            assert!(scanner.push(&request).is_err(), "index {index}");
274        }
275    }
276
277    #[test]
278    #[should_panic(expected = "already complete")]
279    fn push_after_completion_panics() {
280        let request = request_bytes();
281        let mut scanner = MessageStreamScanner::new();
282        scanner.push(&request).unwrap();
283        let _ = scanner.push(&[0]);
284    }
285}
ssp2/stream.rs

ssp2/
stream.rs
