fix(adapters): subscriptions escape the 30s total timeout + total SSE byte cap (FWD-15, FWD-14)
FWD-15: forward_stream now sends through SharedHttpClient::stream_client — a client derived from the same config with the total request timeout removed and connect + read timeouts retained. reqwest 0.13's per-request override can lengthen a client-level total timeout but never clear it (request-scoped None falls back to the client default), so the derived client is the only correct mechanism. Both clients rebuild-and-swap together atomically (FWD-12). A healthy >30s subscription survives; the read timeout stays as the staleness guard, matching the gateway's deadline: None dispatch contract (alkcall ADR-021). FWD-14: the streaming branch enforces a total streamed-bytes cap per subscription (HttpClientConfig::stream_total_byte_cap, default 1 GiB), accumulated across every chunk fed to the SSE parser; exceeding it terminates with a single terminal HTTP_413 error envelope. The SSE line-cap check moved before extend_from_slice so the reassembly buffer can never exceed the cap. Removing the total timeout without this cap would open an unbounded-memory window, so both land together. Wire tests: keepalive trickle past a scaled total-timeout deadline keeps delivering; over-cap stream terminates with exactly one terminal error; parser boundary tests for pre-extend cap checks. Verified: cargo test (302+5), --all-features (373+41), clippy --all-targets -D warnings (default + all-features), fmt --check, doc --no-deps clean.
This commit is contained in:
+331
-23
@@ -5,6 +5,41 @@
|
||||
//! SSE frame parser for streaming (SSE → `ResponseEnvelope`) handlers
|
||||
//! (ADR-049).
|
||||
//!
|
||||
//! # Timeout / byte-bound policy for streaming forwards (FWD-15, FWD-14)
|
||||
//!
|
||||
//! The two send paths impose deliberately different bounds, mirroring
|
||||
//! the gateway's dispatch contract (alkcall ADR-021: Once-op invokes
|
||||
//! carry a 30 s deadline; `HandlerKind::Stream` invokes set
|
||||
//! `deadline: None` — subscriptions are unbounded in *time* by design):
|
||||
//!
|
||||
//! - **`forward`** (request/response) sends through the shared client's
|
||||
//! total request timeout (30 s default), so a forwarded call fails
|
||||
//! before its caller does.
|
||||
//! - **`forward_stream`** (subscriptions) sends through
|
||||
//! [`SharedHttpClient::stream_client`] — the same config minus the
|
||||
//! total request timeout, connect + read timeouts retained. A healthy
|
||||
//! subscription longer than 30 s survives; the read timeout remains
|
||||
//! the stall guard on upstream staleness (it resets on every body
|
||||
//! byte, so a keep-alive-emitting source lives as long as it keeps
|
||||
//! the connection warm — and quiet silence past the read timeout
|
||||
//! still terminates it). reqwest 0.13's per-request timeout override
|
||||
//! can lengthen but never clear a client-level total timeout, hence
|
||||
//! the derived client rather than an extension override.
|
||||
//!
|
||||
//! Unbounded time without a byte bound would let a hostile upstream
|
||||
//! stream well-formed 1-MiB-line events forever (the 1 MiB SSE line cap
|
||||
//! bounds one line, not the stream), so the streaming branch also
|
||||
//! enforces a total streamed-bytes cap per subscription —
|
||||
//! [`crate::client::HttpClientConfig::stream_total_byte_cap`], 1 GiB by
|
||||
//! default, accumulated across every chunk fed to the SSE parser.
|
||||
//! Exceeding it terminates the stream with a single terminal error
|
||||
//! envelope (the stream-ends semantics: one error frame, then end —
|
||||
//! matching the other terminal arms). The line-cap check runs before
|
||||
//! the buffer grows, so the reassembly buffer can never exceed the
|
||||
//! line cap.
|
||||
//!
|
||||
//! # Credentials and input routing
|
||||
//!
|
||||
//! The forwarding handler is the no-env-vars credential injection point
|
||||
//! (ADR-014): it reads `OperationContext.capabilities`, never
|
||||
//! `std::env::var`. Imported error codes are `HTTP_<status>` to avoid
|
||||
@@ -820,10 +855,11 @@ pub(crate) fn forward_stream(
|
||||
|
||||
let request_id_stream = request_id.clone();
|
||||
let error_status_codes_stream = error_status_codes.clone();
|
||||
let stream_byte_cap = http_client.config().stream_total_byte_cap;
|
||||
|
||||
let init = async move {
|
||||
let request_builder = http_client
|
||||
.client()
|
||||
.stream_client()
|
||||
.request(http_method, url.as_str())
|
||||
.headers(headers)
|
||||
.header(ACCEPT, "text/event-stream");
|
||||
@@ -879,32 +915,67 @@ pub(crate) fn forward_stream(
|
||||
let request_id_inner = request_id.clone();
|
||||
Box::pin(
|
||||
stream::unfold(
|
||||
(response.bytes_stream(), SseParser::new(), false),
|
||||
move |(mut bytes, mut parser, broken)| {
|
||||
(
|
||||
response.bytes_stream(),
|
||||
SseParser::new(),
|
||||
false,
|
||||
0u64,
|
||||
),
|
||||
move |(mut bytes, mut parser, broken, mut total_bytes)| {
|
||||
let request_id = request_id_inner.clone();
|
||||
async move {
|
||||
if broken {
|
||||
return None;
|
||||
}
|
||||
match bytes.next().await {
|
||||
Some(Ok(chunk)) => match parser.feed(&chunk, false) {
|
||||
Ok(events) => {
|
||||
let envelopes: Vec<ResponseEnvelope> = events
|
||||
.into_iter()
|
||||
.map(|e| sse_event_envelope(e, &request_id))
|
||||
.collect();
|
||||
Some((envelopes, (bytes, parser, false)))
|
||||
}
|
||||
Err(err) => {
|
||||
let error = CallError::internal(format!(
|
||||
"SSE parse error: {err}"
|
||||
));
|
||||
Some((
|
||||
Some(Ok(chunk)) => {
|
||||
let chunk_len = chunk.len() as u64;
|
||||
if stream_byte_cap > 0
|
||||
&& total_bytes.saturating_add(chunk_len)
|
||||
> stream_byte_cap
|
||||
{
|
||||
let error = CallError::new(
|
||||
"HTTP_413",
|
||||
format!(
|
||||
"upstream SSE stream exceeded the {stream_byte_cap}-byte total streamed-bytes cap on a subscription operation"
|
||||
),
|
||||
false,
|
||||
);
|
||||
return Some((
|
||||
vec![ResponseEnvelope::error(
|
||||
request_id, error,
|
||||
)],
|
||||
(bytes, parser, true),
|
||||
))
|
||||
(bytes, parser, true, total_bytes),
|
||||
));
|
||||
}
|
||||
total_bytes = total_bytes.saturating_add(chunk_len);
|
||||
match parser.feed(&chunk, false) {
|
||||
Ok(events) => {
|
||||
let envelopes: Vec<ResponseEnvelope> =
|
||||
events
|
||||
.into_iter()
|
||||
.map(|e| {
|
||||
sse_event_envelope(
|
||||
e, &request_id,
|
||||
)
|
||||
})
|
||||
.collect();
|
||||
Some((
|
||||
envelopes,
|
||||
(bytes, parser, false, total_bytes),
|
||||
))
|
||||
}
|
||||
Err(err) => {
|
||||
let error = CallError::internal(format!(
|
||||
"SSE parse error: {err}"
|
||||
));
|
||||
Some((
|
||||
vec![ResponseEnvelope::error(
|
||||
request_id, error,
|
||||
)],
|
||||
(bytes, parser, true, total_bytes),
|
||||
))
|
||||
}
|
||||
}
|
||||
},
|
||||
Some(Err(err)) => {
|
||||
@@ -913,7 +984,7 @@ pub(crate) fn forward_stream(
|
||||
));
|
||||
Some((
|
||||
vec![ResponseEnvelope::error(request_id, error)],
|
||||
(bytes, parser, true),
|
||||
(bytes, parser, true, total_bytes),
|
||||
))
|
||||
}
|
||||
None => match parser.feed(&[], true) {
|
||||
@@ -922,7 +993,10 @@ pub(crate) fn forward_stream(
|
||||
.into_iter()
|
||||
.map(|e| sse_event_envelope(e, &request_id))
|
||||
.collect();
|
||||
Some((envelopes, (bytes, parser, true)))
|
||||
Some((
|
||||
envelopes,
|
||||
(bytes, parser, true, total_bytes),
|
||||
))
|
||||
}
|
||||
_ => None,
|
||||
},
|
||||
@@ -963,7 +1037,12 @@ pub(crate) enum SseParseError {
|
||||
/// buffer. A single event (all `data:` lines plus framing) must fit
|
||||
/// within this budget; a stream emitting a longer partial line — or an
|
||||
/// unterminated event — trips `SseParseError::BufferOverflow` instead
|
||||
/// of buffering without bound.
|
||||
/// of buffering without bound. The check runs *before* the buffer
|
||||
/// takes a chunk's bytes, so the reassembly buffer can never exceed
|
||||
/// the cap (FWD-14). This bounds one *line*, not the
|
||||
/// stream; the per-subscription total is
|
||||
/// [`HttpClientConfig::stream_total_byte_cap`], enforced by the
|
||||
/// `forward_stream` unfold across every `feed`.
|
||||
pub(crate) const SSE_EVENT_BUFFER_CAP: usize = 1024 * 1024;
|
||||
|
||||
/// Incremental byte-level SSE frame parser.
|
||||
@@ -999,10 +1078,10 @@ impl SseParser {
|
||||
/// `eof`, also dispatches a pending event if it carries data
|
||||
/// lines, and flushes the buffer.
|
||||
pub(crate) fn feed(&mut self, chunk: &[u8], eof: bool) -> Result<Vec<SseEvent>, SseParseError> {
|
||||
self.buf.extend_from_slice(chunk);
|
||||
if self.buf.len() > SSE_EVENT_BUFFER_CAP {
|
||||
if self.buf.len().saturating_add(chunk.len()) > SSE_EVENT_BUFFER_CAP {
|
||||
return Err(SseParseError::BufferOverflow);
|
||||
}
|
||||
self.buf.extend_from_slice(chunk);
|
||||
let mut events = Vec::new();
|
||||
let mut start = 0usize;
|
||||
while let Some(nl) = self.buf[start..].iter().position(|&b| b == b'\n') {
|
||||
@@ -1492,6 +1571,66 @@ mod tests {
|
||||
builder.body(body).expect("static response builds")
|
||||
}
|
||||
|
||||
/// Spawns a raw-TCP responder whose response head is written from a
|
||||
/// status/content-type pair, then hands the socket to an async
|
||||
/// writer so a test can trickle SSE frames over time (the
|
||||
/// wire-level seam the FWD-15/FWD-14 tests need: a stream that stays
|
||||
/// open past a deadline, or dribbles bytes toward a cap).
|
||||
async fn spawn_sse_responder_with_writer<F, Fut>(head: &str, writer: F) -> String
|
||||
where
|
||||
F: FnOnce(tokio::net::tcp::OwnedWriteHalf) -> Fut + Send + 'static,
|
||||
Fut: std::future::Future<Output = ()> + Send,
|
||||
{
|
||||
let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
|
||||
.await
|
||||
.expect("bind");
|
||||
let addr = listener.local_addr().expect("local addr");
|
||||
let head = head.to_string();
|
||||
tokio::spawn(async move {
|
||||
use tokio::io::{AsyncReadExt, AsyncWriteExt};
|
||||
let Ok((mut sock, _)) = listener.accept().await else {
|
||||
return;
|
||||
};
|
||||
let mut buf = vec![0u8; 8192];
|
||||
loop {
|
||||
let read = sock.read(&mut buf).await.unwrap_or(0);
|
||||
if read == 0 || String::from_utf8_lossy(&buf).contains("\r\n\r\n") {
|
||||
break;
|
||||
}
|
||||
}
|
||||
let _ = sock.write_all(head.as_bytes()).await;
|
||||
let _ = sock.flush().await;
|
||||
let (_, write_half) = sock.into_split();
|
||||
writer(write_half).await;
|
||||
});
|
||||
format!("http://{addr}")
|
||||
}
|
||||
|
||||
fn streaming_client(total_byte_cap: u64) -> TestArc<SharedHttpClient> {
|
||||
TestArc::new(
|
||||
SharedHttpClient::new(crate::client::HttpClientConfig {
|
||||
stream_total_byte_cap: total_byte_cap,
|
||||
..crate::client::HttpClientConfig::default()
|
||||
})
|
||||
.expect("client builds"),
|
||||
)
|
||||
}
|
||||
|
||||
/// Builds a client whose configured request/read timeout is the
|
||||
/// (test-scaled) old 30 s deadline: the FWD-15 wire test sends
|
||||
/// through it and asserts the stream outlives that deadline.
|
||||
fn client_with_timeout(timeout: Duration) -> TestArc<SharedHttpClient> {
|
||||
TestArc::new(
|
||||
SharedHttpClient::new(crate::client::HttpClientConfig {
|
||||
request_timeout: Some(timeout),
|
||||
connect_timeout: Some(Duration::from_secs(5)),
|
||||
read_timeout: Some(timeout),
|
||||
..crate::client::HttpClientConfig::default()
|
||||
})
|
||||
.expect("client builds"),
|
||||
)
|
||||
}
|
||||
|
||||
async fn call_forward(base_url: &str, ctx: OperationContext) -> ResponseEnvelope {
|
||||
call_forward_authed(base_url, ctx, &None).await
|
||||
}
|
||||
@@ -1834,4 +1973,173 @@ mod tests {
|
||||
other => panic!("expected HTTP_500, got {other:?}"),
|
||||
}
|
||||
}
|
||||
|
||||
/// FWD-15 wire test (deadline-scaled): the responder trickles
|
||||
/// `: keepalive` comments and one `data:` event past the configured
|
||||
/// total request timeout (a scaled stand-in for the 30 s default
|
||||
/// the streaming path used to inherit); the subscription must still
|
||||
/// be delivering events after that deadline has passed. With the
|
||||
/// fix, `forward_stream` sends through the derived no-total-timeout
|
||||
/// client, so the stream survives; without it, reqwest 0.13's total
|
||||
/// timeout rides into the body stream and kills the subscription at
|
||||
/// the deadline.
|
||||
#[tokio::test]
|
||||
async fn stream_survives_past_the_total_request_timeout() {
|
||||
let timeout = Duration::from_millis(500);
|
||||
let keepalive = Duration::from_millis(200);
|
||||
let total_keepalives = 6u32;
|
||||
let start = std::time::Instant::now();
|
||||
let head = "HTTP/1.1 200 OK\r\ncontent-type: text/event-stream\r\n\r\n";
|
||||
let base = spawn_sse_responder_with_writer(head, move |mut sock| async move {
|
||||
use tokio::io::AsyncWriteExt;
|
||||
for _ in 0..total_keepalives {
|
||||
tokio::time::sleep(keepalive).await;
|
||||
let _ = sock.write_all(b": keepalive\n\n").await;
|
||||
let _ = sock.flush().await;
|
||||
}
|
||||
let _ = sock.write_all(b"data: {\"late\":true}\n\n").await;
|
||||
let _ = sock.flush().await;
|
||||
let _ = sock.shutdown().await;
|
||||
})
|
||||
.await;
|
||||
let client = client_with_timeout(timeout);
|
||||
let stream = forward_stream(
|
||||
&client,
|
||||
&base,
|
||||
"/x",
|
||||
"GET",
|
||||
&None,
|
||||
&TestHashMap::new(),
|
||||
"svc",
|
||||
&serde_json::json!({"type": "object"}),
|
||||
&[],
|
||||
json!({}),
|
||||
noop_context(),
|
||||
);
|
||||
tokio::pin!(stream);
|
||||
let crossed = tokio::time::timeout(Duration::from_secs(5), stream.next())
|
||||
.await
|
||||
.expect("the stream must deliver within the test budget")
|
||||
.expect("stream must not end without the data event");
|
||||
assert!(
|
||||
start.elapsed() > timeout,
|
||||
"the event must arrive after the old total-request deadline (elapsed {:?}, timeout {:?})",
|
||||
start.elapsed(),
|
||||
timeout
|
||||
);
|
||||
match crossed.result {
|
||||
Ok(value) => assert_eq!(value, json!({"late": true})),
|
||||
other => panic!("expected the post-deadline data event, got {other:?}"),
|
||||
}
|
||||
assert!(
|
||||
tokio::time::timeout(Duration::from_millis(500), stream.next())
|
||||
.await
|
||||
.unwrap_or(None)
|
||||
.is_none(),
|
||||
"responder closed after the data event; stream must end"
|
||||
);
|
||||
}
|
||||
|
||||
/// FWD-14 wire test: a stream whose total bytes exceed the
|
||||
/// configured per-subscription cap terminates with exactly one
|
||||
/// terminal error envelope (the stream-ends semantics: error frame,
|
||||
/// then end).
|
||||
#[tokio::test]
|
||||
async fn stream_exceeding_total_byte_cap_terminates_with_one_error() {
|
||||
let cap = 4096u64;
|
||||
let chunk = vec![b'a'; 1024];
|
||||
let head = "HTTP/1.1 200 OK\r\ncontent-type: text/event-stream\r\n\r\n";
|
||||
let base = spawn_sse_responder_with_writer(head, move |mut sock| async move {
|
||||
use tokio::io::AsyncWriteExt;
|
||||
for _ in 0..64 {
|
||||
let _ = sock.write_all(b"data: ").await;
|
||||
let _ = sock.write_all(&chunk).await;
|
||||
let _ = sock.write_all(b"\n\n").await;
|
||||
let _ = sock.flush().await;
|
||||
}
|
||||
let _ = sock.shutdown().await;
|
||||
})
|
||||
.await;
|
||||
let client = streaming_client(cap);
|
||||
let stream = forward_stream(
|
||||
&client,
|
||||
&base,
|
||||
"/x",
|
||||
"GET",
|
||||
&None,
|
||||
&TestHashMap::new(),
|
||||
"svc",
|
||||
&serde_json::json!({"type": "object"}),
|
||||
&[],
|
||||
json!({}),
|
||||
noop_context(),
|
||||
);
|
||||
let envelopes = collect_stream(stream).await;
|
||||
assert!(!envelopes.is_empty(), "events before the cap still flow");
|
||||
assert!(
|
||||
envelopes[..envelopes.len() - 1]
|
||||
.iter()
|
||||
.all(|e| e.result.is_ok()),
|
||||
"every envelope before the terminal one is an event"
|
||||
);
|
||||
let terminal = envelopes.last().expect("terminal envelope present");
|
||||
match &terminal.result {
|
||||
Err(err) => {
|
||||
assert_eq!(err.code, "HTTP_413");
|
||||
assert!(err.message.contains("total streamed-bytes cap"));
|
||||
}
|
||||
other => panic!("expected the terminal cap error, got {other:?}"),
|
||||
}
|
||||
let ok_count = envelopes[..envelopes.len() - 1].len();
|
||||
assert_eq!(
|
||||
envelopes.len(),
|
||||
ok_count + 1,
|
||||
"exactly one terminal error envelope after the last event"
|
||||
);
|
||||
}
|
||||
|
||||
/// FWD-14 parser-boundary test: for a never-newline feed the
|
||||
/// cap check fires *before* the buffer takes the chunk's bytes, so
|
||||
/// the buffered bytes stay at or below the cap — the pre-fix shape
|
||||
/// (extend first, check after) buffered past the cap before erroring.
|
||||
/// A full-cap buffer remains legal (the cap is inclusive) as long as
|
||||
/// the arriving chunk completes a line.
|
||||
#[test]
|
||||
fn line_cap_trips_before_the_buffer_takes_the_overshooting_chunk() {
|
||||
let mut parser = SseParser::new();
|
||||
let seed = vec![b'x'; SSE_EVENT_BUFFER_CAP + 1];
|
||||
let oversized = parser.feed(&seed, false);
|
||||
assert!(
|
||||
matches!(oversized, Err(SseParseError::BufferOverflow)),
|
||||
"a single over-cap line trips at the pre-extend check"
|
||||
);
|
||||
let mut parser = SseParser::new();
|
||||
let half = vec![b'x'; SSE_EVENT_BUFFER_CAP / 2];
|
||||
let first = parser.feed(&half, false);
|
||||
assert!(first.is_ok(), "partial line under the cap buffers fine");
|
||||
let second = parser.feed(&seed, false);
|
||||
assert!(
|
||||
matches!(second, Err(SseParseError::BufferOverflow)),
|
||||
"the chunk that would push past the cap is rejected before extend"
|
||||
);
|
||||
let mut parser = SseParser::new();
|
||||
let at_cap = vec![b'x'; SSE_EVENT_BUFFER_CAP - 2];
|
||||
let ok = parser.feed(&at_cap, false);
|
||||
assert!(ok.is_ok(), "a partial line under the cap is legal");
|
||||
let framing = parser.feed(b"\n\n", false);
|
||||
assert!(
|
||||
framing.is_ok(),
|
||||
"the newline pair completes the event without tripping the cap"
|
||||
);
|
||||
let next = parser.feed(&vec![b'y'; SSE_EVENT_BUFFER_CAP], false);
|
||||
assert!(
|
||||
next.is_ok(),
|
||||
"the dispatched event drained the buffer; a fresh full-cap line is legal again"
|
||||
);
|
||||
let over = parser.feed(b"z", false);
|
||||
assert!(
|
||||
matches!(over, Err(SseParseError::BufferOverflow)),
|
||||
"one byte past a full buffer still trips before extend"
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user