diff --git a/crates/base/src/worker/supervisor/strategy_per_worker.rs b/crates/base/src/worker/supervisor/strategy_per_worker.rs index 516e1882e..d3f756ff3 100644 --- a/crates/base/src/worker/supervisor/strategy_per_worker.rs +++ b/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, + 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(); diff --git a/crates/base/test_cases/main/index.ts b/crates/base/test_cases/main/index.ts index edf88ffa0..63110fc89 100644 --- a/crates/base/test_cases/main/index.ts +++ b/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", + ), }, }; diff --git a/crates/base/tests/integration_tests.rs b/crates/base/tests/integration_tests.rs index 1e76dfbab..6d41d7729 100644 --- a/crates/base/tests/integration_tests.rs +++ b/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") }) diff --git a/types/global.d.ts b/types/global.d.ts index a00ee6dc2..fc571ed3f 100644 --- a/types/global.d.ts +++ b/types/global.d.ts @@ -60,6 +60,7 @@ interface UserWorkerCreateContext { useReadSyncFileAPI?: boolean | null; supervisor?: { requestAbsentTimeoutMs?: number | null; + cleanupIdleGraceMs?: number | null; }; otel?: { [attribute: string]: string;