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}