From 0ec8fcb1231ab01fc536cada3389ead1b7015d20 Mon Sep 17 00:00:00 2001 From: Nyannyacha Date: Fri, 24 Apr 2026 03:26:43 +0000 Subject: [PATCH 1/3] fix(base/server): return HTTP error responses instead of dropping connection in WorkerService --- crates/base/src/server.rs | 25 +++++++++++++++++-------- 1 file changed, 17 insertions(+), 8 deletions(-) diff --git a/crates/base/src/server.rs b/crates/base/src/server.rs index b982fd4a2..ba06d7177 100644 --- a/crates/base/src/server.rs +++ b/crates/base/src/server.rs @@ -221,20 +221,29 @@ impl Service> for WorkerService { let res = match res_rx.await { Ok(res) => res, - Err(err) => { + Err(_) => { metric_src.incl_handled_requests(); - return Err(err.into()); + return Ok( + Response::builder() + .status(http_v02::StatusCode::INTERNAL_SERVER_ERROR) + .header("x-served-by", concat!(env!("CARGO_PKG_NAME"), "/server")) + .body(Body::empty()) + .unwrap(), + ); } }; - // If the token has already been canceled, drop the socket connection - // without sending a response. + // If the token has already been canceled, return 503 instead of + // dropping the socket connection without a response. if cancel.is_cancelled() { error!("connection aborted (uri: {:?})", req_uri.to_string()); - return Err(anyhow!(std::io::Error::new( - std::io::ErrorKind::ConnectionAborted, - "connection aborted" - ))); + return Ok( + Response::builder() + .status(http_v02::StatusCode::SERVICE_UNAVAILABLE) + .header("x-served-by", concat!(env!("CARGO_PKG_NAME"), "/server")) + .body(Body::empty()) + .unwrap(), + ); } let res = match res { From 6d1f6efce9f07bf3de4de52a5703c662c87c483e Mon Sep 17 00:00:00 2001 From: Nyannyacha Date: Fri, 24 Apr 2026 03:29:32 +0000 Subject: [PATCH 2/3] stamp: meow --- crates/base/src/server.rs | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/crates/base/src/server.rs b/crates/base/src/server.rs index ba06d7177..0b33426d7 100644 --- a/crates/base/src/server.rs +++ b/crates/base/src/server.rs @@ -71,6 +71,8 @@ mod signal { pub use tokio::signal::unix; } +const SERVED_BY: &str = concat!(env!("CARGO_PKG_NAME"), "/server"); + pub enum ServerEvent { ConnectionError(hyper_v014::Error), #[cfg(debug_assertions)] @@ -226,7 +228,7 @@ impl Service> for WorkerService { return Ok( Response::builder() .status(http_v02::StatusCode::INTERNAL_SERVER_ERROR) - .header("x-served-by", concat!(env!("CARGO_PKG_NAME"), "/server")) + .header("x-served-by", SERVED_BY) .body(Body::empty()) .unwrap(), ); @@ -240,7 +242,7 @@ impl Service> for WorkerService { return Ok( Response::builder() .status(http_v02::StatusCode::SERVICE_UNAVAILABLE) - .header("x-served-by", concat!(env!("CARGO_PKG_NAME"), "/server")) + .header("x-served-by", SERVED_BY) .body(Body::empty()) .unwrap(), ); From 4146ba13e5031b4051d05d24b32f3dcfa0858be6 Mon Sep 17 00:00:00 2001 From: Nyannyacha Date: Fri, 24 Apr 2026 04:05:34 +0000 Subject: [PATCH 3/3] stamp: update tests --- crates/base/tests/integration_tests.rs | 74 +++++++------------------- 1 file changed, 18 insertions(+), 56 deletions(-) diff --git a/crates/base/tests/integration_tests.rs b/crates/base/tests/integration_tests.rs index c6b6f5751..4ab4188de 100644 --- a/crates/base/tests/integration_tests.rs +++ b/crates/base/tests/integration_tests.rs @@ -3,7 +3,6 @@ use std::borrow::Cow; use std::collections::HashMap; -use std::error::Error; use std::io; use std::io::BufRead; use std::io::Cursor; @@ -1267,25 +1266,16 @@ async fn req_failure_case_op_cancel_from_server_due_to_cpu_resource_limit() { 120 * MB, None, |resp| async { - if let Err(err) = resp { - assert_connection_aborted(err); - } else { - let res = resp.unwrap(); - - assert_eq!(res.status().as_u16(), 500); - - let res = res.json::().await; - - assert!(res.is_ok()); - - let msg = res.unwrap().msg; + let res = resp.unwrap(); - assert!( - msg - == "WorkerRequestCancelled: request has been cancelled by supervisor" - || msg == "broken pipe" - ); - } + assert_eq!(res.status().as_u16(), 503); + assert_eq!( + res + .headers() + .get("x-served-by") + .map(|v| v.to_str().unwrap()), + Some(concat!(env!("CARGO_PKG_NAME"), "/server")) + ); }, ) .await; @@ -1299,28 +1289,16 @@ async fn req_failure_case_op_cancel_from_server_due_to_cpu_resource_limit_2() { 10 * MB, Some("image/png"), |resp| async { - if let Err(err) = resp { - assert_connection_aborted(err); - } else { - let res = resp.unwrap(); - - assert_eq!(res.status().as_u16(), 500); - - let res = res.json::().await; - - assert!(res.is_ok()); - - let msg = res.unwrap().msg; + let res = resp.unwrap(); - assert!( - !msg.starts_with("TypeError: request body receiver not connected") - ); - assert!( - msg - == "WorkerRequestCancelled: request has been cancelled by supervisor" - || msg == "broken pipe" - ); - } + assert_eq!(res.status().as_u16(), 503); + assert_eq!( + res + .headers() + .get("x-served-by") + .map(|v| v.to_str().unwrap()), + Some(concat!(env!("CARGO_PKG_NAME"), "/server")) + ); }, ) .await; @@ -4642,19 +4620,3 @@ fn new_localhost_tls(secure: bool) -> Option { Tls::new(SECURE_PORT, TLS_LOCALHOST_KEY, TLS_LOCALHOST_CERT).unwrap() }) } - -fn assert_connection_aborted(err: reqwest::Error) { - let source = err.source(); - let hyper_err = source - .and_then(|err| err.downcast_ref::()) - .unwrap(); - - if hyper_err.is_incomplete_message() { - return; - } - - let cause = hyper_err.source().unwrap(); - let cause = cause.downcast_ref::().unwrap(); - - assert_eq!(cause.kind(), std::io::ErrorKind::ConnectionReset); -}