Skip to content

fix(base/server): return HTTP error responses instead of dropping connection in WorkerService #696

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
nyannyacha merged 3 commits into from
Apr 24, 2026
Merged

fix(base/server): return HTTP error responses instead of dropping connection in WorkerService #696

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
0ec8fcb
fix(base/server): return HTTP error responses instead of dropping con…
nyannyacha Apr 24, 2026
6d1f6ef
stamp: meow
nyannyacha Apr 24, 2026
4146ba1
stamp: update tests
nyannyacha Apr 24, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
27 changes: 19 additions & 8 deletions crates/base/src/server.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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)]
Expand Down Expand Up @@ -221,20 +223,29 @@ impl Service<Request<Body>> 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", SERVED_BY)
.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", SERVED_BY)
.body(Body::empty())
.unwrap(),
);
}

let res = match res {
Expand Down
74 changes: 18 additions & 56 deletions crates/base/tests/integration_tests.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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::<ErrorResponsePayload>().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;
Expand All @@ -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::<ErrorResponsePayload>().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;
Expand Down Expand Up @@ -4642,19 +4620,3 @@ fn new_localhost_tls(secure: bool) -> Option<Tls> {
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::<hyper::Error>())
.unwrap();

if hyper_err.is_incomplete_message() {
return;
}

let cause = hyper_err.source().unwrap();
let cause = cause.downcast_ref::<std::io::Error>().unwrap();

assert_eq!(cause.kind(), std::io::ErrorKind::ConnectionReset);
}
Loading