From b62b7c8f2d70579fe677551695dea6a764d12e83 Mon Sep 17 00:00:00 2001 From: Hare Date: Fri, 18 Sep 2026 16:45:40 +0900 Subject: [PATCH] worker: await cleanup on automatic shutdown --- crates/worker/src/controller.rs | 28 ++++++++++++++++++++++++++-- 1 file changed, 26 insertions(+), 2 deletions(-) diff --git a/crates/worker/src/controller.rs b/crates/worker/src/controller.rs index a137134b2..2d3e5ad09 100644 --- a/crates/worker/src/controller.rs +++ b/crates/worker/src/controller.rs @@ -1719,6 +1719,15 @@ async fn controller_loop( ) .await; if shutdown { + while worker.has_pending_compaction_cleanup() { + if let Err(error) = worker.retry_pending_compaction_cleanup().await { + let _ = working_event_tx.send(Event::Error { + code: worker_error_code(&error), + message: error.to_string(), + }); + tokio::time::sleep(Duration::from_millis(50)).await; + } + } let _ = working_event_tx.send(Event::Shutdown); break; } @@ -3502,9 +3511,24 @@ mod tests { } #[tokio::test] - async fn pending_compaction_cleanup_from_run_never_publishes_idle() { + async fn pending_compaction_cleanup_preserves_shutdown_request_for_outer_barrier() { let mut env = make_env().await; + let method_tx = env._method_tx.clone(); + env.shared_state + .transition(WorkerState::Busy(WorkerBusyState::Run( + WorkerRunState::Running, + ))); + tokio::spawn(async move { + tokio::time::sleep(Duration::from_millis(10)).await; + method_tx + .send(Method::Shutdown { + command: WorkerCommandEnvelope::new(1), + }) + .await + .expect("send shutdown"); + }); let worker_future = async { + tokio::time::sleep(Duration::from_millis(50)).await; Err::(WorkerError::CompactionCleanupPending { source: crate::runtime::worker_allocation::ScopeLockError::UnknownWorker( "worker".into(), @@ -3531,7 +3555,7 @@ mod tests { .await; assert_eq!(status, WorkerStatus::Running); - assert!(!shutdown); + assert!(shutdown); assert!(!may_drain_pending); }