From 57d905046f2dac434853425969eee90e9894897f Mon Sep 17 00:00:00 2001 From: Michael Fara Date: Wed, 9 Sep 2026 12:17:52 +0000 Subject: [PATCH 1/2] Sync request-path and TLS fixes Publish the sanitized implementation changes from internal commits a4f7696 and ab3318b. This includes H2 hot-path sizing, bridge generation sideband, bounded decompression fast paths, configuration-safe CGI routing shortcuts, and lossless monoio-rustls read-ahead preservation. Signed-off-by: Michael Fara --- crates/hj-h2/src/server/mod.rs | 4 + crates/hj-h2/src/server/send.rs | 21 +- crates/httpjet/src/pipeline/e2e.rs | 50 +++ crates/httpjet/src/pipeline/mod.rs | 83 ++++- crates/httpjet/src/pipeline/suffix_routing.rs | 32 +- crates/httpjet/src/serving_generation.rs | 5 +- crates/httpjet/src/state.rs | 21 ++ crates/httpjet/src/uring/bridge.rs | 41 ++- crates/httpjet/src/uring/generation_test.rs | 6 +- crates/httpjet/src/uring/h3.rs | 14 +- crates/httpjet/src/uring/mod.rs | 325 ++++++++++++++---- crates/httpjet/src/uring/otel_test.rs | 1 + crates/httpjet/src/uring/request_body.rs | 106 +++++- vendor/monoio-rustls/src/stream.rs | 85 +++++ 14 files changed, 654 insertions(+), 140 deletions(-) diff --git a/crates/hj-h2/src/server/mod.rs b/crates/hj-h2/src/server/mod.rs index 36bfd67..424d3f9 100644 --- a/crates/hj-h2/src/server/mod.rs +++ b/crates/hj-h2/src/server/mod.rs @@ -789,7 +789,11 @@ pub(super) struct OutQueue { mod completion_tests; impl OutQueue { + #[inline] fn finish_responses(&mut self, success: bool) { + if self.completions.is_empty() { + return; + } for completion in self.completions.drain(..) { completion.finish(if success { hj_core::ResponseEnd::Complete diff --git a/crates/hj-h2/src/server/send.rs b/crates/hj-h2/src/server/send.rs index db5d8f3..770a83e 100644 --- a/crates/hj-h2/src/server/send.rs +++ b/crates/hj-h2/src/server/send.rs @@ -140,7 +140,14 @@ pub(super) fn begin_response( block_scratch: &mut Vec, ) { let (mut head, body) = response.into_parts(); - let mut completion = head.extensions.remove::(); + // ResponseCompletion is opt-in (currently OpenTelemetry). The normal + // production response has no extensions, so avoid a TypeId/hash-table + // lookup on every H2 response. + let mut completion = if head.extensions.is_empty() { + None + } else { + head.extensions.remove::() + }; // §8.2.2: connection-specific ("hop-by-hop") fields are illegal on an h2 response — strip // them before encoding so a backend that emits e.g. `Connection`/`Transfer-Encoding` // (PHP over LSAPI, a proxied upstream) can't produce a malformed frame stream. @@ -223,7 +230,9 @@ pub(super) fn begin_response( let headers_only = |out: &mut OutQueue, completion: Option| { out.frames(|b| write_field_block(b, stream_id, flags::END_STREAM, block, mf)); - out.completions.extend(completion); + if let Some(completion) = completion { + out.completions.push(completion); + } }; let headers_open = |out: &mut OutQueue| { out.frames(|b| write_field_block(b, stream_id, 0, block, mf)); @@ -490,7 +499,9 @@ fn pump_one_frame( st.window -= n as i64; if last { st.done = true; - out.completions.extend(st.completion.take()); + if let Some(completion) = st.completion.take() { + out.completions.push(completion); + } } return true; } @@ -507,7 +518,9 @@ fn pump_one_frame( .write(b) }); st.done = true; - out.completions.extend(st.completion.take()); + if let Some(completion) = st.completion.take() { + out.completions.push(completion); + } return true; } false diff --git a/crates/httpjet/src/pipeline/e2e.rs b/crates/httpjet/src/pipeline/e2e.rs index 45913f0..a86283f 100644 --- a/crates/httpjet/src/pipeline/e2e.rs +++ b/crates/httpjet/src/pipeline/e2e.rs @@ -375,6 +375,56 @@ async fn static_get_serves_litespeed_etag_and_revalidates_to_304() { assert_eq!(body_bytes(resp.into_body()).as_ref(), b"hello after edit\n"); } +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn cgi_route_summary_reloads_and_missing_processor_fails_closed() { + let root = temp_root("cgi-route-summary"); + const SOURCE: &[u8] = b" Option { /// while keeping cache-safety, compression and transport headers under httpjet's /// final control. Every full and on-core response funnel calls this helper. async fn apply_response_transforms(state: &ServerState, ctx: &ReqCtx, resp: &mut Response) { - state.extensions.run_response_transforms(ctx, resp).await; + // The distributed binary ships an empty compile-time registry. Avoid + // constructing and polling its no-op async dispatcher on every response. + if !state.extensions.is_empty() { + state.extensions.run_response_transforms(ctx, resp).await; + } for transform in &state.transforms { transform.transform(ctx, resp).await; } } +/// Default-build entry point: return the pipeline future directly so every +/// request does not pay for a redundant outer async state machine. The OTel +/// build keeps its wrapper below because it may need to establish trace context +/// around that future. +#[cfg(not(feature = "otel"))] +#[allow(clippy::too_many_arguments)] +pub fn handle<'a>( + state: Arc, + listener: &'a str, + peer_ip: IpAddr, + local_addr: std::net::SocketAddr, + peer_port: u16, + is_tls: bool, + peer_unix: bool, + mtls_required: bool, + tls: Option, + proto: Proto, + sni: Option<&'a str>, + req: Request, +) -> impl std::future::Future + 'a { + handle_inner( + state, + listener, + peer_ip, + local_addr, + peer_port, + is_tls, + peer_unix, + mtls_required, + tls, + proto, + sni, + req, + ) +} + +#[cfg(feature = "otel")] +#[allow(clippy::too_many_arguments)] pub async fn handle( state: Arc, listener: &str, @@ -1116,7 +1158,6 @@ pub async fn handle( sni: Option<&str>, req: Request, ) -> Response { - #[cfg(feature = "otel")] if crate::otel::enabled() { let mut req = req; let transport = req @@ -1466,22 +1507,26 @@ async fn handle_inner( ctx.set_env("HTTP_ACCEPT_ENCODING", ae.to_string()); } set_redirect_guard(&mut ctx, req.uri(), &req_host); - match state.extensions.run_pre_handlers(&ctx, &req).await { - Ok(hj_extension::PreHandlerDecision::Continue) => { - dispatch(&state, host_foreign, req_host, &mut ctx, req).await - } - Ok(hj_extension::PreHandlerDecision::Respond(response)) => response, - Err(error) => { - let status = error.status(); - tracing::warn!( - request_id = %ctx.request_id, - vhost = %ctx.vhost_name, - status = status.as_u16(), - error = %error, - "compile-time pre-handler extension failed" - ); - error_page(status) + if state.extensions.has_pre_handlers() { + match state.extensions.run_pre_handlers(&ctx, &req).await { + Ok(hj_extension::PreHandlerDecision::Continue) => { + dispatch(&state, host_foreign, req_host, &mut ctx, req).await + } + Ok(hj_extension::PreHandlerDecision::Respond(response)) => response, + Err(error) => { + let status = error.status(); + tracing::warn!( + request_id = %ctx.request_id, + vhost = %ctx.vhost_name, + status = status.as_u16(), + error = %error, + "compile-time pre-handler extension failed" + ); + error_page(status) + } } + } else { + dispatch(&state, host_foreign, req_host, &mut ctx, req).await } }; @@ -3018,7 +3063,9 @@ async fn dispatch( // ---- 7. Suffix routing: LSAPI (php/html) or static ------------------- // Reuse the split resolved once above (B5) — `cur_path` is unchanged since. if let Some((script_abs, script_name, path_info)) = script_split { - if let Some(handler_name) = fastcgi_handler_for_script(ctx, &script_abs) { + if state.has_cgi_script_routes + && let Some(handler_name) = fastcgi_handler_for_script(ctx, &script_abs) + { let Some(handler) = state .fastcgi_handler(&ctx.vhost_name, handler_name) .cloned() diff --git a/crates/httpjet/src/pipeline/suffix_routing.rs b/crates/httpjet/src/pipeline/suffix_routing.rs index eb0656e..b1e657e 100644 --- a/crates/httpjet/src/pipeline/suffix_routing.rs +++ b/crates/httpjet/src/pipeline/suffix_routing.rs @@ -66,22 +66,26 @@ pub(super) fn split_script_path( // resolution and authorization path. Dispatch later selects FastCGI only // when the named processor is explicitly type=fcgi; unsupported CGI still // resolves as executable and fails 503 rather than serving source bytes. - let cgi_suffixes: Vec = ctx - .vhost - .script_handlers - .iter() - .filter(|handler| handler.kind == hj_core::config::ContextKind::Cgi) - .map(|handler| handler.suffix.to_ascii_lowercase()) - .collect(); - let php_suffixes = if cgi_suffixes - .iter() - .all(|suffix| php_suffixes.contains(suffix)) - { + let php_suffixes = if !state.has_cgi_script_routes { php_suffixes } else { - let mut combined = php_suffixes.into_owned(); - combined.extend(cgi_suffixes); - std::borrow::Cow::Owned(combined) + let cgi_suffixes: Vec = ctx + .vhost + .script_handlers + .iter() + .filter(|handler| handler.kind == hj_core::config::ContextKind::Cgi) + .map(|handler| handler.suffix.to_ascii_lowercase()) + .collect(); + if cgi_suffixes + .iter() + .all(|suffix| php_suffixes.contains(suffix)) + { + php_suffixes + } else { + let mut combined = php_suffixes.into_owned(); + combined.extend(cgi_suffixes); + std::borrow::Cow::Owned(combined) + } }; // Hot-path gate: only chains that actually carry a `SetHandler`/`AddHandler`/ // `AddType` directive pay the per-prefix scope-match cost. Bool-field scan over diff --git a/crates/httpjet/src/serving_generation.rs b/crates/httpjet/src/serving_generation.rs index d19451b..f90bab9 100644 --- a/crates/httpjet/src/serving_generation.rs +++ b/crates/httpjet/src/serving_generation.rs @@ -64,7 +64,8 @@ impl From>> for ServingView { } } -/// Selected once before dispatch and carried through a fast-path miss into the -/// Tokio bridge. Remote request headers cannot manufacture this extension. +/// Selected once before dispatch and carried through a fast-path miss beside +/// the request in the transport bridge context. Remote request data cannot +/// manufacture or replace this snapshot. #[derive(Clone)] pub(crate) struct RequestGeneration(pub(crate) Arc); diff --git a/crates/httpjet/src/state.rs b/crates/httpjet/src/state.rs index a620e0e..4aeb2cd 100644 --- a/crates/httpjet/src/state.rs +++ b/crates/httpjet/src/state.rs @@ -219,6 +219,10 @@ pub struct ServerState { /// Opt-in FastCGI handlers keyed by `(vhost scope, processor name)`. /// A scoped processor always wins over a global processor of the same name. pub fastcgi: HashMap<(Option, String), Arc>, + /// True when any loaded vhost declares an explicit CGI/FastCGI suffix route. + /// Kept independently of `fastcgi`: a declared route whose processor is + /// missing must still resolve as executable and fail closed with 503. + pub(crate) has_cgi_script_routes: bool, /// Reverse-proxy engine for this config generation. Reload retains unchanged /// upstream Arcs while obsolete named definitions drain with the old state. pub proxy: Arc, @@ -494,6 +498,7 @@ struct ConfigDerived { inline_rules: HashMap>, ext_by_name: HashMap, php_suffixes: HashSet, + has_cgi_script_routes: bool, acl: Arc, client_throttle: hj_acl::ClientThrottle, compress: Arc, @@ -515,6 +520,19 @@ fn build_config_derived( .as_ref() .map(|p| p.suffixes.iter().map(|s| s.to_ascii_lowercase()).collect()) .unwrap_or_default(); + // This summarizes DECLARED suffix routes, not successfully constructed + // FastCGI processors. Keeping those concepts separate preserves the + // source-disclosure guard: a route that names a missing processor must + // still enter script dispatch and return 503. The common configuration has + // no CGI routes, so request routing can skip both per-vhost suffix scans. + let has_cgi_script_routes = server.vhosts.values().any(|declaration| { + declaration.config.as_ref().is_some_and(|vhost| { + vhost + .script_handlers + .iter() + .any(|handler| handler.kind == hj_core::config::ContextKind::Cgi) + }) + }); // Pre-parse each vhost's inline rewrite rules once. let mut inline_rules = HashMap::new(); @@ -580,6 +598,7 @@ fn build_config_derived( inline_rules, ext_by_name, php_suffixes, + has_cgi_script_routes, acl, client_throttle, compress, @@ -946,6 +965,7 @@ impl ServerState { static_handler: cd.static_handler, lsapi, fastcgi, + has_cgi_script_routes: cd.has_cgi_script_routes, proxy, rewrite_cache: Arc::new(HtaccessCache::new()), inline_rules: cd.inline_rules, @@ -1092,6 +1112,7 @@ impl ServerState { // ---- runtime half: carried forward (proxy filtered to new config) ---- lsapi: old.lsapi.clone(), fastcgi, + has_cgi_script_routes: cd.has_cgi_script_routes, proxy, // Candidate construction must not clear live caches. Separate generations // also prevent an in-flight old request repopulating the new rule memo. diff --git a/crates/httpjet/src/uring/bridge.rs b/crates/httpjet/src/uring/bridge.rs index 73dda50..64ecc93 100644 --- a/crates/httpjet/src/uring/bridge.rs +++ b/crates/httpjet/src/uring/bridge.rs @@ -22,10 +22,12 @@ use tokio_util::sync::CancellationToken; use hj_core::{Body, Proto, Request, Response}; +use crate::serving_generation::RequestGeneration; + /// Per-request connection context the pipeline needs (peer/local addr, protocol, /// TLS + mTLS state, SNI, SSL_* params). Carried across the runtime boundary -/// alongside the request. Built once per connection (TLS metadata is per-handshake) -/// and cloned per request. +/// alongside the request. Its connection metadata is built once per handshake +/// and cloned per request; dispatch may then attach a request-generation snapshot. #[derive(Clone)] pub(crate) struct BridgeCtx { pub peer: std::net::SocketAddr, @@ -47,6 +49,11 @@ pub(crate) struct BridgeCtx { /// SSL_* CGI params (protocol/cipher/client-cert) for the LSAPI env, mirroring /// the tokio TLS path's `extract_tls_meta`. pub tls: Option, + /// Application generation selected by the transport before dispatch. This + /// travels beside the request instead of through `http::Extensions`, so a + /// normal bridged request does not allocate an extension map merely to pin + /// its reload snapshot. + pub request_generation: Option, } impl BridgeCtx { @@ -63,6 +70,7 @@ impl BridgeCtx { mtls_required: false, sni: None, tls: None, + request_generation: None, } } @@ -82,6 +90,7 @@ impl BridgeCtx { mtls_required: false, sni: None, tls: None, + request_generation: None, } } } @@ -516,13 +525,24 @@ pub(crate) fn service_unavailable_resp() -> BridgeResp { } } +#[inline] +fn take_response_completion( + extensions: &mut http::Extensions, +) -> Option { + if extensions.is_empty() { + None + } else { + extensions.remove::() + } +} + fn full_resp(mut parts: http::response::Parts, body: Bytes, bw_rate: Option) -> BridgeResp { let status = parts.status; observe_response_head(&mut parts, status); // The Full arms own `parts` outright — move the header map instead of cloning it per // bridged response. BridgeResp { - completion: parts.extensions.remove::(), + completion: take_response_completion(&mut parts.extensions), status: parts.status, headers: parts.headers, body: BridgeBody::Full(body), @@ -549,7 +569,7 @@ pub(crate) async fn fast_response(r: Response, direct_file: bool) -> BridgeResp let status = parts.status; observe_response_head(&mut parts, status); BridgeResp { - completion: parts.extensions.remove::(), + completion: take_response_completion(&mut parts.extensions), status: parts.status, headers: strip_framing(parts.headers), body: BridgeBody::File(file), @@ -570,7 +590,7 @@ pub(crate) async fn fast_response(r: Response, direct_file: bool) -> BridgeResp fn bad_gateway_for(mut parts: http::response::Parts) -> BridgeResp { observe_response_head(&mut parts, http::StatusCode::BAD_GATEWAY); let mut response = bad_gateway(); - response.completion = parts.extensions.remove::(); + response.completion = take_response_completion(&mut parts.extensions); response } @@ -634,7 +654,7 @@ async fn forward_response(r: Response, resp: oneshot::Sender, direct let status = parts.status; observe_response_head(&mut parts, status); let _ = resp.send(BridgeResp { - completion: parts.extensions.remove::(), + completion: take_response_completion(&mut parts.extensions), status: parts.status, headers: strip_framing(parts.headers), body: BridgeBody::File(f), @@ -680,7 +700,7 @@ async fn forward_file( let headers = strip_framing(parts.headers); let (tx, rrx) = mpsc::channel(STREAM_CHANNEL_DEPTH); let _ = resp.send(BridgeResp { - completion: parts.extensions.remove::(), + completion: take_response_completion(&mut parts.extensions), status: parts.status, headers, body: BridgeBody::Stream { @@ -749,7 +769,7 @@ async fn forward_stream( let headers = strip_framing(parts.headers); let (tx, rrx) = mpsc::channel(STREAM_CHANNEL_DEPTH); let _ = resp.send(BridgeResp { - completion: parts.extensions.remove::(), + completion: take_response_completion(&mut parts.extensions), status: parts.status, headers, body: BridgeBody::Stream { rx: rrx, len: None }, @@ -770,7 +790,7 @@ async fn forward_stream( let headers = strip_framing(parts.headers); let (tx, rrx) = mpsc::channel(STREAM_CHANNEL_DEPTH); let _ = resp.send(BridgeResp { - completion: parts.extensions.remove::(), + completion: take_response_completion(&mut parts.extensions), status: parts.status, headers, body: BridgeBody::Stream { rx: rrx, len: None }, @@ -792,7 +812,7 @@ async fn forward_stream( let status = parts.status; observe_response_head(&mut parts, status); let _ = resp.send(BridgeResp { - completion: parts.extensions.remove::(), + completion: take_response_completion(&mut parts.extensions), status: parts.status, headers: parts.headers, body: BridgeBody::Full(Bytes::from(acc)), @@ -1323,6 +1343,7 @@ mod tests { mtls_required: false, sni: None, tls: None, + request_generation: None, }; let resp = bridge.dispatch(req, ctx).await.expect("bridged response"); assert_eq!(resp.status, http::StatusCode::OK); diff --git a/crates/httpjet/src/uring/generation_test.rs b/crates/httpjet/src/uring/generation_test.rs index 49c85af..294a87f 100644 --- a/crates/httpjet/src/uring/generation_test.rs +++ b/crates/httpjet/src/uring/generation_test.rs @@ -941,9 +941,9 @@ async fn bridge_uses_request_snapshot_across_application_publication() { "127.0.0.1:8080".parse().unwrap(), Proto::Http1, ); - let mut pinned = make_request(); - pinned.extensions_mut().insert(RequestGeneration(old)); - let response = bridge.dispatch_response(pinned, ctx.clone()).await; + let mut pinned_ctx = ctx.clone(); + pinned_ctx.request_generation = Some(RequestGeneration(old)); + let response = bridge.dispatch_response(make_request(), pinned_ctx).await; assert_eq!(response.status(), http::StatusCode::OK); let (body, truncated) = bridge::buffer_body(response.into_body()).await; assert!(!truncated); diff --git a/crates/httpjet/src/uring/h3.rs b/crates/httpjet/src/uring/h3.rs index a9984c1..43d6a6b 100644 --- a/crates/httpjet/src/uring/h3.rs +++ b/crates/httpjet/src/uring/h3.rs @@ -3458,9 +3458,6 @@ async fn handle_h3_request( Err(_) => return H3Outcome::full(h3_error(http::StatusCode::BAD_REQUEST)), }; hj_core::coalesce_cookie_crumbs(req.headers_mut()); - if let Some(generation) = request_generation { - req.extensions_mut().insert(generation); - } let ctx = BridgeCtx { peer, local, @@ -3471,6 +3468,7 @@ async fn handle_h3_request( mtls_required: require_client_cert, sni, tls, + request_generation, }; // A HEAD response must carry no DATA (RFC 9114): even if the pipeline streams a body, // emit headers only and don't open a streamed body. @@ -4019,7 +4017,7 @@ mod h3_codec_tests { #[tokio::test] async fn accepted_quic_view_pins_limits_and_dispatch_generation() { - use crate::serving_generation::{RequestGeneration, ServingView}; + use crate::serving_generation::ServingView; use std::sync::atomic::Ordering; let root = std::env::temp_dir().join(format!( "hj-quic-generation-{}-{}", @@ -4076,13 +4074,9 @@ mod h3_codec_tests { // publication travels through dispatch, rather than reloading live state. let seen = Arc::new(AtomicU64::new(0)); let observed = seen.clone(); - let bridge = crate::uring::bridge::spawn_on_current(2, move |req, _| { + let bridge = crate::uring::bridge::spawn_on_current(2, move |_req, ctx| { observed.store( - req.extensions() - .get::() - .unwrap() - .0 - .generation, + ctx.request_generation.as_ref().unwrap().0.generation, Ordering::SeqCst, ); async { http::Response::new(hj_core::Body::Empty) } diff --git a/crates/httpjet/src/uring/mod.rs b/crates/httpjet/src/uring/mod.rs index cd053e8..ba8f13d 100644 --- a/crates/httpjet/src/uring/mod.rs +++ b/crates/httpjet/src/uring/mod.rs @@ -103,17 +103,13 @@ impl CoreHandler { #[cfg(feature = "otel")] fn trace_request( &self, + state: &ServerState, ctx: &BridgeCtx, req: &mut hj_core::Request, ) -> Option { if !crate::otel::enabled() { return None; } - let state = req - .extensions() - .get::() - .map(|snapshot| arc_swap::Guard::from_inner(snapshot.0.clone())) - .unwrap_or_else(|| self.holder.load()); let direct_peer = !ctx.peer_unix && state .server @@ -130,29 +126,37 @@ impl CoreHandler { async fn dispatch_h1( &self, - ctx: BridgeCtx, - mut req: hj_core::Request, + state: Arc, + mut ctx: BridgeCtx, + req: hj_core::Request, upgrade: bool, ) -> Option { - req.extensions_mut() - .insert(RequestGeneration(self.holder.load_full())); - if !upgrade && let Some(response) = self.fast(&ctx, &req).await { + if !upgrade && let Some(response) = self.fast(&state, &ctx, &req).await { return Some(bridge::fast_response(response, ctx.direct_file_egress).await); } + // Only a request that crosses to the Tokio runtime carries its selected + // generation. Keep it in the bridge sideband so ordinary misses do not + // allocate and hash an `http::Extensions` map. + ctx.request_generation = Some(RequestGeneration(state)); #[cfg(feature = "otel")] crate::otel::execution_path(false); self.bridge.dispatch(req, ctx).await } - async fn dispatch_h2(&self, ctx: BridgeCtx, mut req: hj_core::Request) -> hj_core::Response { - req.extensions_mut() - .insert(RequestGeneration(self.holder.load_full())); + #[cfg(feature = "otel")] + async fn dispatch_h2( + &self, + mut ctx: BridgeCtx, + mut req: hj_core::Request, + ) -> hj_core::Response { + let state = self.holder.load_full(); #[cfg(feature = "otel")] - let trace = self.trace_request(&ctx, &mut req); + let trace = self.trace_request(&state, &ctx, &mut req); let future = async { - if let Some(response) = self.fast(&ctx, &req).await { + if let Some(response) = self.fast(&state, &ctx, &req).await { return response; } + ctx.request_generation = Some(RequestGeneration(state)); #[cfg(feature = "otel")] crate::otel::execution_path(false); self.bridge.dispatch_response(req, ctx).await @@ -168,17 +172,17 @@ impl CoreHandler { } /// Try the on-core cache-hit fast path; `Some(resp)` if served without the bridge. - async fn fast(&self, ctx: &BridgeCtx, req: &hj_core::Request) -> Option { - let st = req - .extensions() - .get::() - .map(|snapshot| snapshot.0.clone()) - .unwrap_or_else(|| self.holder.load_full()); + async fn fast( + &self, + st: &Arc, + ctx: &BridgeCtx, + req: &hj_core::Request, + ) -> Option { // Stamp Date here (insert-if-absent): the page cache strips the stored Date // expecting the serve boundary to re-add one, and the uring writers never do — the // tokio path stamps at server::stamp_date, this is its on-core fast-path twin. crate::pipeline::fast_serve( - &st, + st, &self.listener_name, ctx.peer.ip(), ctx.local, @@ -726,10 +730,10 @@ fn build_pipeline_bridge( // for every bridged request (the closure runs concurrently across tokio workers). let lname = listener_name; let view = holder.into(); - bridge::spawn_on_current_with_admission(admission, move |mut req, ctx: BridgeCtx| { - let state = req - .extensions_mut() - .remove::() + bridge::spawn_on_current_with_admission(admission, move |req, mut ctx: BridgeCtx| { + let state = ctx + .request_generation + .take() .map(|snapshot| snapshot.0) .unwrap_or_else(|| view.load_full()); let lname = lname.clone(); @@ -737,24 +741,32 @@ fn build_pipeline_bridge( // H1 already calls the same helper before bridging so it can retain // its historical refusal/connection semantics. H2/H3 arrive here // as one lease-backed full body; a second H1 pass is a no-op after - // successful decoding removed Content-Encoding. - let req = match request_body::finish_bridged_request( - req, - &state.body_budget, - state.serve_config.max_req_body_size, - state.request_decompression, - ) - .await - { - Ok(req) => req, - Err(status) => { - return hj_core::stamp_date( - http::Response::builder() - .status(status) - .body(hj_core::Body::Empty) - .expect("static request-decompression response"), - ); + // successful decoding removed Content-Encoding. Keep the decoder + // future out of the common bridge task entirely: its codec state is + // large, while an allocation on the explicitly encoded path is both + // bounded and negligible beside decompression. + let policy = state.request_decompression; + let req = if request_body::needs_bridged_decompression(&req, policy) { + match Box::pin(request_body::finish_bridged_request( + req, + &state.body_budget, + state.serve_config.max_req_body_size, + policy, + )) + .await + { + Ok(req) => req, + Err(status) => { + return hj_core::stamp_date( + http::Response::builder() + .status(status) + .body(hj_core::Body::Empty) + .expect("static request-decompression response"), + ); + } } + } else { + req }; // Stamp Date (insert-if-absent) on EVERY bridged response (H1/H2/H3): the uring // writers + native h2/h3 encoders don't add it and the cache strips the stored @@ -965,10 +977,10 @@ fn per_core_https( /// Terminate TLS on a monoio io_uring connection, enforce mTLS, then serve H1/H2 /// (by ALPN) over the encrypted stream via the bridge. The connection metadata -/// (SNI, ALPN proto, SSL_* params, client-cert presence) is extracted by detaching -/// the rustls `ServerConnection` (only public accessor), draining any pipelined -/// post-handshake plaintext into the handler prefix (lossless), then reconstructing -/// the stream to serve. +/// (SNI, ALPN proto, SSL_* params, client-cert presence) is borrowed from the +/// rustls `ServerConnection`. Any decrypted post-handshake plaintext becomes the +/// handler prefix, while the TLS adapter and its raw-ciphertext read-ahead stay +/// intact across the direct-write I/O wrapper transition. async fn handle_tls_bridged( mut stream: TcpStream, mut peer: SocketAddr, @@ -1025,7 +1037,7 @@ async fn handle_tls_bridged( } } let handshake_timeout = state.serve_config.header_read_timeout; - let tls = match handshake_timeout { + let mut tls = match handshake_timeout { Some(d) => match monoio::time::timeout(d, acceptor.accept(stream)).await { Ok(Ok(t)) => t, Ok(Err(e)) => { @@ -1045,22 +1057,22 @@ async fn handle_tls_bridged( } }, }; - // Detach the session to read the handshake metadata (the `session` field is not - // otherwise accessible), drain any early app-data, then rebuild the stream. - let (io, mut session) = tls.into_parts(); - let sni: Option> = session.server_name().map(Arc::from); - let proto = match session.alpn_protocol() { + let sni: Option> = tls.session().server_name().map(Arc::from); + let proto = match tls.session().alpn_protocol() { Some(b"h2") => Proto::Http2, _ => Proto::Http1, }; - let has_client_cert = session.peer_certificates().is_some_and(|c| !c.is_empty()); + let has_client_cert = tls + .session() + .peer_certificates() + .is_some_and(|c| !c.is_empty()); // Full-vs-resumed split sizes the resumption win the client-verify // `NoServerSessions` posture forfeits (every CF-cycled origin connection is a // full handshake today). Counted once, at handshake completion. { use std::sync::atomic::Ordering; let m = core.holder.load(); - match session.handshake_kind() { + match tls.session().handshake_kind() { Some(rustls::HandshakeKind::Resumed) => { m.metrics .tls_handshakes_resumed @@ -1074,7 +1086,7 @@ async fn handle_tls_bridged( None => {} } } - let tls_params = hj_tls::tls_params_from_conn(&session); + let tls_params = hj_tls::tls_params_from_conn(tls.session()); // Application-layer mTLS (clientVerify=2): refuse a non-internal peer that // presented no valid client cert — mirrors server.rs::mtls_refused exactly. // On a RESUMED handshake `peer_certificates` is the chain rustls reinstated @@ -1091,7 +1103,7 @@ async fn handle_tls_bridged( let mut prefix: Vec = Vec::new(); { use std::io::Read; - let mut reader = session.reader(); + let mut reader = tls.session_mut().reader(); let mut buf = [0u8; 8192]; loop { match reader.read(&mut buf) { @@ -1111,6 +1123,7 @@ async fn handle_tls_bridged( mtls_required: require_client_cert, sni, tls: tls_params, + request_generation: None, }; // kTLS path: upgrade the socket to kernel-TLS and serve plaintext over the RAW fd @@ -1119,11 +1132,16 @@ async fn handle_tls_bridged( // schedule + no KeyUpdate on 1.2 ⇒ a 1.2 connection just falls through to userspace). #[cfg(feature = "ktls")] if let Some(kl) = conn_key_log { - let is_tls13 = session.protocol_version() == Some(rustls::ProtocolVersion::TLSv1_3); - let suite = session.negotiated_cipher_suite(); + let is_tls13 = tls.session().protocol_version() == Some(rustls::ProtocolVersion::TLSv1_3); + let suite = tls.session().negotiated_cipher_suite(); if is_tls13 { match (kl.secrets(), suite) { (Some((rx, tx)), Some(suite)) => { + // kTLS needs ownership of the raw socket and rustls connection to + // extract traffic secrets. Keep this destructive split confined to + // the committed kernel-TLS path; every userspace fallback retains + // the adapter's raw-ciphertext read-ahead via `map_io` below. + let (io, session) = tls.into_parts(); use std::os::fd::AsRawFd; let fd = io.as_raw_fd(); // Read the true post-handshake record sequence for each direction (tickets @@ -1180,11 +1198,10 @@ async fn handle_tls_bridged( // Wrap the socket so monoio-rustls writes the encrypted bytes via a direct write(2) // syscall (not an io_uring write) — matching tokio's write path on loopback bulk egress. - // rustls/aws-lc-rs still does the AEAD; this only changes the socket write. - let stream = monoio_rustls::ServerTlsStream::new( - directio::DirectWriteSocket::new_for(io, proto != Proto::Http2), - session, - ); + // rustls/aws-lc-rs still does the AEAD; this only changes the socket write. Mapping + // the I/O in place is essential: the adapter may have read ciphertext beyond the + // final handshake record which rustls has not consumed yet. + let stream = tls.map_io(|io| directio::DirectWriteSocket::new_for(io, proto != Proto::Http2)); match proto { Proto::Http2 => serve_h2_bridged(stream, prefix, ctx, core, shutdown, None).await, _ => handle_h1_bridged(stream, prefix, ctx, core, shutdown, None).await, @@ -1346,10 +1363,33 @@ async fn serve_h2_bridged( // rate is connection-wide here; the per-context override is an H1 refinement). h2_cfg.bandwidth_limit = state.serve_config.bandwidth_limit; + // H2 clones its service captures for every stream. Keep the connection-local + // handler behind one Arc so that does not clone every Arc-bearing field in + // CoreHandler (in particular the live and pinned ServerState generations). + let core = Arc::new(core); let service = move |req: hj_core::Request| { let core = core.clone(); let ctx = ctx.clone(); - async move { core.dispatch_h2(ctx, req).await } + #[cfg(feature = "otel")] + { + async move { core.dispatch_h2(ctx, req).await } + } + #[cfg(not(feature = "otel"))] + { + // Keep the ordinary build's fast/bridge state machine directly in + // the service future. An extra async dispatch wrapper made every + // FuturesUnordered node substantially larger on the H2 hot path. + async move { + let req = req; + let mut ctx = ctx; + let state = core.holder.load_full(); + if let Some(response) = core.fast(&state, &ctx, &req).await { + return response; + } + ctx.request_generation = Some(RequestGeneration(state)); + core.bridge.dispatch_response(req, ctx).await + } + } }; // `ktls_fd` (Some only for a kTLS connection) lets the h2 flush writev plaintext directly // from the OutQueue to the kernel-TLS socket (zero-copy); None ⇒ the coalesce path. @@ -1398,13 +1438,6 @@ async fn handle_h1_bridged( let mut throttle = hj_http::BandwidthThrottle::new(core.holder.load().serve_config.bandwidth_limit); loop { - let state = core.holder.load(); - // Request-size caps from the LiteSpeed config (maxReqHeaderSize/maxReqBodySize), - // matching the tokio path — mirror hyper's `max_buf_size` 8 KiB floor for the head. - let max_head = state.serve_config.max_req_header_size.max(8192); - let max_body = state.serve_config.max_req_body_size; - // maxKeepAliveReq: 0 = unlimited; else close the connection after N requests. - let max_keepalive = state.serve_config.max_keepalive_requests; // Graceful drain: at a clean request boundary (no buffered bytes = idle // keep-alive), wait for either the next request OR the shutdown signal — on // shutdown, close the idle connection promptly instead of holding it open. @@ -1421,7 +1454,9 @@ async fn handle_h1_bridged( // keeps idle origin connections well past a 5s keepAliveTimeout, and an // unpadded H1 idle wait closes them almost immediately (constant // reconnect + TLS-handshake churn). H2 has padded to >=90s for a while. - let keep_alive_timeout = state + let keep_alive_timeout = core + .holder + .load() .serve_config .keep_alive_timeout .map(|t| t.max(std::time::Duration::from_secs(90))); @@ -1439,6 +1474,17 @@ async fn handle_h1_bridged( } } } + // Select the request's generation only after idle waiting has received + // bytes. Compatible SIGHUPs must therefore remain visible to the next + // request on a persistent H1 connection. This exact Arc is then used + // for intake limits, the on-core fast path and any bridge dispatch. + let state = core.holder.load_full(); + // Request-size caps from the LiteSpeed config (maxReqHeaderSize/maxReqBodySize), + // matching the tokio path — mirror hyper's `max_buf_size` 8 KiB floor for the head. + let max_head = state.serve_config.max_req_header_size.max(8192); + let max_body = state.serve_config.max_req_body_size; + // maxKeepAliveReq: 0 = unlimited; else close the connection after N requests. + let max_keepalive = state.serve_config.max_keepalive_requests; // Parse a complete request head (drops the borrow before mutating `acc`). let request_start = std::time::Instant::now(); let header_read_timeout = state.serve_config.header_read_timeout; @@ -1702,8 +1748,8 @@ async fn handle_h1_bridged( upgrade_ready = Some(ready); } #[cfg(feature = "otel")] - let mut trace = core.trace_request(&ctx, &mut req); - let dispatch = core.dispatch_h1(ctx.clone(), req, upgrade_ready.is_some()); + let mut trace = core.trace_request(&state, &ctx, &mut req); + let dispatch = core.dispatch_h1(state, ctx.clone(), req, upgrade_ready.is_some()); #[cfg(feature = "otel")] let response = match &trace { Some(trace) => crate::otel::in_context(trace.context(), dispatch).await, @@ -3039,6 +3085,26 @@ Sec-WebSocket-Key: dGhlIHNhbXBsZSBub25jZQ==\r\n\ .unwrap() }); let bridge = bridge::spawn_bridge(1, |mut req: hj_core::Request, _ctx| async move { + if req.uri().path() == "/body" { + use http_body_util::BodyExt; + + let body = req + .into_body() + .collect() + .await + .expect("collect test request body") + .to_bytes(); + let valid = body.len() == 64 * 1024 && body.iter().all(|byte| *byte == b'x'); + return hj_core::text_response( + if valid { + http::StatusCode::OK + } else { + http::StatusCode::BAD_REQUEST + }, + body.len().to_string(), + ); + } + if req.uri().path() == "/reject" { return http::Response::builder() .status(http::StatusCode::FORBIDDEN) @@ -4145,6 +4211,115 @@ Sec-WebSocket-Key: dGhlIHNhbXBsZSBub25jZQ==\r\n\ client.join().unwrap(); } + #[test] + fn h1_tls_preserves_read_ahead_after_coalesced_finished_and_large_post() { + use std::io::{Read, Write}; + + let (_tokio_runtime, core) = websocket_test_core(); + let _ = rustls::crypto::aws_lc_rs::default_provider().install_default(); + let certified = rcgen::generate_simple_self_signed(vec!["localhost".to_string()]).unwrap(); + let cert = rustls::pki_types::CertificateDer::from(certified.cert.der().to_vec()); + let key = rustls::pki_types::PrivateKeyDer::try_from(certified.signing_key.serialize_der()) + .unwrap(); + let mut server_config = rustls::ServerConfig::builder() + .with_no_client_auth() + .with_single_cert(vec![cert.clone()], key) + .unwrap(); + server_config.alpn_protocols = vec![b"http/1.1".to_vec()]; + let acceptor = monoio_rustls::TlsAcceptor::from(Arc::new(server_config)); + + let mut roots = rustls::RootCertStore::empty(); + roots.add(cert).unwrap(); + let mut client_config = rustls::ClientConfig::builder() + .with_root_certificates(roots) + .with_no_client_auth(); + client_config.alpn_protocols = vec![b"http/1.1".to_vec()]; + let client_config = Arc::new(client_config); + + let std_listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); + let local = std_listener.local_addr().unwrap(); + std_listener.set_nonblocking(true).unwrap(); + let client = std::thread::spawn(move || { + let mut socket = std::net::TcpStream::connect(local).unwrap(); + socket + .set_read_timeout(Some(std::time::Duration::from_secs(5))) + .unwrap(); + socket + .set_write_timeout(Some(std::time::Duration::from_secs(5))) + .unwrap(); + let name = rustls::pki_types::ServerName::try_from("localhost") + .unwrap() + .to_owned(); + let mut connection = rustls::ClientConnection::new(client_config, name).unwrap(); + connection.set_buffer_limit(Some(128 * 1024)); + + let mut request = format!( + "POST /body HTTP/1.1\r\nHost: localhost\r\nContent-Length: {}\r\nConnection: close\r\n\r\n", + 64 * 1024 + ) + .into_bytes(); + request.resize(request.len() + 64 * 1024, b'x'); + // rustls buffers pre-handshake plaintext. Once the server flight is + // processed, the next write contains the client Finished followed by + // application records, reproducing a client that pipelines its POST. + connection.writer().write_all(&request).unwrap(); + + let mut initial_flight = Vec::new(); + while connection.wants_write() { + assert!(connection.write_tls(&mut initial_flight).unwrap() > 0); + } + socket.write_all(&initial_flight).unwrap(); + socket.flush().unwrap(); + + let mut sent_coalesced_large_flight = false; + while connection.is_handshaking() { + let read = connection.read_tls(&mut socket).unwrap(); + assert!(read > 0, "server closed during TLS handshake"); + connection.process_new_packets().unwrap(); + + if connection.wants_write() { + let mut flight = Vec::new(); + while connection.wants_write() { + assert!(connection.write_tls(&mut flight).unwrap() > 0); + } + sent_coalesced_large_flight |= flight.len() > 16 * 1024; + socket.write_all(&flight).unwrap(); + socket.flush().unwrap(); + } + } + assert!( + sent_coalesced_large_flight, + "fixture must coalesce Finished with more than one adapter read of POST data" + ); + + let mut stream = rustls::StreamOwned::new(connection, socket); + let head = read_h1_head(&mut stream); + assert!(head.starts_with(b"HTTP/1.1 200 OK\r\n")); + let mut body = [0u8; 5]; + stream.read_exact(&mut body).unwrap(); + assert_eq!(&body, b"65536"); + }); + + let mut runtime = build_core_runtime().unwrap(); + runtime.block_on(async move { + let listener = TcpListener::from_std(std_listener).unwrap(); + let (stream, peer) = listener.accept().await.unwrap(); + handle_tls_bridged( + stream, + peer, + local, + core, + acceptor, + false, + None, + CancellationToken::new(), + ListenerBinding::default(), + ) + .await; + }); + client.join().unwrap(); + } + #[cfg(feature = "ktls")] #[test] fn ktls_h1_sendfile_preserves_range_across_mid_transfer_key_update() { diff --git a/crates/httpjet/src/uring/otel_test.rs b/crates/httpjet/src/uring/otel_test.rs index 6b3dec5..028a182 100644 --- a/crates/httpjet/src/uring/otel_test.rs +++ b/crates/httpjet/src/uring/otel_test.rs @@ -84,6 +84,7 @@ fn traced_fast_and_bridged_requests() { mtls_required: false, sni: None, tls: None, + request_generation: None, }; // Drive the same H2 service dispatcher with both on-core and bridged requests. runtime.block_on(async { diff --git a/crates/httpjet/src/uring/request_body.rs b/crates/httpjet/src/uring/request_body.rs index 7357e33..87d7507 100644 --- a/crates/httpjet/src/uring/request_body.rs +++ b/crates/httpjet/src/uring/request_body.rs @@ -94,6 +94,26 @@ impl ContentCoding { } } +#[inline] +fn enabled_content_coding( + headers: &HeaderMap, + policy: RequestDecompression, +) -> Option { + ContentCoding::from_headers(headers).filter(|coding| policy.allows(*coding)) +} + +/// Whether a bridged request needs the asynchronous collect/decode path. +/// +/// Keep this synchronous probe outside the bridge task's ordinary future so +/// requests without an enabled coding do not carry decoder state at all. +#[inline] +pub(super) fn needs_bridged_decompression( + req: &hj_core::Request, + policy: RequestDecompression, +) -> bool { + enabled_content_coding(req.headers(), policy).is_some() +} + pub(super) fn finish_body( headers: &mut HeaderMap, data: Vec, @@ -127,6 +147,14 @@ pub(super) async fn finish_bridged_request( max_body: usize, policy: RequestDecompression, ) -> Result { + // The overwhelmingly common request has no supported Content-Encoding. + // Preserve its body and parts verbatim instead of collecting, unboxing and + // rebuilding the request merely for decode_body() to return the same bytes. + // This also makes H1's second bridge-side pass free after finish_body() + // removed a coding decoded on the monoio intake path. + if !needs_bridged_decompression(&req, policy) { + return Ok(req); + } let (mut parts, body) = req.into_parts(); let encoded = body .collect() @@ -151,12 +179,14 @@ fn decode_body( max_body: usize, policy: RequestDecompression, ) -> Result { - let Some(coding) = ContentCoding::from_headers(headers) else { - return Ok(encoded); - }; - if encoded.is_empty() || !policy.allows(coding) { + // Bodyless requests dominate GET traffic. Avoid even probing the header + // map for Content-Encoding when there is nothing a decoder could consume. + if encoded.is_empty() { return Ok(encoded); } + let Some(coding) = enabled_content_coding(headers, policy) else { + return Ok(encoded); + }; // Charge the codec before constructing it. This makes concurrent decoder // windows participate in the same process-wide request-body ledger as the @@ -620,4 +650,72 @@ mod tests { drop(body); assert_eq!(budget.in_flight(), 0); } + + #[tokio::test] + async fn disabled_or_unencoded_bridged_body_is_not_polled_or_rebuilt() { + use std::pin::Pin; + use std::task::{Context, Poll}; + + struct MustNotPoll; + impl http_body::Body for MustNotPoll { + type Data = Bytes; + type Error = hj_core::BoxError; + + fn poll_frame( + self: Pin<&mut Self>, + _cx: &mut Context<'_>, + ) -> Poll, Self::Error>>> { + panic!("unencoded bridge fast path must preserve the body without polling it") + } + } + + let budget = Arc::new(BodyBufferBudget::new(1024)); + for coding in [None, Some("unknown"), Some("br"), Some("zstd")] { + let body = http_body_util::BodyExt::boxed(MustNotPoll); + let mut builder = http::Request::builder() + .method("POST") + .uri("/opaque") + .header("x-test", "retained"); + if let Some(coding) = coding { + builder = builder.header(header::CONTENT_ENCODING, coding); + } + let mut req = builder.body(body).unwrap(); + req.extensions_mut().insert(42_u32); + assert!(!needs_bridged_decompression( + &req, + RequestDecompression::default() + )); + + let req = finish_bridged_request(req, &budget, 1024, RequestDecompression::default()) + .await + .unwrap(); + assert_eq!(req.uri(), "/opaque"); + assert_eq!(req.headers()["x-test"], "retained"); + assert_eq!(req.extensions().get::(), Some(&42)); + } + + for (coding, policy) in [ + ("gzip", RequestDecompression::default()), + ( + "br", + RequestDecompression { + brotli: true, + zstd: false, + }, + ), + ( + "zstd", + RequestDecompression { + brotli: false, + zstd: true, + }, + ), + ] { + let req = http::Request::builder() + .header(header::CONTENT_ENCODING, coding) + .body(hj_core::empty_incoming()) + .unwrap(); + assert!(needs_bridged_decompression(&req, policy)); + } + } } diff --git a/vendor/monoio-rustls/src/stream.rs b/vendor/monoio-rustls/src/stream.rs index 9e56431..3416bf0 100644 --- a/vendor/monoio-rustls/src/stream.rs +++ b/vendor/monoio-rustls/src/stream.rs @@ -69,6 +69,40 @@ impl Stream { (self.io, self.session) } + /// Borrow the rustls connection without detaching it from the transport + /// adapter's read-ahead and pending-write buffers. + #[inline] + pub fn session(&self) -> &C { + &self.session + } + + /// Mutably borrow the rustls connection without detaching it from the + /// transport adapter's read-ahead and pending-write buffers. + #[inline] + pub fn session_mut(&mut self) -> &mut C { + &mut self.session + } + + /// Replace only the underlying I/O object, retaining the rustls connection + /// and both adapter buffers exactly as they stood before the mapping. + /// + /// This is the lossless way to wrap an accepted stream after its handshake: + /// [`Stream::into_parts`] intentionally returns only the public I/O/session + /// pair and therefore cannot preserve ciphertext already read ahead by the + /// adapter. + #[inline] + pub fn map_io(self, map: F) -> Stream + where + F: FnOnce(IO) -> IO2, + { + Stream { + io: map(self.io), + session: self.session, + r_buffer: self.r_buffer, + w_buffer: self.w_buffer, + } + } + pub(crate) fn map_conn C2>(self, f: F) -> Stream { Stream { io: self.io, @@ -79,6 +113,57 @@ impl Stream { } } +#[cfg(test)] +mod transition_tests { + use std::future::Future; + use std::io::{Read, Write}; + use std::sync::Arc; + use std::task::{Context, Poll, Waker}; + + use super::{Stream, WriteBuffer}; + + fn ready(future: F) -> F::Output { + let mut future = std::pin::pin!(future); + let mut context = Context::from_waker(Waker::noop()); + match future.as_mut().poll(&mut context) { + Poll::Ready(output) => output, + Poll::Pending => panic!("in-memory I/O unexpectedly yielded"), + } + } + + #[test] + fn map_io_preserves_session_and_adapter_buffers() { + let session = Arc::new("session-marker"); + let mut stream = Stream::new("original-io", session.clone()); + + let mut read_ahead: &[u8] = b"ciphertext-read-ahead"; + assert_eq!( + ready(stream.r_buffer.do_io(&mut read_ahead)).unwrap(), + b"ciphertext-read-ahead".len() + ); + stream.w_buffer.write_all(b"pending-ciphertext").unwrap(); + + let mut mapped = stream.map_io(|io| { + assert_eq!(io, "original-io"); + "wrapped-io" + }); + assert_eq!(mapped.io, "wrapped-io"); + assert!(Arc::ptr_eq(mapped.session(), &session)); + + let mut received = [0u8; 21]; + mapped.r_buffer.read_exact(&mut received).unwrap(); + assert_eq!(&received, b"ciphertext-read-ahead"); + match &mapped.w_buffer { + WriteBuffer::Safe(buffer) => assert_eq!( + buffer.buffer.as_ref().expect("write buffer").len(), + b"pending-ciphertext".len() + ), + #[cfg(feature = "unsafe_io")] + WriteBuffer::Unsafe(_) => panic!("safe constructor selected unsafe buffer"), + } + } +} + #[derive(Debug)] enum WriteBuffer { Safe(SafeWriteBuffer), From ef796345aee50040bd35a8fdc0adacd5a6d5f35c Mon Sep 17 00:00:00 2001 From: Michael Fara Date: Wed, 9 Sep 2026 12:26:07 +0000 Subject: [PATCH 2/2] Raise CI descriptor limit for handoff test Signed-off-by: Michael Fara --- .github/workflows/ci.yml | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index bccb3dc..9b88083 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -66,11 +66,13 @@ jobs: - name: Run workspace tests # The LSAPI supervisor suite signals and reaps detached descendants. # GitHub's runner blocks those operations for its unprivileged user. + # sudo may also reset RLIMIT_NOFILE to 1024, but the TCP handoff test + # deliberately holds 1024 queued descriptors plus its control sockets. run: | cargo_bin="$(rustup which cargo)" rustc_bin="$(rustup which rustc)" rustdoc_bin="$(rustup which rustdoc)" - sudo env \ + sudo prlimit --nofile=65536:65536 -- env \ "HOME=$HOME" \ "CARGO_HOME=$HOME/.cargo" \ "PATH=$PATH" \