fix(tasks): Make every background task cancellable, queued or running (#177)

Background tasks often couldn't be cancelled, or cancelling didn't stop them:
- A cancel that arrived before the backend registered the request was dropped (build while saving files, multi-repository fetch/pull between repositories). The host now remembers early cancels (bounded, 2 min), and request ids start at a random value per page.
- There was no queue and no common cancel path: backend-reported tasks (LSP progress, deployment transfers) couldn't be cancelled, and tool installs, VCS operations and update checks had no cancel button in Background Tasks.
- Only the direct child process was killed, and JDK / tool archives were extracted without checking for cancellation.

Tool installs and JDK downloads now share a downloads queue; cancelling a queued task removes it without starting any work. Every task cancels through one path that shows "Cancelling…" and then removes it; tasks that can't be interrupted show a disabled cancel button with the reason. IdeEvent::Task carries `cancellable` with a cancel_task command, cancellable LSP progress sends window/workDoneProgress/cancel, installers, curl and tar run in their own process group and are killed as a group, and extraction checks for cancellation between entries.

Claude-Session: https://claude.ai/code/session_01Dq6jQQn8ZhRfkkyj3PRutA
This commit is contained in:
l0ng-ai
2026-10-29 09:05:12 +08:00
committed by GitHub
parent c6621c5c36
commit 1c46a1cd04
40 changed files with 888 additions and 148 deletions
Generated
+1
View File
@@ -4479,6 +4479,7 @@ dependencies = [
"hex",
"ide-core",
"ide-language",
"libc",
"parking_lot",
"serde",
"serde_json",
+1 -1
View File
@@ -504,7 +504,7 @@ fn install_binary(b: &Binary, dir: &Path, cancel: &CancellationToken, emit: Emit
ArchiveKind::Raw | ArchiveKind::Gz => parents.iter().fold(dir.to_path_buf(), |p, s| p.join(s)),
_ => dir.to_path_buf(),
};
archive::extract(&file, kind, &dest, exe_name)?;
archive::extract(&file, kind, &dest, exe_name, cancel)?;
let _ = fs::remove_dir_all(&downloads);
let exe = segments.iter().fold(dir.to_path_buf(), |p, s| p.join(s));
if !exe.is_file() {
+2
View File
@@ -50,6 +50,8 @@ pub enum IdeEvent {
message: Option<String>,
fraction: Option<f32>,
done: bool,
/// The UI may stop it with `cancel_task(id)` (see [`crate::task`]).
cancellable: bool,
},
/// Project indexing state changed (dumb mode on/off).
IndexState {
+1
View File
@@ -15,6 +15,7 @@ pub mod process;
pub mod project_dir;
pub mod remote_settings;
pub mod settings;
pub mod task;
pub mod text;
pub mod workflow_settings;
+74
View File
@@ -0,0 +1,74 @@
//! Cancelling background tasks the backend reports itself ([`IdeEvent::Task`](crate::IdeEvent)
//! with `cancellable`), such as a language server's work-done progress. Requests the UI
//! starts are cancelled through their `RequestId` instead.
use crate::TaskId;
use std::collections::HashMap;
use std::sync::{Mutex, MutexGuard};
type Cancel = Box<dyn FnOnce() + Send>;
static CANCELS: Mutex<Option<HashMap<TaskId, Cancel>>> = Mutex::new(None);
fn cancels() -> MutexGuard<'static, Option<HashMap<TaskId, Cancel>>> {
CANCELS.lock().unwrap_or_else(|e| e.into_inner())
}
/// Keeps task `id` cancellable with [`cancel`] until dropped (when the task ends).
#[must_use = "the task stops being cancellable when this is dropped"]
pub struct CancelHandle(TaskId);
impl Drop for CancelHandle {
fn drop(&mut self) {
if let Some(map) = cancels().as_mut() {
map.remove(&self.0);
}
}
}
/// Makes task `id` cancellable: [`cancel`] runs `f` once, while the handle lives.
pub fn on_cancel(id: TaskId, f: impl FnOnce() + Send + 'static) -> CancelHandle {
cancels().get_or_insert_with(HashMap::new).insert(id, Box::new(f));
CancelHandle(id)
}
/// Cancels task `id`; false when it is not cancellable (or already ended / cancelled).
pub fn cancel(id: TaskId) -> bool {
let f = cancels().as_mut().and_then(|map| map.remove(&id));
match f {
Some(f) => {
f();
true
}
None => false,
}
}
#[cfg(test)]
mod tests {
use super::*;
use std::sync::Arc;
use std::sync::atomic::{AtomicUsize, Ordering};
#[test]
fn cancel_runs_the_callback_once_while_the_handle_lives() {
let hits = Arc::new(AtomicUsize::new(0));
let id = TaskId::next();
let h = hits.clone();
let handle = on_cancel(id, move || {
h.fetch_add(1, Ordering::SeqCst);
});
assert!(cancel(id));
assert!(!cancel(id));
assert_eq!(hits.load(Ordering::SeqCst), 1);
drop(handle);
let other = TaskId::next();
let h = hits.clone();
drop(on_cancel(other, move || {
h.fetch_add(1, Ordering::SeqCst);
}));
assert!(!cancel(other), "an ended task is not cancelled");
assert_eq!(hits.load(Ordering::SeqCst), 1);
}
}
+1
View File
@@ -378,6 +378,7 @@ async fn on_event(mgr: &Arc<DebugManager>, s: &Arc<Session>, event: &str, body:
message: str_("message"),
fraction: body.get("percentage").and_then(Value::as_f64).map(|p| (p / 100.0) as f32),
done,
cancellable: false,
});
}
"hotcodereplace" => s.emit(DebugEvent::HotCodeReplace {
+8 -12
View File
@@ -5,6 +5,7 @@
pub mod ops;
mod project;
mod requests;
pub mod rpc;
pub mod shell_env;
@@ -15,7 +16,7 @@ use ide_language::{LanguageRegistry, Syntax};
use ide_run::ProcessManager;
use ide_terminal::TerminalManager;
use ide_workspace::SettingsStore;
use parking_lot::{Mutex, RwLock};
use parking_lot::RwLock;
use std::collections::HashMap;
use std::path::PathBuf;
use std::sync::{Arc, OnceLock};
@@ -54,7 +55,7 @@ pub struct Host {
pub runtime: tokio::runtime::Handle,
/// One project per window, keyed by window label.
projects: RwLock<HashMap<String, Arc<Project>>>,
requests: Mutex<HashMap<RequestId, CancellationToken>>,
requests: requests::Requests,
}
impl Host {
@@ -79,7 +80,7 @@ impl Host {
owner,
runtime,
projects: RwLock::default(),
requests: Mutex::default(),
requests: Default::default(),
}
}
@@ -139,21 +140,16 @@ impl Host {
// ---- cancellable requests: the UI allocates ids, `cancel_request` cancels ----
pub fn begin_request(&self, id: RequestId) -> CancellationToken {
let token = CancellationToken::new();
if let Some(old) = self.requests.lock().insert(id, token.clone()) {
old.cancel();
}
token
self.requests.begin(id)
}
pub fn end_request(&self, id: RequestId) {
self.requests.lock().remove(&id);
self.requests.end(id);
}
/// Cancels request `id`, also when it has not begun yet (it then starts out cancelled).
pub fn cancel_request(&self, id: RequestId) {
if let Some(token) = self.requests.lock().remove(&id) {
token.cancel();
}
self.requests.cancel(id);
}
pub(crate) fn insert_project(&self, window: &str, project: Arc<Project>) {
+116
View File
@@ -0,0 +1,116 @@
//! Cancellable requests: the UI allocates ids, `cancel_request` cancels.
//!
//! A cancel can arrive before the command that owns the id has begun: the UI shows
//! a task (with Cancel) and then awaits other work first (saving documents, a
//! queue, a branch lookup), or the command waits for something before it gets to
//! `begin`. Such a cancel is remembered for a while so the request starts out
//! cancelled instead of running to the end.
use ide_core::{CancellationToken, RequestId};
use parking_lot::Mutex;
use std::collections::{HashMap, VecDeque};
use std::time::{Duration, Instant};
/// How long a cancel for a request that has not begun is remembered.
const EARLY_CANCEL_TTL: Duration = Duration::from_secs(120);
/// At most this many are remembered (ids of finished requests end up here too).
const EARLY_CANCEL_MAX: usize = 1024;
#[derive(Default)]
pub(crate) struct Requests {
inner: Mutex<Inner>,
}
#[derive(Default)]
struct Inner {
running: HashMap<RequestId, CancellationToken>,
/// Cancelled before they began, oldest first.
early: VecDeque<(RequestId, Instant)>,
}
impl Inner {
fn prune(&mut self, now: Instant) {
while self.early.front().is_some_and(|(_, at)| now.duration_since(*at) > EARLY_CANCEL_TTL)
|| self.early.len() > EARLY_CANCEL_MAX
{
self.early.pop_front();
}
}
}
impl Requests {
pub(crate) fn begin(&self, id: RequestId) -> CancellationToken {
let token = CancellationToken::new();
let mut inner = self.inner.lock();
inner.prune(Instant::now());
if let Some(i) = inner.early.iter().position(|(early, _)| *early == id) {
inner.early.remove(i);
token.cancel();
return token;
}
if let Some(old) = inner.running.insert(id, token.clone()) {
old.cancel();
}
token
}
pub(crate) fn end(&self, id: RequestId) {
self.inner.lock().running.remove(&id);
}
pub(crate) fn cancel(&self, id: RequestId) {
let mut inner = self.inner.lock();
match inner.running.remove(&id) {
Some(token) => token.cancel(),
None => {
let now = Instant::now();
inner.early.push_back((id, now));
inner.prune(now);
}
}
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn cancel_reaches_a_running_request() {
let r = Requests::default();
let token = r.begin(RequestId(1));
assert!(!token.is_cancelled());
r.cancel(RequestId(1));
assert!(token.is_cancelled());
}
#[test]
fn a_cancel_before_begin_cancels_the_request_when_it_begins() {
let r = Requests::default();
r.cancel(RequestId(7));
assert!(r.begin(RequestId(7)).is_cancelled());
// Only once: the id is not cancelled for good.
assert!(!r.begin(RequestId(7)).is_cancelled());
assert!(!r.begin(RequestId(8)).is_cancelled());
}
#[test]
fn early_cancels_are_bounded() {
let r = Requests::default();
for i in 0..(EARLY_CANCEL_MAX as u32 + 10) {
r.cancel(RequestId(i));
}
assert_eq!(r.inner.lock().early.len(), EARLY_CANCEL_MAX);
assert!(!r.begin(RequestId(0)).is_cancelled(), "the oldest were dropped");
assert!(r.begin(RequestId(EARLY_CANCEL_MAX as u32 + 9)).is_cancelled());
}
#[test]
fn a_finished_request_is_not_cancelled_again() {
let r = Requests::default();
let token = r.begin(RequestId(3));
r.end(RequestId(3));
r.cancel(RequestId(3));
assert!(!token.is_cancelled());
}
}
+2 -1
View File
@@ -13,7 +13,7 @@
use crate::Host;
use crate::ops::{self, Emit};
use ide_core::{IdeError, IdeEvent, IdeResult, RequestId};
use ide_core::{IdeError, IdeEvent, IdeResult, RequestId, TaskId};
use ide_jsonrpc::{Handler, Peer, codes};
use serde::de::DeserializeOwned;
use serde::{Deserialize, Serialize};
@@ -165,6 +165,7 @@ impl Dispatcher {
c.host.cancel_request(c.arg::<RequestId>("id")?);
Ok(())
})
.add("cancel_task", |c| async move { Ok(ide_core::task::cancel(c.arg::<TaskId>("id")?)) })
.add("fs_read_dir", |c| async move { ops::fs::read_dir(&c.host, &c.window, c.arg("path")?) })
.add("fs_create_file", |c| async move { ops::fs::create_file(&c.host, &c.window, c.arg("path")?) })
.add("fs_create_dir", |c| async move { ide_vfs::create_dir(&c.arg::<PathBuf>("path")?) })
+4
View File
@@ -199,6 +199,7 @@ impl Shared {
message: None,
fraction: None,
done: true,
cancellable: false,
});
self.emit_state();
}
@@ -214,6 +215,7 @@ impl Shared {
message: Some("Scanning files…".into()),
fraction: None,
done: false,
cancellable: false,
});
let filter = self.filter();
@@ -260,6 +262,7 @@ impl Shared {
message: Some(format!("{done} / {total} files")),
fraction: Some(done as f32 / total.max(1) as f32),
done: false,
cancellable: false,
});
}
}
@@ -271,6 +274,7 @@ impl Shared {
message: None,
fraction: Some(1.0),
done: true,
cancellable: false,
});
self.emit_state();
let d = self.data.read();
+26 -2
View File
@@ -30,6 +30,18 @@ impl ClientHandler {
self.server.running_gen(self.generation).map(|r| r.caps.read().encoding).unwrap_or_default()
}
/// Cancelling task `task` asks this server instance to stop the work of progress `token`.
fn cancel_handle(&self, task: TaskId, token: &Value) -> ide_core::task::CancelHandle {
let (server, generation, token) = (self.server.clone(), self.generation, token.clone());
ide_core::task::on_cancel(task, move || {
if let Some(run) = server.running_gen(generation)
&& let Err(e) = run.peer.notify("window/workDoneProgress/cancel", json!({ "token": token }))
{
tracing::debug!("{}: cannot cancel progress: {e}", server.name);
}
})
}
fn progress(&self, mgr: &LspManager, params: Value) {
let token = match &params["token"] {
Value::String(s) => s.clone(),
@@ -38,6 +50,7 @@ impl ClientHandler {
let value = &params["value"];
let message = value.get("message").and_then(Value::as_str).map(str::to_owned);
let fraction = value.get("percentage").and_then(Value::as_f64).map(|p| (p / 100.0).clamp(0.0, 1.0) as f32);
let cancellable = value.get("cancellable").and_then(Value::as_bool);
let event = {
let mut progress = self.server.progress.lock();
match value.get("kind").and_then(Value::as_str) {
@@ -49,20 +62,30 @@ impl ClientHandler {
.map(str::to_owned)
.unwrap_or_else(|| self.server.name.clone());
let task = TaskId::next();
progress.insert(token, Progress { task, title: title.clone(), message: message.clone() });
IdeEvent::Task { id: task, title, message, fraction, done: false }
let cancel = (cancellable == Some(true)).then(|| self.cancel_handle(task, &params["token"]));
let cancellable = cancel.is_some();
progress.insert(token, Progress { task, title: title.clone(), message: message.clone(), cancel });
IdeEvent::Task { id: task, title, message, fraction, done: false, cancellable }
}
Some("report") => {
let Some(p) = progress.get_mut(&token) else { return };
if message.is_some() {
p.message = message;
}
match cancellable {
Some(true) if p.cancel.is_none() => {
p.cancel = Some(self.cancel_handle(p.task, &params["token"]))
}
Some(false) => p.cancel = None,
_ => {}
}
IdeEvent::Task {
id: p.task,
title: p.title.clone(),
message: p.message.clone(),
fraction,
done: false,
cancellable: p.cancel.is_some(),
}
}
Some("end") => {
@@ -73,6 +96,7 @@ impl ClientHandler {
message: message.or(p.message),
fraction: Some(1.0),
done: true,
cancellable: false,
}
}
_ => return,
+8 -1
View File
@@ -553,7 +553,14 @@ impl LspManager {
/// Ends progress tasks and withdraws diagnostics of a dead instance.
fn clear_server_state(&self, server: &Server) {
for (id, title) in server.take_progress() {
self.sink.emit(ide_core::IdeEvent::Task { id, title, message: None, fraction: None, done: true });
self.sink.emit(ide_core::IdeEvent::Task {
id,
title,
message: None,
fraction: None,
done: true,
cancellable: false,
});
}
server.progress_ended();
for path in server.take_diagnostic_paths() {
+2
View File
@@ -67,6 +67,8 @@ pub struct Progress {
pub task: TaskId,
pub title: String,
pub message: Option<String>,
/// Set while the server accepts `window/workDoneProgress/cancel` for it.
pub cancel: Option<ide_core::task::CancelHandle>,
}
pub struct Server {
+31
View File
@@ -728,6 +728,37 @@ async fn progress_tasks_and_busy_status() {
assert!(tasks[2].3);
}
#[tokio::test(flavor = "multi_thread")]
async fn cancelling_a_cancellable_progress_task_asks_the_server_to_stop() {
let h = harness(&[("mock", incremental_caps())]);
h.open("a.mk", "x");
h.wait_status("mock", ServerStatus::Ready).await;
let s = h.launcher.state("mock");
let peer = s.peer();
let begin = |token: Value, cancellable: bool| json!({ "token": token, "value": { "kind": "begin", "title": "Building", "cancellable": cancellable } });
peer.notify("$/progress", begin(json!(7), true)).unwrap();
peer.notify("$/progress", begin(json!("plain"), false)).unwrap();
let task_of = |want: bool| {
h.events.lock().iter().find_map(|e| match e {
IdeEvent::Task { id, cancellable, done: false, .. } if *cancellable == want => Some(*id),
_ => None,
})
};
eventually(|| task_of(true).is_some() && task_of(false).is_some()).await;
let (cancellable, plain) = (task_of(true).unwrap(), task_of(false).unwrap());
assert!(!ide_core::task::cancel(plain), "the server did not offer to cancel it");
assert!(ide_core::task::cancel(cancellable));
eventually(|| !s.notified("window/workDoneProgress/cancel").is_empty()).await;
assert_eq!(s.notified("window/workDoneProgress/cancel"), [json!({ "token": 7 })]);
// Once it ended there is nothing left to cancel.
peer.notify("$/progress", json!({ "token": 7, "value": { "kind": "end" } })).unwrap();
eventually(|| {
h.events.lock().iter().any(|e| matches!(e, IdeEvent::Task { id, done: true, .. } if *id == cancellable))
})
.await;
assert!(!ide_core::task::cancel(cancellable));
}
#[tokio::test(flavor = "multi_thread")]
async fn angular_project_loading_shows_as_progress() {
let h = harness(&[("mock", incremental_caps())]);
+3
View File
@@ -20,5 +20,8 @@ zip = { version = "=2.4.2", default-features = false, features = ["deflate"] }
sha2 = "0.10"
hex = "0.4"
[target.'cfg(unix)'.dependencies]
libc = "0.2"
[dev-dependencies]
tempfile.workspace = true
+48 -24
View File
@@ -1,7 +1,7 @@
//! Archive extraction (zip / vsix, tar.gz, tar.xz, single-file gzip) and
//! executable discovery inside extracted trees.
use ide_core::{IdeError, IdeResult};
use ide_core::{CancellationToken, IdeError, IdeResult};
use std::fs::{self, File};
use std::io::{self, BufReader};
use std::path::{Component, Path, PathBuf};
@@ -41,18 +41,29 @@ fn io_err(context: &str, e: impl std::fmt::Display) -> IdeError {
IdeError::new(ide_core::error::ErrorKind::Io, format!("{context}: {e}"))
}
fn check(cancel: &CancellationToken) -> IdeResult<()> {
if cancel.is_cancelled() { Err(IdeError::cancelled()) } else { Ok(()) }
}
/// Unpacks `archive` into `dest` (created if needed). `Gz` / `Raw` payloads
/// are written to `dest/<single_name>`.
pub fn extract(archive: &Path, kind: ArchiveKind, dest: &Path, single_name: &str) -> IdeResult<()> {
/// are written to `dest/<single_name>`. Stops between entries once `cancel` fires.
pub fn extract(
archive: &Path,
kind: ArchiveKind,
dest: &Path,
single_name: &str,
cancel: &CancellationToken,
) -> IdeResult<()> {
check(cancel)?;
fs::create_dir_all(dest)?;
match kind {
ArchiveKind::Zip => extract_zip(archive, dest),
ArchiveKind::Zip => extract_zip(archive, dest, cancel),
ArchiveKind::TarGz => {
let gz = flate2::read::GzDecoder::new(BufReader::new(File::open(archive)?));
extract_tar(tar::Archive::new(gz), dest)
extract_tar(tar::Archive::new(gz), dest, cancel)
}
ArchiveKind::TarXz => extract_with_tar(archive, dest, "-xJf"),
ArchiveKind::TarBz2 => extract_with_tar(archive, dest, "-xjf"),
ArchiveKind::TarXz => extract_with_tar(archive, dest, "-xJf", cancel),
ArchiveKind::TarBz2 => extract_with_tar(archive, dest, "-xjf", cancel),
ArchiveKind::Gz => {
let mut gz = flate2::read::GzDecoder::new(BufReader::new(File::open(archive)?));
let target = dest.join(single_name);
@@ -68,10 +79,11 @@ pub fn extract(archive: &Path, kind: ArchiveKind, dest: &Path, single_name: &str
}
}
fn extract_tar<R: io::Read>(mut archive: tar::Archive<R>, dest: &Path) -> IdeResult<()> {
fn extract_tar<R: io::Read>(mut archive: tar::Archive<R>, dest: &Path, cancel: &CancellationToken) -> IdeResult<()> {
archive.set_preserve_permissions(true);
archive.set_overwrite(true);
for entry in archive.entries().map_err(|e| io_err("reading archive", e))? {
check(cancel)?;
let mut entry = entry.map_err(|e| io_err("reading archive", e))?;
// `unpack_in` refuses entries escaping `dest` (returns Ok(false)).
entry.unpack_in(dest).map_err(|e| io_err("extracting", e))?;
@@ -79,9 +91,10 @@ fn extract_tar<R: io::Read>(mut archive: tar::Archive<R>, dest: &Path) -> IdeRes
Ok(())
}
fn extract_zip(archive: &Path, dest: &Path) -> IdeResult<()> {
fn extract_zip(archive: &Path, dest: &Path, cancel: &CancellationToken) -> IdeResult<()> {
let mut zip = zip::ZipArchive::new(BufReader::new(File::open(archive)?)).map_err(|e| io_err("reading zip", e))?;
for i in 0..zip.len() {
check(cancel)?;
let mut file = zip.by_index(i).map_err(|e| io_err("reading zip", e))?;
let Some(rel) = file.enclosed_name() else { continue };
if !is_safe_relative(&rel) {
@@ -116,17 +129,13 @@ fn extract_zip(archive: &Path, dest: &Path) -> IdeResult<()> {
/// `.tar.xz` / `.tar.bz2` go through the system `tar` (bsdtar on macOS/Windows, GNU tar
/// + xz / bzip2 on Linux).
fn extract_with_tar(archive: &Path, dest: &Path, flags: &str) -> IdeResult<()> {
fn extract_with_tar(archive: &Path, dest: &Path, flags: &str, cancel: &CancellationToken) -> IdeResult<()> {
let tar = which::which("tar").map_err(|_| IdeError::unavailable("`tar` is required to unpack this archive"))?;
let out = ide_core::process::prepare(&mut std::process::Command::new(tar))
.arg(flags)
.arg(archive)
.arg("-C")
.arg(dest)
.output()
.map_err(|e| io_err("running tar", e))?;
if !out.status.success() {
return Err(io_err("tar failed", String::from_utf8_lossy(&out.stderr).trim()));
let mut cmd = std::process::Command::new(tar);
cmd.arg(flags).arg(archive).arg("-C").arg(dest);
let (ok, output, _) = crate::process::run_collect(cmd, cancel, &|_| {})?;
if !ok {
return Err(io_err("tar failed", output.trim()));
}
Ok(())
}
@@ -269,6 +278,21 @@ mod tests {
assert_eq!(ArchiveKind::from_name("marksman.exe"), ArchiveKind::Raw);
}
#[test]
fn extraction_stops_when_cancelled() {
let tmp = tempfile::tempdir().unwrap();
let archive = tmp.path().join("a.tar.gz");
tar_gz(&archive, &[("one", b"1", 0o644), ("two", b"2", 0o644)]);
let cancel = CancellationToken::new();
cancel.cancel();
let dest = tmp.path().join("out");
for kind in [ArchiveKind::TarGz, ArchiveKind::Zip, ArchiveKind::TarXz, ArchiveKind::Raw] {
let err = extract(&archive, kind, &dest, "x", &cancel).unwrap_err();
assert_eq!(err.kind, ide_core::error::ErrorKind::Cancelled, "{kind:?}");
}
assert!(!dest.join("one").exists());
}
#[test]
fn extracts_tar_gz_and_locates_binary() {
let tmp = tempfile::tempdir().unwrap();
@@ -282,7 +306,7 @@ mod tests {
],
);
let dest = tmp.path().join("out");
extract(&archive, ArchiveKind::TarGz, &dest, "x").unwrap();
extract(&archive, ArchiveKind::TarGz, &dest, "x", &CancellationToken::new()).unwrap();
assert_eq!(fs::read_to_string(dest.join("main.lua")).unwrap(), "-- entry");
let found = locate(&dest, &["lua-language-server".into()]).unwrap();
assert_eq!(found, dest.join("bin/lua-language-server"));
@@ -306,7 +330,7 @@ mod tests {
],
);
let dest = tmp.path().join("out");
extract(&archive, ArchiveKind::Zip, &dest, "x").unwrap();
extract(&archive, ArchiveKind::Zip, &dest, "x", &CancellationToken::new()).unwrap();
let found = locate(&dest, &["codelldb".into()]).unwrap();
assert_eq!(found, dest.join("extension/adapter/codelldb"));
#[cfg(unix)]
@@ -323,7 +347,7 @@ mod tests {
let archive = tmp.path().join("evil.zip");
zip_file(&archive, &[("../escape.txt", b"x", 0o644), ("ok.txt", b"y", 0o644)]);
let dest = tmp.path().join("out");
extract(&archive, ArchiveKind::Zip, &dest, "x").unwrap();
extract(&archive, ArchiveKind::Zip, &dest, "x", &CancellationToken::new()).unwrap();
assert!(!tmp.path().join("escape.txt").exists());
assert!(dest.join("ok.txt").exists());
}
@@ -336,12 +360,12 @@ mod tests {
gz.write_all(b"taplo-binary").unwrap();
gz.finish().unwrap();
let dest = tmp.path().join("bin");
extract(&gz_path, ArchiveKind::Gz, &dest, "taplo").unwrap();
extract(&gz_path, ArchiveKind::Gz, &dest, "taplo", &CancellationToken::new()).unwrap();
assert_eq!(fs::read(dest.join("taplo")).unwrap(), b"taplo-binary");
let raw = tmp.path().join("marksman-macos");
fs::write(&raw, b"mm").unwrap();
extract(&raw, ArchiveKind::Raw, &dest, "marksman").unwrap();
extract(&raw, ArchiveKind::Raw, &dest, "marksman", &CancellationToken::new()).unwrap();
assert_eq!(fs::read(dest.join("marksman")).unwrap(), b"mm");
#[cfg(unix)]
{
+5 -1
View File
@@ -290,9 +290,13 @@ impl Tools {
let archive = fetch_verified(&http, &dl, &staging, cancel, emit)?;
emit(InstallEvent::Line { text: "Extracting…".into() });
let pkg = staging.join("pkg");
archive::extract(&archive, archive::ArchiveKind::from_name(&dl.name), &pkg, "java")?;
archive::extract(&archive, archive::ArchiveKind::from_name(&dl.name), &pkg, "java", cancel)?;
let home = find_jdk_home(&pkg, self.platform)
.ok_or_else(|| IdeError::new(ErrorKind::Io, format!("no JDK found in {}", dl.name)))?;
// Last point to stop: from here the JDK replaces what is at `target`.
if cancel.is_cancelled() {
return Err(IdeError::cancelled());
}
if target.exists() {
std::fs::remove_dir_all(&target)?;
}
+2 -2
View File
@@ -985,9 +985,9 @@ impl Tools {
let exe_name = self.platform.exe(main);
match kind {
archive::ArchiveKind::Gz | archive::ArchiveKind::Raw => {
archive::extract(&archive, kind, &pkg.join("bin"), &exe_name)?
archive::extract(&archive, kind, &pkg.join("bin"), &exe_name, cancel)?
}
_ => archive::extract(&archive, kind, &pkg, &exe_name)?,
_ => archive::extract(&archive, kind, &pkg, &exe_name, cancel)?,
}
let names: Vec<String> = locate
.iter()
+60 -2
View File
@@ -12,7 +12,35 @@ use std::time::{Duration, Instant};
pub type Emit<'a> = &'a dyn Fn(InstallEvent);
/// Starts the child as the leader of its own process group, so [`kill`] reaches what it
/// starts in turn (`npm` → `node`, `go install` → the compiler, `pip` → build backends).
fn own_group(cmd: &mut Command) -> &mut Command {
#[cfg(unix)]
{
use std::os::unix::process::CommandExt;
cmd.process_group(0);
}
cmd
}
/// Kills the child and everything it started.
fn kill(child: &mut Child) {
#[cfg(unix)]
if let Ok(pid) = i32::try_from(child.id()) {
// SAFETY: plain syscall; the child leads its own process group (`own_group`) and is not reaped yet.
unsafe {
libc::kill(-pid, libc::SIGKILL);
}
}
#[cfg(windows)]
{
let mut taskkill = Command::new("taskkill");
ide_core::process::prepare(&mut taskkill)
.args(["/T", "/F", "/PID", &child.id().to_string()])
.stdout(Stdio::null())
.stderr(Stdio::null());
let _ = taskkill.status();
}
let _ = child.kill();
let _ = child.wait();
}
@@ -48,7 +76,7 @@ pub fn run_streamed(cmd: Command, cancel: &CancellationToken, emit: Emit) -> Ide
/// Like [`run_streamed`] but returns `(success, output, status text)` instead of failing on a non-zero exit.
pub fn run_collect(mut cmd: Command, cancel: &CancellationToken, emit: Emit) -> IdeResult<(bool, String, String)> {
emit(InstallEvent::Line { text: format!("$ {}", describe(&cmd)) });
ide_core::process::prepare(&mut cmd).stdin(Stdio::null()).stdout(Stdio::piped()).stderr(Stdio::piped());
own_group(ide_core::process::prepare(&mut cmd)).stdin(Stdio::null()).stdout(Stdio::piped()).stderr(Stdio::piped());
let mut child = cmd.spawn().map_err(|e| IdeError::unavailable(format!("cannot run {}: {e}", describe(&cmd))))?;
let (tx, rx) = mpsc::channel::<String>();
let pipes: [Option<Box<dyn Read + Send>>; 2] = [
@@ -168,7 +196,7 @@ impl Http {
fn base(&self, url: &str) -> Command {
let mut cmd = Command::new(&self.curl);
ide_core::process::prepare(&mut cmd);
own_group(ide_core::process::prepare(&mut cmd));
cmd.args(["-fsSL", "--retry", "2", "--connect-timeout", "20", "-A", "openidea"]);
if self.https_only {
cmd.args(["--proto", "=https", "--proto-redir", "=https"]);
@@ -346,6 +374,36 @@ mod tests {
assert!(started.elapsed() < Duration::from_secs(5));
}
/// `npm` / `go` / `pip` run the real work in children of their own: cancelling
/// kills those too, not only the process the installer started.
#[cfg(unix)]
#[test]
fn cancellation_kills_what_the_child_started() {
let dir = tempfile::tempdir().unwrap();
let pid_file = dir.path().join("pid");
let cancel = CancellationToken::new();
let c2 = cancel.clone();
let watch = pid_file.clone();
std::thread::spawn(move || {
while !watch.exists() {
std::thread::sleep(Duration::from_millis(10));
}
std::thread::sleep(Duration::from_millis(50));
c2.cancel();
});
let mut cmd = Command::new("sh");
cmd.arg("-c").arg(format!("sleep 30 & echo $! > '{}'; wait", pid_file.display()));
let err = run_streamed(cmd, &cancel, &|_| {}).unwrap_err();
assert_eq!(err.kind, ErrorKind::Cancelled);
let pid: i32 = std::fs::read_to_string(&pid_file).unwrap().trim().parse().unwrap();
let deadline = Instant::now() + Duration::from_secs(5);
// SAFETY: signal 0 only checks that the process exists.
while unsafe { libc::kill(pid, 0) } == 0 {
assert!(Instant::now() < deadline, "the grandchild {pid} survived the cancel");
std::thread::sleep(Duration::from_millis(20));
}
}
#[cfg(unix)]
#[test]
fn quiet_run_times_out() {
+9 -1
View File
@@ -1,6 +1,6 @@
use crate::state::AppState;
use ide_core::settings::Settings;
use ide_core::{EventSink, IdeError, IdeResult, RequestId};
use ide_core::{EventSink, IdeError, IdeResult, RequestId, TaskId};
use ide_language::LanguageInfo;
use ide_workspace::RecentProject;
use ide_workspace::settings::SettingsScope;
@@ -225,6 +225,14 @@ pub fn cancel_request(state: State<'_, AppState>, id: RequestId) {
state.cancel_request(id);
}
/// Stops a background task the backend reported with `cancellable` set; false when it
/// has ended or can't be cancelled.
#[tauri::command]
#[specta::specta]
pub fn cancel_task(id: TaskId) -> bool {
ide_core::task::cancel(id)
}
#[derive(Serialize, Deserialize, Type)]
#[serde(rename_all = "camelCase")]
pub enum LogLevel {
+18 -8
View File
@@ -1,20 +1,30 @@
use crate::state::AppState;
use ide_core::{IdeEvent, IdeResult, IpcPath, RequestId, SharedSink, TaskId};
use ide_core::{CancellationToken, IdeEvent, IdeResult, IpcPath, RequestId, SharedSink, TaskId};
use ide_remote::{DeployedPath, SyncEntry, SyncStep, TransferReport};
use ide_ssh::RemoteEntry;
use std::path::PathBuf;
use tauri::{State, Window};
/// Reports a transfer as a background task (status bar progress).
fn progress(sink: SharedSink, title: &str) -> (impl FnMut(&str, usize, usize) + Send, impl FnOnce()) {
/// Reports a transfer as a background task (status bar progress) whose Cancel fires `token`.
fn progress(
sink: SharedSink,
title: &str,
token: CancellationToken,
) -> (impl FnMut(&str, usize, usize) + Send, impl FnOnce()) {
let id = TaskId::next();
let cancel = token;
let handle = ide_core::task::on_cancel(id, move || cancel.cancel());
let title = title.to_owned();
let (s, t) = (sink.clone(), title.clone());
let report = move |path: &str, i: usize, n: usize| {
let fraction = if n == 0 { None } else { Some(i as f32 / n as f32) };
s.emit(IdeEvent::Task { id, title: t.clone(), message: Some(path.to_owned()), fraction, done: false });
let message = Some(path.to_owned());
s.emit(IdeEvent::Task { id, title: t.clone(), message, fraction, done: false, cancellable: true });
};
let done = move || {
drop(handle);
sink.emit(IdeEvent::Task { id, title, message: None, fraction: Some(1.0), done: true, cancellable: false })
};
let done = move || sink.emit(IdeEvent::Task { id, title, message: None, fraction: Some(1.0), done: true });
(report, done)
}
@@ -34,7 +44,7 @@ pub async fn deploy_upload(
) -> IdeResult<TransferReport> {
let project = state.project(window.label())?;
let token = state.begin_request(request);
let (mut report, done) = progress(project.sink.clone(), "Uploading");
let (mut report, done) = progress(project.sink.clone(), "Uploading", token.clone());
let r = state
.remote
.upload(&project.settings.effective(), &project.root, server.as_deref(), &local(paths), token, &mut report)
@@ -56,7 +66,7 @@ pub async fn deploy_download(
) -> IdeResult<TransferReport> {
let project = state.project(window.label())?;
let token = state.begin_request(request);
let (mut report, done) = progress(project.sink.clone(), "Downloading");
let (mut report, done) = progress(project.sink.clone(), "Downloading", token.clone());
let r = state
.remote
.download(&project.settings.effective(), &project.root, server.as_deref(), &local(paths), token, &mut report)
@@ -108,7 +118,7 @@ pub async fn deploy_sync(
) -> IdeResult<TransferReport> {
let project = state.project(window.label())?;
let token = state.begin_request(request);
let (mut report, done) = progress(project.sink.clone(), "Synchronizing");
let (mut report, done) = progress(project.sink.clone(), "Synchronizing", token.clone());
let r = state
.remote
.sync(&project.settings.effective(), &project.root, server.as_deref(), &steps, token, &mut report)
+1
View File
@@ -71,6 +71,7 @@ pub fn builder() -> tauri_specta::Builder<tauri::Wry> {
app::project_rename,
app::project_roots,
app::cancel_request,
app::cancel_task,
app::log_frontend,
agent::agent_subscribe,
agent::agent_providers,
+8 -1
View File
@@ -56,6 +56,11 @@ export const commands = {
*/
projectRoots: () => typedError<ProjectRoots, IdeError>(__TAURI_INVOKE("project_roots")),
cancelRequest: (id: RequestId) => __TAURI_INVOKE<void>("cancel_request", { id }),
/**
* Stops a background task the backend reported with `cancellable` set; false when it
* has ended or can't be cancelled.
*/
cancelTask: (id: TaskId) => __TAURI_INVOKE<boolean>("cancel_task", { id }),
/** Routes frontend logs into the backend `tracing` pipeline. */
logFrontend: (level: LogLevel, message: string) => __TAURI_INVOKE<void>("log_frontend", { level, message }),
/** Streams the project's agent events to this window; returns every task. */
@@ -3237,7 +3242,9 @@ export type IdeEvent =
/** Full replacement of diagnostics for one file from one producer. */
{ type: "diagnostics"; path: IpcPath; source: string; items: Diagnostic[] } |
/** Progress of a background task; `fraction` is `None` when indeterminate. */
{ type: "task"; id: TaskId; title: string; message: string | null; fraction: number | null; done: boolean } |
{ type: "task"; id: TaskId; title: string; message: string | null; fraction: number | null; done: boolean;
/** The UI may stop it with `cancel_task(id)` (see [`crate::task`]). */
cancellable: boolean } |
/** Project indexing state changed (dumb mode on/off). */
{ type: "indexState"; indexing: boolean; files: number; symbols: number } |
/** Git status / branch information should be refreshed. */
+8 -2
View File
@@ -39,9 +39,13 @@ export function checkForUpdates(): Promise<void> {
async function runCheck() {
const title = t("app.update.checking");
upsertTask({ id: CHECK_TASK_ID, title, message: null, fraction: null, done: false });
// The check only reads: cancelling drops its answer (the request finishes on its own).
let abandon = () => {};
const abandoned = new Promise<"cancelled">((resolve) => (abandon = () => resolve("cancelled")));
upsertTask({ id: CHECK_TASK_ID, title, message: null, fraction: null, done: false, cancel: () => abandon() });
try {
const r = await call(commands.appCheckUpdate());
const r = await Promise.race([call(commands.appCheckUpdate()), abandoned]);
if (r === "cancelled") return;
if (!r.enabled) {
notifications().notify({
level: "info",
@@ -122,6 +126,8 @@ async function runInstall(version: string) {
fraction,
done: false,
cancel: p.type === "installing" ? undefined : cancel,
// Replacing the app happens in one step that must not be cut short.
uncancellable: p.type === "installing" ? t("app.update.installUncancellable") : undefined,
});
};
// The notification's action is spent: progress lives in the status bar from here.
+5 -2
View File
@@ -23,7 +23,7 @@ import {
import { downloadJdkActions } from "@/platform/jdk";
import { useProject } from "@/platform/project";
import { editorService, notifications } from "@/workbench/contributions";
import { removeTask, upsertTask } from "@/workbench/tasks";
import { cancelTask, removeTask, upsertTask, useTasks } from "@/workbench/tasks";
import { profilesList, type GradleRunOptions, type MavenRunOptions } from "./model";
export interface BuildSettings {
@@ -231,7 +231,10 @@ export async function syncTool(tool: BuildTool): Promise<void> {
export function cancelSync(tool: BuildTool) {
const s = useBuild.getState().syncing[tool];
if (s) void commands.cancelRequest(s.request);
if (!s) return;
const taskId = `build-sync:${tool}`;
if (useTasks.getState().tasks.some((t) => t.id === taskId && t.cancel)) cancelTask(taskId);
else void commands.cancelRequest(s.request);
}
export function resetBuild() {
+5 -2
View File
@@ -25,7 +25,7 @@ import {
} from "@/platform/ipc";
import { useProject } from "@/platform/project";
import { maybeEditorService, notifications, toolWindowService } from "@/workbench/contributions";
import { removeTask, upsertTask } from "@/workbench/tasks";
import { cancelTask, removeTask, upsertTask, useTasks } from "@/workbench/tasks";
import { appendOutput, finishSummary, firstError, messageFiles, type BuildStatus, type ConsoleLine } from "./model";
export const BUILD_TASK_ID = "build:compile";
@@ -248,7 +248,10 @@ function showFailure() {
export function stopBuild() {
const { request } = useBuildRun.getState();
if (request != null) backend.cancel(request);
if (request == null) return;
// Through the task, so its progress says "Cancelling…" until the build has stopped.
if (useTasks.getState().tasks.some((t) => t.id === BUILD_TASK_ID && t.cancel)) cancelTask(BUILD_TASK_ID);
else backend.cancel(request);
}
/** Runs the last build again (the tool window's rerun button). */
@@ -70,6 +70,22 @@ describe("tools store", () => {
expect(useTools.getState().tools.find((t) => t.name === "gopls")?.status).toBe("missing");
});
it("installs one at a time; one cancelled while it waits never starts", async () => {
backend.stepMs = 10;
await refreshTools();
const install = vi.spyOn(backend, "install");
const first = installTool("gopls");
const queued = installTool("vtsls");
expect(useTasks.getState().tasks.find((t) => t.id === "tools:vtsls")).toMatchObject({ queued: true });
cancelInstall("vtsls");
expect(useTasks.getState().tasks.find((t) => t.id === "tools:vtsls")).toBeUndefined();
expect(await queued).toBeNull();
expect(useTools.getState().logs["vtsls"]?.error).toBe("Cancelled");
expect((await first)?.status).toBe("installed");
expect(install.mock.calls.map((c) => c[0])).toEqual(["gopls"]);
expect(notes.filter((n) => n.level === "error")).toHaveLength(0);
});
it("uninstalls", async () => {
await refreshTools();
expect(await uninstallTool("rust-analyzer")).toBe(true);
+14 -9
View File
@@ -17,10 +17,12 @@ import {
requiredJdk,
TEMURIN,
type DownloadJdkArgs,
type JdkDownload,
} from "@/platform/jdk";
import { useProject } from "@/platform/project";
import { notifications } from "@/workbench/contributions";
import { upsertTask } from "@/workbench/tasks";
import { queueTask, upsertTask } from "@/workbench/tasks";
import { DOWNLOAD_QUEUE } from "./store";
import { tx } from "@/platform/i18n";
const JAVA_SETTINGS = "languages.java";
@@ -42,16 +44,19 @@ export function downloadJdkFor(args: DownloadJdkArgs): Promise<Sdk | null> {
const title = `Downloading JDK ${args.version}`;
const taskId = `jdk-download:${args.version}`;
const key = `jdk-download-${args.version}`;
const cancel = () => job.cancel();
const job = downloadJdk(TEMURIN, args.version, null, (e) => {
const message = e.type === "line" ? e.text : e.message;
const fraction = e.type === "progress" ? e.fraction : null;
upsertTask({ id: taskId, title, message, fraction, done: false, cancel });
});
upsertTask({ id: taskId, title, message: "Eclipse Temurin", fraction: null, done: false, cancel });
const p = (async () => {
try {
const sdk = await job.promise;
// Waits for the downloads ahead of it; cancelling it there means it never starts.
const sdk = await queueTask(DOWNLOAD_QUEUE, { id: taskId, title }, () => {
const cancel = () => job.cancel();
upsertTask({ id: taskId, title, message: "Eclipse Temurin", fraction: null, done: false, cancel });
const job: JdkDownload = downloadJdk(TEMURIN, args.version, null, (e) => {
const message = e.type === "line" ? e.text : e.message;
const fraction = e.type === "progress" ? e.fraction : null;
upsertTask({ id: taskId, title, message, fraction, done: false, cancel });
});
return job.promise;
});
if (args.languageServer) await chooseForLanguageServer(sdk.path);
notifications().notify({
group: "tools",
+23 -17
View File
@@ -4,11 +4,13 @@ import type { RequestId, ToolInfo } from "@/bindings";
import { executeAction } from "@/platform/actions";
import { isCancelled, nextRequestId } from "@/platform/ipc";
import { notifications } from "@/workbench/contributions";
import { upsertTask } from "@/workbench/tasks";
import { cancelTask, queueTask, upsertTask, useTasks } from "@/workbench/tasks";
import { toolsBackend } from "./backend";
import { tx } from "@/platform/i18n";
export const SETTINGS_PAGE_ID = "languages.servers";
/** Tool installs and JDK downloads take turns (one network download / package manager at a time). */
export const DOWNLOAD_QUEUE = "downloads";
const MAX_LOG_LINES = 400;
export interface InstallJob {
@@ -100,24 +102,25 @@ export function installTool(name: string, opts: { notificationKey?: string } = {
const taskId = `tools:${name}`;
const key = opts.notificationKey ?? `tools-install-${name}`;
useTools.setState((s) => ({ jobs: { ...s.jobs, [name]: { request, lines: [], fraction: null, message: null } } }));
upsertTask({ id: taskId, title: tx`Installing ${display}`, message: null, fraction: null, done: false });
const title = tx`Installing ${display}`;
const cancel = () => toolsBackend().cancel(request);
const task = (message: string | null, fraction: number | null) =>
upsertTask({ id: taskId, title, message, fraction, done: false, cancel });
const p = (async () => {
try {
const info = await toolsBackend().install(name, request, (e) => {
if (e.type === "line") {
patchJob(name, (j) => ({ ...j, lines: [...j.lines, e.text].slice(-MAX_LOG_LINES), message: e.text }));
upsertTask({ id: taskId, title: tx`Installing ${display}`, message: e.text, fraction: null, done: false });
} else {
patchJob(name, (j) => ({ ...j, fraction: e.fraction, message: e.message }));
upsertTask({
id: taskId,
title: tx`Installing ${display}`,
message: e.message,
fraction: e.fraction,
done: false,
});
}
// Downloads and package managers run one at a time; the rest wait, cancellable.
const info = await queueTask(DOWNLOAD_QUEUE, { id: taskId, title }, () => {
task(null, null);
return toolsBackend().install(name, request, (e) => {
if (e.type === "line") {
patchJob(name, (j) => ({ ...j, lines: [...j.lines, e.text].slice(-MAX_LOG_LINES), message: e.text }));
task(e.text, null);
} else {
patchJob(name, (j) => ({ ...j, fraction: e.fraction, message: e.message }));
task(e.message, e.fraction);
}
});
});
upsertTool(info);
endJob(name, null);
@@ -151,7 +154,7 @@ export function installTool(name: string, opts: { notificationKey?: string } = {
});
return null;
} finally {
upsertTask({ id: taskId, title: tx`Installing ${display}`, message: null, fraction: 1, done: true });
upsertTask({ id: taskId, title, message: null, fraction: 1, done: true });
running.delete(name);
}
})();
@@ -159,7 +162,10 @@ export function installTool(name: string, opts: { notificationKey?: string } = {
return p;
}
/** Cancels the install of `name`, queued or running (as the task's Cancel does). */
export function cancelInstall(name: string) {
const taskId = `tools:${name}`;
if (useTasks.getState().tasks.some((t) => t.id === taskId)) return cancelTask(taskId);
const job = useTools.getState().jobs[name];
if (job) toolsBackend().cancel(job.request);
}
+26 -17
View File
@@ -6,9 +6,10 @@ import { Channel } from "@tauri-apps/api/core";
import { create } from "zustand";
import type { GitOperation, RemoteOp } from "@/bindings";
import { tx } from "@/platform/i18n";
import { call, commands, emitLocalIdeEvent, isCancelled, nextRequestId } from "@/platform/ipc";
import { call, commands, isCancelled, nextRequestId } from "@/platform/ipc";
import { editorsFor } from "@/editor/api";
import { maybeEditorService, notifications } from "@/workbench/contributions";
import { cancelTask, upsertTaskSoon, useTasks } from "@/workbench/tasks";
import { progressFraction } from "./model";
import { currentRepo, refreshNow, repoName, useVcs } from "./store";
@@ -129,7 +130,10 @@ export function lastRemoteFailure(): string | null {
}
export function cancelRemote() {
if (activeRemote !== null) void commands.cancelRequest(activeRemote);
if (activeRemote === null) return;
const taskId = `vcs:${activeRemote}`;
if (useTasks.getState().tasks.some((t) => t.id === taskId)) cancelTask(taskId);
else void commands.cancelRequest(activeRemote);
}
/** Every repository of the project (Fetch and Update Project act on all of them, as in IDEA). */
@@ -154,11 +158,12 @@ export async function runRemote(
const [progressText, doneTitle] = REMOTE_TITLE[op.op];
const progressTitle = tx(progressText);
const request = nextRequestId();
const taskId = 900_000 + request;
const taskId = `vcs:${request}`;
activeRemote = request;
useVcs.setState({ remoteBusy: progressTitle });
const cancel = () => void commands.cancelRequest(request);
const task = (message: string | null, fraction: number | null, done: boolean) =>
emitLocalIdeEvent({ type: "task", id: taskId, title: progressTitle, message, fraction, done });
upsertTaskSoon({ id: taskId, title: progressTitle, message, fraction, done, cancel });
task(detail ?? null, null, false);
notifications().notify({
group: "vcs",
@@ -220,13 +225,16 @@ export async function runRemote(
const cancelled = isCancelled(e);
lastRemoteError = cancelled ? null : errorMessage(e);
logConsole(cancelled ? "info" : "err", cancelled ? `${doneTitle} cancelled` : errorMessage(e));
notifications().notify({
group: "vcs",
level: cancelled ? "warning" : "error",
key: "vcs.remote",
title: cancelled ? `${doneTitle} cancelled` : `${doneTitle} failed`,
body: cancelled ? undefined : errorMessage(e),
});
// The user asked for it: the progress balloon just goes away.
if (cancelled) notifications().expire?.("vcs.remote");
else
notifications().notify({
group: "vcs",
level: "error",
key: "vcs.remote",
title: `${doneTitle} failed`,
body: errorMessage(e),
});
return false;
} finally {
task(null, 1, true);
@@ -320,12 +328,13 @@ export async function runGitOperation(op: GitOperation, opts: GitOperationOption
}
await saveAll();
const request = nextRequestId();
const taskId = 900_000 + request;
const taskId = `vcs:${request}`;
activeRemote = request;
const progress = tx(opts.progress);
useVcs.setState({ remoteBusy: progress });
const cancel = () => void commands.cancelRequest(request);
const task = (message: string | null, done: boolean) =>
emitLocalIdeEvent({ type: "task", id: taskId, title: progress, message, fraction: done ? 1 : null, done });
upsertTaskSoon({ id: taskId, title: progress, message, fraction: done ? 1 : null, done, cancel });
task(null, false);
notifications().notify({
group: "vcs",
@@ -357,14 +366,14 @@ export async function runGitOperation(op: GitOperation, opts: GitOperationOption
logConsole(cancelled ? "info" : "err", cancelled ? `${opts.failure}: cancelled` : message);
notifications().expire?.("vcs.operation");
await refreshNow();
const handled = !cancelled && opts.onError ? await opts.onError(message) : false;
const handled = cancelled || (opts.onError ? await opts.onError(message) : false);
if (!handled)
notifications().notify({
group: "vcs",
level: cancelled ? "warning" : "error",
level: "error",
key: "vcs.operation",
title: cancelled ? `${opts.progress} cancelled` : opts.failure,
body: cancelled ? undefined : message,
title: opts.failure,
body: message,
});
return false;
} finally {
+3 -1
View File
@@ -34,7 +34,9 @@ export function isCancelled(e: unknown): boolean {
return e instanceof IpcError && e.kind === "cancelled";
}
let nextId = 1;
// Ids are shared by every window of the backend and outlive a reload there (a cancel that
// arrives before its request began is remembered), so each page starts at a random point.
let nextId = 1 + Math.floor(Math.random() * 0x7000_0000);
/** Allocates an id for a cancellable request (see `commands.cancelRequest`). */
export function nextRequestId(): RequestId {
return nextId++ as RequestId;
+1
View File
@@ -66,6 +66,7 @@
"app.update.installed": "OpenIDEA {version} is installed",
"app.update.installedBody": "It is used after OpenIDEA restarts.",
"app.update.installingMessage": "Installing…",
"app.update.installUncancellable": "The update is being installed; this step can't be cancelled",
"app.update.none": "OpenIDEA {version} is up to date",
"app.update.progress": "{done} MB of {total} MB",
"app.update.progressUnknown": "{done} MB",
+6 -1
View File
@@ -4644,5 +4644,10 @@
"Couldn't copy the task": "无法复制任务",
"Copy to a New Task From Here": "从此处复制为新任务",
"Run Again From Here": "从此处重新运行",
"The messages after this one are replaced. Files the agent changed stay as they are.": "这条之后的消息会被替换。Agent 改动过的文件保持不变。"
"The messages after this one are replaced. Files the agent changed stay as they are.": "这条之后的消息会被替换。Agent 改动过的文件保持不变。",
"Cancelling…": "正在取消…",
"Code insight needs the index; it can't be stopped": "代码洞察依赖索引,无法停止",
"Waiting for other tasks": "正在等待其他任务",
"This task can't be cancelled": "此任务无法取消",
"Cancelled": "已取消"
}
+1
View File
@@ -66,6 +66,7 @@
"app.update.installed": "OpenIDEA {version} 已安装",
"app.update.installedBody": "重启 OpenIDEA 后生效。",
"app.update.installingMessage": "正在安装…",
"app.update.installUncancellable": "正在安装更新,此步骤无法取消",
"app.update.none": "OpenIDEA {version} 已是最新版本",
"app.update.progress": "{done} MB / {total} MB",
"app.update.progressUnknown": "{done} MB",
+6
View File
@@ -90,6 +90,12 @@
text-overflow: ellipsis;
white-space: nowrap;
}
.taskCancelOff {
display: inline-flex;
}
.taskCancelOff > button {
pointer-events: none;
}
.taskPct {
font-family: var(--font-mono);
font-size: var(--fs-small);
+20 -6
View File
@@ -1,10 +1,10 @@
import { useMemo, useRef, useState } from "react";
import clsx from "clsx";
import { BarButton, IconButton, Popup, ProgressBar, showContextMenu, T } from "@/ui";
import { BarButton, IconButton, Popup, ProgressBar, showContextMenu, T, tooltipAttrs } from "@/ui";
import { useContributions, useStatusInfo, type StatusItemContribution } from "./contributions";
import { ErrorBoundary } from "./ErrorBoundary";
import { isStatusItemVisible, statusBarMenu, useStatusWidgetOverrides } from "./statusBarWidgets";
import { useTasks } from "./tasks";
import { cancelTask, canCancel, useTasks, type BackgroundTask } from "./tasks";
import styles from "./StatusBar.module.css";
import { tx } from "@/platform/i18n";
@@ -71,13 +71,29 @@ export function StatusInfo() {
);
}
/** A task's cancel button: working, "Cancelling…", or disabled with the reason it can't be stopped. */
function TaskCancel({ task: t }: { task: BackgroundTask }) {
if (t.done) return null;
if (canCancel(t))
return <IconButton icon="stop" size="small" title={tx`Cancel ${t.title}`} onClick={() => cancelTask(t.id)} />;
const reason = t.cancelling ? tx("Cancelling…") : t.uncancellable;
if (!reason) return null;
// A disabled button gets no pointer events; the wrapper shows why.
return (
<span className={styles.taskCancelOff} {...tooltipAttrs(reason)}>
<IconButton icon="stop" size="small" title={reason} disabled />
</span>
);
}
/** Built-in: background task progress; click lists all running tasks. */
export function TaskStatus() {
const tasks = useTasks((s) => s.tasks);
const [open, setOpen] = useState(false);
const ref = useRef<HTMLButtonElement>(null);
const running = tasks.filter((t) => !t.done);
const current = running[running.length - 1] ?? tasks[tasks.length - 1];
// A task that is working wins over the queued ones waiting behind it.
const current = running.findLast((t) => !t.queued) ?? running[running.length - 1] ?? tasks[tasks.length - 1];
if (!current) return null;
const more = running.length - 1;
return (
@@ -119,9 +135,7 @@ export function TaskStatus() {
<T>{t.title}</T>
</span>
{t.fraction != null && <span className={styles.taskPct}>{Math.round(t.fraction * 100)}%</span>}
{t.cancel && !t.done && (
<IconButton icon="stop" size="small" title={tx`Cancel ${t.title}`} onClick={() => t.cancel?.()} />
)}
<TaskCancel task={t} />
</div>
<ProgressBar value={t.done ? 1 : t.fraction} aria-label={t.title} />
{t.message && <div className={styles.taskMessage}>{t.message}</div>}
+19 -2
View File
@@ -86,12 +86,29 @@ describe("background tasks", () => {
message: null,
fraction: null,
done: false,
cancellable: false,
});
emitLocalIdeEvent({
type: "task",
id: 7,
title: "Building index",
message: null,
fraction: 0.5,
done: false,
cancellable: false,
});
emitLocalIdeEvent({ type: "task", id: 7, title: "Building index", message: null, fraction: 0.5, done: false });
});
expect(screen.queryByText("Building index")).toBeNull();
act(() => {
emitLocalIdeEvent({ type: "task", id: 6, title: "Publish Diagnostics", message: null, fraction: 1, done: true });
emitLocalIdeEvent({
type: "task",
id: 6,
title: "Publish Diagnostics",
message: null,
fraction: 1,
done: true,
cancellable: false,
});
vi.advanceTimersByTime(600);
});
expect(screen.queryByText("Publish Diagnostics")).toBeNull();
+5 -5
View File
@@ -8,7 +8,7 @@ import { notify } from "./notificationStore";
import { NOTIFICATIONS_TOOL_WINDOW } from "./Notifications";
import { activateToolWindow } from "./toolwindows/store";
import { toggleTheme } from "./theme";
import { useTasks } from "./tasks";
import { cancelTask, canCancel, useTasks } from "./tasks";
import { showStatusInfo } from "./contributions";
export async function openProjectDialog() {
@@ -39,13 +39,13 @@ export async function openProjectDialog() {
/** Stop Background Processes (⇧⌘F2): the one running cancellable task, else a choice of them. */
export function stopBackgroundProcesses() {
const running = useTasks.getState().tasks.filter((t) => t.cancel && !t.done);
const running = useTasks.getState().tasks.filter(canCancel);
if (!running.length) return showStatusInfo("No background processes to stop");
if (running.length === 1) return running[0]!.cancel!();
if (running.length === 1) return cancelTask(running[0]!.id);
showContextMenu({ x: window.innerWidth / 2 - 160, y: 96 }, [
...running.map((t) => ({ id: String(t.id), label: t.title, icon: "stop", run: () => t.cancel!() })),
...running.map((t) => ({ id: String(t.id), label: t.title, icon: "stop", run: () => cancelTask(t.id) })),
{ kind: "separator" as const },
{ id: "all", label: "Stop All", run: () => running.forEach((t) => t.cancel!()) },
{ id: "all", label: "Stop All", run: () => running.forEach((t) => cancelTask(t.id)) },
]);
}
+139
View File
@@ -0,0 +1,139 @@
import { afterEach, describe, expect, it, vi } from "vitest";
import { act, fireEvent, render, screen } from "@testing-library/react";
const cancelTaskCommand = vi.fn(async (_id: number) => true);
vi.mock("@/platform/ipc", async (orig) => {
const actual = await orig<typeof import("@/platform/ipc")>();
return { ...actual, commands: { ...actual.commands, cancelTask: cancelTaskCommand } };
});
const { emitLocalIdeEvent, isCancelled } = await import("@/platform/ipc");
const { cancelTask, installTaskEvents, queueTask, setIndexing, upsertTask, useTasks } = await import("./tasks");
const { TaskStatus } = await import("./StatusBar");
const task = (id: string) => useTasks.getState().tasks.find((t) => t.id === id);
function deferred() {
let resolve!: () => void;
const promise = new Promise<void>((r) => (resolve = r));
return { promise, resolve };
}
afterEach(() => {
useTasks.setState({ tasks: [], indexing: null });
vi.useRealTimers();
vi.clearAllMocks();
});
describe("task queues", () => {
it("runs one task at a time; the others wait, listed as queued", async () => {
const first = deferred();
const started: string[] = [];
const a = queueTask("q1", { id: "a", title: "A" }, async () => {
started.push("a");
upsertTask({ id: "a", title: "A", message: null, fraction: null, done: false });
await first.promise;
return "a";
});
const b = queueTask("q1", { id: "b", title: "B" }, async () => {
started.push("b");
return "b";
});
expect(started).toEqual(["a"]);
expect(task("b")).toMatchObject({ queued: true, message: "Waiting for other tasks" });
first.resolve();
await expect(a).resolves.toBe("a");
await expect(b).resolves.toBe("b");
expect(started).toEqual(["a", "b"]);
});
it("cancelling a queued task drops it at once: its work never starts and its caller sees a cancellation", async () => {
const first = deferred();
const work = vi.fn(async () => "never");
const running = queueTask("q2", { id: "run", title: "Running" }, () => first.promise);
const queued = queueTask("q2", { id: "wait", title: "Waiting" }, work);
const next = queueTask("q2", { id: "next", title: "Next" }, async () => "next");
cancelTask("wait");
await expect(queued).rejects.toSatisfy(isCancelled);
expect(task("wait")).toBeUndefined();
first.resolve();
await running;
await expect(next).resolves.toBe("next");
expect(work).not.toHaveBeenCalled();
});
it("the queue's Cancel is gone once the work starts (the work brings its own)", async () => {
const first = deferred();
const second = deferred();
void queueTask("q3", { id: "x", title: "X" }, () => first.promise);
const y = queueTask("q3", { id: "y", title: "Y" }, () => second.promise);
first.resolve();
await vi.waitFor(() => expect(task("y")?.queued).toBe(false));
expect(task("y")?.cancel).toBeUndefined();
second.resolve();
await y;
});
});
describe("cancelling", () => {
it("shows Cancelling… until the task ends, then removes it without the finished bar", () => {
vi.useFakeTimers();
const cancel = vi.fn();
upsertTask({ id: "sync", title: "Syncing", message: "Resolving", fraction: 0.3, done: false, cancel });
cancelTask("sync");
cancelTask("sync");
expect(cancel).toHaveBeenCalledOnce();
expect(task("sync")).toMatchObject({ cancelling: true, message: "Cancelling…" });
upsertTask({ id: "sync", title: "Syncing", message: "Downloading", fraction: 0.4, done: false, cancel });
expect(task("sync")?.message).toBe("Cancelling…");
upsertTask({ id: "sync", title: "Syncing", message: null, fraction: 1, done: true });
expect(task("sync")).toBeUndefined();
});
it("the popup's cancel buttons: working, Cancelling…, and disabled with the reason", () => {
const cancel = vi.fn();
upsertTask({ id: "dl", title: "Downloading JDK 25", message: null, fraction: null, done: false, cancel });
upsertTask({
id: "inst",
title: "Installing update",
message: null,
fraction: null,
done: false,
uncancellable: "This step can't be cancelled",
});
render(<TaskStatus />);
fireEvent.click(screen.getByRole("button", { name: /Background tasks/ }));
fireEvent.click(screen.getByRole("button", { name: "Cancel Downloading JDK 25" }));
expect(cancel).toHaveBeenCalledOnce();
expect(screen.getByRole("button", { name: "Cancelling…" })).toBeDisabled();
const off = screen.getByRole("button", { name: "This step can't be cancelled" });
expect(off).toBeDisabled();
expect(off.parentElement).toHaveAttribute("data-tooltip", "This step can't be cancelled");
});
it("indexing says why it can't be stopped", () => {
setIndexing(true, 10, 20);
expect(task("indexing")).toMatchObject({ uncancellable: expect.any(String) });
expect(task("indexing")?.cancel).toBeUndefined();
});
it("backend tasks: cancellable ones cancel through cancel_task, the others say they can't", () => {
vi.useFakeTimers();
const uninstall = installTaskEvents();
const base = { type: "task" as const, message: null, fraction: null, done: false };
act(() => {
emitLocalIdeEvent({ ...base, id: 41, title: "Building workspace", cancellable: true });
emitLocalIdeEvent({ ...base, id: 42, title: "Loading", cancellable: false });
vi.advanceTimersByTime(600);
});
cancelTask(41);
expect(cancelTaskCommand).toHaveBeenCalledWith(41);
expect(useTasks.getState().tasks.find((t) => t.id === 42)).toMatchObject({
cancel: undefined,
uncancellable: "This task can't be cancelled",
});
uninstall();
});
});
+160 -28
View File
@@ -1,6 +1,6 @@
/** Background task progress (from `task` / `indexState` backend events). */
import { create } from "zustand";
import { onIdeEvent } from "@/platform/ipc";
import { commands, IpcError, onIdeEvent } from "@/platform/ipc";
import { tx } from "@/platform/i18n";
export interface BackgroundTask {
@@ -11,8 +11,14 @@ export interface BackgroundTask {
fraction: number | null;
done: boolean;
startedAt: number;
/** Stops the task (shown as a cancel button in the Background Tasks popup). */
/** Stops the task (shown as a cancel button in the Background Tasks popup). Call it through `cancelTask`. */
cancel?: () => void;
/** Why the task can't be stopped (a disabled cancel button with this tooltip), when it has no `cancel`. */
uncancellable?: string;
/** Cancel was requested: shown as "Cancelling…" until the task ends, then removed at once. */
cancelling?: boolean;
/** Waiting for the tasks ahead of it in its queue (`queueTask`); cancelling drops it. */
queued?: boolean;
}
interface TaskState {
@@ -27,22 +33,47 @@ const DONE_LINGER_MS = 600;
const SHOW_DELAY_MS = 500;
export const INDEX_TASK_ID = "indexing";
export function upsertTask(t: Omit<BackgroundTask, "startedAt">) {
export type TaskUpdate = Omit<BackgroundTask, "startedAt" | "cancelling" | "queued">;
export function upsertTask(t: TaskUpdate) {
let cancelled = false;
useTasks.setState((s) => {
const i = s.tasks.findIndex((x) => x.id === t.id);
const tasks = [...s.tasks];
if (i >= 0) tasks[i] = { ...tasks[i]!, ...t };
else if (!t.done) tasks.push({ ...t, startedAt: Date.now() });
if (i >= 0) {
cancelled = !!tasks[i]!.cancelling;
// Progress that arrives after Cancel keeps saying "Cancelling…".
tasks[i] = { ...tasks[i]!, ...t, queued: false, message: cancelled ? tasks[i]!.message : t.message };
} else if (!t.done) tasks.push({ ...t, startedAt: Date.now() });
else return s;
return { tasks };
});
if (t.done) setTimeout(() => removeTask(t.id), DONE_LINGER_MS);
// A cancelled task leaves without the finished (full) progress bar.
if (t.done) {
if (cancelled) removeTask(t.id);
else setTimeout(() => removeTask(t.id), DONE_LINGER_MS);
}
}
export function removeTask(id: BackgroundTask["id"]) {
useTasks.setState((s) => ({ tasks: s.tasks.filter((t) => t.id !== id) }));
}
/** Whether `t` offers a working Cancel now. */
export function canCancel(t: BackgroundTask): boolean {
return !!t.cancel && !t.done && !t.cancelling;
}
/** Cancels task `id`: it shows "Cancelling…" until its producer ends it. */
export function cancelTask(id: BackgroundTask["id"]) {
const t = useTasks.getState().tasks.find((x) => x.id === id);
if (!t || !canCancel(t)) return;
useTasks.setState((s) => ({
tasks: s.tasks.map((x) => (x.id === id ? { ...x, cancelling: true, message: tx("Cancelling…") } : x)),
}));
t.cancel!();
}
export function setIndexing(indexing: boolean, files: number, symbols: number) {
useTasks.setState({ indexing: indexing ? { files, symbols } : null });
upsertTask({
@@ -51,36 +82,137 @@ export function setIndexing(indexing: boolean, files: number, symbols: number) {
message: indexing ? tx`${files.toLocaleString()} files · ${symbols.toLocaleString()} symbols` : null,
fraction: null,
done: !indexing,
uncancellable: tx("Code insight needs the index; it can't be stopped"),
});
}
export function installTaskEvents(): () => void {
const pending = new Map<BackgroundTask["id"], ReturnType<typeof setTimeout>>();
const latest = new Map<BackgroundTask["id"], Omit<BackgroundTask, "startedAt">>();
const a = onIdeEvent("task", (e) => {
const t = { id: e.id, title: e.title, message: e.message, fraction: e.fraction, done: e.done };
if (useTasks.getState().tasks.some((x) => x.id === t.id)) return upsertTask(t);
if (t.done) {
clearTimeout(pending.get(t.id));
pending.delete(t.id);
latest.delete(t.id);
return;
// ---------------------------------------------------------------------------
// Queues: tasks that run one at a time, the rest waiting (listed, cancellable).
/** The error a task cancelled while queued rejects with (`isCancelled` is true for it). */
export function queuedCancelled(): IpcError {
return new IpcError({ kind: "cancelled", message: "Cancelled" });
}
interface Waiter {
id: BackgroundTask["id"];
start: () => void;
}
interface Queue {
limit: number;
running: number;
waiting: Waiter[];
}
const queues = new Map<string, Queue>();
function startNext(q: Queue) {
while (q.running < q.limit && q.waiting.length) {
const w = q.waiting.shift()!;
// The queue's Cancel is spent; the work sets its own (or says why it has none).
useTasks.setState((s) => ({
tasks: s.tasks.map((t) => (t.id === w.id ? { ...t, queued: false, cancel: undefined } : t)),
}));
w.start();
}
}
/**
* Runs `work` once the tasks queued before it in `queue` are done (`limit` run at a time).
* Until then task `task.id` is listed as queued, and its Cancel drops it: `work` never
* starts and the promise rejects with a cancellation (`isCancelled`). Once started,
* `work` reports its own progress with `upsertTask` under `task.id` and ends the task.
*/
export function queueTask<T>(
queue: string,
task: { id: BackgroundTask["id"]; title: string },
work: () => Promise<T>,
limit = 1,
): Promise<T> {
let q = queues.get(queue);
if (!q) queues.set(queue, (q = { limit, running: 0, waiting: [] }));
const slot = q;
const run = async () => {
slot.running++;
try {
return await work();
} finally {
slot.running--;
startNext(slot);
}
latest.set(t.id, t);
if (pending.has(t.id)) return;
pending.set(
t.id,
setTimeout(() => {
pending.delete(t.id);
const last = latest.get(t.id);
latest.delete(t.id);
if (last) upsertTask(last);
}, SHOW_DELAY_MS),
);
};
if (slot.running < slot.limit && !slot.waiting.length) return run();
return new Promise<T>((resolve, reject) => {
const waiter: Waiter = { id: task.id, start: () => void run().then(resolve, reject) };
slot.waiting.push(waiter);
const cancel = () => {
const i = slot.waiting.indexOf(waiter);
if (i < 0) return;
slot.waiting.splice(i, 1);
removeTask(task.id);
reject(queuedCancelled());
};
const queued: BackgroundTask = {
id: task.id,
title: task.title,
message: tx("Waiting for other tasks"),
fraction: null,
done: false,
startedAt: Date.now(),
queued: true,
cancel,
};
useTasks.setState((s) => ({ tasks: [...s.tasks.filter((t) => t.id !== task.id), queued] }));
});
}
// ---------------------------------------------------------------------------
// Tasks that often finish at once
const pending = new Map<BackgroundTask["id"], ReturnType<typeof setTimeout>>();
const latest = new Map<BackgroundTask["id"], TaskUpdate>();
/** `upsertTask`, except that a task ending within `SHOW_DELAY_MS` of its start is never shown. */
export function upsertTaskSoon(t: TaskUpdate) {
if (useTasks.getState().tasks.some((x) => x.id === t.id)) return upsertTask(t);
if (t.done) {
clearTimeout(pending.get(t.id));
pending.delete(t.id);
latest.delete(t.id);
return;
}
latest.set(t.id, t);
if (pending.has(t.id)) return;
pending.set(
t.id,
setTimeout(() => {
pending.delete(t.id);
const last = latest.get(t.id);
latest.delete(t.id);
if (last) upsertTask(last);
}, SHOW_DELAY_MS),
);
}
export function installTaskEvents(): () => void {
const notCancellable = tx("This task can't be cancelled");
const a = onIdeEvent("task", (e) =>
upsertTaskSoon({
id: e.id,
title: e.title,
message: e.message,
fraction: e.fraction,
done: e.done,
cancel: e.cancellable ? () => void commands.cancelTask(e.id).catch(() => {}) : undefined,
uncancellable: e.cancellable ? undefined : notCancellable,
}),
);
const b = onIdeEvent("indexState", (e) => setIndexing(e.indexing, e.files, e.symbols));
return () => {
pending.forEach(clearTimeout);
pending.clear();
latest.clear();
a();
b();
};