From 8824952c4b70a9525edeb77697ec7bfab8d9d719 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?S=C3=A9bastien=20Han?= Date: Tue, 1 Sep 2026 09:03:41 +0200 Subject: [PATCH 1/3] feat(filter): add a bounded, incremental SSE codec MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add a provider-neutral, bounded, incremental Server-Sent Events (SSE) codec to praxis-filter, exposed as `praxis_filter::sse`. - `SseDecoder` frames records from streamed body chunks under configurable byte/field limits (`SseLimits`), tolerating cross-chunk splits, CR/CRLF/LF terminators, and a leading UTF-8 BOM, and poisons itself on a limit violation. - `SseRecord`/`SseField` model fields in wire order with typed accessors (`data`/`event`/`id`/`retry`/`is_event`); a validated `SseRecordBuilder` rejects framing-breaking values. - `encode`/`encode_into` serialize records to canonical wire bytes. Pingora keeps ownership of transport streaming, backpressure, and H1/H2 framing; provider JSON, `[DONE]`, and lifecycle concerns stay in the consumer. Leaf library with no consumers yet. Signed-off-by: Sébastien Han --- filter/src/lib.rs | 1 + filter/src/sse/decoder.rs | 964 ++++++++++++++++++++++++++++++++++++++ filter/src/sse/encoder.rs | 157 +++++++ filter/src/sse/mod.rs | 195 ++++++++ filter/src/sse/record.rs | 471 +++++++++++++++++++ 5 files changed, 1788 insertions(+) create mode 100644 filter/src/sse/decoder.rs create mode 100644 filter/src/sse/encoder.rs create mode 100644 filter/src/sse/mod.rs create mode 100644 filter/src/sse/record.rs diff --git a/filter/src/lib.rs b/filter/src/lib.rs index 38652acf5..e03b90f3b 100644 --- a/filter/src/lib.rs +++ b/filter/src/lib.rs @@ -57,6 +57,7 @@ pub(crate) mod path_match; mod pipeline; mod registry; mod results; +pub mod sse; mod tcp_filter; pub use actions::{FilterAction, Rejection, StreamingResponseBody, StreamingTerminalResponse, TerminalResponse}; diff --git a/filter/src/sse/decoder.rs b/filter/src/sse/decoder.rs new file mode 100644 index 000000000..ca1a19536 --- /dev/null +++ b/filter/src/sse/decoder.rs @@ -0,0 +1,964 @@ +// SPDX-License-Identifier: MIT +// Copyright (c) 2026 Praxis Contributors + +//! Bounded, incremental SSE decoder and its limits / batch / error types. + +use bytes::Bytes; + +use super::record::{SseField, SseRecord}; + +// ----------------------------------------------------------------------------- +// Constants +// ----------------------------------------------------------------------------- + +/// UTF-8 byte order mark, stripped once at stream start. +const BOM: [u8; 3] = [0xEF, 0xBB, 0xBF]; + +/// Lifecycle state of an `SseDecoder`. +#[derive(Clone, Copy, Debug)] +enum DecoderState { + /// Normal operation. + Active, + /// `finish` has been called; `push` reports `Finished`. + Finished, + /// A limit violation occurred; the stored error is re-reported until reset. + Poisoned(SseDecodeError), +} + +/// Bounds on retained memory. Enforced continuously as bytes accumulate. +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub struct SseLimits { + /// Max bytes for a single field line held across chunks. + pub max_line_bytes: usize, + /// Max total retained bytes for one in-progress record: the sum of every + /// field's value plus every `Unknown` field name, so large unknown names + /// cannot bypass the limit. + pub max_record_bytes: usize, + /// Max number of fields in one record (bounds per-field allocations so many + /// tiny fields cannot evade `max_record_bytes`). + pub max_fields_per_record: usize, +} + +impl Default for SseLimits { + fn default() -> Self { + Self { + max_line_bytes: 10_485_760, // 10 MiB + max_record_bytes: 10_485_760, // 10 MiB + max_fields_per_record: 4096, // far above any real record + } + } +} + +/// Result of one `SseDecoder::push` or `SseDecoder::finish`. +/// +/// Carries records completed *before* any error, so a size violation never +/// discards earlier records from the same chunk. A limit violation poisons the +/// decoder; an `error` of `Finished` instead signals a `push` after `finish` and +/// does not poison. +#[derive(Debug, Default, PartialEq, Eq)] +#[must_use] +pub struct SseBatch { + /// Records completed by this call, in order. + pub records: Vec, + /// The error that stopped decoding this call, if any. + pub error: Option, +} + +/// Decode-time errors. The three limit violations poison the decoder (the error +/// is re-reported until `reset`); `Finished` is not a poison. +#[derive(Clone, Copy, Debug, Eq, PartialEq, thiserror::Error)] +pub enum SseDecodeError { + /// A single field line exceeded `max_line_bytes`. Poisons. + #[error("SSE line exceeded limit: {size} bytes > {limit} byte limit")] + LineTooLong { + /// Observed line size in bytes. + size: usize, + /// Configured `max_line_bytes`. + limit: usize, + }, + /// The in-progress record's retained bytes (values + `Unknown` names) + /// exceeded `max_record_bytes`. Poisons. + #[error("SSE record exceeded limit: {size} bytes > {limit} byte limit")] + RecordTooLarge { + /// Observed retained record size in bytes. + size: usize, + /// Configured `max_record_bytes`. + limit: usize, + }, + /// The in-progress record exceeded `max_fields_per_record`. Poisons. + #[error("SSE record exceeded field limit: {count} fields > {limit} field limit")] + TooManyFields { + /// Observed field count. + count: usize, + /// Configured `max_fields_per_record`. + limit: usize, + }, + /// `push` was called after `finish`. Does not poison. + #[error("SSE decoder push called after finish")] + Finished, +} + +/// Bounded, incremental SSE record decoder. +/// +/// Feed body chunks with `push`; each call returns the records that chunk +/// completed. Only the in-progress line and record are retained between calls. +/// Call `finish` at end of stream. +#[derive(Debug)] +pub struct SseDecoder { + /// Bytes of the current, not-yet-terminated field line. + line_buf: Vec, + /// Fields accumulated for the in-progress record, in wire order. + fields: Vec, + /// Set when the previous chunk ended on a bare `\r`, so a leading `\n` in + /// the next chunk completes the `\r\n` pair. + prev_cr: bool, + /// Retained-memory bounds applied while decoding. + limits: SseLimits, + /// Retained bytes of the in-progress record: field values plus `Unknown` + /// names. Reset to 0 when a record is dispatched. + record_bytes: usize, + /// Current lifecycle state. + state: DecoderState, + /// Leading BOM bytes matched so far (0..=3), before resolution. + bom_len: usize, + /// Whether the leading-BOM check has completed (stripped or ruled out). + bom_resolved: bool, +} + +impl Default for SseDecoder { + fn default() -> Self { + Self::new() + } +} + +impl SseDecoder { + /// Create a decoder with default limits. + #[must_use] + pub fn new() -> Self { + Self::with_limits(SseLimits::default()) + } + + /// Create a decoder with explicit limits. + #[must_use] + pub fn with_limits(limits: SseLimits) -> Self { + Self { + line_buf: Vec::new(), + fields: Vec::new(), + prev_cr: false, + limits, + record_bytes: 0, + state: DecoderState::Active, + bom_len: 0, + bom_resolved: false, + } + } + + /// Feed one body chunk; returns the records it completed and an optional + /// error. + pub fn push(&mut self, chunk: &[u8]) -> SseBatch { + if let DecoderState::Finished = self.state { + return SseBatch { + records: Vec::new(), + error: Some(SseDecodeError::Finished), + }; + } + if let DecoderState::Poisoned(err) = self.state { + return SseBatch { + records: Vec::new(), + error: Some(err), + }; + } + let mut records = Vec::new(); + match self.parse(chunk, &mut records) { + Ok(()) => SseBatch { records, error: None }, + Err(err) => { + self.state = DecoderState::Poisoned(err); + SseBatch { + records, + error: Some(err), + } + }, + } + } + + /// `true` only in the poisoned state (after a limit violation). + #[must_use] + pub fn is_poisoned(&self) -> bool { + matches!(self.state, DecoderState::Poisoned(_)) + } + + /// Signal end of stream, transitioning `Active` to `Finished`. + /// + /// Processes any buffered partial line as if terminated and, if the + /// in-progress block accumulated any field, returns it as a single trailing + /// record. Idempotent: a second call returns an empty batch. While poisoned, + /// re-reports the limit error. + pub fn finish(&mut self) -> SseBatch { + if let DecoderState::Poisoned(err) = self.state { + return SseBatch { + records: Vec::new(), + error: Some(err), + }; + } + if let DecoderState::Finished = self.state { + return SseBatch::default(); + } + let mut records = Vec::new(); + match self.flush(&mut records) { + Ok(()) => { + self.state = DecoderState::Finished; + SseBatch { records, error: None } + }, + Err(err) => { + self.state = DecoderState::Poisoned(err); + SseBatch { + records, + error: Some(err), + } + }, + } + } + + /// Flush a trailing partial line and dispatch any accumulated fields. + fn flush(&mut self, records: &mut Vec) -> Result<(), SseDecodeError> { + if !self.bom_resolved && self.bom_len > 0 { + let prior = self.bom_len; + self.bom_resolved = true; + self.bom_len = 0; + self.feed(BOM.get(..prior).unwrap_or_default(), records)?; + } + if !self.line_buf.is_empty() { + // `process_line` returns `Some` only for an empty `line_buf` (a blank + // line ends a record); here the buffer is non-empty (BOM replay above + // only pushes bytes, never a terminator), so it commits the trailing + // line as a field and returns `None` — the record is emitted by the + // `!self.fields.is_empty()` block below. The `Some` arm is defensive. + if let Some(record) = self.process_line()? { + records.push(record); + } + self.line_buf.clear(); + } + if !self.fields.is_empty() { + records.push(SseRecord::from_fields(std::mem::take(&mut self.fields))); + self.record_bytes = 0; + } + Ok(()) + } + + /// Reset to `Active`, clearing buffers and any `Finished`/`Poisoned` state. + pub fn reset(&mut self) { + self.line_buf.clear(); + self.fields.clear(); + self.record_bytes = 0; + self.prev_cr = false; + self.state = DecoderState::Active; + self.bom_len = 0; + self.bom_resolved = false; + } + + /// `true` only in the `Finished` state (after `finish`). + #[must_use] + pub fn is_finished(&self) -> bool { + matches!(self.state, DecoderState::Finished) + } + + /// Strip a single leading UTF-8 BOM (possibly split across chunks) before + /// delegating to `feed`. + fn parse(&mut self, chunk: &[u8], records: &mut Vec) -> Result<(), SseDecodeError> { + if self.bom_resolved { + return self.feed(chunk, records); + } + let prior = self.bom_len; + let mut matched = prior; + let mut i = 0; + while matched < BOM.len() { + let Some(&b) = chunk.get(i) else { + self.bom_len = matched; // wait for more bytes + return Ok(()); + }; + if BOM.get(matched) == Some(&b) { + matched += 1; + i += 1; + } else { + self.bom_resolved = true; + self.bom_len = 0; + self.feed(BOM.get(..prior).unwrap_or_default(), records)?; + return self.feed(chunk, records); + } + } + self.bom_resolved = true; + self.bom_len = 0; + self.feed(chunk.get(i..).unwrap_or_default(), records) + } + + /// Consume a byte slice, appending completed records to `records`. + fn feed(&mut self, bytes: &[u8], records: &mut Vec) -> Result<(), SseDecodeError> { + if bytes.is_empty() { + return Ok(()); // empty chunk must preserve prev_cr + } + let mut i = 0; + if self.prev_cr && bytes.first() == Some(&b'\n') { + i = 1; + } + self.prev_cr = false; + while let Some(&b) = bytes.get(i) { + match b { + b'\n' => self.end_line(records)?, + b'\r' => self.handle_cr(bytes, &mut i, records)?, + _ => self.push_byte(b)?, + } + i += 1; + } + Ok(()) + } + + /// Handle a `\r`: end the line and pair a following `\n` (this chunk or the + /// next) so `\r\n` counts as one line ending. + fn handle_cr(&mut self, bytes: &[u8], i: &mut usize, records: &mut Vec) -> Result<(), SseDecodeError> { + self.end_line(records)?; + match bytes.get(*i + 1) { + Some(&b'\n') => *i += 1, + Some(_) => {}, + None => self.prev_cr = true, + } + Ok(()) + } + + /// Append a data byte to the current line, enforcing `max_line_bytes`. + fn push_byte(&mut self, b: u8) -> Result<(), SseDecodeError> { + self.line_buf.push(b); + if self.line_buf.len() > self.limits.max_line_bytes { + return Err(SseDecodeError::LineTooLong { + size: self.line_buf.len(), + limit: self.limits.max_line_bytes, + }); + } + Ok(()) + } + + /// Process the current line as terminated, then clear the line buffer. + fn end_line(&mut self, records: &mut Vec) -> Result<(), SseDecodeError> { + if let Some(record) = self.process_line()? { + records.push(record); + } + self.line_buf.clear(); + Ok(()) + } + + /// Interpret the current line: blank dispatches a record; otherwise commit a + /// field. + fn process_line(&mut self) -> Result, SseDecodeError> { + if self.line_buf.is_empty() { + if self.fields.is_empty() { + return Ok(None); + } + let record = SseRecord::from_fields(std::mem::take(&mut self.fields)); + self.record_bytes = 0; + return Ok(Some(record)); + } + let field = self.classify_line(); + self.push_field(field)?; + Ok(None) + } + + /// Derive the field for the current (non-empty) line per the event-stream + /// interpretation rules. + fn classify_line(&self) -> SseField { + let line = self.line_buf.as_slice(); + match line.iter().position(|&b| b == b':') { + Some(colon) => { + let name = line.get(..colon).unwrap_or_default(); + let rest = line.get(colon + 1..).unwrap_or_default(); + let value = rest.strip_prefix(b" ").unwrap_or(rest); + classify(name, value) + }, + None => classify(line, b""), + } + } + + /// Append a field, enforcing `max_fields_per_record` and `max_record_bytes`. + fn push_field(&mut self, field: SseField) -> Result<(), SseDecodeError> { + if self.fields.len() >= self.limits.max_fields_per_record { + return Err(SseDecodeError::TooManyFields { + count: self.fields.len() + 1, + limit: self.limits.max_fields_per_record, + }); + } + let new_bytes = self.record_bytes + field_cost(&field); + if new_bytes > self.limits.max_record_bytes { + return Err(SseDecodeError::RecordTooLarge { + size: new_bytes, + limit: self.limits.max_record_bytes, + }); + } + self.record_bytes = new_bytes; + self.fields.push(field); + Ok(()) + } +} + +/// Classify a name/value pair into a typed field, copying bytes out of the line +/// buffer. An empty name means a `:`-prefixed comment line. +fn classify(name: &[u8], value: &[u8]) -> SseField { + match name { + b"" => SseField::Comment(Bytes::copy_from_slice(value)), + b"event" => SseField::Event(Bytes::copy_from_slice(value)), + b"data" => SseField::Data(Bytes::copy_from_slice(value)), + b"id" => SseField::Id(Bytes::copy_from_slice(value)), + b"retry" => SseField::Retry(Bytes::copy_from_slice(value)), + other => SseField::Unknown { + name: Bytes::copy_from_slice(other), + value: Bytes::copy_from_slice(value), + }, + } +} + +/// Retained-byte cost of a field: value length, plus name length for `Unknown`. +fn field_cost(field: &SseField) -> usize { + match field { + SseField::Unknown { name, value } => name.len() + value.len(), + SseField::Event(value) + | SseField::Data(value) + | SseField::Id(value) + | SseField::Retry(value) + | SseField::Comment(value) => value.len(), + } +} + +#[cfg(test)] +#[expect(clippy::allow_attributes, reason = "blanket test suppressions")] +#[allow( + clippy::unwrap_used, + clippy::expect_used, + clippy::indexing_slicing, + clippy::panic, + reason = "tests" +)] +mod tests { + use super::*; + + #[test] + fn default_limits_match_spec() { + let limits = SseLimits::default(); + assert_eq!(limits.max_line_bytes, 10_485_760, "default line limit is 10 MiB"); + assert_eq!(limits.max_record_bytes, 10_485_760, "default record limit is 10 MiB"); + assert_eq!(limits.max_fields_per_record, 4096, "default field cap is 4096"); + } + + #[test] + fn batch_default_is_empty() { + let batch = SseBatch::default(); + assert!(batch.records.is_empty(), "default batch has no records"); + assert_eq!(batch.error, None, "default batch has no error"); + } + + #[test] + fn decode_errors_are_copy_and_boxable() { + let err = SseDecodeError::LineTooLong { size: 20, limit: 10 }; + let copy = err; + assert_eq!(err, copy, "SseDecodeError is Copy"); + assert!(err.to_string().contains("20"), "Display includes the size"); + assert!(err.to_string().contains("10"), "Display includes the limit"); + let _boxed: Box = err.into(); + } + + // ----- Test Utilities ----- + + // Module-scoped BOM constant for the tests below. + const BOM_BYTES: &[u8] = &[0xEF, 0xBB, 0xBF]; + + // Push one chunk, assert no error, return the completed records. + fn push_ok(decoder: &mut SseDecoder, chunk: &[u8]) -> Vec { + let batch = decoder.push(chunk); + assert_eq!(batch.error, None, "unexpected decode error"); + batch.records + } + + // Decode the whole input in one push then finish; return all records. + fn decode_whole(input: &[u8]) -> Vec { + let mut decoder = SseDecoder::new(); + let mut records = decoder.push(input).records; + records.extend(decoder.finish().records); + records + } + + // Decode the input split at byte offsets a and b; return all records. + fn decode_split(input: &[u8], a: usize, b: usize) -> Vec { + let mut decoder = SseDecoder::new(); + let mut records = Vec::new(); + for part in [&input[..a], &input[a..b], &input[b..]] { + records.extend(decoder.push(part).records); + } + records.extend(decoder.finish().records); + records + } + + // Assert every 3-way split of the input decodes identically to the whole. + fn assert_all_splits_match(input: &[u8]) { + let expected = decode_whole(input); + let len = input.len(); + for a in 0..=len { + for b in a..=len { + assert_eq!( + decode_split(input, a, b), + expected, + "split ({a},{b}) diverged for {input:?}" + ); + } + } + } + + #[test] + fn single_record() { + let mut decoder = SseDecoder::new(); + let records = push_ok(&mut decoder, b"data: hello\n\n"); + assert_eq!(records.len(), 1, "one complete record"); + assert_eq!(records[0].data(), Bytes::from_static(b"hello"), "data decoded"); + } + + #[test] + fn event_type_and_multiple_records_in_one_chunk() { + let mut decoder = SseDecoder::new(); + let records = push_ok(&mut decoder, b"event: a\ndata: 1\n\ndata: 2\n\n"); + assert_eq!(records.len(), 2, "two records in one chunk"); + assert_eq!(records[0].event(), Some(b"a".as_slice()), "first event type"); + assert_eq!(records[0].data(), Bytes::from_static(b"1"), "first data"); + assert_eq!(records[1].event(), None, "second has no event"); + assert_eq!(records[1].data(), Bytes::from_static(b"2"), "second data"); + } + + #[test] + fn multiline_data_joined() { + let mut decoder = SseDecoder::new(); + let records = push_ok(&mut decoder, b"data: a\ndata: b\n\n"); + assert_eq!(records[0].data(), Bytes::from_static(b"a\nb"), "multi-line data joined"); + } + + #[test] + fn record_split_across_chunks() { + let mut decoder = SseDecoder::new(); + assert!( + push_ok(&mut decoder, b"data: hel").is_empty(), + "partial line yields nothing" + ); + assert!( + push_ok(&mut decoder, b"lo\n").is_empty(), + "line without blank yields nothing" + ); + let records = push_ok(&mut decoder, b"\n"); + assert_eq!( + records[0].data(), + Bytes::from_static(b"hello"), + "record completes on blank line" + ); + } + + #[test] + fn empty_blocks_ignored() { + let mut decoder = SseDecoder::new(); + assert!( + push_ok(&mut decoder, b"\n\n\n").is_empty(), + "blank lines dispatch nothing" + ); + } + + #[test] + fn field_kinds_and_optional_space() { + let mut decoder = SseDecoder::new(); + let records = push_ok( + &mut decoder, + b": ping\nevent: e\ndata:nospace\nid: 7\nretry: 500\nx-foo: bar\n\n", + ); + let fields = records[0].fields(); + assert_eq!( + fields[0], + SseField::Comment(Bytes::from_static(b"ping")), + "comment field" + ); + assert_eq!(fields[1], SseField::Event(Bytes::from_static(b"e")), "event field"); + assert_eq!( + fields[2], + SseField::Data(Bytes::from_static(b"nospace")), + "no leading space to strip" + ); + assert_eq!(fields[3], SseField::Id(Bytes::from_static(b"7")), "id field"); + assert_eq!(fields[4], SseField::Retry(Bytes::from_static(b"500")), "retry field"); + assert_eq!( + fields[5], + SseField::Unknown { + name: Bytes::from_static(b"x-foo"), + value: Bytes::from_static(b"bar") + }, + "unknown field preserved" + ); + } + + #[test] + fn colon_less_lines_classified_by_name() { + let mut decoder = SseDecoder::new(); + let records = push_ok(&mut decoder, b"data\nevent\nid\nretry\nnope\n\n"); + let fields = records[0].fields(); + assert_eq!(fields[0], SseField::Data(Bytes::new()), "bare data name"); + assert_eq!(fields[1], SseField::Event(Bytes::new()), "bare event name"); + assert_eq!(fields[2], SseField::Id(Bytes::new()), "bare id name"); + assert_eq!(fields[3], SseField::Retry(Bytes::new()), "bare retry name"); + assert_eq!( + fields[4], + SseField::Unknown { + name: Bytes::from_static(b"nope"), + value: Bytes::new() + }, + "bare unknown name" + ); + } + + #[test] + fn data_with_empty_value_is_event() { + let mut decoder = SseDecoder::new(); + let records = push_ok(&mut decoder, b"data: \n\n"); + assert_eq!(records[0].data(), Bytes::new(), "empty data value"); + assert!(records[0].is_event(), "empty-value data still dispatches an event"); + } + + #[test] + fn crlf_line_endings() { + let mut decoder = SseDecoder::new(); + let records = push_ok(&mut decoder, b"data: hello\r\n\r\n"); + assert_eq!(records[0].data(), Bytes::from_static(b"hello"), "CRLF endings handled"); + } + + #[test] + fn bare_cr_line_ending() { + let mut decoder = SseDecoder::new(); + let records = push_ok(&mut decoder, b"data: hello\r\r"); + assert_eq!( + records[0].data(), + Bytes::from_static(b"hello"), + "bare CR endings handled" + ); + } + + #[test] + fn line_too_long_returns_error() { + let limits = SseLimits { + max_line_bytes: 4, + ..SseLimits::default() + }; + let mut decoder = SseDecoder::with_limits(limits); + let batch = decoder.push(b"data: toolong\n\n"); + assert!( + matches!(batch.error, Some(SseDecodeError::LineTooLong { size, limit }) if size > limit), + "over-long line reports LineTooLong" + ); + } + + #[test] + fn too_many_fields_returns_error() { + let limits = SseLimits { + max_fields_per_record: 2, + ..SseLimits::default() + }; + let mut decoder = SseDecoder::with_limits(limits); + let batch = decoder.push(b"data: 1\ndata: 2\ndata: 3\n\n"); + assert!( + matches!(batch.error, Some(SseDecodeError::TooManyFields { count, limit }) if count > limit), + "exceeding field cap reports TooManyFields" + ); + } + + #[test] + fn crlf_split_across_chunks() { + let mut decoder = SseDecoder::new(); + assert!(push_ok(&mut decoder, b"data: hello\r").is_empty(), "trailing CR waits"); + let records = push_ok(&mut decoder, b"\n\r\n"); + assert_eq!( + records[0].data(), + Bytes::from_static(b"hello"), + "leading LF completes CRLF" + ); + } + + #[test] + fn empty_chunk_preserves_pending_cr() { + let mut decoder = SseDecoder::new(); + assert!(push_ok(&mut decoder, b"event: keep\r").is_empty(), "trailing CR waits"); + assert!( + push_ok(&mut decoder, b"").is_empty(), + "empty chunk must preserve prev_cr" + ); + let records = push_ok(&mut decoder, b"\ndata: x\r\n\r\n"); + assert_eq!( + records[0].event(), + Some(b"keep".as_slice()), + "event survived the boundary" + ); + assert_eq!( + records[0].data(), + Bytes::from_static(b"x"), + "data decoded after boundary" + ); + } + + #[test] + fn mixed_endings_in_one_record() { + let mut decoder = SseDecoder::new(); + let records = push_ok(&mut decoder, b"event: a\rdata: b\ndata: c\r\n\n"); + assert_eq!(records[0].event(), Some(b"a".as_slice()), "CR-terminated event line"); + assert_eq!( + records[0].data(), + Bytes::from_static(b"b\nc"), + "mixed LF/CRLF data joined" + ); + } + + #[test] + fn record_too_large_counts_values() { + let limits = SseLimits { + max_record_bytes: 6, + ..SseLimits::default() + }; + let mut decoder = SseDecoder::with_limits(limits); + let batch = decoder.push(b"data: aaaa\ndata: bbbb\n\n"); + assert!( + matches!(batch.error, Some(SseDecodeError::RecordTooLarge { size, limit }) if size > limit), + "summed data values over the cap report RecordTooLarge" + ); + } + + #[test] + fn record_too_large_counts_unknown_names() { + let limits = SseLimits { + max_record_bytes: 8, + ..SseLimits::default() + }; + let mut decoder = SseDecoder::with_limits(limits); + let batch = decoder.push(b"aaaaa\nbbbbb\n\n"); + assert!( + matches!(batch.error, Some(SseDecodeError::RecordTooLarge { .. })), + "unknown-field names must count toward max_record_bytes" + ); + } + + #[test] + fn record_bytes_reset_between_records() { + let limits = SseLimits { + max_record_bytes: 3, + ..SseLimits::default() + }; + let mut decoder = SseDecoder::with_limits(limits); + let records = push_ok(&mut decoder, b"data: ab\n\ndata: cd\n\n"); + assert_eq!(records.len(), 2, "byte budget must reset after each dispatched record"); + } + + #[test] + fn records_before_overflow_returned_then_poisoned() { + let limits = SseLimits { + max_line_bytes: 8, + ..SseLimits::default() + }; + let mut decoder = SseDecoder::with_limits(limits); + let batch = decoder.push(b"data: ok\n\ndata: waytoolong\n\n"); + assert_eq!(batch.records.len(), 1, "record completed before the overflow is kept"); + assert_eq!( + batch.records[0].data(), + Bytes::from_static(b"ok"), + "kept record is correct" + ); + assert!( + matches!(batch.error, Some(SseDecodeError::LineTooLong { .. })), + "overflow reported after the kept record" + ); + assert!(decoder.is_poisoned(), "limit violation poisons the decoder"); + } + + #[test] + fn poisoned_decoder_re_reports_and_yields_no_records() { + let limits = SseLimits { + max_line_bytes: 4, + ..SseLimits::default() + }; + let mut decoder = SseDecoder::with_limits(limits); + let first_err = decoder.push(b"data: toolong\n\n").error.unwrap(); + let second = decoder.push(b"data: x\n\n"); + assert!(second.records.is_empty(), "poisoned decoder yields no records"); + assert_eq!( + second.error, + Some(first_err), + "poisoned decoder re-reports the same error" + ); + assert!(decoder.is_poisoned(), "stays poisoned"); + } + + #[test] + fn healthy_decoder_is_not_poisoned() { + let mut decoder = SseDecoder::new(); + let records = push_ok(&mut decoder, b"data: hi\n\n"); + assert_eq!(records.len(), 1, "healthy decode"); + assert!(!decoder.is_poisoned(), "no poison on clean input"); + } + + #[test] + fn finish_flushes_trailing_record_without_blank_line() { + let mut decoder = SseDecoder::new(); + assert!(push_ok(&mut decoder, b"data: tail\n").is_empty(), "no blank line yet"); + let batch = decoder.finish(); + assert_eq!(batch.error, None, "clean finish"); + assert_eq!(batch.records.len(), 1, "trailing record flushed"); + assert_eq!( + batch.records[0].data(), + Bytes::from_static(b"tail"), + "trailing data correct" + ); + assert!(decoder.is_finished(), "finish transitions to Finished"); + assert!(!decoder.is_poisoned(), "clean finish is not poison"); + } + + #[test] + fn finish_flushes_unterminated_final_line() { + let mut decoder = SseDecoder::new(); + assert!(push_ok(&mut decoder, b"data: tail").is_empty(), "no terminator at all"); + let batch = decoder.finish(); + assert_eq!( + batch.records[0].data(), + Bytes::from_static(b"tail"), + "unterminated line flushed" + ); + } + + #[test] + fn finish_is_idempotent_and_push_after_finish_errors() { + let mut decoder = SseDecoder::new(); + let records = push_ok(&mut decoder, b"data: x\n\n"); + assert_eq!(records.len(), 1, "record before finish"); + let first = decoder.finish(); + assert!(first.records.is_empty(), "nothing pending to flush"); + assert_eq!(first.error, None, "first finish is clean"); + + let second = decoder.finish(); + assert_eq!(second, SseBatch::default(), "second finish is an empty batch"); + + let after = decoder.push(b"data: y\n\n"); + assert!(after.records.is_empty(), "push after finish yields nothing"); + assert_eq!( + after.error, + Some(SseDecodeError::Finished), + "push after finish reports Finished" + ); + } + + #[test] + fn finish_on_poisoned_reports_limit_error() { + let limits = SseLimits { + max_line_bytes: 4, + ..SseLimits::default() + }; + let mut decoder = SseDecoder::with_limits(limits); + let err = decoder.push(b"data: toolong\n\n").error.unwrap(); + let batch = decoder.finish(); + assert_eq!(batch.error, Some(err), "finish re-reports the poison error"); + assert!(decoder.is_poisoned(), "still poisoned after finish"); + } + + #[test] + fn reset_resumes_after_finish() { + let mut decoder = SseDecoder::new(); + assert!( + decoder.finish().records.is_empty(), + "finish on empty stream yields nothing" + ); + assert!(decoder.is_finished(), "finished before reset"); + decoder.reset(); + assert!(!decoder.is_finished(), "reset clears Finished"); + let records = push_ok(&mut decoder, b"data: again\n\n"); + assert_eq!( + records[0].data(), + Bytes::from_static(b"again"), + "decoding resumes after reset" + ); + } + + #[test] + fn bom_stripped_at_stream_start() { + let mut decoder = SseDecoder::new(); + let mut chunk = BOM_BYTES.to_vec(); + chunk.extend_from_slice(b"data: hi\n\n"); + let records = push_ok(&mut decoder, &chunk); + assert_eq!(records[0].data(), Bytes::from_static(b"hi"), "leading BOM stripped"); + } + + #[test] + fn bom_split_across_chunks() { + let mut decoder = SseDecoder::new(); + assert!(push_ok(&mut decoder, &[0xEF]).is_empty(), "BOM byte 1 buffered"); + assert!(push_ok(&mut decoder, &[0xBB]).is_empty(), "BOM byte 2 buffered"); + assert!(push_ok(&mut decoder, &[0xBF]).is_empty(), "BOM byte 3 buffered"); + let records = push_ok(&mut decoder, b"data: hi\n\n"); + assert_eq!(records[0].data(), Bytes::from_static(b"hi"), "split BOM stripped"); + } + + #[test] + fn partial_bom_mismatch_is_preserved_as_data() { + let mut decoder = SseDecoder::new(); + assert!(push_ok(&mut decoder, &[0xEF]).is_empty(), "possible BOM start buffered"); + assert!(push_ok(&mut decoder, &[0x41]).is_empty(), "divergent byte buffered"); + let batch = decoder.finish(); + assert_eq!( + batch.records[0].fields()[0], + SseField::Unknown { + name: Bytes::copy_from_slice(&[0xEF, 0x41]), + value: Bytes::new() + }, + "non-BOM bytes are preserved as data" + ); + } + + #[test] + fn feff_mid_stream_is_ordinary_data() { + let mut decoder = SseDecoder::new(); + let mut chunk = b"data: a\n\n".to_vec(); + chunk.extend_from_slice(BOM_BYTES); + chunk.extend_from_slice(b"\n\n"); + let records = push_ok(&mut decoder, &chunk); + assert_eq!(records.len(), 2, "two records"); + assert_eq!( + records[1].fields()[0], + SseField::Unknown { + name: Bytes::copy_from_slice(BOM_BYTES), + value: Bytes::new() + }, + "mid-stream BOM bytes are data, not stripped" + ); + } + + #[test] + fn bom_stripped_again_after_reset() { + let mut decoder = SseDecoder::new(); + assert!(decoder.finish().records.is_empty(), "empty finish"); + decoder.reset(); + let mut chunk = BOM_BYTES.to_vec(); + chunk.extend_from_slice(b"data: fresh\n\n"); + let records = push_ok(&mut decoder, &chunk); + assert_eq!( + records[0].data(), + Bytes::from_static(b"fresh"), + "reset re-arms BOM stripping" + ); + } + + #[test] + fn every_split_point_matches_unsplit_decode() { + let corpus: &[&[u8]] = &[ + b"data: hello\n\n", + b"event: e\ndata: 1\ndata: 2\n\ndata: 3\n\n", + b"data: a\r\nid: 1\r\n\r\n", + b": comment\ndata: x\n\n", + b"retry: 500\n\ndata: y\n\n", + &[0xEF, 0xBB, 0xBF, b'd', b'a', b't', b'a', b':', b' ', b'z', b'\n', b'\n'], + b"data: a\rid: 1\r\r", + &[b'd', b'a', b't', b'a', b':', b' ', 0xFF, 0xFE, b'\n', b'\n'], + ]; + for input in corpus { + assert_all_splits_match(input); + } + } +} diff --git a/filter/src/sse/encoder.rs b/filter/src/sse/encoder.rs new file mode 100644 index 000000000..dbfbf1a60 --- /dev/null +++ b/filter/src/sse/encoder.rs @@ -0,0 +1,157 @@ +// SPDX-License-Identifier: MIT +// Copyright (c) 2026 Praxis Contributors + +//! Canonical SSE encoder for locally generated records. + +use bytes::{Bytes, BytesMut}; + +use super::record::{SseField, SseRecord}; + +/// Serialize a record to canonical event-stream bytes. +/// +/// Infallible: an `SseRecord` is validated at construction, so it cannot express +/// a value that would break framing. +#[must_use] +pub fn encode(record: &SseRecord) -> Bytes { + let mut out = BytesMut::new(); + encode_into(record, &mut out); + out.freeze() +} + +/// Append a record's canonical bytes to a caller-provided buffer, letting the +/// caller serialize several records into one buffer. +pub fn encode_into(record: &SseRecord, out: &mut BytesMut) { + for field in record.fields() { + match field { + SseField::Event(value) => write_named(out, b"event", value), + SseField::Data(value) => write_named(out, b"data", value), + SseField::Id(value) => write_named(out, b"id", value), + SseField::Retry(value) => write_named(out, b"retry", value), + SseField::Comment(value) => write_named(out, b"", value), + SseField::Unknown { name, value } => write_named(out, name, value), + } + } + out.extend_from_slice(b"\n"); +} + +/// Emit `NAME: VALUE\n` (or `: VALUE\n` for an empty name). +fn write_named(out: &mut BytesMut, name: &[u8], value: &[u8]) { + out.extend_from_slice(name); + out.extend_from_slice(b": "); + out.extend_from_slice(value); + out.extend_from_slice(b"\n"); +} + +#[cfg(test)] +#[expect(clippy::allow_attributes, reason = "blanket test suppressions")] +#[allow( + clippy::unwrap_used, + clippy::expect_used, + clippy::indexing_slicing, + clippy::panic, + reason = "tests" +)] +mod tests { + use super::*; + use crate::sse::SseDecoder; + + #[test] + fn encodes_canonical_fields_in_order() { + let record = SseRecord::builder() + .event("message") + .data("hello") + .data("world") + .comment("hb") + .build() + .unwrap(); + assert_eq!( + encode(&record), + Bytes::from_static(b"event: message\ndata: hello\ndata: world\n: hb\n\n"), + "canonical encoding in wire order with a trailing blank line" + ); + } + + #[test] + fn encodes_empty_value_with_single_space() { + let record = SseRecord::builder().data("").build().unwrap(); + assert_eq!( + encode(&record), + Bytes::from_static(b"data: \n\n"), + "empty value still emits one space after the colon" + ); + } + + #[test] + fn encodes_unknown_field() { + let record = SseRecord::builder() + .field(SseField::Unknown { + name: Bytes::from_static(b"x-foo"), + value: Bytes::from_static(b"bar"), + }) + .build() + .unwrap(); + assert_eq!( + encode(&record), + Bytes::from_static(b"x-foo: bar\n\n"), + "unknown field uses its name" + ); + } + + #[test] + fn encode_into_appends_multiple_records() { + let a = SseRecord::builder().data("a").build().unwrap(); + let b = SseRecord::builder().data("b").build().unwrap(); + let mut out = BytesMut::new(); + encode_into(&a, &mut out); + encode_into(&b, &mut out); + assert_eq!( + out.freeze(), + Bytes::from_static(b"data: a\n\ndata: b\n\n"), + "encode_into appends without a separator" + ); + } + + #[test] + fn round_trip_is_field_stable() { + let record = SseRecord::builder() + .event("msg") + .data("l1") + .data("l2") + .id("id1") + .retry(4200) + .comment("keepalive") + .field(SseField::Unknown { + name: Bytes::from_static(b"x-trace"), + value: Bytes::from_static(b"t"), + }) + .build() + .unwrap(); + let bytes = encode(&record); + let mut decoder = SseDecoder::new(); + let batch = decoder.push(&bytes); + assert_eq!(batch.error, None, "round-trip decodes cleanly"); + assert_eq!(batch.records.len(), 1, "one record round-trips"); + assert_eq!( + batch.records[0].fields(), + record.fields(), + "fields are byte-stable across round-trip" + ); + } + + #[test] + fn round_trip_preserves_non_utf8_and_nul_in_id() { + let record = SseRecord::builder() + .data(Bytes::copy_from_slice(&[0xFF, 0xFE, b'x'])) + .id(Bytes::copy_from_slice(b"a\0b")) + .build() + .unwrap(); + let bytes = encode(&record); + let mut decoder = SseDecoder::new(); + let batch = decoder.push(&bytes); + assert_eq!( + batch.records[0].fields(), + record.fields(), + "non-UTF-8 data and NUL-in-id survive a round-trip" + ); + } +} diff --git a/filter/src/sse/mod.rs b/filter/src/sse/mod.rs new file mode 100644 index 000000000..f4e74eaf0 --- /dev/null +++ b/filter/src/sse/mod.rs @@ -0,0 +1,195 @@ +// SPDX-License-Identifier: MIT +// Copyright (c) 2026 Praxis Contributors + +//! Bounded, incremental Server-Sent Events (SSE) codec. +//! +//! Pingora owns transport streaming, backpressure, and H1/H2 framing; this +//! module adds only application-level SSE record framing on top of a byte +//! stream. Provider JSON typing, `[DONE]`, event budgets, lifecycle rules, +//! accumulation, and replay stay in the consumer. +//! +//! # Decode from a streamed body +//! +//! Feed each Pingora body chunk to the decoder: +//! +//! ``` +//! use praxis_filter::sse::{SseBatch, SseDecoder}; +//! +//! let mut decoder = SseDecoder::new(); +//! let SseBatch { records, error } = decoder.push(b"data: {\"n\":1}\n\ndata: {"); +//! assert!(error.is_none()); +//! assert_eq!(records.len(), 1); +//! assert_eq!(records[0].data().as_ref(), br#"{"n":1}"#); +//! +//! // the second record completes on a later chunk +//! let batch = decoder.push(b"\"n\":2}\n\n"); +//! assert_eq!(batch.records[0].data().as_ref(), br#"{"n":2}"#); +//! +//! // at end of stream +//! let tail = decoder.finish(); +//! assert!(tail.records.is_empty()); +//! ``` +//! +//! # Encode locally generated records +//! +//! Build records and serialize them to canonical bytes: +//! +//! ``` +//! use praxis_filter::sse::{SseRecord, encode}; +//! +//! let record = SseRecord::builder() +//! .event("message") +//! .data("hello") +//! .build() +//! .expect("valid record"); +//! assert_eq!(encode(&record).as_ref(), b"event: message\ndata: hello\n\n"); +//! ``` +//! +//! # Decode inside a streaming filter +//! +//! Run the decoder from `on_response_body` under +//! [`BodyMode::Stream`](crate::BodyMode): forward each chunk unchanged and feed +//! it to a per-request decoder for inspection. The decoder is request-scoped +//! state, so it lives in the filter context, not in the shared filter. +//! +//! ``` +//! use async_trait::async_trait; +//! use bytes::Bytes; +//! use praxis_filter::{ +//! BodyAccess, BodyMode, FilterAction, FilterError, HttpFilter, HttpFilterContext, +//! sse::SseDecoder, +//! }; +//! +//! // A private newtype keeps this decoder distinct from other filters' state. +//! struct SseState(SseDecoder); +//! +//! struct SseInspector; +//! +//! #[async_trait] +//! impl HttpFilter for SseInspector { +//! fn name(&self) -> &'static str { +//! "sse-inspector" +//! } +//! +//! async fn on_request( +//! &self, +//! _ctx: &mut HttpFilterContext<'_>, +//! ) -> Result { +//! Ok(FilterAction::Continue) +//! } +//! +//! // Receive body chunks read-only, streamed one at a time (the default). +//! fn response_body_access(&self) -> BodyAccess { +//! BodyAccess::ReadOnly +//! } +//! +//! fn response_body_mode(&self) -> BodyMode { +//! BodyMode::Stream +//! } +//! +//! fn on_response_body( +//! &self, +//! ctx: &mut HttpFilterContext<'_>, +//! body: &mut Option, +//! end_of_stream: bool, +//! ) -> Result { +//! if ctx.get_filter_state::().is_none() { +//! ctx.insert_filter_state(SseState(SseDecoder::new())); +//! } +//! let SseState(decoder) = ctx +//! .get_filter_state_mut::() +//! .expect("just inserted the decoder state"); +//! +//! // Inspect by reference; `body` is left untouched and forwarded as is. +//! if let Some(chunk) = body.as_ref() { +//! let batch = decoder.push(chunk); +//! if let Some(err) = batch.error { +//! return Err(err.into()); // a limit violation poisons the decoder +//! } +//! // Provider-neutral inspection, e.g. events completed by this chunk. +//! let _event_count = batch.records.iter().filter(|r| r.is_event()).count(); +//! } +//! if end_of_stream { +//! let tail = decoder.finish(); +//! if let Some(err) = tail.error { +//! return Err(err.into()); // finish re-reports a poisoned decoder +//! } +//! // A trailing record with no blank-line terminator surfaces here. +//! let _final_events = tail.records.iter().filter(|r| r.is_event()).count(); +//! } +//! Ok(FilterAction::Continue) +//! } +//! } +//! +//! // The hook runs inside Pingora with a real context; construct the filter +//! // here to type-check the integration. +//! let filter = SseInspector; +//! assert_eq!(filter.name(), "sse-inspector"); +//! ``` +//! +//! # Encode for a streaming response body +//! +//! Serialize locally generated records as canonical SSE bytes from a +//! [`StreamingResponseBody`](crate::StreamingResponseBody), e.g. replaying a stored response: +//! +//! ``` +//! use async_trait::async_trait; +//! use bytes::Bytes; +//! use praxis_filter::{ +//! FilterError, StreamingResponseBody, +//! sse::{SseRecord, encode}, +//! }; +//! +//! struct ReplayBody { +//! records: std::vec::IntoIter, +//! } +//! +//! #[async_trait] +//! impl StreamingResponseBody for ReplayBody { +//! async fn next_chunk(&mut self) -> Result, FilterError> { +//! Ok(self.records.next().as_ref().map(encode)) +//! } +//! +//! async fn suppress(&mut self) -> Result<(), FilterError> { +//! self.records = Vec::new().into_iter(); +//! Ok(()) +//! } +//! +//! async fn cancel(&mut self) { +//! self.records = Vec::new().into_iter(); +//! } +//! } +//! +//! let records = vec![ +//! SseRecord::builder() +//! .data("hello") +//! .build() +//! .expect("valid record"), +//! SseRecord::builder() +//! .event("done") +//! .data("bye") +//! .build() +//! .expect("valid record"), +//! ]; +//! let mut body = ReplayBody { +//! records: records.into_iter(), +//! }; +//! +//! tokio::runtime::Runtime::new() +//! .expect("runtime") +//! .block_on(async { +//! let first = body.next_chunk().await.expect("chunk").expect("record"); +//! assert_eq!(first.as_ref(), b"data: hello\n\n"); +//! let second = body.next_chunk().await.expect("chunk").expect("record"); +//! assert_eq!(second.as_ref(), b"event: done\ndata: bye\n\n"); +//! assert!(body.next_chunk().await.expect("chunk").is_none()); +//! }); +//! ``` + +mod decoder; +mod encoder; +mod record; + +pub use decoder::{SseBatch, SseDecodeError, SseDecoder, SseLimits}; +pub use encoder::{encode, encode_into}; +pub use record::{SseBuildError, SseField, SseRecord, SseRecordBuilder}; diff --git a/filter/src/sse/record.rs b/filter/src/sse/record.rs new file mode 100644 index 000000000..e2f9f3163 --- /dev/null +++ b/filter/src/sse/record.rs @@ -0,0 +1,471 @@ +// SPDX-License-Identifier: MIT +// Copyright (c) 2026 Praxis Contributors + +//! Provider-neutral SSE field and record types. + +use bytes::{Bytes, BytesMut}; + +/// One field line within a record. +/// +/// Values are raw bytes; the codec performs no UTF-8 normalization. A +/// well-formed value never contains CR or LF (those are line terminators), and +/// an `Unknown` name is non-empty and free of `:`, CR, and LF. These invariants +/// are enforced when a record is *built* (see `SseRecordBuilder`) and hold for +/// every record the decoder produces. +#[derive(Clone, Debug, Eq, PartialEq)] +pub enum SseField { + /// An `event` field naming the record's event type. + Event(Bytes), + /// A `data` field; repeats form multi-line data. + Data(Bytes), + /// An `id` field carrying the last-event-id token. + Id(Bytes), + /// A `retry` field carrying the reconnection time in milliseconds. + Retry(Bytes), + /// A `:`-prefixed comment line; value is the text after the colon. + Comment(Bytes), + /// A field whose name is not `event`/`data`/`id`/`retry` and is not a + /// comment. The name is non-empty and never equals a known field name, so + /// an encode-then-decode round-trip preserves the variant. + Unknown { + /// The field name (non-empty, colon-free, not a known name). + name: Bytes, + /// The field value. + value: Bytes, + }, +} + +/// A framing block delimited by a blank line. +/// +/// A record may be a dispatched `EventSource` event (at least one `Data` field), +/// or a comment-only heartbeat, an id/retry-only block, or an unknown-field +/// block. Fields are retained in wire order. +/// +/// There is no public constructor and no `Default`: an empty record cannot +/// round-trip (the decoder ignores empty blocks). Build one with +/// `SseRecord::builder`; the decoder is the only other source. +#[derive(Clone, Debug, Eq, PartialEq)] +pub struct SseRecord { + /// Fields in wire order. + fields: Vec, +} + +impl SseRecord { + /// Construct directly from ordered fields. Crate-internal: the decoder and + /// `SseRecordBuilder::build` are the only callers, and both uphold the field + /// invariants documented on `SseField`. + pub(crate) fn from_fields(fields: Vec) -> Self { + Self { fields } + } + + /// The record's fields, in wire order. + #[must_use] + pub fn fields(&self) -> &[SseField] { + &self.fields + } + + /// `Data` field values joined with a single `b'\n'`; empty when the record + /// has no `Data` field. + #[must_use] + pub fn data(&self) -> Bytes { + let mut values = self.fields.iter().filter_map(|field| match field { + SseField::Data(value) => Some(value), + _ => None, + }); + let Some(first) = values.next() else { + return Bytes::new(); + }; + let Some(second) = values.next() else { + return first.clone(); + }; + let mut out = BytesMut::new(); + out.extend_from_slice(first); + for value in std::iter::once(second).chain(values) { + out.extend_from_slice(b"\n"); + out.extend_from_slice(value); + } + out.freeze() + } + + /// Effective event type: the value of the last `Event` field, if any. + #[must_use] + pub fn event(&self) -> Option<&[u8]> { + self.fields.iter().rev().find_map(|field| match field { + SseField::Event(value) => Some(value.as_ref()), + _ => None, + }) + } + + /// Effective id: the last `Id` field whose value has no NUL. Later + /// NUL-containing id lines are ignored and do not clear an earlier id. + #[must_use] + pub fn id(&self) -> Option<&[u8]> { + self.fields.iter().rev().find_map(|field| match field { + SseField::Id(value) if !value.contains(&0) => Some(value.as_ref()), + _ => None, + }) + } + + /// Effective retry (ms): the last `Retry` field that is all ASCII digits and + /// fits in `u64`. Later invalid retry lines are ignored. + #[must_use] + pub fn retry(&self) -> Option { + self.fields.iter().rev().find_map(|field| match field { + SseField::Retry(value) => parse_retry(value), + _ => None, + }) + } + + /// Whether this record has at least one `Data` field, i.e. whether an + /// `EventSource` client would dispatch an event for it. + #[must_use] + pub fn is_event(&self) -> bool { + self.fields.iter().any(|field| matches!(field, SseField::Data(_))) + } + + /// Start building a record for local generation / encoding. + #[must_use = "call `.build()` to finalize the record"] + pub fn builder() -> SseRecordBuilder { + SseRecordBuilder::default() + } +} + +/// Parse an SSE `retry` value: all-ASCII-digit `u64`, else `None`. +fn parse_retry(value: &[u8]) -> Option { + if value.is_empty() || !value.iter().all(u8::is_ascii_digit) { + return None; + } + std::str::from_utf8(value).ok()?.parse::().ok() +} + +/// Validated construction for local generation / encoding. Setters are +/// infallible (they accumulate fields); validation happens in `build`. +#[derive(Debug, Default)] +#[must_use] +pub struct SseRecordBuilder { + /// Accumulated fields, validated on `build`. + fields: Vec, +} + +impl SseRecordBuilder { + /// Append an `event` field. + #[must_use = "builder methods should be chained"] + pub fn event(mut self, value: impl Into) -> Self { + self.fields.push(SseField::Event(value.into())); + self + } + + /// Append a `data` field. Repeat for multi-line data. + #[must_use = "builder methods should be chained"] + pub fn data(mut self, value: impl Into) -> Self { + self.fields.push(SseField::Data(value.into())); + self + } + + /// Append an `id` field. + #[must_use = "builder methods should be chained"] + pub fn id(mut self, value: impl Into) -> Self { + self.fields.push(SseField::Id(value.into())); + self + } + + /// Append a `retry` field (milliseconds); always numeric and valid. + #[must_use = "builder methods should be chained"] + pub fn retry(mut self, ms: u64) -> Self { + self.fields.push(SseField::Retry(Bytes::from(ms.to_string()))); + self + } + + /// Append a comment field. + #[must_use = "builder methods should be chained"] + pub fn comment(mut self, value: impl Into) -> Self { + self.fields.push(SseField::Comment(value.into())); + self + } + + /// Append an arbitrary field for full control over kind and order. + #[must_use = "builder methods should be chained"] + pub fn field(mut self, field: SseField) -> Self { + self.fields.push(field); + self + } + + /// Validate and finalize. + /// + /// # Errors + /// + /// Returns `SseBuildError` if any value contains CR or LF, any `Unknown` + /// name is empty / contains `:`/CR/LF / equals a known field name, or the + /// record has no fields. + pub fn build(self) -> Result { + if self.fields.is_empty() { + return Err(SseBuildError::EmptyRecord); + } + for field in &self.fields { + validate_field(field)?; + } + Ok(SseRecord::from_fields(self.fields)) + } +} + +/// Whether the slice contains a raw CR or LF (which would break framing). +fn has_newline(bytes: &[u8]) -> bool { + bytes.iter().any(|&b| matches!(b, b'\r' | b'\n')) +} + +/// Whether `name` equals one of the typed field names. +fn is_known_name(name: &[u8]) -> bool { + matches!(name, b"event" | b"data" | b"id" | b"retry") +} + +/// Validate one field's framing safety for `build`. +fn validate_field(field: &SseField) -> Result<(), SseBuildError> { + match field { + SseField::Event(value) + | SseField::Data(value) + | SseField::Id(value) + | SseField::Retry(value) + | SseField::Comment(value) => { + if has_newline(value) { + return Err(SseBuildError::NewlineInValue); + } + }, + SseField::Unknown { name, value } => { + if has_newline(value) { + return Err(SseBuildError::NewlineInValue); + } + let bad_name = + name.is_empty() || name.iter().any(|&b| matches!(b, b':' | b'\r' | b'\n')) || is_known_name(name); + if bad_name { + return Err(SseBuildError::InvalidFieldName); + } + }, + } + Ok(()) +} + +/// Build-time validation errors for `SseRecordBuilder::build`. +#[derive(Clone, Copy, Debug, Eq, PartialEq, thiserror::Error)] +pub enum SseBuildError { + /// A field value contained a raw CR or LF, which would break framing. + #[error("SSE field value contains a raw CR or LF")] + NewlineInValue, + /// An `Unknown` field name was empty, contained `:`/CR/LF, or equaled a + /// known field name. + #[error("SSE unknown field name is empty, contains ':'/CR/LF, or is a known name")] + InvalidFieldName, + /// The record had no fields; empty records cannot be represented. + #[error("SSE record has no fields")] + EmptyRecord, +} + +#[cfg(test)] +#[expect(clippy::allow_attributes, reason = "blanket test suppressions")] +#[allow( + clippy::unwrap_used, + clippy::expect_used, + clippy::indexing_slicing, + clippy::panic, + reason = "tests" +)] +mod tests { + use super::*; + + #[test] + fn field_variants_hold_their_bytes() { + let data = SseField::Data(Bytes::from_static(b"hello")); + assert_eq!( + data, + SseField::Data(Bytes::from_static(b"hello")), + "Data variant should compare by its bytes" + ); + + let unknown = SseField::Unknown { + name: Bytes::from_static(b"x-trace"), + value: Bytes::from_static(b"abc"), + }; + match unknown { + SseField::Unknown { name, value } => { + assert_eq!(name, Bytes::from_static(b"x-trace"), "unknown name should round-trip"); + assert_eq!(value, Bytes::from_static(b"abc"), "unknown value should round-trip"); + }, + _ => panic!("expected Unknown variant"), + } + } + + #[test] + fn data_joins_multiline_with_single_newline() { + let record = SseRecord::from_fields(vec![ + SseField::Data(Bytes::from_static(b"line1")), + SseField::Data(Bytes::from_static(b"line2")), + ]); + assert_eq!( + record.data(), + Bytes::from_static(b"line1\nline2"), + "multi-line data should join with a single newline" + ); + assert!(record.is_event(), "record with data should be an event"); + } + + #[test] + fn data_single_field_shares_backing_storage() { + // A heap value: copying would reallocate to a new address, so pointer + // equality is what proves the sole-data path shares the field's buffer. + let value = Bytes::from(b"only".to_vec()); + let value_ptr = value.as_ptr(); + let record = SseRecord::from_fields(vec![SseField::Data(value)]); + let data = record.data(); + assert_eq!(data, Bytes::from_static(b"only"), "content preserved"); + assert_eq!( + data.as_ptr(), + value_ptr, + "a sole data field shares the field's backing storage without copying" + ); + assert!(record.is_event(), "single data field is an event"); + } + + #[test] + fn data_is_empty_when_no_data_fields() { + let record = SseRecord::from_fields(vec![SseField::Comment(Bytes::from_static(b"hi"))]); + assert_eq!(record.data(), Bytes::new(), "comment-only record has empty data"); + assert!(!record.is_event(), "comment-only record is not an event"); + } + + #[test] + fn event_returns_last_event_value() { + let record = SseRecord::from_fields(vec![ + SseField::Event(Bytes::from_static(b"first")), + SseField::Event(Bytes::from_static(b"second")), + ]); + assert_eq!( + record.event(), + Some(b"second".as_slice()), + "event should be the last value" + ); + } + + #[test] + fn id_ignores_later_nul_containing_value() { + let record = SseRecord::from_fields(vec![ + SseField::Id(Bytes::from_static(b"good")), + SseField::Id(Bytes::from_static(b"ba\0d")), + ]); + assert_eq!( + record.id(), + Some(b"good".as_slice()), + "NUL-containing id must be ignored" + ); + } + + #[test] + fn retry_parses_last_valid_numeric() { + let record = SseRecord::from_fields(vec![ + SseField::Retry(Bytes::from_static(b"1000")), + SseField::Retry(Bytes::from_static(b"3000")), + ]); + assert_eq!(record.retry(), Some(3000), "retry should be the last numeric value"); + } + + #[test] + fn retry_ignores_non_digits_and_overflow() { + let non_digit = SseRecord::from_fields(vec![SseField::Retry(Bytes::from_static(b"1a"))]); + assert_eq!(non_digit.retry(), None, "non-digit retry is invalid"); + + let plus = SseRecord::from_fields(vec![SseField::Retry(Bytes::from_static(b"+5"))]); + assert_eq!(plus.retry(), None, "signed retry is invalid"); + + let overflow = SseRecord::from_fields(vec![SseField::Retry(Bytes::from_static(b"99999999999999999999999999"))]); + assert_eq!(overflow.retry(), None, "overflowing retry is invalid"); + } + + #[test] + fn retry_falls_back_to_earlier_valid_when_last_invalid() { + let record = SseRecord::from_fields(vec![ + SseField::Retry(Bytes::from_static(b"1000")), + SseField::Retry(Bytes::from_static(b"oops")), + ]); + assert_eq!(record.retry(), Some(1000), "should skip the invalid trailing retry"); + } + + #[test] + fn builder_builds_valid_record_in_order() { + let record = SseRecord::builder() + .event("message") + .data("hello") + .data("world") + .id("42") + .retry(3000) + .comment("hb") + .build() + .unwrap(); + assert_eq!(record.event(), Some(b"message".as_slice()), "event preserved"); + assert_eq!( + record.data(), + Bytes::from_static(b"hello\nworld"), + "multi-line data preserved" + ); + assert_eq!(record.id(), Some(b"42".as_slice()), "id preserved"); + assert_eq!(record.retry(), Some(3000), "retry preserved"); + assert_eq!(record.fields().len(), 6, "all six fields retained in order"); + } + + #[test] + fn builder_rejects_empty_record() { + assert_eq!( + SseRecord::builder().build().unwrap_err(), + SseBuildError::EmptyRecord, + "empty builder must not produce a record" + ); + } + + #[test] + fn builder_rejects_newline_in_value() { + assert_eq!( + SseRecord::builder().data("a\nb").build().unwrap_err(), + SseBuildError::NewlineInValue, + "LF in a value breaks framing" + ); + assert_eq!( + SseRecord::builder().event("a\rb").build().unwrap_err(), + SseBuildError::NewlineInValue, + "CR in a value breaks framing" + ); + assert_eq!( + SseRecord::builder().comment("a\nb").build().unwrap_err(), + SseBuildError::NewlineInValue, + "LF in a comment breaks framing" + ); + } + + fn assert_invalid_unknown_name(name: &[u8], msg: &str) { + let err = SseRecord::builder() + .field(SseField::Unknown { + name: Bytes::copy_from_slice(name), + value: Bytes::from_static(b"v"), + }) + .build() + .unwrap_err(); + assert_eq!(err, SseBuildError::InvalidFieldName, "{msg}"); + } + + #[test] + fn builder_rejects_bad_unknown_names() { + assert_invalid_unknown_name(b"", "empty unknown name is invalid"); + assert_invalid_unknown_name(b"a:b", "colon in unknown name is invalid"); + assert_invalid_unknown_name(b"a\nb", "newline in unknown name is invalid"); + assert_invalid_unknown_name(b"data", "known name in Unknown would change variant"); + } + + #[test] + fn builder_allows_valid_unknown_and_nul_in_id() { + let record = SseRecord::builder() + .field(SseField::Unknown { + name: Bytes::from_static(b"x-trace"), + value: Bytes::from_static(b"abc"), + }) + .id(Bytes::from_static(b"n\0ul")) + .build() + .unwrap(); + assert_eq!(record.fields().len(), 2, "valid unknown and NUL-in-id are accepted"); + } +} From d81b62e98df72d139ae8fdcc0c10580392a89b3e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?S=C3=A9bastien=20Han?= Date: Tue, 1 Sep 2026 15:14:27 +0200 Subject: [PATCH 2/3] style(filter): order SSE test utilities last in test modules MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Test helper fns and constants must come after all #[test] functions in a #[cfg(test)] module, and separator comments must use the full-width canonical form. - decoder.rs: move BOM_BYTES, push_ok, decode_whole, decode_split, and assert_all_splits_match below the last test; replace the short-form "// ----- Test Utilities -----" separator with the full-width form. - record.rs: move assert_invalid_unknown_name below the last test under a full-width "// Test Utilities" separator. Signed-off-by: Sébastien Han --- filter/src/sse/decoder.rs | 94 ++++++++++++++++++++------------------- filter/src/sse/record.rs | 26 ++++++----- 2 files changed, 63 insertions(+), 57 deletions(-) diff --git a/filter/src/sse/decoder.rs b/filter/src/sse/decoder.rs index ca1a19536..2e01759e6 100644 --- a/filter/src/sse/decoder.rs +++ b/filter/src/sse/decoder.rs @@ -462,52 +462,6 @@ mod tests { let _boxed: Box = err.into(); } - // ----- Test Utilities ----- - - // Module-scoped BOM constant for the tests below. - const BOM_BYTES: &[u8] = &[0xEF, 0xBB, 0xBF]; - - // Push one chunk, assert no error, return the completed records. - fn push_ok(decoder: &mut SseDecoder, chunk: &[u8]) -> Vec { - let batch = decoder.push(chunk); - assert_eq!(batch.error, None, "unexpected decode error"); - batch.records - } - - // Decode the whole input in one push then finish; return all records. - fn decode_whole(input: &[u8]) -> Vec { - let mut decoder = SseDecoder::new(); - let mut records = decoder.push(input).records; - records.extend(decoder.finish().records); - records - } - - // Decode the input split at byte offsets a and b; return all records. - fn decode_split(input: &[u8], a: usize, b: usize) -> Vec { - let mut decoder = SseDecoder::new(); - let mut records = Vec::new(); - for part in [&input[..a], &input[a..b], &input[b..]] { - records.extend(decoder.push(part).records); - } - records.extend(decoder.finish().records); - records - } - - // Assert every 3-way split of the input decodes identically to the whole. - fn assert_all_splits_match(input: &[u8]) { - let expected = decode_whole(input); - let len = input.len(); - for a in 0..=len { - for b in a..=len { - assert_eq!( - decode_split(input, a, b), - expected, - "split ({a},{b}) diverged for {input:?}" - ); - } - } - } - #[test] fn single_record() { let mut decoder = SseDecoder::new(); @@ -961,4 +915,52 @@ mod tests { assert_all_splits_match(input); } } + + // ------------------------------------------------------------------------- + // Test Utilities + // ------------------------------------------------------------------------- + + // Module-scoped BOM constant for the tests below. + const BOM_BYTES: &[u8] = &[0xEF, 0xBB, 0xBF]; + + // Push one chunk, assert no error, return the completed records. + fn push_ok(decoder: &mut SseDecoder, chunk: &[u8]) -> Vec { + let batch = decoder.push(chunk); + assert_eq!(batch.error, None, "unexpected decode error"); + batch.records + } + + // Decode the whole input in one push then finish; return all records. + fn decode_whole(input: &[u8]) -> Vec { + let mut decoder = SseDecoder::new(); + let mut records = decoder.push(input).records; + records.extend(decoder.finish().records); + records + } + + // Decode the input split at byte offsets a and b; return all records. + fn decode_split(input: &[u8], a: usize, b: usize) -> Vec { + let mut decoder = SseDecoder::new(); + let mut records = Vec::new(); + for part in [&input[..a], &input[a..b], &input[b..]] { + records.extend(decoder.push(part).records); + } + records.extend(decoder.finish().records); + records + } + + // Assert every 3-way split of the input decodes identically to the whole. + fn assert_all_splits_match(input: &[u8]) { + let expected = decode_whole(input); + let len = input.len(); + for a in 0..=len { + for b in a..=len { + assert_eq!( + decode_split(input, a, b), + expected, + "split ({a},{b}) diverged for {input:?}" + ); + } + } + } } diff --git a/filter/src/sse/record.rs b/filter/src/sse/record.rs index e2f9f3163..808d990bf 100644 --- a/filter/src/sse/record.rs +++ b/filter/src/sse/record.rs @@ -437,17 +437,6 @@ mod tests { ); } - fn assert_invalid_unknown_name(name: &[u8], msg: &str) { - let err = SseRecord::builder() - .field(SseField::Unknown { - name: Bytes::copy_from_slice(name), - value: Bytes::from_static(b"v"), - }) - .build() - .unwrap_err(); - assert_eq!(err, SseBuildError::InvalidFieldName, "{msg}"); - } - #[test] fn builder_rejects_bad_unknown_names() { assert_invalid_unknown_name(b"", "empty unknown name is invalid"); @@ -468,4 +457,19 @@ mod tests { .unwrap(); assert_eq!(record.fields().len(), 2, "valid unknown and NUL-in-id are accepted"); } + + // ------------------------------------------------------------------------- + // Test Utilities + // ------------------------------------------------------------------------- + + fn assert_invalid_unknown_name(name: &[u8], msg: &str) { + let err = SseRecord::builder() + .field(SseField::Unknown { + name: Bytes::copy_from_slice(name), + value: Bytes::from_static(b"v"), + }) + .build() + .unwrap_err(); + assert_eq!(err, SseBuildError::InvalidFieldName, "{msg}"); + } } From 2a12cd203b791a6687454efa9ac7889103701ea3 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?S=C3=A9bastien=20Han?= Date: Tue, 1 Sep 2026 16:42:40 +0200 Subject: [PATCH 3/3] refactor(filter): tighten SSE decoder limit defaults and docs MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Address review feedback on the SSE decoder limits: - Extract the SseLimits default magic numbers into named DEFAULT_MAX_* constants, matching the convention used elsewhere in the codebase. - Lower the default max_line_bytes from 10 MiB to 1 MiB; callers that need more can raise it via SseLimits. - Expand the max_record_bytes doc to explain that the in-progress line is bounded separately by max_line_bytes, so peak per-record memory is max_record_bytes + max_line_bytes. - Add the standard section separators for each top-level type (DecoderState, SseLimits, SseBatch, SseDecodeError, SseDecoder). Signed-off-by: Sébastien Han --- filter/src/sse/decoder.rs | 47 +++++++++++++++++++++++++++++++++------ 1 file changed, 40 insertions(+), 7 deletions(-) diff --git a/filter/src/sse/decoder.rs b/filter/src/sse/decoder.rs index 2e01759e6..1a59ae0fe 100644 --- a/filter/src/sse/decoder.rs +++ b/filter/src/sse/decoder.rs @@ -14,6 +14,17 @@ use super::record::{SseField, SseRecord}; /// UTF-8 byte order mark, stripped once at stream start. const BOM: [u8; 3] = [0xEF, 0xBB, 0xBF]; +/// Default for `SseLimits::max_line_bytes`. +const DEFAULT_MAX_LINE_BYTES: usize = 1_048_576; // 1 MiB +/// Default for `SseLimits::max_record_bytes`. +const DEFAULT_MAX_RECORD_BYTES: usize = 10_485_760; // 10 MiB +/// Default for `SseLimits::max_fields_per_record`. +const DEFAULT_MAX_FIELDS_PER_RECORD: usize = 4096; // far above any real record + +// ----------------------------------------------------------------------------- +// DecoderState +// ----------------------------------------------------------------------------- + /// Lifecycle state of an `SseDecoder`. #[derive(Clone, Copy, Debug)] enum DecoderState { @@ -25,14 +36,24 @@ enum DecoderState { Poisoned(SseDecodeError), } +// ----------------------------------------------------------------------------- +// SseLimits +// ----------------------------------------------------------------------------- + /// Bounds on retained memory. Enforced continuously as bytes accumulate. #[derive(Clone, Copy, Debug, Eq, PartialEq)] pub struct SseLimits { /// Max bytes for a single field line held across chunks. pub max_line_bytes: usize, - /// Max total retained bytes for one in-progress record: the sum of every - /// field's value plus every `Unknown` field name, so large unknown names - /// cannot bypass the limit. + /// Max bytes retained for the committed fields of one in-progress record: + /// the sum of every field's value, plus the name of every `Unknown` field + /// so large unknown names cannot bypass the limit. Resets when the record + /// is dispatched. + /// + /// The line currently being accumulated is *not* counted until it commits; + /// it is bounded separately by `max_line_bytes`. Peak memory the decoder + /// can hold for a single record is therefore + /// `max_record_bytes + max_line_bytes`. pub max_record_bytes: usize, /// Max number of fields in one record (bounds per-field allocations so many /// tiny fields cannot evade `max_record_bytes`). @@ -42,13 +63,17 @@ pub struct SseLimits { impl Default for SseLimits { fn default() -> Self { Self { - max_line_bytes: 10_485_760, // 10 MiB - max_record_bytes: 10_485_760, // 10 MiB - max_fields_per_record: 4096, // far above any real record + max_line_bytes: DEFAULT_MAX_LINE_BYTES, + max_record_bytes: DEFAULT_MAX_RECORD_BYTES, + max_fields_per_record: DEFAULT_MAX_FIELDS_PER_RECORD, } } } +// ----------------------------------------------------------------------------- +// SseBatch +// ----------------------------------------------------------------------------- + /// Result of one `SseDecoder::push` or `SseDecoder::finish`. /// /// Carries records completed *before* any error, so a size violation never @@ -64,6 +89,10 @@ pub struct SseBatch { pub error: Option, } +// ----------------------------------------------------------------------------- +// SseDecodeError +// ----------------------------------------------------------------------------- + /// Decode-time errors. The three limit violations poison the decoder (the error /// is re-reported until `reset`); `Finished` is not a poison. #[derive(Clone, Copy, Debug, Eq, PartialEq, thiserror::Error)] @@ -98,6 +127,10 @@ pub enum SseDecodeError { Finished, } +// ----------------------------------------------------------------------------- +// SseDecoder +// ----------------------------------------------------------------------------- + /// Bounded, incremental SSE record decoder. /// /// Feed body chunks with `push`; each call returns the records that chunk @@ -440,7 +473,7 @@ mod tests { #[test] fn default_limits_match_spec() { let limits = SseLimits::default(); - assert_eq!(limits.max_line_bytes, 10_485_760, "default line limit is 10 MiB"); + assert_eq!(limits.max_line_bytes, 1_048_576, "default line limit is 1 MiB"); assert_eq!(limits.max_record_bytes, 10_485_760, "default record limit is 10 MiB"); assert_eq!(limits.max_fields_per_record, 4096, "default field cap is 4096"); }