fix: return an error instead of a response when a request idle timeout occurs for requests sent to the user worker (#679)

nyannyacha · web-flow · commit 76c2581a1677 · 2026-04-01T11:06:19.000+01:00
diff --git a/crates/base/src/server.rs b/crates/base/src/server.rs
@@ -4,6 +4,7 @@ use std::future::Future;
 use std::net::SocketAddr;
 use std::pin::Pin;
 use std::str;
+use std::sync::atomic::AtomicBool;
 use std::sync::Arc;
 use std::task::Poll;
 use std::time::Duration;
@@ -196,6 +197,7 @@ impl Service<Request<Body>> for WorkerService {
         req,
         res_tx,
         conn_token: Some(cancel.clone()),
+        idle_timed_out: Arc::new(AtomicBool::new(false)),
       };
 
       worker_req_tx.send(msg)?;
diff --git a/crates/base/src/utils/test_utils.rs b/crates/base/src/utils/test_utils.rs
@@ -3,6 +3,7 @@
 use std::marker::PhantomPinned;
 use std::path::Path;
 use std::path::PathBuf;
+use std::sync::atomic::AtomicBool;
 use std::sync::Arc;
 use std::task::ready;
 use std::task::Poll;
@@ -297,6 +298,7 @@ impl TestBed {
       req,
       res_tx,
       conn_token: Some(conn_token.clone()),
+      idle_timed_out: Arc::new(AtomicBool::new(false)),
     });
 
     let Ok(res) = res_rx.await else {
diff --git a/crates/base/src/worker/utils.rs b/crates/base/src/worker/utils.rs
@@ -1,4 +1,7 @@
 use std::collections::HashMap;
+use std::sync::atomic::AtomicBool;
+use std::sync::atomic::Ordering;
+use std::sync::Arc;
 
 use anyhow::anyhow;
 use anyhow::bail;
@@ -63,10 +66,12 @@ pub async fn send_user_worker_request(
 ) -> Result<Response<Body>, anyhow::Error> {
   let (res_tx, res_rx) =
     oneshot::channel::<Result<Response<Body>, hyper_v014::Error>>();
+  let idle_timed_out = Arc::new(AtomicBool::new(false));
   let msg = WorkerRequestMsg {
     req,
     res_tx,
     conn_token,
+    idle_timed_out: idle_timed_out.clone(),
   };
 
   // send the message to worker
@@ -88,7 +93,9 @@ pub async fn send_user_worker_request(
 
   match res {
     Ok(v) => {
-      // send the response back to the caller
+      if idle_timed_out.load(Ordering::Relaxed) {
+        bail!(WorkerError::RequestIdleTimeout)
+      }
       Ok(v)
     }
 
diff --git a/crates/base/src/worker/worker_surface_creation.rs b/crates/base/src/worker/worker_surface_creation.rs
@@ -35,6 +35,7 @@ use super::WorkerSurface;
 mod request {
   use std::future::pending;
   use std::io::ErrorKind;
+  use std::sync::atomic::Ordering;
   use std::sync::Arc;
   use std::time::Duration;
 
@@ -90,6 +91,7 @@ mod request {
       mut req,
       res_tx,
       conn_token,
+      idle_timed_out,
     } = msg;
 
     let _ = duplex_stream_tx.send((theirs, conn_token.clone()));
@@ -155,6 +157,7 @@ mod request {
     let res = tokio::select! {
       resp = request_sender.send_request(req) => resp,
       _ = maybe_cancel_fut => {
+        idle_timed_out.store(true, Ordering::Relaxed);
         Ok(emit_status_code(
           http_v02::StatusCode::GATEWAY_TIMEOUT,
           None,
diff --git a/crates/base/test_cases/main/index.ts b/crates/base/test_cases/main/index.ts
@@ -111,6 +111,14 @@ Deno.serve(async (req: Request) => {
       // 	return await callWorker();
       // }
 
+      if (e instanceof Deno.errors.WorkerRequestIdleTimeout) {
+        const error = { msg: e.toString() };
+        return new Response(
+          JSON.stringify(error),
+          { status: 504, headers: { "Content-Type": "application/json" } },
+        );
+      }
+
       const error = { msg: e.toString() };
       return new Response(
         JSON.stringify(error),
diff --git a/crates/base/tests/integration_tests.rs b/crates/base/tests/integration_tests.rs
@@ -12,6 +12,7 @@ use std::net::Ipv4Addr;
 use std::net::SocketAddr;
 use std::path::PathBuf;
 use std::str::FromStr;
+use std::sync::atomic::AtomicBool;
 use std::sync::Arc;
 use std::time::Duration;
 
@@ -265,6 +266,7 @@ async fn test_not_trigger_pku_sigsegv_due_to_jit_compilation_non_cli() {
     req,
     res_tx,
     conn_token: Some(conn_token.clone()),
+    idle_timed_out: Arc::new(AtomicBool::new(false)),
   };
 
   let _ = surface.msg_tx.send(msg);
@@ -549,6 +551,7 @@ async fn test_main_worker_user_worker_mod_evaluate_exception() {
     req,
     res_tx,
     conn_token: Some(conn_token.clone()),
+    idle_timed_out: Arc::new(AtomicBool::new(false)),
   };
 
   let _ = surface.msg_tx.send(msg);
@@ -1893,7 +1896,15 @@ async fn test_request_idle_timeout_no_streamed_response(
     request_builder,
     maybe_tls,
     (|resp| async {
-      assert_eq!(resp.unwrap().status().as_u16(), StatusCode::GATEWAY_TIMEOUT);
+      let resp = resp.unwrap();
+      assert_eq!(resp.status().as_u16(), StatusCode::GATEWAY_TIMEOUT);
+      let body = resp.bytes().await.unwrap();
+      assert!(
+        std::str::from_utf8(&body)
+          .unwrap()
+          .contains("WorkerRequestIdleTimeout"),
+        "expected WorkerRequestIdleTimeout in body"
+      );
     }),
     TerminationToken::new()
   );
diff --git a/examples/main/index.ts b/examples/main/index.ts
@@ -240,6 +240,15 @@ Deno.serve(async (req: Request) => {
       if (e instanceof Deno.errors.WorkerAlreadyRetired) {
         return await callWorker();
       }
+      if (e instanceof Deno.errors.WorkerRequestIdleTimeout) {
+        return new Response(
+          JSON.stringify({ msg: e.toString() }),
+          {
+            status: STATUS_CODE.GatewayTimeout,
+            headers,
+          },
+        );
+      }
       if (e instanceof Deno.errors.WorkerRequestCancelled) {
         headers.append("Connection", "close");
 
diff --git a/ext/runtime/js/errors.js b/ext/runtime/js/errors.js
@@ -29,6 +29,7 @@ const InvalidWorkerResponse = buildErrorClass("InvalidWorkerResponse");
 const InvalidWorkerCreation = buildErrorClass("InvalidWorkerCreation");
 const WorkerRequestCancelled = buildErrorClass("WorkerRequestCancelled");
 const WorkerAlreadyRetired = buildErrorClass("WorkerAlreadyRetired");
+const WorkerRequestIdleTimeout = buildErrorClass("WorkerRequestIdleTimeout");
 const NotFound = buildErrorClass("NotFound");
 const PermissionDenied = buildErrorClass("PermissionDenied");
 const ConnectionRefused = buildErrorClass("ConnectionRefused");
@@ -78,6 +79,7 @@ function registerErrors() {
   core.registerErrorClass("InvalidWorkerCreation", InvalidWorkerCreation);
   core.registerErrorClass("WorkerRequestCancelled", WorkerRequestCancelled);
   core.registerErrorClass("WorkerAlreadyRetired", WorkerAlreadyRetired);
+  core.registerErrorClass("WorkerRequestIdleTimeout", WorkerRequestIdleTimeout);
   core.registerErrorClass("NotFound", NotFound);
   core.registerErrorClass("PermissionDenied", PermissionDenied);
   core.registerErrorClass("ConnectionRefused", ConnectionRefused);
diff --git a/ext/workers/context.rs b/ext/workers/context.rs
@@ -1,5 +1,6 @@
 use std::collections::HashMap;
 use std::path::PathBuf;
+use std::sync::atomic::AtomicBool;
 use std::sync::atomic::AtomicUsize;
 use std::sync::Arc;
 use std::sync::RwLock;
@@ -310,4 +311,5 @@ pub struct WorkerRequestMsg {
   pub req: Request<Body>,
   pub res_tx: oneshot::Sender<Result<Response<Body>, hyper_v014::Error>>,
   pub conn_token: Option<CancellationToken>,
+  pub idle_timed_out: Arc<AtomicBool>,
 }
diff --git a/ext/workers/errors.rs b/ext/workers/errors.rs
@@ -6,4 +6,6 @@ pub enum WorkerError {
   RequestCancelledBySupervisor,
   #[error("request cannot be handled because the worker has already retired")]
   WorkerAlreadyRetired,
+  #[error("request timed out")]
+  RequestIdleTimeout,
 }
diff --git a/ext/workers/lib.rs b/ext/workers/lib.rs
@@ -634,6 +634,12 @@ pub async fn op_user_worker_fetch_send(
         Some(err @ WorkerError::WorkerAlreadyRetired) => {
           return Err(custom_error("WorkerAlreadyRetired", err.to_string()));
         }
+        Some(err @ WorkerError::RequestIdleTimeout) => {
+          return Err(custom_error(
+            "WorkerRequestIdleTimeout",
+            err.to_string(),
+          ));
+        }
 
         None => {
           return Err(custom_error("InvalidWorkerResponse", err.to_string()));
diff --git a/types/global.d.ts b/types/global.d.ts
@@ -154,6 +154,10 @@ interface WorkerHeapStatisticsWithServicePath {
 interface RuntimeMetrics {
   mainWorkerHeapStats: HeapStatistics;
   eventWorkerHeapStats?: HeapStatistics;
+  activeUserWorkersCount: number;
+  retiredUserWorkersCount: number;
+  receivedRequestsCount: number;
+  handledRequestsCount: number;
 }
 
 interface MemInfo {
@@ -252,6 +256,7 @@ declare namespace Deno {
   export namespace errors {
     class WorkerRequestCancelled extends Error {}
     class WorkerAlreadyRetired extends Error {}
+    class WorkerRequestIdleTimeout extends Error {}
 
     /** Thrown when an outbound HTTP request is blocked by the rate limiter. */
     class RateLimitError extends Error {

Original file line number	Diff line number	Diff line change
`@@ -6,4 +6,6 @@ pub enum WorkerError {`
`6`	`6`	`RequestCancelledBySupervisor,`
`7`	`7`	`#[error("request cannot be handled because the worker has already retired")]`
`8`	`8`	`WorkerAlreadyRetired,`
	`9`	`+ #[error("request timed out")]`
	`10`	`+ RequestIdleTimeout,`
`9`	`11`	`}`