From 2b4bf4b7bdfd6bcae9025e070477c3929660b621 Mon Sep 17 00:00:00 2001 From: Hare Date: Fri, 30 Oct 2026 20:33:57 +0900 Subject: [PATCH] fix: log Workdir cleanup busy reasons --- crates/worker-runtime/src/runtime.rs | 106 ++++++++++++++++-- .../src/working_directory/occupancy.rs | 91 ++++++++++++++- 2 files changed, 183 insertions(+), 14 deletions(-) diff --git a/crates/worker-runtime/src/runtime.rs b/crates/worker-runtime/src/runtime.rs index 57b70bc8e..03c968e20 100644 --- a/crates/worker-runtime/src/runtime.rs +++ b/crates/worker-runtime/src/runtime.rs @@ -101,7 +101,18 @@ impl Drop for WorkdirOperationGuard { } } -fn workdir_operation_busy() -> RuntimeError { +fn workdir_operation_busy(id: &str, action: &str, blocking: &WorkdirOperation) -> RuntimeError { + let (busy_reason, active_use_count) = match blocking { + WorkdirOperation::Use(count) => ("active_use", *count), + WorkdirOperation::Cleanup => ("cleanup_in_progress", 0), + }; + tracing::warn!(target: "yoi::workdir_cleanup", + working_directory_id = id, + action, + busy_source = "runtime_operation", + busy_reason, + active_use_count, + "Workdir operation blocked by busy resource"); RuntimeError::WorkingDirectory( crate::working_directory::WorkingDirectoryDiagnostic::rejected( "working_directory_cleanup_resource_busy", @@ -1148,8 +1159,12 @@ impl Runtime { let backend = { let mut state = self.lock()?; state.ensure_running()?; - if state.workdir_operations.contains_key(working_directory_id) { - return Err(workdir_operation_busy()); + if let Some(blocking) = state.workdir_operations.get(working_directory_id) { + return Err(workdir_operation_busy( + working_directory_id, + "cleanup_working_directory", + blocking, + )); } if let Some(worker_id) = state.worker_id_for_workdir(working_directory_id) { return Err(RuntimeError::InvalidRequest(format!( @@ -6318,11 +6333,8 @@ impl RuntimeState { } fn ensure_workdir_not_removing(&self, id: &str) -> Result<(), RuntimeError> { - if matches!( - self.workdir_operations.get(id), - Some(WorkdirOperation::Cleanup) - ) { - return Err(workdir_operation_busy()); + if let Some(blocking @ WorkdirOperation::Cleanup) = self.workdir_operations.get(id) { + return Err(workdir_operation_busy(id, "admit_use", blocking)); } Ok(()) } @@ -9345,6 +9357,84 @@ mod tests { .block_on(subscription.recv()) } + #[test] + fn busy_admission_logs_runtime_use_and_cleanup_blockers() { + use std::io::{Read, Seek, SeekFrom}; + + let mut output = tempfile::tempfile().unwrap(); + let writer = output.try_clone().unwrap(); + let subscriber = tracing_subscriber::fmt() + .with_ansi(false) + .without_time() + .with_writer(move || writer.try_clone().unwrap()) + .finish(); + tracing::subscriber::with_default(subscriber, || { + tracing::callsite::rebuild_interest_cache(); + let (runtime, backend) = runtime_and_backend(); + let first = runtime.reserve_workdir_use(["workdir-used"]).unwrap(); + let second = runtime.reserve_workdir_use(["workdir-used"]).unwrap(); + assert_workdir_busy( + runtime + .cleanup_working_directory("workdir-used") + .unwrap_err(), + ); + drop((first, second)); + assert!(runtime.inner.lock().unwrap().workdir_operations.is_empty()); + + let (entered, release) = WorkdirEffectGate::install(&backend.cleanup_gate); + let cleanup_runtime = runtime.clone(); + let cleanup = std::thread::spawn(move || { + cleanup_runtime.cleanup_working_directory("workdir-removing") + }); + entered + .recv_timeout(std::time::Duration::from_secs(5)) + .unwrap(); + assert_workdir_busy( + runtime + .cleanup_working_directory("workdir-removing") + .unwrap_err(), + ); + assert_workdir_busy( + runtime + .reserve_workdir_use(["workdir-removing"]) + .unwrap_err(), + ); + release.send(()).unwrap(); + assert_workdir_busy(cleanup.join().unwrap().unwrap_err()); + assert!(runtime.inner.lock().unwrap().workdir_operations.is_empty()); + }); + output.seek(SeekFrom::Start(0)).unwrap(); + let mut log = String::new(); + output.read_to_string(&mut log).unwrap(); + let events: Vec<_> = log + .lines() + .filter(|line| line.contains("busy_source=")) + .collect(); + assert_eq!(events.len(), 3, "{log}"); + for (event, (id, action, reason, count)) in events.iter().zip([ + ("workdir-used", "cleanup_working_directory", "active_use", 2), + ( + "workdir-removing", + "cleanup_working_directory", + "cleanup_in_progress", + 0, + ), + ("workdir-removing", "admit_use", "cleanup_in_progress", 0), + ]) { + for field in [ + "WARN".to_string(), + "yoi::workdir_cleanup".to_string(), + format!("working_directory_id=\"{id}\""), + format!("action=\"{action}\""), + "busy_source=\"runtime_operation\"".to_string(), + format!("busy_reason=\"{reason}\""), + format!("active_use_count={count}"), + ] { + assert!(event.contains(&field), "missing {field}: {event}"); + } + } + } + #[test] fn workdir_removal_reserves_only_its_target_and_releases_on_failure() { let (runtime, backend) = runtime_and_backend(); diff --git a/crates/worker-runtime/src/working_directory/occupancy.rs b/crates/worker-runtime/src/working_directory/occupancy.rs index c3e477bb5..031d1cda7 100644 --- a/crates/worker-runtime/src/working_directory/occupancy.rs +++ b/crates/worker-runtime/src/working_directory/occupancy.rs @@ -28,7 +28,19 @@ pub(super) struct Lease { id: String, } -fn busy() -> WorkingDirectoryDiagnostic { +fn busy(id: &str, action: &str, blocking: &Entry) -> WorkingDirectoryDiagnostic { + let (busy_reason, active_use_count) = match blocking { + Entry::Use(count) => ("active_use", *count), + Entry::Create => ("creation_in_progress", 0), + Entry::Cleanup => ("cleanup_in_progress", 0), + }; + tracing::warn!(target: "yoi::workdir_cleanup", + working_directory_id = id, + action, + busy_source = "materializer_occupancy", + busy_reason, + active_use_count, + "Workdir operation blocked by busy resource"); WorkingDirectoryDiagnostic::new( "working_directory_cleanup_resource_busy", "A filesystem resource is busy or changed during removal. Release the resource and retry.", @@ -42,7 +54,9 @@ impl Occupancy { .lock() .unwrap_or_else(|error| error.into_inner()); match entries.get_mut(id) { - Some(Entry::Cleanup | Entry::Create) => return Err(busy()), + Some(blocking @ (Entry::Cleanup | Entry::Create)) => { + return Err(busy(id, "acquire_use", blocking)); + } Some(Entry::Use(count)) => *count += 1, None => { entries.insert(id.to_string(), Entry::Use(1)); @@ -101,8 +115,8 @@ impl Occupancy { .entries .lock() .unwrap_or_else(|error| error.into_inner()); - if entries.contains_key(id) { - return Err(busy()); + if let Some(blocking) = entries.get(id) { + return Err(busy(id, "acquire_creation", blocking)); } entries.insert(id.to_string(), Entry::Create); Ok(Lease { @@ -116,8 +130,8 @@ impl Occupancy { .entries .lock() .unwrap_or_else(|error| error.into_inner()); - if entries.contains_key(id) { - return Err(busy()); + if let Some(blocking) = entries.get(id) { + return Err(busy(id, "acquire_cleanup", blocking)); } entries.insert(id.to_string(), Entry::Cleanup); Ok(Lease { @@ -131,6 +145,71 @@ impl Occupancy { mod tests { use super::*; + #[test] + fn busy_admission_logs_blocker_and_use_count_without_changing_error() { + use std::io::{Read, Seek, SeekFrom}; + + let mut output = tempfile::tempfile().unwrap(); + let writer = output.try_clone().unwrap(); + let subscriber = tracing_subscriber::fmt() + .with_ansi(false) + .without_time() + .with_writer(move || writer.try_clone().unwrap()) + .finish(); + tracing::subscriber::with_default(subscriber, || { + tracing::callsite::rebuild_interest_cache(); + let occupancy = Occupancy::default(); + for (id, leases) in [ + ( + "workdir-used", + vec![ + occupancy.acquire_use("workdir-used").unwrap(), + occupancy.acquire_use("workdir-used").unwrap(), + ], + ), + ( + "workdir-creating", + vec![occupancy.acquire_creation("workdir-creating").unwrap()], + ), + ( + "workdir-removing", + vec![occupancy.acquire_cleanup("workdir-removing").unwrap()], + ), + ] { + let error = occupancy.acquire_cleanup(id).unwrap_err(); + assert_eq!(error.code, "working_directory_cleanup_resource_busy"); + assert_eq!( + error.message, + "A filesystem resource is busy or changed during removal. Release the resource and retry." + ); + drop(leases); + drop(occupancy.acquire_cleanup(id).unwrap()); + } + }); + output.seek(SeekFrom::Start(0)).unwrap(); + let mut log = String::new(); + output.read_to_string(&mut log).unwrap(); + let events: Vec<_> = log.lines().collect(); + assert_eq!(events.len(), 3, "{log}"); + for (event, (id, reason, count)) in events.iter().zip([ + ("workdir-used", "active_use", 2), + ("workdir-creating", "creation_in_progress", 0), + ("workdir-removing", "cleanup_in_progress", 0), + ]) { + for field in [ + "WARN".to_string(), + "yoi::workdir_cleanup".to_string(), + format!("working_directory_id=\"{id}\""), + "action=\"acquire_cleanup\"".to_string(), + "busy_source=\"materializer_occupancy\"".to_string(), + format!("busy_reason=\"{reason}\""), + format!("active_use_count={count}"), + ] { + assert!(event.contains(&field), "missing {field}: {event}"); + } + } + } + #[test] fn use_and_cleanup_leases_release_during_unwind() { let occupancy = Occupancy::default();