diff --git a/src/adapters/forward.rs b/src/adapters/forward.rs index a851431..a2584e1 100644 --- a/src/adapters/forward.rs +++ b/src/adapters/forward.rs @@ -2361,6 +2361,54 @@ mod tests { assert_eq!(envelopes[3].result.clone().unwrap(), json!({"n": 1})); } + /// SSE framing edge: a CRLF split across TCP chunks (chunk 1 ends + /// with the `\r`, chunk 2 opens with the `\n`) still frames as one + /// line — the reassembly window waits for the `\n` before parsing, + /// not treating the `\r` as a terminator (so the bare `\r` holds + /// until the `\n` arrives, then frames a single line). + #[test] + fn sse_parser_crlf_split_across_chunks_frames_one_line() { + let mut parser = SseParser::new(); + let first = parser.feed(b"data: {\"n\":1}\r", false).expect("chunk 1"); + assert!(first.is_empty(), "the bare \\r does not dispatch"); + let second = parser + .feed(b"\ndata: {\"n\":2}\r\n\r\n", false) + .expect("chunk 2"); + assert_eq!(second.len(), 1, "the joined CRLF framed exactly one event"); + assert_eq!( + second[0].data, "{\"n\":1}\n{\"n\":2}", + "the \\r held as line content until the arriving \\n framed it, \ + joining both data lines per WHATWG accumulation" + ); + let rest = parser.feed(b"data: {\"n\":2}\n\n", true).expect("tail"); + assert_eq!(rest.len(), 1); + assert_eq!(rest[0].data, "{\"n\":2}", "the second frame is unaffected"); + } + + /// A line that is not valid UTF-8 is dropped (WHATWG decode-failure + /// semantics for this parser): no event, no panic, and the framing + /// continues — the following valid frame dispatches normally. + #[test] + fn sse_parser_drops_invalid_utf8_lines_and_keeps_framing() { + let mut parser = SseParser::new(); + let events = parser + .feed(b"data: \xff\xfe\xfd\n\ndata: {\"ok\":true}\n\n", false) + .expect("ascii framing is valid"); + assert_eq!(events.len(), 1, "the invalid-UTF8 data line was dropped"); + assert_eq!(events[0].data, "{\"ok\":true}"); + assert_eq!(events[0].event, None); + + let mut parser = SseParser::new(); + let events = parser + .feed(b"data: good\r\ndata: \xf0\x28\x8c\x28\r\n\r\n", false) + .expect("CRLF framing is valid"); + assert_eq!(events.len(), 1); + assert_eq!( + events[0].data, "good", + "the invalid line contributes nothing to the joined data" + ); + } + /// 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 @@ -2983,6 +3031,50 @@ mod tests { } } + /// The streaming analog of the Once-path build-error envelope: a + /// `forward_stream` invocation with an invalid input yields exactly + /// one `INVALID_INPUT` envelope and the stream ends, before any + /// upstream contact — the responder counts requests and the test + /// asserts the count stays zero (a regression returning an empty + /// stream, silently swallowing the error, would be the worst + /// failure shape for a subscriptions consumer). + #[tokio::test] + async fn stream_build_error_yields_one_invalid_input_envelope_with_zero_upstream_contact() { + let base = spawn_responder(TestArc::new(|_parts| { + panic!("the rejected invocation must never reach the upstream"); + })) + .await; + for input in [ + json!({"debug": true}), + json!({"id": {"deeply": {"nested": "object"}}}), + ] { + let stream = forward_stream( + &minimal_client(), + &base, + "/x/{id}", + "GET", + &None, + &TestHashMap::new(), + "svc", + &serde_json::json!({ + "type": "object", + "properties": {"id": {"type": "string"}} + }), + &[], + input, + noop_context(), + ); + let envelopes = collect_stream(stream).await; + assert_eq!(envelopes.len(), 1, "one envelope, then the stream ends"); + match &envelopes[0].result { + Err(err) => { + assert_eq!(err.code, "INVALID_INPUT", "message was: {}", err.message); + } + other => panic!("expected one INVALID_INPUT envelope, got {other:?}"), + } + } + } + /// COV-10 (b): the responder emits one complete `data:` frame, lets /// the client consume it, then kills the connection before the /// declared body length is met — a genuine mid-stream transport diff --git a/src/adapters/from_jsonschema.rs b/src/adapters/from_jsonschema.rs index 8870fff..208a007 100644 --- a/src/adapters/from_jsonschema.rs +++ b/src/adapters/from_jsonschema.rs @@ -591,6 +591,63 @@ mod tests { assert_eq!(collected[1].request_id, "req-13"); } + /// The streaming analog of the Once-path build-error envelope: a + /// Sub op invoked with an invalid input (undeclared key, or a + /// non-scalar value bound to a path placeholder) yields exactly one + /// `INVALID_INPUT` envelope, then the stream ends — pre-send, so + /// the upstream receives zero requests. An empty stream (a silently + /// swallowed error) would be the worst failure shape for a + /// subscriptions consumer; the non-empty, single-error shape is the + /// assertion. + #[tokio::test] + async fn integration_sse_subscription_invalid_input_yields_one_error_envelope_and_ends() { + for (input, expected_fragment) in [ + ( + serde_json::json!({"debug": true}), + "not declared by the operation's input schema", + ), + ( + serde_json::json!({"id": {"nested": "object"}}), + "must be a scalar", + ), + ] { + let base = spawn_echo_server(200, "data: {\"n\":1}\n\n", "text/event-stream").await; + let adapter = FromJsonSchema::new( + test_spec("svc/stream", OperationType::Sub), + test_config("svc", &base), + "/stream/{id}".to_string(), + "POST".to_string(), + test_http_client(), + ) + .unwrap(); + let bundles = adapter.import().await.unwrap(); + let registration = &bundles[0]; + let ctx = noop_context("req-13b", Capabilities::new()); + let stream = match ®istration.handler { + HandlerKind::Stream(h) => h(input.clone(), ctx), + _ => panic!("expected Stream handler"), + }; + let collected: Vec = stream.collect().await; + assert_eq!( + collected.len(), + 1, + "exactly one error envelope, then the stream ends (never an empty-200-subscribe)" + ); + match &collected[0].result { + Err(e) => { + assert_eq!(e.code, "INVALID_INPUT", "message was: {}", e.message); + assert!( + e.message.contains(expected_fragment), + "message `{}` lacks `{expected_fragment}`", + e.message + ); + } + other => panic!("expected INVALID_INPUT, got {other:?}"), + } + assert_eq!(collected[0].request_id, "req-13b"); + } + } + #[test] fn no_env_vars_read_in_build_request() { std::env::set_var("OPENAI_API_KEY", "should-not-be-used"); diff --git a/src/adapters/openapi_spec.rs b/src/adapters/openapi_spec.rs index 5f30ff3..09d7c41 100644 --- a/src/adapters/openapi_spec.rs +++ b/src/adapters/openapi_spec.rs @@ -1204,6 +1204,41 @@ mod tests { assert!(props.get("name").is_some()); } + /// The structural-reject arms at the very top of `from_value` + /// (before any path walking): a non-object document, a + /// missing-`info` object, and a non-object `paths` each fail with + /// `SchemaParse` naming the missing/mistyped member. + #[test] + fn from_value_structural_rejects_name_the_missing_member() { + for (raw, expected_fragment) in [ + (json!("just a string"), "must be a JSON object"), + (json!([1, 2, 3]), "must be a JSON object"), + (json!({"paths": {}}), "missing `info`"), + ( + json!({"info": {"title": "T", "version": "1"}}), + "missing `paths`", + ), + ( + json!({ + "info": {"title": "T", "version": "1"}, + "paths": ["/x"] + }), + "`paths` must be a JSON object", + ), + ] { + match OpenAPISpec::from_value(raw) { + Err(AdapterError::SchemaParse { message }) => { + assert!( + message.contains(expected_fragment), + "message `{message}` lacks `{expected_fragment}`" + ); + } + Ok(_) => panic!("the malformed document must be rejected"), + other => panic!("expected SchemaParse, got {other:?}"), + } + } + } + #[test] fn request_body_self_ref_fails_import_not_silent_bodyless_op() { let doc = r##"{ diff --git a/src/server/adapter.rs b/src/server/adapter.rs index 2ade19c..855e73c 100644 --- a/src/server/adapter.rs +++ b/src/server/adapter.rs @@ -765,6 +765,29 @@ mod tests { let _ = server_task.await; } + /// The accept-path connection-failure arm: a `Connection` whose + /// single underlying stream is already gone (closed by the dial + /// side) yields `StreamError::ConnectionClosed` from `accept_bi`, + /// which `stream_error_to_handler` maps to + /// `HandlerError::ConnectionClosed` — the error the consumer's + /// accept loop observes for a peer that died before the first + /// stream arrived. + #[tokio::test] + async fn accept_bi_failure_maps_to_handler_error_connection_closed() { + let adapter = HttpAdapter::new(provider(), empty_registry()); + let (client, server) = tokio::io::duplex(64 * 1024); + let conn = Connection::from_bidi(server, b"http/1.1".to_vec(), None); + let auth = AuthContext::anonymous(b"http/1.1"); + + drop(client); + conn.close(0, "gone before the first stream"); + let result = ProtocolHandler::handle(&adapter, conn, &auth).await; + match result { + Err(HandlerError::ConnectionClosed) => {} + other => panic!("expected HandlerError::ConnectionClosed, got {other:?}"), + } + } + #[tokio::test] async fn healthz_served_by_the_adapter_over_duplex() { let adapter = HttpAdapter::new(provider(), empty_registry()); diff --git a/src/websocket/byte_adapter.rs b/src/websocket/byte_adapter.rs index dd4501b..83817a4 100644 --- a/src/websocket/byte_adapter.rs +++ b/src/websocket/byte_adapter.rs @@ -1188,6 +1188,54 @@ mod tests { .expect("pump emitted the cap-sized chunk"); } + /// WS-14's rejection leg: a single `AsyncWrite` call one byte *over* + /// `PENDING_BUFFER_CAP` fails synchronously with `InvalidData`, + /// naming the cap, before any byte enters the write queue — the mux + /// sees the stream error directly instead of the fault reaching the + /// wire (the = cap acceptance is the test above; this is the + /// over-cap rejection the WS-14 move introduced). + #[tokio::test] + async fn tungstenite_write_one_byte_over_the_cap_is_rejected_with_invalid_data() { + let (client_io, mut server_io) = tokio::io::duplex(64); + let ws = tokio_tungstenite::WebSocketStream::from_raw_socket( + client_io, + tokio_tungstenite::tungstenite::protocol::Role::Client, + None, + ) + .await; + let (mut stream, _pumps) = split_tungstenite_to_bytes(ws); + + let err = stream + .write_all(&vec![0u8; PENDING_BUFFER_CAP + 1]) + .await + .expect_err("the over-cap write must be rejected"); + assert_eq!(err.kind(), io::ErrorKind::InvalidData); + assert!( + err.to_string().contains("pending buffer cap"), + "error names the cap: {err}" + ); + assert!( + err.to_string().contains(&PENDING_BUFFER_CAP.to_string()), + "error carries the cap value: {err}" + ); + + // The rejection must not poison the stream: an exactly-at-cap + // write still flows through the pump afterwards. + let mut chunk = vec![0u8; PENDING_BUFFER_CAP]; + chunk[4..8].copy_from_slice(&(PENDING_BUFFER_CAP as u32 - 8).to_be_bytes()); + stream + .write_all(&chunk) + .await + .expect("at-cap write after the rejection is accepted"); + stream.flush().await.expect("flush"); + tokio::time::timeout(std::time::Duration::from_secs(30), async { + let mut buf = [0u8; 2]; + server_io.read_exact(&mut buf).await.expect("chunk read") + }) + .await + .expect("pump still emits after a rejected over-cap write"); + } + /// A valid chunk (length field ≤ `MAX_CHUNK_LEN`) still parses and /// flushes through the pump after the caps landed — the validation /// must not false-positive on well-framed traffic. (The pump-side @@ -1268,6 +1316,54 @@ mod tests { assert_eq!(&buf, b"from-peer"); } + /// The read-pump's demux-gone break: with the byte-stream side + /// (the `read_rx` holder) dropped while the WS is still open, the + /// next inbound binary message fails its `read_tx.send` and ends + /// the pump — it must not park on the now-unreceivable channel. + /// The peer holds the socket open (no EOF, idle window far beyond + /// the test budget), so the only exit for the pump is the failed + /// send itself; a regression that parks here hangs until the test + /// deadline rather than ending the task. + #[tokio::test] + async fn read_pump_breaks_when_the_byte_stream_side_is_dropped() { + let (client_io, server_io) = tokio::io::duplex(1 << 16); + let ws = tokio_tungstenite::WebSocketStream::from_raw_socket( + client_io, + tokio_tungstenite::tungstenite::protocol::Role::Client, + None, + ) + .await; + let (_stream, pumps) = split_tungstenite_to_bytes(ws); + drop(_stream); + + let peer = tokio::spawn(async move { + let peer = tokio_tungstenite::WebSocketStream::from_raw_socket( + server_io, + tokio_tungstenite::tungstenite::protocol::Role::Server, + None, + ) + .await; + let (mut sink, mut reader) = peer.split(); + use futures::{SinkExt, StreamExt}; + sink.send(tokio_tungstenite::tungstenite::Message::Binary( + b"after-drop".to_vec().into(), + )) + .await + .expect("peer send"); + let _ = tokio::time::timeout(std::time::Duration::from_secs(10), reader.next()).await; + }); + + let ended = tokio::time::timeout(std::time::Duration::from_secs(5), pumps.read_task) + .await + .expect("the pump must end on the demux-gone break, not park"); + assert!( + ended.is_ok(), + "the pump ends by itself (the break), not by abort: {ended:?}" + ); + + peer.abort(); + } + /// The lossless read-EOF signal on the tungstenite path (COV-03, /// the WS-11 constraint): the peer disappearing surfaces on /// `pumps.read_eof()` as `watch` = true — observable both before diff --git a/src/websocket/upgrade.rs b/src/websocket/upgrade.rs index af102b1..804a077 100644 --- a/src/websocket/upgrade.rs +++ b/src/websocket/upgrade.rs @@ -341,15 +341,6 @@ pub struct WsTimeouts { pub write: Option, } -impl Default for WsTimeouts { - fn default() -> Self { - Self { - idle: Some(crate::websocket::DEFAULT_WS_IDLE_TIMEOUT), - write: None, - } - } -} - /// The upgrade handler. Requires the resolved identity in request /// extensions (stashed by [`ws_bearer_auth`]) — a WS session without /// an identity cannot run `AccessControl::check`. diff --git a/tasks/infra/review-002-fu-stream-error-coverage.md b/tasks/infra/review-002-fu-stream-error-coverage.md index 175ee32..a532f22 100644 --- a/tasks/infra/review-002-fu-stream-error-coverage.md +++ b/tasks/infra/review-002-fu-stream-error-coverage.md @@ -1,7 +1,7 @@ --- id: review-002-fu-stream-error-coverage name: Streaming error-arm coverage — forward_stream invalid input, PEM read-failure, over-cap poll_write mirror (post-bulk coverage gap) -status: pending +status: done depends_on: [] scope: narrow risk: low diff --git a/tests/client_tls.rs b/tests/client_tls.rs index f19ac2c..f38d68b 100644 --- a/tests/client_tls.rs +++ b/tests/client_tls.rs @@ -374,6 +374,73 @@ async fn reload_to_a_ca_bundle_backed_client_succeeds() { tokio::time::sleep(Duration::from_millis(1)).await; } +/// COV-11/CLI-03 read-failure arms (the sync `SharedHttpClient::new` +/// path): a `ca_bundle` path that does not exist fails the build with +/// `CaBundleRead` carrying the path — the unreadable-file complement to +/// the parse-failure tests above. +#[test] +fn nonexistent_ca_bundle_path_fails_ca_bundle_read_with_path() { + let path = std::env::temp_dir().join(format!( + "alkhttp-pem-read-{}-{}-missing.pem", + std::process::id(), + uuid::Uuid::new_v4() + )); + let error = SharedHttpClient::new(client_config(Some(path.clone()), None)) + .expect_err("a nonexistent CA path must fail the build"); + match error { + HttpClientBuildError::CaBundleRead { path: p, .. } => { + assert_eq!(p, path, "the error names the unreadable path"); + } + other => panic!("expected CaBundleRead, got {other:?}"), + } +} + +/// COV-11/CLI-03 read-failure arm (the async `reload` path): a +/// `client_cert` key path that does not exist fails the reload with +/// `ClientCertRead` carrying the unreadable path, and the held clients +/// stay on the previous generation. +#[tokio::test] +async fn reload_with_nonexistent_client_cert_path_fails_client_cert_read() { + let pki = TestPki::generate(); + let (_ca, _cert, dir) = pki.write_config_files(false); + + let http = SharedHttpClient::new(HttpClientConfig::default()).expect("initial client"); + let missing_key = std::env::temp_dir().join(format!( + "alkhttp-pem-read-{}-{}-missing-key.pem", + std::process::id(), + uuid::Uuid::new_v4() + )); + let missing_cert = std::env::temp_dir().join(format!( + "alkhttp-pem-read-{}-{}-missing-cert.pem", + std::process::id(), + uuid::Uuid::new_v4() + )); + let error = http + .reload(client_config( + None, + Some(ClientCertConfig { + cert_pem: missing_cert.clone(), + key_pem: missing_key.clone(), + }), + )) + .await + .expect_err("a nonexistent client-cert key path must fail the reload"); + match error { + HttpClientBuildError::ClientCertRead { path: p, .. } => { + assert_eq!( + p, missing_cert, + "the error names the unreadable cert path (read before the key)" + ); + } + other => panic!("expected ClientCertRead, got {other:?}"), + } + assert!( + http.config().client_cert.is_none(), + "the reload failure keeps the previous generation's clients+config (FWD-12)" + ); + cleanup_dir(&dir); +} + /// COV-11/CLI-03 parse-failure arms: a `ca_bundle` file that parses as /// PEM framing but carries a corrupt section fails the build with /// `CaBundleParse`, carrying the offending path. (Purely non-PEM text