Skip to main content

microvms_core/session/
sse.rs

1// SPDX-License-Identifier: Apache-2.0
2//! Incremental Server-Sent Events parsing, and the typed events the daemon emits.
3//!
4//! # Incremental because the transport picks the read boundaries
5//!
6//! A single `data:` line routinely arrives split across two chunks, and a parser that
7//! assumes a read boundary is a frame boundary loses output at exactly the moment the
8//! stream gets busy. So bytes are buffered until a blank line proves a frame is
9//! complete, and a trailing partial frame stays in the buffer.
10//!
11//! Only the subset the daemon emits is implemented: named events with one `data:` line
12//! of JSON, plus the `:` keepalive comments axum sends every fifteen seconds. `id:` and
13//! `retry:` are ignored rather than half-supported — the daemon does not send them, and
14//! the byte offset is the resume cursor rather than `Last-Event-ID`.
15//!
16//! # Hand-rolled rather than a dependency
17//!
18//! The daemon's SSE is two field names and three event names, and the interesting part
19//! of a client's stream handling is the cursor rather than the framing. An eventsource
20//! crate would bring a reconnect policy keyed on `Last-Event-ID`, which is the wrong
21//! cursor: it resumes at an event, and this protocol resumes at a byte.
22
23use base64::Engine as _;
24
25use crate::error::{Error, ErrorKind, WireKind};
26
27/// One complete SSE frame: its event name and its raw `data:` text.
28#[derive(Clone, Debug, Eq, PartialEq)]
29pub struct Frame {
30    pub event: String,
31    pub data: String,
32}
33
34/// The longest frame terminator, in bytes. See [`find_frame_end`].
35const MAX_TERMINATOR_LEN: usize = 4;
36
37/// The ceiling on bytes held for a frame that has not terminated.
38///
39/// # Why a ceiling at all
40///
41/// Without one, a peer that sends bytes and never a blank line makes this buffer grow
42/// until the process dies. That is not hypothetical for this client: the thing on the
43/// other end of the stream is a *proxy* in front of the daemon, and a proxy that answers
44/// a stream request with a non-SSE body — an HTML error page, a redirect chain, a chunked
45/// response it keeps writing to — produces exactly that shape. There is no frame boundary
46/// coming, so no other guard in the parser ever fires.
47///
48/// # Why 4 MiB
49///
50/// Sized from the daemon's actual frame sizes, with a large multiple of headroom.
51///
52/// One `output` event carries one read from a child's pipe, and the daemon reads into a
53/// 16 KiB scratch buffer (`Capped::new` in `agentd/src/exec.rs`). Base64 inflates that by
54/// 4/3, so the biggest `data:` line the daemon emits is about 22 KiB plus a short JSON
55/// envelope — call it 24 KiB. A reattach that replays the 1 MiB stream ring does not
56/// change the number: the ring is replayed as its individual chunks, each its own frame,
57/// each still one 16 KiB read. `gap` and `exit` frames are tens of bytes.
58///
59/// 4 MiB is roughly 170 frames' worth of that, which leaves room for a proxy that
60/// coalesces reads, for a future daemon with a larger scratch buffer, and for the
61/// `stream_buffer_bytes` ring being raised to a few MiB — while staying small enough to be
62/// irrelevant against the 512 MiB baseline VM this client talks to. A caller that hits it
63/// is not reading a big frame; it is reading something that is not SSE.
64///
65/// The bound is on the *residue*: [`SseParser::feed`] drains every complete frame first, so
66/// a single read that legitimately carries many megabytes of finished frames is not
67/// refused. What is refused is a residue that keeps growing across feeds.
68pub const MAX_PENDING_BYTES: usize = 4 * 1024 * 1024;
69
70/// Feed bytes, get whole frames out. Holds a partial frame across feeds.
71#[derive(Debug, Default)]
72pub struct SseParser {
73    buffer: Vec<u8>,
74    /// How much of `buffer` the terminator scan has already ruled out.
75    ///
76    /// The scan resumes here rather than at byte zero. Rescanning from zero on every feed
77    /// is quadratic in the size of a partial frame, and the buffer is largest exactly when
78    /// the stream is busiest: 800 unterminated 8 KiB reads took 31 seconds to scan and now
79    /// take milliseconds.
80    scanned: usize,
81}
82
83impl SseParser {
84    pub fn new() -> Self {
85        Self::default()
86    }
87
88    /// Every frame `chunk` completed. Never returns a partial one.
89    ///
90    /// Fails with [`ErrorKind::Protocol`] once the undelimited residue passes
91    /// [`MAX_PENDING_BYTES`] — see that constant for why, and for why the check is on what
92    /// is left after draining rather than on what arrived.
93    pub fn feed(&mut self, chunk: &[u8]) -> Result<Vec<Frame>, Error> {
94        self.buffer.extend_from_slice(chunk);
95        let mut frames = Vec::new();
96        loop {
97            // Back up by one byte less than the longest terminator, because a terminator
98            // straddling a chunk boundary has its first bytes inside the already-scanned
99            // region. Backing up to zero instead is what made this quadratic.
100            let from = self.scanned.saturating_sub(MAX_TERMINATOR_LEN - 1);
101            let Some((body, consumed)) = find_frame_end(&self.buffer, from) else {
102                self.scanned = self.buffer.len();
103                break;
104            };
105            let raw = self.buffer[..body].to_vec();
106            self.buffer.drain(..consumed);
107            // Nothing past the terminator has been looked at yet — the search stopped at
108            // the first match — so the next pass starts over on what is left.
109            self.scanned = 0;
110            if let Some(frame) = parse_frame(&raw) {
111                frames.push(frame);
112            }
113        }
114
115        if self.buffer.len() > MAX_PENDING_BYTES {
116            let held = self.buffer.len();
117            // Dropped rather than kept: the stream is over, and holding megabytes that no
118            // longer have a reader is the second half of the same bug.
119            self.buffer = Vec::new();
120            self.scanned = 0;
121            return Err(Error::new(
122                ErrorKind::Protocol,
123                format!(
124                    "the event stream sent {held} bytes with no frame terminator, over the \
125                     {MAX_PENDING_BYTES}-byte ceiling on undelimited data. The daemon's largest \
126                     frame is one 16 KiB pipe read base64-encoded, about 24 KiB, so a residue \
127                     this size means the response body is not server-sent events — most often \
128                     a proxy answering with an error page instead of the stream."
129                ),
130            ));
131        }
132        Ok(frames)
133    }
134
135    /// Bytes held for an incomplete frame. Diagnostic; a healthy stream sits at 0.
136    pub fn pending(&self) -> usize {
137        self.buffer.len()
138    }
139}
140
141/// Locates the first frame terminator at or after `from`, returning the body length and
142/// bytes consumed.
143///
144/// A frame ends at a blank line, and the wire spelling of that varies: `\n\n`,
145/// `\r\n\r\n`, `\r\r`. All three are accepted because a proxy in the path may rewrite
146/// line endings, and picking one spelling would work right up until it did not.
147fn find_frame_end(buffer: &[u8], from: usize) -> Option<(usize, usize)> {
148    let tail = buffer.get(from..)?;
149    let mut earliest: Option<(usize, usize)> = None;
150    for terminator in [
151        b"\r\n\r\n".as_slice(),
152        b"\n\n".as_slice(),
153        b"\r\r".as_slice(),
154    ] {
155        let Some(idx) = find(tail, terminator) else {
156            continue;
157        };
158        let idx = from + idx;
159        // A shorter terminator can be found at a lower index inside a longer one
160        // (`\n\n` sits at index+1 of `\r\n\r\n`), so the earliest start wins and ties
161        // break toward the longer match.
162        if earliest.is_none_or(|(start, _)| idx < start) {
163            earliest = Some((idx, terminator.len()));
164        }
165    }
166    let (start, width) = earliest?;
167    Some((start, start + width))
168}
169
170fn find(haystack: &[u8], needle: &[u8]) -> Option<usize> {
171    haystack
172        .windows(needle.len())
173        .position(|window| window == needle)
174}
175
176/// Turns one frame's lines into a [`Frame`], or `None` if it carries no data.
177///
178/// `None` for a keepalive comment and for a frame with no `data:` line. A frame whose
179/// data is not JSON is *not* dropped here — that judgement belongs to [`decode`], which
180/// knows which event names carry JSON.
181fn parse_frame(raw: &[u8]) -> Option<Frame> {
182    let mut event = String::new();
183    let mut data_lines: Vec<String> = Vec::new();
184    let normalized: Vec<u8> = {
185        // `\r\n` and a bare `\r` both become `\n`, so one split handles every spelling.
186        let mut out = Vec::with_capacity(raw.len());
187        let mut iter = raw.iter().copied().peekable();
188        while let Some(byte) = iter.next() {
189            if byte == b'\r' {
190                if iter.peek() == Some(&b'\n') {
191                    iter.next();
192                }
193                out.push(b'\n');
194            } else {
195                out.push(byte);
196            }
197        }
198        out
199    };
200
201    for line in normalized.split(|byte| *byte == b'\n') {
202        // A leading colon is a comment. axum's keepalive is exactly this.
203        if line.is_empty() || line.starts_with(b":") {
204            continue;
205        }
206        let (name, value) = match find(line, b":") {
207            Some(idx) => (&line[..idx], &line[idx + 1..]),
208            None => (line, b"".as_slice()),
209        };
210        // "If value starts with a single space, remove it" — one space only, since a
211        // payload may legitimately begin with whitespace.
212        let value = value.strip_prefix(b" ").unwrap_or(value);
213        match String::from_utf8_lossy(name).as_ref() {
214            "event" => event = String::from_utf8_lossy(value).into_owned(),
215            "data" => data_lines.push(String::from_utf8_lossy(value).into_owned()),
216            _ => {}
217        }
218    }
219
220    if data_lines.is_empty() {
221        return None;
222    }
223    Some(Frame {
224        event,
225        data: data_lines.join("\n"),
226    })
227}
228
229/// One decoded stream event.
230///
231/// The payloads are the `protocol` crate's, so a field renamed on the daemon side
232/// breaks this build. The one thing added on top is the decoded bytes of an output
233/// event: base64 is how arbitrary bytes cross a JSON string, and a caller wants the
234/// bytes.
235///
236/// Not `Clone`, because [`protocol::exec::ExitEvent`] is not — and adding the derive
237/// there would be an edit to a crate this lane does not own. No consumer needs it: a
238/// stream item is moved to its handler once.
239#[derive(Debug)]
240pub enum ExecEvent {
241    /// Output bytes, with the offset they start at.
242    Output {
243        stream: protocol::exec::StreamKind,
244        offset: u64,
245        data: Vec<u8>,
246    },
247    /// A byte range that is gone for good.
248    ///
249    /// Surfaced as a typed event rather than logged, because the alternative is reading
250    /// a truncated log as a complete one. `from` is inclusive and `to` exclusive, so
251    /// `to` is where a cursor resumes.
252    Gap { from: u64, to: u64 },
253    /// The terminal event. Its absence is what distinguishes a cut connection from a
254    /// finished command — the byte sequences are otherwise identical.
255    Exit(protocol::exec::ExitEvent),
256}
257
258/// Maps one frame onto a typed event.
259///
260/// Three outcomes, and the middle one is the reason this returns a nested `Result`:
261///
262/// * `Ok(Some(event))` — a recognized event.
263/// * `Ok(None)` — a frame this client does not dispatch. An unknown event name, or a
264///   payload that will not parse. Dropped rather than raised, matching the daemon,
265///   which degrades a serialization failure to `{}` on purpose instead of taking the
266///   connection down: one bad frame must not end an otherwise live stream.
267/// * `Err(_)` — an `output` event whose base64 will not decode. Distinct from the
268///   above because it is unambiguously corruption of bytes a caller asked for, and
269///   silently dropping output is the failure this whole protocol is shaped to prevent.
270pub fn decode(frame: &Frame) -> Result<Option<ExecEvent>, Error> {
271    match frame.event.as_str() {
272        protocol::exec::EVENT_OUTPUT => {
273            let Ok(payload) = serde_json::from_str::<protocol::exec::OutputEvent>(&frame.data)
274            else {
275                return Ok(None);
276            };
277            let data = base64::engine::general_purpose::STANDARD
278                .decode(&payload.output)
279                .map_err(|err| {
280                    Error::wire(
281                        WireKind::ProtocolError,
282                        format!(
283                            "an output event at offset {} carried undecodable base64, so \
284                             those bytes cannot be delivered: {err}",
285                            payload.offset
286                        ),
287                    )
288                })?;
289            Ok(Some(ExecEvent::Output {
290                stream: payload.stream,
291                offset: payload.offset,
292                data,
293            }))
294        }
295        protocol::exec::EVENT_GAP => {
296            let Ok(payload) = serde_json::from_str::<protocol::exec::GapEvent>(&frame.data) else {
297                return Ok(None);
298            };
299            Ok(Some(ExecEvent::Gap {
300                from: payload.from,
301                to: payload.to,
302            }))
303        }
304        protocol::exec::EVENT_EXIT => {
305            let Ok(payload) = serde_json::from_str::<protocol::exec::ExitEvent>(&frame.data) else {
306                return Ok(None);
307            };
308            Ok(Some(ExecEvent::Exit(payload)))
309        }
310        _ => Ok(None),
311    }
312}
313
314#[cfg(test)]
315mod tests {
316    use super::*;
317
318    fn output_frame(offset: u64, bytes: &[u8]) -> String {
319        let encoded = base64::engine::general_purpose::STANDARD.encode(bytes);
320        format!(
321            "event: output\ndata: {{\"offset\":{offset},\"stream\":\"stdout\",\
322             \"output\":\"{encoded}\"}}\n\n"
323        )
324    }
325
326    /// The property the whole parser exists for: a frame split across reads survives.
327    ///
328    /// Byte-by-byte is the strongest version of the split — every possible boundary is
329    /// exercised at once, including inside the `event:` name and inside the base64.
330    #[test]
331    fn a_frame_split_across_every_possible_read_boundary_still_parses() {
332        let wire = output_frame(0, b"hello");
333        let mut parser = SseParser::new();
334        let mut frames = Vec::new();
335        for byte in wire.as_bytes() {
336            frames.extend(parser.feed(&[*byte]).expect("under the pending ceiling"));
337        }
338        assert_eq!(frames.len(), 1, "the split frame did not reassemble");
339        assert_eq!(parser.pending(), 0, "a complete stream left bytes behind");
340
341        let event = decode(&frames[0]).expect("decodes").expect("is an event");
342        match event {
343            ExecEvent::Output { offset, data, .. } => {
344                assert_eq!(offset, 0);
345                assert_eq!(data, b"hello");
346            }
347            other => panic!("expected output, got {other:?}"),
348        }
349    }
350
351    /// Several frames in one read all come out, in order.
352    #[test]
353    fn one_read_carrying_three_frames_yields_three() {
354        let wire = format!(
355            "{}{}{}",
356            output_frame(0, b"a"),
357            output_frame(1, b"b"),
358            "event: exit\ndata: {\"exit_code\":0,\"signal\":null,\"truncated\":false,\
359             \"writers_may_be_alive\":false,\"offset\":2}\n\n"
360        );
361        let mut parser = SseParser::new();
362        let frames = parser
363            .feed(wire.as_bytes())
364            .expect("under the pending ceiling");
365        assert_eq!(frames.len(), 3);
366        assert!(matches!(
367            decode(&frames[2]).expect("decodes"),
368            Some(ExecEvent::Exit(_))
369        ));
370    }
371
372    /// A trailing partial frame is held, not emitted and not lost.
373    #[test]
374    fn a_trailing_partial_frame_is_held_until_it_completes() {
375        let wire = output_frame(0, b"xyz");
376        let (head, tail) = wire.split_at(wire.len() - 4);
377        let mut parser = SseParser::new();
378        assert!(
379            parser
380                .feed(head.as_bytes())
381                .expect("under the pending ceiling")
382                .is_empty(),
383            "an incomplete frame was emitted"
384        );
385        assert!(parser.pending() > 0);
386        let frames = parser
387            .feed(tail.as_bytes())
388            .expect("under the pending ceiling");
389        assert_eq!(frames.len(), 1);
390    }
391
392    /// The scan cursor does not skip a terminator that straddles a chunk boundary.
393    ///
394    /// This is the way a resumed scan gets it wrong: the previous feed ended part-way
395    /// through the blank line, so the terminator's first bytes are inside the region
396    /// already ruled out. Every split of every spelling is exercised, which is what makes
397    /// the back-up-by-terminator-length-minus-one arithmetic pinned rather than sampled.
398    ///
399    /// **Falsification** — resume the scan at `self.scanned` with no back-up and the
400    /// `\r\n\r\n` and `\r\r` splits stop producing a frame at all.
401    #[test]
402    fn a_terminator_straddling_a_chunk_boundary_is_still_found() {
403        for terminator in ["\n\n", "\r\n\r\n", "\r\r"] {
404            let wire = format!("event: gap\ndata: {{\"from\":0,\"to\":4}}{terminator}");
405            for split in 1..wire.len() {
406                let mut parser = SseParser::new();
407                let mut frames = Vec::new();
408                frames.extend(
409                    parser
410                        .feed(&wire.as_bytes()[..split])
411                        .expect("under the pending ceiling"),
412                );
413                frames.extend(
414                    parser
415                        .feed(&wire.as_bytes()[split..])
416                        .expect("under the pending ceiling"),
417                );
418                assert_eq!(
419                    frames.len(),
420                    1,
421                    "{terminator:?} split at {split} did not reassemble"
422                );
423                assert_eq!(
424                    parser.pending(),
425                    0,
426                    "{terminator:?} split at {split} left bytes behind"
427                );
428                assert!(matches!(
429                    decode(&frames[0]).expect("decodes"),
430                    Some(ExecEvent::Gap { from: 0, to: 4 })
431                ));
432            }
433        }
434    }
435
436    /// An undelimited stream is refused at the ceiling rather than buffered forever.
437    ///
438    /// [`ErrorKind::Protocol`] and therefore **not retryable**, which is the half that
439    /// matters to `exec::Attach`: a reconnect into the same non-SSE body would refill the
440    /// buffer `max_reconnects` times over.
441    ///
442    /// **Falsification** — drop the ceiling check and the parser accepts every chunk, so
443    /// `expect_err` panics.
444    #[test]
445    fn an_unterminated_stream_is_refused_at_the_pending_ceiling() {
446        let mut parser = SseParser::new();
447        // 8 KiB with no blank line anywhere, which is what a proxy's HTML error page or a
448        // redirect chain looks like to this parser.
449        let chunk = vec![b'x'; 8 * 1024];
450        let mut fed = 0usize;
451        loop {
452            match parser.feed(&chunk) {
453                Ok(frames) => {
454                    assert!(frames.is_empty(), "undelimited data produced a frame");
455                    fed += chunk.len();
456                    assert!(
457                        parser.pending() <= MAX_PENDING_BYTES,
458                        "the buffer passed the ceiling without erroring: {}",
459                        parser.pending()
460                    );
461                    assert!(
462                        fed <= MAX_PENDING_BYTES + chunk.len(),
463                        "the ceiling never fired"
464                    );
465                }
466                Err(err) => {
467                    assert_eq!(err.kind(), ErrorKind::Protocol);
468                    assert!(
469                        !err.retryable(),
470                        "a reconnect would refill the buffer from the same body"
471                    );
472                    assert_eq!(
473                        err.wire_kind(),
474                        None,
475                        "no status produced this; it is a client-side ceiling"
476                    );
477                    let message = err.to_string();
478                    assert!(
479                        message.contains(&MAX_PENDING_BYTES.to_string()),
480                        "the message must name the ceiling: {message}"
481                    );
482                    assert!(
483                        message.contains("no frame terminator"),
484                        "the message must say what was wrong: {message}"
485                    );
486                    assert_eq!(
487                        parser.pending(),
488                        0,
489                        "the refused bytes are dropped, not held for a reader that is gone"
490                    );
491                    return;
492                }
493            }
494        }
495    }
496
497    /// Scanning is linear in the buffer, not quadratic.
498    ///
499    /// The measured shape of the defect: 800 unterminated 8 KiB reads — 6.4 MiB, one read
500    /// under the ceiling per feed — rescanned from byte zero every time.
501    ///
502    /// **The budget is 10 s for a workload that takes ~0.75 s fixed and ~160 s broken.**
503    /// Deliberately nowhere near the fixed figure, because the two populations are 200x
504    /// apart and the only way to lose that discrimination is to make the threshold tight
505    /// enough for load to cross it. It was 500 ms and did exactly that: measured
506    /// 2026-08-09 on a 16-core box at load average 26, the fixed path took 706-847 ms
507    /// across nine runs and this assertion failed every time — a red gate over code with
508    /// no defect in it, which is the failure mode that teaches people to skip a gate.
509    /// A 13x margin over the fixed path still leaves a 16x margin under the broken one.
510    ///
511    /// **Falsification** — replace `from` with `0` in `feed` and this takes 160 s
512    /// (verified 2026-08-09), so the guard fails by two orders of magnitude rather than
513    /// by a hair.
514    #[test]
515    fn scanning_an_unterminated_buffer_stays_linear() {
516        let chunk = vec![b'x'; 8 * 1024];
517        let started = std::time::Instant::now();
518        let mut parser = SseParser::new();
519        for _ in 0..800 {
520            // A feed that trips the ceiling resets the buffer, which would hide the
521            // quadratic scan — so the parser is rebuilt instead of being allowed to grow
522            // past it, and each iteration still scans a buffer the old code rescanned
523            // whole.
524            if parser.pending() + chunk.len() > MAX_PENDING_BYTES {
525                parser = SseParser::new();
526            }
527            parser.feed(&chunk).expect("under the pending ceiling");
528        }
529        let elapsed = started.elapsed();
530        assert!(
531            elapsed < std::time::Duration::from_secs(10),
532            "800 x 8 KiB unterminated feeds took {elapsed:?}, over the 10s budget; the \
533             quadratic rescan is back (the linear path is ~0.75s even under heavy load, \
534             and the quadratic one is ~160s)"
535        );
536    }
537
538    /// All three blank-line spellings terminate a frame, because a proxy may rewrite
539    /// line endings and picking one spelling works right up until it does not.
540    #[test]
541    fn every_blank_line_spelling_terminates_a_frame() {
542        for terminator in ["\n\n", "\r\n\r\n", "\r\r"] {
543            let wire = format!("event: gap\r\ndata: {{\"from\":0,\"to\":4}}{terminator}");
544            let mut parser = SseParser::new();
545            let frames = parser
546                .feed(wire.as_bytes())
547                .expect("under the pending ceiling");
548            assert_eq!(frames.len(), 1, "{terminator:?} did not terminate a frame");
549            assert!(matches!(
550                decode(&frames[0]).expect("decodes"),
551                Some(ExecEvent::Gap { from: 0, to: 4 })
552            ));
553        }
554    }
555
556    /// A keepalive is a comment with no data, and it must not surface as an event —
557    /// otherwise fifteen seconds of silence looks like output.
558    #[test]
559    fn a_keepalive_comment_is_not_a_frame() {
560        let mut parser = SseParser::new();
561        assert!(
562            parser
563                .feed(b":\n\n")
564                .expect("under the pending ceiling")
565                .is_empty()
566        );
567        assert!(
568            parser
569                .feed(b": keep-alive-text\n\n")
570                .expect("under the pending ceiling")
571                .is_empty()
572        );
573        assert_eq!(parser.pending(), 0);
574
575        // And it does not disturb a real frame that follows it on the same read.
576        let wire = format!(":\n\n{}", output_frame(0, b"real"));
577        let mut parser = SseParser::new();
578        assert_eq!(
579            parser
580                .feed(wire.as_bytes())
581                .expect("under the pending ceiling")
582                .len(),
583            1
584        );
585    }
586
587    /// One space after the colon is stripped, a second is payload.
588    ///
589    /// The daemon writes `data: {...}`, and a parser that stripped all leading
590    /// whitespace would corrupt any payload that meant to start with a space.
591    #[test]
592    fn exactly_one_leading_space_is_stripped_from_a_field_value() {
593        let mut parser = SseParser::new();
594        let frames = parser
595            .feed(b"event:  gap\ndata:  {\"from\":1,\"to\":2}\n\n")
596            .expect("under the pending ceiling");
597        assert_eq!(frames.len(), 1);
598        assert_eq!(frames[0].event, " gap", "more than one space was stripped");
599        assert_eq!(frames[0].data, " {\"from\":1,\"to\":2}");
600        // serde_json tolerates the leading space, so the event still decodes — but the
601        // event *name* no longer matches, which is the failure this would cause.
602        assert!(
603            decode(&frames[0]).expect("decodes").is_none(),
604            "a mis-stripped name must not resolve to a known event"
605        );
606    }
607
608    /// An unknown event name and unparseable data are both dropped, matching the
609    /// daemon's own degradation. One bad frame must not end a live stream.
610    #[test]
611    fn an_unrecognized_or_unparseable_frame_is_dropped_rather_than_raised() {
612        let mut parser = SseParser::new();
613        let frames = parser
614            .feed(
615                b"event: something-new\ndata: {}\n\n\
616              event: gap\ndata: not json at all\n\n",
617            )
618            .expect("under the pending ceiling");
619        assert_eq!(frames.len(), 2);
620        for frame in &frames {
621            assert!(
622                decode(frame).expect("no error").is_none(),
623                "{frame:?} should be dropped"
624            );
625        }
626    }
627
628    /// Undecodable base64 in an *output* event is an error, not a drop.
629    ///
630    /// The one place a drop would be wrong: those are bytes the caller asked for, and
631    /// silently losing output is the failure the gap event exists to make impossible.
632    #[test]
633    fn undecodable_output_base64_is_an_error_rather_than_a_silent_drop() {
634        let mut parser = SseParser::new();
635        let frames = parser.feed(
636            b"event: output\ndata: {\"offset\":7,\"stream\":\"stdout\",\"output\":\"!!!!\"}\n\n",
637        ).expect("under the pending ceiling");
638        let err = decode(&frames[0]).expect_err("bad base64 must be reported");
639        assert!(err.to_string().contains('7'), "{err}");
640    }
641
642    /// A multi-line `data:` payload joins with newlines, as the spec says.
643    #[test]
644    fn a_multi_line_data_payload_joins_with_newlines() {
645        let mut parser = SseParser::new();
646        let frames = parser
647            .feed(b"event: gap\ndata: {\"from\":0,\ndata: \"to\":5}\n\n")
648            .expect("under the pending ceiling");
649        assert_eq!(frames.len(), 1);
650        assert_eq!(frames[0].data, "{\"from\":0,\n\"to\":5}");
651        assert!(matches!(
652            decode(&frames[0]).expect("decodes"),
653            Some(ExecEvent::Gap { from: 0, to: 5 })
654        ));
655    }
656}