diff --git a/crates/ide-agent/src/lib.rs b/crates/ide-agent/src/lib.rs index a65400a6e..7116047f1 100644 --- a/crates/ide-agent/src/lib.rs +++ b/crates/ide-agent/src/lib.rs @@ -58,6 +58,8 @@ pub(crate) const LIMITS_UPDATE: &str = "openidea_limits"; const MAX_EMBED: u64 = 256 * 1024; /// How long a probe waits for the agent to answer `initialize`. const PROBE_TIMEOUT: Duration = Duration::from_secs(30); +/// How long deleting a task waits for its agent to exit before removing the worktree. +const EXIT_WAIT: Duration = Duration::from_secs(5); /// The agent tasks of one project. pub struct AgentManager { @@ -839,7 +841,7 @@ impl AgentManager { /// Forgets the task; with `remove_worktree` also deletes its worktree (the branch stays). pub async fn delete(&self, id: &str, remove_worktree: bool) -> IdeResult<()> { let task = self.task(id)?; - task.kill(); + let live = task.kill(); let (isolation, cwd, root) = { let s = task.state.lock(); ( @@ -851,6 +853,11 @@ impl AgentManager { // A copy of the task (a fork) may work in the same worktree. let shared = self.tasks.read().values().any(|t| t.id != id && t.state.lock().task.cwd.0 == cwd); if remove_worktree && !shared && isolation == Isolation::Worktree && cwd.exists() { + // On Windows the agent's processes lock the folder (their working directory, + // open files) until they have exited. + if let Some(peer) = live { + let _ = tokio::time::timeout(EXIT_WAIT, peer.exited()).await; + } let worktrees = self.shared.paths.worktrees_dir.clone(); blocking(move || worktree::remove(&root, &worktrees, &cwd)).await?; } @@ -1605,11 +1612,12 @@ impl Task { } } - fn kill(&self) { + /// Stops the agent; returns its connection, to wait for its process to exit. + fn kill(&self) -> Option { self.cancel_permissions(); - if let Some(live) = self.live.lock().take() { - live.peer.kill(); - } + let live = self.live.lock().take()?; + live.peer.kill(); + Some(live.peer) } } diff --git a/crates/ide-agent/src/tests.rs b/crates/ide-agent/src/tests.rs index e33338a77..0b30157c4 100644 --- a/crates/ide-agent/src/tests.rs +++ b/crates/ide-agent/src/tests.rs @@ -1109,3 +1109,74 @@ async fn idle_agents_stop_and_resume_their_session_on_the_next_message() { assert!(live(&mgr, &task.id)); mgr.shutdown(); } + +/// Deleting a task whose agent still runs in the worktree removes both and leaves no +/// process behind. On Windows the running agent would lock the folder (os error 32); +/// that the wait comes first is checked by `ide-jsonrpc`'s `exited` tests. +#[cfg(unix)] +#[tokio::test(flavor = "multi_thread")] +async fn deleting_a_task_waits_for_its_agent_to_exit_before_removing_the_worktree() { + let tmp = tempfile::tempdir().unwrap(); + let root = dunce::canonicalize(tmp.path()).unwrap(); + let git = |args: &[&str]| { + let out = std::process::Command::new("git") + .args(["-c", "user.name=t", "-c", "user.email=t@t", "-c", "commit.gpgsign=false"]) + .args(args) + .current_dir(&root) + .output() + .unwrap(); + assert!(out.status.success(), "{}", String::from_utf8_lossy(&out.stderr)); + }; + git(&["init", "-q", "-b", "main"]); + std::fs::write(root.join("file.txt"), "original\n").unwrap(); + git(&["add", "."]); + git(&["commit", "-qm", "initial"]); + + let agent = Arc::new(FakeAgent::default()); + let pids: Arc>> = Arc::default(); + let (fake, started) = (agent.clone(), pids.clone()); + // The fake agent, served on behalf of a real process that exits only when killed. + let connector: Connector = Arc::new(move |_spec, cwd, client, _stderr| { + let mut cmd = tokio::process::Command::new("sleep"); + cmd.arg("30").current_dir(cwd).kill_on_drop(true); + let child = cmd.spawn()?; + started.lock().push(child.id().unwrap()); + let (a, b) = duplex(1 << 20); + let (ar, aw) = tokio::io::split(a); + let (br, bw) = tokio::io::split(b); + // `connect_with_child` frames with Content-Length. + fake.servers.lock().push(Peer::connect(br, bw, fake.clone())); + Ok(Peer::connect_with_child(ar, aw, client, child)) + }); + let mgr = AgentManager::with_connector( + AgentPaths { + project_root: root.clone(), + store_dir: root.join("store"), + worktrees_dir: root.join("worktrees"), + models_file: root.join("models.json"), + custom_agents_dir: None, + home_dir: None, + }, + tokio::runtime::Handle::current(), + connector, + ); + let task = mgr + .create(NewAgentTask { + project_root: None, + prompt: text("Fix the lexer"), + provider: "claude".into(), + config: vec![], + isolation: Isolation::Worktree, + base: None, + }) + .await + .unwrap(); + wait_for(&mgr, &task.id, AgentStatus::NeedsInput).await; + let pid = *pids.lock().last().expect("the agent was started"); + + mgr.delete(&task.id, true).await.unwrap(); + assert!(!task.cwd.0.exists()); + // Killed and reaped: not even a zombie is left. + let alive = std::process::Command::new("kill").args(["-0", &pid.to_string()]).status().unwrap().success(); + assert!(!alive, "the agent (pid {pid}) is still running"); +} diff --git a/crates/ide-agent/src/worktree.rs b/crates/ide-agent/src/worktree.rs index b0e2f62fa..6956cb90b 100644 --- a/crates/ide-agent/src/worktree.rs +++ b/crates/ide-agent/src/worktree.rs @@ -5,7 +5,7 @@ use ide_core::{IdeError, IdeResult, IpcPath}; use ide_vcs::{ChangeKind, Repo}; use std::collections::BTreeMap; use std::path::{Path, PathBuf}; -use std::time::UNIX_EPOCH; +use std::time::{Duration, UNIX_EPOCH}; pub(crate) struct Created { pub path: PathBuf, @@ -65,7 +65,27 @@ pub(crate) fn remove(project_root: &Path, worktrees_dir: &Path, path: &Path) -> if !path.starts_with(worktrees_dir) || path == worktrees_dir { return Err(err); } - std::fs::remove_dir_all(path).map_err(|e| IdeError::internal(e.to_string())) + retrying(REMOVE_ATTEMPTS, REMOVE_PAUSE, || std::fs::remove_dir_all(path)) + .map_err(|e| IdeError::internal(format!("{}: {e}", path.display()))) +} + +/// On Windows a folder can't be deleted while a process still has a file in it open +/// (os error 32): an agent's helper that outlived it, an antivirus scan, the search indexer. +/// Those let go within moments, so removing is retried for a few seconds. +const REMOVE_ATTEMPTS: u32 = 10; +const REMOVE_PAUSE: Duration = Duration::from_millis(300); + +fn retrying(attempts: u32, pause: Duration, mut op: impl FnMut() -> std::io::Result<()>) -> std::io::Result<()> { + for _ in 1..attempts { + match op() { + Err(e) if e.kind() != std::io::ErrorKind::NotFound => std::thread::sleep(pause), + _ => return Ok(()), + } + } + match op() { + Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(()), + r => r, + } } /// Files under `cwd` that differ from `base` (untracked files included), with line counts. @@ -149,6 +169,29 @@ fn unique_slug(base: &str, taken: impl Fn(&str) -> bool) -> String { mod tests { use super::*; + #[test] + fn removing_is_retried_while_a_file_is_still_in_use() { + use std::io::{Error, ErrorKind}; + // A sharing violation as Windows reports it (os error 32), a vanished folder, other errors. + let in_use = || Error::from_raw_os_error(32); + for (fails, outcome) in [(0, true), (3, true), (REMOVE_ATTEMPTS - 1, true), (REMOVE_ATTEMPTS, false)] { + let mut calls = 0; + let r = retrying(REMOVE_ATTEMPTS, Duration::ZERO, || { + calls += 1; + if calls <= fails { Err(in_use()) } else { Ok(()) } + }); + assert_eq!(r.is_ok(), outcome, "{fails} failures"); + assert_eq!(calls, (fails + 1).min(REMOVE_ATTEMPTS)); + } + let mut calls = 0; + let gone = retrying(REMOVE_ATTEMPTS, Duration::ZERO, || { + calls += 1; + if calls == 1 { Err(in_use()) } else { Err(Error::from(ErrorKind::NotFound)) } + }); + assert!(gone.is_ok(), "deleted meanwhile: done"); + assert_eq!(calls, 2); + } + #[test] fn slugs_are_branch_safe() { assert_eq!(slugify("Fix lexer UTF-16 columns!"), "fix-lexer-utf-16-columns"); diff --git a/crates/ide-jsonrpc/src/lib.rs b/crates/ide-jsonrpc/src/lib.rs index c04331ad9..929fcef30 100644 --- a/crates/ide-jsonrpc/src/lib.rs +++ b/crates/ide-jsonrpc/src/lib.rs @@ -243,6 +243,9 @@ struct Inner { pid: Option, /// Fired when the reader task has stopped (EOF, error or kill). reader_done: CancellationToken, + /// Fired once the child process has exited and been reaped (after [`Peer::kill`], + /// also what it started); at once when there is no child. + exited: CancellationToken, } impl Inner { @@ -432,6 +435,7 @@ impl Peer { (child.wait().await.ok(), true) } }; + inner.exited.cancel(); if !killed { // Bounded: a grandchild may keep the pipes open. let drained = async { @@ -510,7 +514,11 @@ impl Peer { kill: CancellationToken::new(), pid, reader_done: CancellationToken::new(), + exited: CancellationToken::new(), }); + if pid.is_none() { + inner.exited.cancel(); + } // Writer: a single task drains the queue so messages never interleave. { @@ -704,6 +712,13 @@ impl Peer { self.inner.close_transport(); self.inner.kill.cancel(); } + + /// Resolves once the child process is gone: exited on its own, or killed with what it + /// started and reaped. [`closed`](Self::closed) can come earlier, at EOF. Until then the + /// child may still hold files open, which on Windows can't be deleted. + pub async fn exited(&self) { + self.inner.exited.cancelled().await; + } } /// Removes the pending entry and tells the remote side to stop when a request diff --git a/crates/ide-jsonrpc/src/tests.rs b/crates/ide-jsonrpc/src/tests.rs index 7db0823fd..033f1405d 100644 --- a/crates/ide-jsonrpc/src/tests.rs +++ b/crates/ide-jsonrpc/src/tests.rs @@ -413,6 +413,27 @@ async fn kill_terminates_process() { tokio::time::timeout(Duration::from_secs(5), peer.closed()).await.unwrap(); } +#[cfg(unix)] +#[tokio::test(flavor = "multi_thread")] +async fn exited_resolves_once_the_killed_process_is_reaped() { + let peer = + Peer::spawn("sleep", &["30".into()], &std::env::temp_dir(), &[], Arc::new(NoopHandler), Box::new(|_| {})) + .unwrap(); + let pid = peer.pid().unwrap().to_string(); + let exited = tokio::time::timeout(Duration::from_millis(200), peer.exited()).await; + assert!(exited.is_err(), "running until killed"); + peer.kill(); + tokio::time::timeout(Duration::from_secs(5), peer.exited()).await.unwrap(); + let alive = std::process::Command::new("kill").args(["-0", &pid]).status().unwrap().success(); + assert!(!alive, "reaped, not even a zombie"); +} + +#[tokio::test(flavor = "multi_thread")] +async fn exited_resolves_at_once_without_a_process() { + let (client, _server) = pair(Arc::new(NoopHandler), Arc::new(Recorder::default())); + tokio::time::timeout(Duration::from_secs(1), client.exited()).await.unwrap(); +} + /// A launcher that prints in the system code page and exits before speaking the /// protocol: `jdtls.bat` on a Chinese Windows when its Python step fails, then /// `pause` prints `请按任意键继续. . .` in GBK on stdout.