From d947e26165da46d87d801b2277ba9152a3931d45 Mon Sep 17 00:00:00 2001 From: limityan Date: Thu, 30 Jul 2026 18:55:45 +0800 Subject: [PATCH] fix(cli): make dispatch storage locks portable Use the existing fs2 dependency for real cross-process locks, recover incomplete event tails with a writable handle, and defer busy retention cleanup safely on Windows. Keep detached worker support limited to Linux and macOS. --- Cargo.lock | 1 + src/apps/cli/Cargo.toml | 1 + src/apps/cli/src/dispatch/mod.rs | 11 +- src/apps/cli/src/dispatch/store.rs | 161 ++++++++++++++++++++++------- 4 files changed, 137 insertions(+), 37 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 34cdd836a2..ee767e87f1 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -909,6 +909,7 @@ dependencies = [ "dirs 6.0.0", "dunce", "flate2", + "fs2", "futures-util", "hex", "libc", diff --git a/src/apps/cli/Cargo.toml b/src/apps/cli/Cargo.toml index 2599212d95..00bcce0284 100644 --- a/src/apps/cli/Cargo.toml +++ b/src/apps/cli/Cargo.toml @@ -81,6 +81,7 @@ tracing = { workspace = true } tracing-subscriber = { workspace = true } flate2 = { workspace = true } futures-util = { workspace = true } +fs2 = { workspace = true } base64 = { workspace = true } minisign-verify = "0.2" reqwest = { workspace = true } diff --git a/src/apps/cli/src/dispatch/mod.rs b/src/apps/cli/src/dispatch/mod.rs index 9fa331ff7f..56a36d3e36 100644 --- a/src/apps/cli/src/dispatch/mod.rs +++ b/src/apps/cli/src/dispatch/mod.rs @@ -751,15 +751,22 @@ mod tests { .write_pid("job-no-match", std::process::id()) .expect("record test pid"); - let error = cancel_in_store( + let terminate_called = std::cell::Cell::new(false); + let error = cancel_in_store_with_process_checks( &store, DispatchCancelRequest { job_id: "job-no-match".to_string(), }, - runner::terminate_worker, + |_pid| true, + |_pid, _job_id| false, + |_pid, _job_id| { + terminate_called.set(true); + Ok(true) + }, ) .expect_err("an unrelated live process must not be treated as cancelled"); assert!(error.to_string().contains("no longer matches")); + assert!(!terminate_called.get()); let state = store.load_state("job-no-match").expect("state"); assert_eq!(state.state, DispatchJobState::Failed); assert!(state.cancel_requested()); diff --git a/src/apps/cli/src/dispatch/store.rs b/src/apps/cli/src/dispatch/store.rs index c4f537d142..a4728073f4 100644 --- a/src/apps/cli/src/dispatch/store.rs +++ b/src/apps/cli/src/dispatch/store.rs @@ -878,7 +878,9 @@ impl DispatchStore { if metadata.file_type().is_symlink() || !metadata.is_dir() { continue; } - let lock = JobLock::exclusive(&job_dir.join(".lock"))?; + let Some(lock) = JobLock::try_exclusive(&job_dir.join(".lock"))? else { + continue; + }; let state = match self.load_state_unlocked(&job_dir) { Ok(state) => state, Err(error) => { @@ -908,10 +910,33 @@ impl DispatchStore { job_id, uuid::Uuid::new_v4().as_simple() )); - fs::rename(&job_dir, &tombstone).with_context(|| { - format!("quarantine expired dispatch job {}", job_dir.display()) - })?; + + // Windows cannot rename a directory while a child lock file is + // open. Terminal states are irreversible, so release that handle + // immediately before the atomic quarantine rename. A concurrent + // opener makes the rename fail and the job is retried later. + #[cfg(windows)] drop(lock); + let rename_result = fs::rename(&job_dir, &tombstone); + #[cfg(not(windows))] + drop(lock); + match rename_result { + Ok(()) => {} + Err(error) if retryable_retention_rename_error(&error) => { + tracing::debug!( + job_id = %job_id, + error_kind = ?error.kind(), + raw_os_error = ?error.raw_os_error(), + "Deferring dispatch retention cleanup for busy job" + ); + continue; + } + Err(error) => { + return Err(error).with_context(|| { + format!("quarantine expired dispatch job {}", job_dir.display()) + }); + } + } fs::remove_dir_all(&tombstone) .with_context(|| format!("remove expired dispatch job {}", tombstone.display()))?; @@ -1020,8 +1045,8 @@ impl DispatchStore { let _lock = FileLock::exclusive(&lock_file)?; let path = job_dir.join(EVENTS_FILE); let mut file = OpenOptions::new() - .append(true) .read(true) + .write(true) .open(&path) .with_context(|| format!("open dispatch events {}", path.display()))?; set_private_file_permissions(&path)?; @@ -1049,6 +1074,7 @@ impl DispatchStore { atomic_write_json(&metadata_path, &event_metadata)?; } let physical_len = truncate_incomplete_event_tail(&mut file)?; + file.seek(SeekFrom::Start(physical_len))?; let current_len = physical_len.saturating_sub(data_start); if current_len .saturating_add(encoded.len() as u64) @@ -1547,45 +1573,27 @@ pub(super) fn remove_file_if_present(path: &Path) { } } -#[cfg(unix)] fn lock_file(file: &File, exclusive: bool) -> Result<()> { - use std::os::fd::AsRawFd; - let operation = if exclusive { - libc::LOCK_EX + let result = if exclusive { + fs2::FileExt::lock_exclusive(file) } else { - libc::LOCK_SH + fs2::FileExt::lock_shared(file) }; - // SAFETY: flock only operates on this live file descriptor. - if unsafe { libc::flock(file.as_raw_fd(), operation) } == 0 { - Ok(()) - } else { - Err(std::io::Error::last_os_error()).context("lock dispatch file") - } + result.context("lock dispatch file") } -#[cfg(unix)] fn try_lock_file_exclusive(file: &File) -> Result { - use std::os::fd::AsRawFd; - // SAFETY: flock only operates on this live file descriptor. - if unsafe { libc::flock(file.as_raw_fd(), libc::LOCK_EX | libc::LOCK_NB) } == 0 { - return Ok(true); - } - let error = std::io::Error::last_os_error(); - if error.kind() == std::io::ErrorKind::WouldBlock { - Ok(false) - } else { - Err(error).context("try lock dispatch file") + match fs2::FileExt::try_lock_exclusive(file) { + Ok(()) => Ok(true), + Err(error) if error.raw_os_error() == fs2::lock_contended_error().raw_os_error() => { + Ok(false) + } + Err(error) => Err(error).context("try lock dispatch file"), } } -#[cfg(not(unix))] -fn lock_file(_file: &File, _exclusive: bool) -> Result<()> { - Ok(()) -} - -#[cfg(not(unix))] -fn try_lock_file_exclusive(_file: &File) -> Result { - Ok(true) +fn retryable_retention_rename_error(error: &std::io::Error) -> bool { + cfg!(windows) && error.kind() == std::io::ErrorKind::PermissionDenied } #[cfg(test)] @@ -1682,6 +1690,7 @@ mod tests { file.write_all(br#"{"type":"jobState","timestamp":"partial""#) .expect("write partial line"); file.sync_all().expect("sync partial line"); + drop(file); let page = store .read_events("job-2", initial.cursor) @@ -2385,6 +2394,88 @@ mod tests { assert!(store.root.join("workspaces/running").exists()); } + #[test] + fn retention_skips_contended_job_without_blocking_and_retries_later() { + let (_dir, store) = store(); + store + .create_job(request("contended"), "Task".to_string()) + .expect("create job"); + store + .mark_state("contended", DispatchJobState::Succeeded, None, None) + .expect("mark terminal"); + let now = chrono::Utc::now(); + let mut expired = store.load_state("contended").expect("expired state"); + expired.finished_at = + Some((now - chrono::Duration::days(TERMINAL_JOB_RETENTION_DAYS + 1)).to_rfc3339()); + let job_dir = store.job_dir("contended").expect("job path"); + atomic_write_json(&job_dir.join(STATE_FILE), &expired).expect("age terminal state"); + + let lock = JobLock::exclusive(&job_dir.join(".lock")).expect("hold job lock"); + let collecting_store = store.clone(); + let (sender, receiver) = std::sync::mpsc::channel(); + let collector = std::thread::spawn(move || { + let result = collecting_store.collect_expired_terminal_jobs(now); + sender.send(result).expect("send retention result"); + }); + let while_contended = receiver.recv_timeout(std::time::Duration::from_secs(1)); + drop(lock); + collector.join().expect("join retention collector"); + + assert_eq!( + while_contended + .expect("retention must not block on a busy job") + .expect("skip contended job"), + 0 + ); + assert!(job_dir.exists()); + assert_eq!( + store + .collect_expired_terminal_jobs(now) + .expect("retry expired job"), + 1 + ); + assert!(!job_dir.exists()); + } + + #[cfg(windows)] + #[test] + fn retention_retries_after_windows_sharing_violation() { + let (_dir, store) = store(); + store + .create_job(request("sharing-violation"), "Task".to_string()) + .expect("create job"); + store + .mark_state("sharing-violation", DispatchJobState::Succeeded, None, None) + .expect("mark terminal"); + let now = chrono::Utc::now(); + let mut expired = store + .load_state("sharing-violation") + .expect("expired state"); + expired.finished_at = + Some((now - chrono::Duration::days(TERMINAL_JOB_RETENTION_DAYS + 1)).to_rfc3339()); + let job_dir = store.job_dir("sharing-violation").expect("job path"); + let state_path = job_dir.join(STATE_FILE); + atomic_write_json(&state_path, &expired).expect("age terminal state"); + + let open_state = File::open(&state_path).expect("hold state file open"); + assert_eq!( + store + .collect_expired_terminal_jobs(now) + .expect("sharing violation must defer cleanup"), + 0 + ); + assert!(job_dir.exists()); + + drop(open_state); + assert_eq!( + store + .collect_expired_terminal_jobs(now) + .expect("retry expired job"), + 1 + ); + assert!(!job_dir.exists()); + } + #[cfg(unix)] #[test] fn job_storage_uses_owner_only_permissions() {