fix(base/worker): guard cleanup-triggered early drop against in-flight requests

crates/base/src/worker/supervisor/strategy_per_worker.rs

@@ -36,9 +36,17 @@ use super::Arguments;
 use super::CPUUsageMetrics;
 use super::V8HandleTerminationData;
 
+// Minimum time a worker must sit at req_ack_count == req_demand before a
+// `TryCleanupIdleWorkers` sweep is allowed to drop it. Guards against
+// dropping a worker whose next request is already in flight but hasn't yet
+// reached it (most prominently right after boot, where demand stays at 0
+// until the first request arrives).
+const DEFAULT_CLEANUP_IDLE_GRACE_MS: u64 = 100;
+
 #[derive(Debug, Default)]
 struct State {
   req_absent_duration: Option<Duration>,
+  cleanup_idle_grace: Duration,
 
   is_worker_entered: bool,
   is_wall_clock_limit_disabled: bool,
@@ -116,6 +124,13 @@ impl State {
       && self.promise.have_all_promises_been_resolved()
   }
 
+  fn has_been_idle_past_grace(&self) -> bool {
+    self
+      .last_req_ack
+      .and_then(|t| t.checked_add(self.cleanup_idle_grace))
+      .is_some_and(|t| t <= SystemTime::now())
+  }
+
   fn can_early_drop(&self) -> bool {
     self.has_resource_alert() && self.have_all_pending_tasks_been_resolved()
   }
@@ -168,6 +183,15 @@ pub async fn supervise(args: Arguments) -> (ShutdownReason, i64) {
           .map(Duration::from_millis)
       }),
 
+    cleanup_idle_grace: runtime_opts
+      .context
+      .as_ref()
+      .and_then(|it| it.get("supervisor"))
+      .and_then(|it| it.get("cleanupIdleGraceMs"))
+      .and_then(|it| it.as_u64())
+      .map(Duration::from_millis)
+      .unwrap_or(Duration::from_millis(DEFAULT_CLEANUP_IDLE_GRACE_MS)),
+
     is_wall_clock_limit_disabled: worker_timeout_ms == 0,
     is_cpu_time_limit_disabled: cpu_time_soft_limit_ms == 0
       && cpu_time_hard_limit_ms == 0,
@@ -471,7 +495,9 @@ pub async fn supervise(args: Arguments) -> (ShutdownReason, i64) {
 
       Some(tx) = early_drop_rx.recv() => {
         let mut acknowledged = false;
-        if state.have_all_pending_tasks_been_resolved() {
+        if state.has_been_idle_past_grace()
+          && state.have_all_pending_tasks_been_resolved()
+        {
           if let Some(func) = dispatch_early_drop_beforeunload_fn.take() {
             early_retire_fn();
             func();

crates/base/test_cases/main/index.ts

@@ -84,6 +84,10 @@ Deno.serve(async (req: Request) => {
           req,
           "x-context-supervisor-request-absent-timeout-ms",
         ),
+        cleanupIdleGraceMs: parseIntFromHeadersOrDefault(
+          req,
+          "x-context-supervisor-cleanup-idle-grace-ms",
+        ),
       },
     };
 

crates/base/tests/integration_tests.rs

@@ -4029,6 +4029,13 @@ async fn test_user_workers_cleanup_idle_workers() {
     .request(|b| {
       b.uri("/sleep-5000ms")
         .header("x-worker-timeout-ms", HeaderValue::from_static("3600000"))
+        // Disable the cleanup-idle grace period so the worker becomes
+        // eligible for cleanup as soon as this request is acknowledged,
+        // matching this test's assumption that it's dropped immediately.
+        .header(
+          "x-context-supervisor-cleanup-idle-grace-ms",
+          HeaderValue::from_static("0"),
+        )
         .body(Body::empty())
         .context("can't make request")
     })

types/global.d.ts

@@ -60,6 +60,7 @@ interface UserWorkerCreateContext {
   useReadSyncFileAPI?: boolean | null;
   supervisor?: {
     requestAbsentTimeoutMs?: number | null;
+    cleanupIdleGraceMs?: number | null;
   };
   otel?: {
     [attribute: string]: string;