feat(adapters): SSE non-JSON payloads carry data+event wrapper (FWD-17)

Option (b) of the FWD-17 decision: payloads decode as JSON when valid
(number stays number, quoted string stays string); non-JSON payloads
surface as {"data": <raw>, "event": <name|null>} instead of
silently degrading to a JSON string. The parser captures the frame's
event: field (WHATWG last-wins) and resets all pending state on every
blank-line dispatch, so an event:-only frame (no data) cannot leak its
name into a later frame. JSON frames surface as themselves even under
a named event; the implicit 'message' default never wraps a payload.

Verification: cargo test (33 forward tests incl. 4 new FWD-17
contract/parser pins), clippy, fmt.
This commit is contained in:
2026-08-30 20:52:44 +00:00
parent 5244dc46e2
commit 3d932be75f
+148 -5
View File
@@ -804,13 +804,29 @@ async fn error_envelope(
ResponseEnvelope::error(request_id, CallError::new(code, message, false))
}
/// Converts a parsed SSE event into a response envelope, JSON-decoding
/// the data payload when possible.
/// Maps a parsed SSE frame to a response envelope under the FWD-17
/// payload contract: a payload that is valid JSON surfaces as the
/// decoded value itself (a number payload stays a number, a
/// quoted-string payload stays a string — the distinction a raw-text
/// fallback would erase); any other payload surfaces as the
/// `{"data": <raw payload>, "event": <event-name|null>}` wrapper, so
/// a non-JSON stream is field-addressable and an upstream's
/// named-event conventions are visible on it. An empty payload is
/// `Null`. JSON payloads intentionally carry no `event` binding: a
/// JSON frame surfaces as itself (shape stability for JSON-first
/// upstreams, and the spec's implicit `event: message` default never
/// wraps a payload).
fn sse_event_envelope(event: SseEvent, request_id: &str) -> ResponseEnvelope {
let parsed = if event.data.trim().is_empty() {
Value::Null
} else {
serde_json::from_str(&event.data).unwrap_or(Value::String(event.data.clone()))
match serde_json::from_str::<Value>(&event.data) {
Ok(value) => value,
Err(_) => serde_json::json!({
"data": event.data,
"event": event.event,
}),
}
};
ResponseEnvelope::ok(request_id, parsed)
}
@@ -1022,9 +1038,14 @@ fn is_sse_content_type(content_type: &str) -> bool {
content_type.split(';').next().unwrap_or_default().trim() == "text/event-stream"
}
/// A parsed SSE event: the `data:` lines joined with `\n`.
/// A parsed SSE event: the `data:` lines joined with `\n`, plus the
/// frame's `event:` field name when the upstream sent one (WHATWG
/// event-stream semantics: the last `event:` line before the blank
/// line wins; absent → `None`, never the implicit `message` default —
/// consumers distinguish "upstream named it" from "default name").
pub(crate) struct SseEvent {
pub(crate) data: String,
pub(crate) event: Option<String>,
}
#[derive(Debug, thiserror::Error)]
@@ -1053,13 +1074,16 @@ pub(crate) const SSE_EVENT_BUFFER_CAP: usize = 1024 * 1024;
/// a chunk boundary is therefore not corrupted. Framing follows the
/// WHATWG `text/event-stream` draft semantics for what the call
/// protocol needs: lines split on `\n` with optional trailing `\r`;
/// `data:` fields accumulate and join with `\n`; `event:`, `id:`, and
/// `data:` fields accumulate and join with `\n`; `event:` is captured
/// per frame (last one before the blank line wins) and surfaced on
/// [`SseEvent::event`] for non-JSON payloads (FWD-17); `id:` and
/// `retry:` fields are accepted and ignored; a blank line dispatches
/// the pending event; a pending event with data is dispatched at EOF.
pub(crate) struct SseParser {
buf: Vec<u8>,
data_lines: Vec<String>,
data_seen: bool,
event_name: Option<String>,
bom_stripped: bool,
}
@@ -1069,6 +1093,7 @@ impl SseParser {
buf: Vec::new(),
data_lines: Vec::new(),
data_seen: false,
event_name: None,
bom_stripped: false,
}
}
@@ -1135,6 +1160,7 @@ impl SseParser {
return if self.data_seen {
self.complete_event()
} else {
self.discard_event();
None
};
}
@@ -1148,6 +1174,8 @@ impl SseParser {
if field == "data" {
self.data_lines.push(value.to_string());
self.data_seen = true;
} else if field == "event" && !value.is_empty() {
self.event_name = Some(value.to_string());
}
None
}
@@ -1156,8 +1184,15 @@ impl SseParser {
self.data_seen = false;
Some(SseEvent {
data: std::mem::take(&mut self.data_lines).join("\n"),
event: self.event_name.take(),
})
}
fn discard_event(&mut self) {
self.data_seen = false;
self.event_name = None;
self.data_lines.clear();
}
}
#[cfg(test)]
@@ -1781,6 +1816,114 @@ mod tests {
assert!(envelopes[1].result.is_ok());
}
/// FWD-17: a valid-JSON payload surfaces as the decoded value
/// itself; a non-JSON payload surfaces as the
/// `{"data", "event"}` wrapper. Numbers-vs-quoted-strings are both
/// valid JSON, so both decode (the raw-text fallback would have
/// erased that distinction).
#[tokio::test]
async fn sse_payload_contract_decodes_json_and_wraps_non_json() {
let base = spawn_responder(TestArc::new(|_parts| {
http_response(
200,
"text/event-stream",
b"data: 123\n\ndata: \"123\"\n\ndata: plain text\n\ndata: {\"n\":1}\n\n".to_vec(),
)
}))
.await;
let stream = forward_stream(
&minimal_client(),
&base,
"/x",
"GET",
&None,
&TestHashMap::new(),
"svc",
&serde_json::json!({"type": "object"}),
&[],
json!({}),
noop_context(),
);
let envelopes = collect_stream(stream).await;
assert_eq!(envelopes.len(), 4);
assert_eq!(envelopes[0].result.clone().unwrap(), json!(123));
assert_eq!(envelopes[1].result.clone().unwrap(), json!("123"));
assert_eq!(
envelopes[2].result.clone().unwrap(),
json!({"data": "plain text", "event": null})
);
assert_eq!(envelopes[3].result.clone().unwrap(), json!({"n": 1}));
}
/// FWD-17: an upstream `event:` field on a non-JSON payload is
/// carried in the wrapper; JSON payloads surface as themselves even
/// under a named event, and the name is reset after dispatch (the
/// second frame must not inherit the first frame's name).
#[tokio::test]
async fn sse_event_field_carrys_on_the_non_json_wrapper_and_resets() {
let base = spawn_responder(TestArc::new(|_parts| {
http_response(
200,
"text/event-stream",
b"event: error\ndata: upstream exploded\n\nevent: custom\ndata: {\"n\":1}\n\ndata: after\n\n"
.to_vec(),
)
}))
.await;
let stream = forward_stream(
&minimal_client(),
&base,
"/x",
"GET",
&None,
&TestHashMap::new(),
"svc",
&serde_json::json!({"type": "object"}),
&[],
json!({}),
noop_context(),
);
let envelopes = collect_stream(stream).await;
assert_eq!(envelopes.len(), 3);
assert_eq!(
envelopes[0].result.clone().unwrap(),
json!({"data": "upstream exploded", "event": "error"})
);
assert_eq!(envelopes[1].result.clone().unwrap(), json!({"n": 1}));
assert_eq!(
envelopes[2].result.clone().unwrap(),
json!({"data": "after", "event": null}),
"the name must not leak across frames"
);
}
/// FWD-17 parser pin: WHATWG last-wins for repeated `event:` lines
/// in one frame; an `event:`-only frame (no data) dispatches
/// nothing but must not leak its name into a later frame.
#[test]
fn sse_parser_last_event_wins_and_name_does_not_leak_across_frames() {
let mut parser = SseParser::new();
let events = parser
.feed(b"event: a\nevent: b\ndata: x\n\n", false)
.expect("ascii only");
assert_eq!(events.len(), 1);
assert_eq!(events[0].event.as_deref(), Some("b"));
assert_eq!(events[0].data, "x");
let events = parser
.feed(b"event: orphan\n\n", false)
.expect("ascii only");
assert!(events.is_empty(), "no data field, nothing dispatched");
let events = parser.feed(b"data: next\n\n", false).expect("ascii only");
assert_eq!(events.len(), 1);
assert_eq!(
events[0].event.as_deref(),
None,
"an event:-only frame must not leak its name"
);
}
#[tokio::test]
async fn bearer_credential_with_control_character_fails_loudly() {
let base = spawn_responder(TestArc::new(|_parts| {