feat(filter): add a bounded, incremental SSE codec

[leseb]

What does this PR do?

Adds 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, tolerating cross-chunk splits, CR/CRLF/LF terminators, and a leading UTF-8 BOM, and it poisons itself on a limit violation so a caller cannot silently lose framing. A validated SseRecordBuilder plus encode/encode_into serialize locally generated SseRecords to canonical wire bytes. It is a leaf codec: Pingora keeps ownership of transport streaming, backpressure, and H1/H2 framing, while provider JSON, [DONE], and lifecycle concerns stay in the consumer. There are no consumers yet, so this adds the library only.

Which issue(s) does this relate to?

Fixes #986

Checklist

• Signed off all commits (git commit -s)
• Tests added or updated
• Documentation updated (if applicable) — full rustdoc plus 4 runnable doctests on the module; no operator/reference docs apply since no filter consumes it yet
• make lint && make test passes locally — make lint and make build pass, and the filter crate's 2114 unit tests + 65 doctests pass. The full-workspace make test is green except a pre-existing, unrelated flaky test tls::watcher::tests::watcher_reloads_on_file_change, which fails only under full-suite parallel load (fixed 2000 ms filesystem-watch window) and passes 262/262 in isolation. This PR touches only filter/src/sse/* and one line of filter/src/lib.rs, so it has no causal path to the TLS watcher.

Does this introduce a breaking change?

No. It adds a new, additive module (praxis_filter::sse) with no changes to existing code and no current consumers.

[praxis-bot-app]

Commit message format: the following commits do not follow conventional commits:

• 408e62d: style(filter): apply nightly rustfmt to SSE lib export and encoder test
• 8f5b20b: style(filter): drop redundant Bytes import in SSE decoder tests

Expected format: type(scope): summary (max 72 chars). Types: build, chore, ci, docs, feat, fix, perf, refactor, test.

[praxis-bot-app]

PR too large: 1788 lines added (limit: 500, excludes Cargo files, tests, docs, examples, and benchmarks). Please split into smaller PRs.

[leseb]

@shaneutt any chance we could squeeze this in 0.5.4? If not, no biggie.

[praxis-bot]

PR Review

Summary: Well-implemented bounded SSE codec with thorough test coverage. The decoder correctly handles all SSE spec edge cases (BOM stripping, CR/CRLF/LF terminators, cross-chunk splits, limit poisoning). The encoder round-trips cleanly with the decoder. Two convention issues found.

Overall: Solid library addition. Minor convention fixes needed.

Severity	Count
Critical	0
Large	0
Medium	2

[shaneutt]

Overall approach looks good, thank you @leseb!

I have comments about some of the specifics to resolve, and then 👍

[shaneutt]

The signature here is something I would wanna get settled early since it's the expensive thing to change once filters depend on it 🤔

Pingora gives body filters a Bytes, and by taking &[u8] we're committing to copying every field value twice: once into line_buf, then again with Bytes::copy_from_slice when the line is classified, plus a fresh Vec<SseField> allocation per record because fields gets mem::taken.

For an LLM token stream that's roughly three allocations per token. If push took Bytes instead, any field that lands entirely inside one chunk (which is nearly all of them in practice) could just be a slice_ref into the chunk, and we'd only fall back to copying when a line actually straddles a chunk boundary.

It would be OK in my book to have this be a follow-up for later and keep the internals as they are for now I just don't want the public signature to rule that out for us to optimize later.

[shaneutt]

Something we might optimize: When we poison line_buf and fields are left as-is so a decoder that just hit a limit is holding onto up to max_line_bytes + max_record_bytes until someone drops it. That's fine if the consumer drops it immediately, but if a filter keeps the decoder in a per-connection ctx (which is the natural place for it) that memory sits for the life of the connection.

Since a poisoned decoder can't produce anything further anyway, I think I would just clear both buffers as part of the transition? 🤔

[shaneutt]

I think this one might be worth changing before anything consumes it:

If the upstream connection drops in the middle of a data: line, the finish method currently takes whatever's buffered, treats it as a terminated line, and hands it back as a normal record.

From the consumer's side that's indistinguishable from a real one, so a half-written JSON payload would get parsed (or forwarded to the client) as though it were complete?

The event-stream spec actually says the opposite: "an incomplete event at EOF is discarded, not dispatched". That feels right, WTDY? If I'm not misunderstanding, we need a mechanism here so the caller doesn't lose framing without knowing.

I'd rather finish say "there was a partial record at EOF" explicitly, either as an SseDecodeError::Truncated alongside the completed records, or as a separate trailing field on SseBatch, and let the caller decide whether that's an error or something to salvage.

[shaneutt]

If you wanted to ensure that this reset doesn't let new fields on the struct slip past in the future, you can use a destructuring instead of the type fields directly:

let Self { line_buf, fields, record_bytes, prev_cr, state, bom_len, bom_resolved, limits = _ } = self;
line_buf.clear();

Notably not using the range operator (..) in the destructuring so any future field added that doesn't get a reset implementation gets a compile-time error.

All this said, I don't think we actually want a reset method here: you would normally implement a reset method for something that's on the hot path with allocations so you can reduce memory thrash, however, that isn't the case here:

with_limits is already allocation-free (Vec::new() doesn't allocate), so constructing a fresh decoder costs the same as resetting one. The only capacity reset could preserve is line_buf (fields is mem::taken on every dispatched record, so its allocation never survives anyhow) and a decoder lives for a whole stream, so construction happens per stream, not per chunk. A couple of line_buf reallocs per streamed response isn't measurable given this context.

Semantically this also is a bit perplexing: the main thing reset does is "unpoison" or "unfinish", but once a limit violation has lost framing, resuming the same stream just yields garbage? The only legitimate use is starting a new stream, and a fresh with_limits(self.limits) does that without offering an escape hatch. With no consumers yet, I might drop reset entirely? If you want to keep the name, make it *self = Self::with_limits(self.limits), which also sidesteps the field-drift problem above, but IDK if that's really worth it over just having the caller make their own?

[shaneutt]

This is the loop that runs for every byte of every streamed response, and right now it's a Vec::push plus a limit check per byte: so scanning ahead for the next "\n/\r" with iter().position and then extend_from_slice-ing the whole run with a single length check per run would be a lot cheaper and slightly simpler to read.

[shaneutt]

I appreciate the generous rustdocs in the mod.rs 🙇

[leseb]

Pulled out of 0.5.4, I don't want to block the release.