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

nyannyacha · web-flow · commit c47d831081d6 · 2026-04-24T13:23:36.000+09:00
* fix(base/server): return HTTP error responses instead of dropping connection in WorkerService

* stamp: meow

* stamp: update tests
diff --git 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)]
@@ -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 {
diff --git 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::<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;
@@ -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;
@@ -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);
-}
