diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index e7d766fe..958afa28 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -7,7 +7,8 @@ on: concurrency: group: ci-${{ github.ref }} - cancel-in-progress: ${{ github.event_name == 'pull_request' }} + # A newer push supersedes the run in progress, on main too. + cancel-in-progress: true env: CARGO_TERM_COLOR: always @@ -29,7 +30,10 @@ jobs: with: { node-version: 22, cache: pnpm } - uses: dtolnay/rust-toolchain@stable with: { components: "clippy, rustfmt" } + # Only main saves the cache (saving takes minutes on Windows); PRs restore main's. - uses: Swatinem/rust-cache@v2 + with: + save-if: ${{ github.ref == 'refs/heads/main' }} - uses: taiki-e/install-action@nextest - name: Linux deps if: runner.os == 'Linux' @@ -45,9 +49,10 @@ jobs: # Test fixtures write LF files; keep Git from converting them on checkout. - run: git config --global core.autocrlf false - run: cargo nextest run --workspace --no-fail-fast + # nextest ran export_bindings, which rewrites src/bindings.ts. - name: Bindings are up to date if: runner.os == 'Linux' - run: pnpm gen:bindings && git diff --exit-code src/bindings.ts + run: git diff --exit-code src/bindings.ts # The remote development server ships as a static Linux binary (release.yml): # keep that build working. @@ -78,8 +83,22 @@ jobs: - run: pnpm lint - run: pnpm test - run: pnpm build - - run: pnpm exec playwright install --with-deps chromium + + e2e: + name: E2E + runs-on: ubuntu-24.04 + # Browsers and their system libraries are preinstalled; keep the tag in step with @playwright/test. + container: mcr.microsoft.com/playwright:v1.64.1-noble + steps: + - uses: actions/checkout@v4 + - uses: pnpm/action-setup@v4 + with: { version: 11 } + - uses: actions/setup-node@v4 + with: { node-version: 22, cache: pnpm } + - run: pnpm install --frozen-lockfile - run: pnpm e2e + env: + HOME: /root - uses: actions/upload-artifact@v4 if: failure() with: diff --git a/crates/ide-agent/src/lib.rs b/crates/ide-agent/src/lib.rs index 209f57ea..ca7ed203 100644 --- a/crates/ide-agent/src/lib.rs +++ b/crates/ide-agent/src/lib.rs @@ -191,6 +191,8 @@ struct Task { dirty: Mutex>, flush_scheduled: AtomicBool, stderr: Mutex>, + /// Set by `delete`; held while saving, so a late save can't bring the file back. + deleted: Mutex, } impl AgentManager { @@ -519,6 +521,7 @@ impl AgentManager { blocking(move || worktree::remove(&root, &cwd)).await?; } self.tasks.write().remove(id); + *task.deleted.lock() = true; let _ = std::fs::remove_file(self.shared.paths.store_dir.join(format!("{id}.json"))); self.shared.emit(AgentEvent::Removed { id: id.to_owned() }); Ok(()) @@ -568,6 +571,7 @@ impl Task { dirty: Mutex::new(BTreeSet::new()), flush_scheduled: AtomicBool::new(false), stderr: Mutex::new(VecDeque::new()), + deleted: Mutex::new(false), }) } @@ -604,6 +608,10 @@ impl Task { }) }; let path = self.shared.paths.store_dir.join(format!("{}.json", self.id)); + let deleted = self.deleted.lock(); + if *deleted { + return; + } if let Err(e) = bytes.map_err(|e| e.to_string()).and_then(|b| write_file(&path, &b)) { tracing::warn!("saving agent task {}: {e}", self.id); } @@ -934,15 +942,6 @@ impl Task { .collect() }) .unwrap_or_default(); - // The tool call it is about arrives as a notification on another thread: let it land first. - if let Some(tool) = params.pointer("/toolCall/toolCallId").and_then(Value::as_str) { - for _ in 0..20 { - if self.state.lock().timeline.has_tool(tool) { - break; - } - std::thread::sleep(Duration::from_millis(10)); - } - } let (tx, rx) = std::sync::mpsc::channel(); let entry = self.push_entry(AgentEntry::Permission { id, title, options: options.clone(), chosen: None }); self.permissions.lock().insert(id, tx); diff --git a/crates/ide-agent/src/timeline.rs b/crates/ide-agent/src/timeline.rs index c9a00a67..e5701235 100644 --- a/crates/ide-agent/src/timeline.rs +++ b/crates/ide-agent/src/timeline.rs @@ -37,10 +37,6 @@ impl Timeline { t } - pub fn has_tool(&self, id: &str) -> bool { - self.tools.contains_key(id) - } - pub fn push(&mut self, entry: AgentEntry) -> usize { self.entries.push(entry); self.entries.len() - 1 diff --git a/crates/ide-jsonrpc/src/lib.rs b/crates/ide-jsonrpc/src/lib.rs index 9f7aa470..ee339784 100644 --- a/crates/ide-jsonrpc/src/lib.rs +++ b/crates/ide-jsonrpc/src/lib.rs @@ -202,7 +202,10 @@ pub mod codes { /// /// Calls run on blocking-capable threads, never on the reader task, so a slow /// handler cannot stall response delivery. Notifications are delivered one at -/// a time in arrival order; requests run concurrently. +/// a time in arrival order; requests run concurrently, each once every +/// notification that arrived before it has been handled. So `on_notification` +/// must not wait for the remote side (a response to our request may depend on +/// its request to us, which would then never start); hand such work to a task. pub trait Handler: Send + Sync + 'static { fn on_notification(&self, method: &str, params: Value); /// Respond to a server→client request (e.g. `workspace/configuration`). @@ -221,14 +224,6 @@ impl Handler for NoopHandler { type Pending = HashMap>>; -/// What the delivery thread hands on, in arrival order: a response is not seen -/// before the notifications the peer sent ahead of it (a streamed result's last -/// values before its request completes). -enum Incoming { - Notification(String, Value), - Response(oneshot::Sender>, Result), -} - struct Inner { out: mpsc::UnboundedSender>, pending: Mutex, @@ -506,27 +501,35 @@ impl Peer { }); } - // Notifications (and the responses after them) are delivered in order on a dedicated thread. + // Notifications are delivered in order on a dedicated thread. Requests start from the same + // queue, so a request handler sees every notification that arrived before it, and responses + // complete from it, so a caller sees the notifications sent before its response (a streamed + // result's last values). let (notif_tx, notif_rx) = std::sync::mpsc::channel::(); { let handler = handler.clone(); + let inner = inner.clone(); + let runtime = tokio::runtime::Handle::current(); std::thread::Builder::new() .name("jsonrpc-notify".into()) .spawn(move || { - while let Ok(incoming) = notif_rx.recv() { - let (method, params) = match incoming { - Incoming::Notification(method, params) => (method, params), + while let Ok(msg) = notif_rx.recv() { + match msg { + Incoming::Notification(method, params) => { + let h = handler.clone(); + let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| { + h.on_notification(&method, params) + })); + if result.is_err() { + tracing::error!("jsonrpc notification handler panicked on {method}"); + } + } + Incoming::Request(id, method, params) => { + runtime.spawn(handle_request(inner.clone(), handler.clone(), id, method, params)); + } Incoming::Response(tx, result) => { let _ = tx.send(result); - continue; } - }; - let h = handler.clone(); - let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| { - h.on_notification(&method, params) - })); - if result.is_err() { - tracing::error!("jsonrpc notification handler panicked on {method}"); } } }) @@ -558,7 +561,7 @@ impl Peer { match msg { Ok(Some(body)) => { seen_message.store(true, Ordering::Relaxed); - dispatch(&inner, &handler, ¬if_tx, &body); + dispatch(&inner, ¬if_tx, &body); } Ok(None) => break, Err(e) => { @@ -695,7 +698,29 @@ fn parse_id(id: &Value) -> Option { } } -fn dispatch(inner: &Arc, handler: &Arc, notif_tx: &std::sync::mpsc::Sender, body: &[u8]) { +enum Incoming { + Notification(String, Value), + Request(Value, String, Value), + Response(oneshot::Sender>, Result), +} + +async fn handle_request(inner: Arc, handler: Arc, id: Value, method: String, params: Value) { + let m = method.clone(); + let result = tokio::task::spawn_blocking(move || handler.on_request(&m, params)).await; + let reply = match result { + Ok(Ok(result)) => json!({ "jsonrpc": "2.0", "id": id, "result": result }), + Ok(Err((code, message))) => { + json!({ "jsonrpc": "2.0", "id": id, "error": { "code": code, "message": message } }) + } + Err(_) => json!({ + "jsonrpc": "2.0", "id": id, + "error": { "code": codes::INTERNAL_ERROR, "message": format!("handler for {method} panicked") } + }), + }; + let _ = inner.send_value(&reply); +} + +fn dispatch(inner: &Arc, notif_tx: &std::sync::mpsc::Sender, body: &[u8]) { tracing::trace!(target: "ide_jsonrpc::wire", "<-- {}", String::from_utf8_lossy(body)); let msg: Value = match serde_json::from_slice(body) { Ok(v) => v, @@ -714,23 +739,7 @@ fn dispatch(inner: &Arc, handler: &Arc, notif_tx: &std::sync let params = obj.remove("params").unwrap_or(Value::Null); match id { Some(id) if !id.is_null() => { - let handler = handler.clone(); - let inner = inner.clone(); - tokio::spawn(async move { - let m = method.clone(); - let result = tokio::task::spawn_blocking(move || handler.on_request(&m, params)).await; - let reply = match result { - Ok(Ok(result)) => json!({ "jsonrpc": "2.0", "id": id, "result": result }), - Ok(Err((code, message))) => { - json!({ "jsonrpc": "2.0", "id": id, "error": { "code": code, "message": message } }) - } - Err(_) => json!({ - "jsonrpc": "2.0", "id": id, - "error": { "code": codes::INTERNAL_ERROR, "message": format!("handler for {method} panicked") } - }), - }; - let _ = inner.send_value(&reply); - }); + let _ = notif_tx.send(Incoming::Request(id, method, params)); } _ => { let _ = notif_tx.send(Incoming::Notification(method, params)); diff --git a/crates/ide-jsonrpc/src/tests.rs b/crates/ide-jsonrpc/src/tests.rs index 58fc3f0d..7db0823f 100644 --- a/crates/ide-jsonrpc/src/tests.rs +++ b/crates/ide-jsonrpc/src/tests.rs @@ -251,6 +251,27 @@ async fn server_to_client_request() { assert!(matches!(err, RpcError::Server { code: codes::METHOD_NOT_FOUND, .. })); } +#[tokio::test(flavor = "multi_thread")] +async fn requests_start_after_the_notifications_before_them() { + // `seen` answers with the notifications handled so far; handling one is slow. + struct Server(Recorder); + impl Handler for Server { + fn on_notification(&self, method: &str, params: Value) { + std::thread::sleep(Duration::from_millis(20)); + self.0.on_notification(method, params) + } + fn on_request(&self, _: &str, _: Value) -> Result { + Ok(json!(self.0.notifications.lock().len())) + } + } + let (client, _server) = pair(Arc::new(NoopHandler), Arc::new(Server(Recorder::default()))); + for n in 1..=5 { + client.notify("open", n).unwrap(); + let seen: usize = client.request("seen", (), None).await.unwrap(); + assert_eq!(seen, n); + } +} + #[tokio::test(flavor = "multi_thread")] async fn slow_handler_does_not_block_responses() { // The client handles a slow server→client request while its own request completes. diff --git a/crates/ide-lsp/tests/mock_server.rs b/crates/ide-lsp/tests/mock_server.rs index 3d414f77..98d7af07 100644 --- a/crates/ide-lsp/tests/mock_server.rs +++ b/crates/ide-lsp/tests/mock_server.rs @@ -72,6 +72,8 @@ struct MockHandler(Arc); impl Handler for MockHandler { fn on_notification(&self, method: &str, params: Value) { let s = &self.0; + // Held throughout: a test that sees the effect (a document, the peer gone) also sees the record. + let mut log = s.notifications.lock(); let uri = params["textDocument"]["uri"].as_str().unwrap_or_default().to_owned(); match method { "textDocument/didOpen" => { @@ -110,7 +112,7 @@ impl Handler for MockHandler { } _ => {} } - s.notifications.lock().push((method.to_owned(), params)); + log.push((method.to_owned(), params)); } fn on_request(&self, method: &str, params: Value) -> Result { @@ -276,14 +278,37 @@ impl Harness { } } +/// Waits for `f`; the bound only keeps a broken test from hanging (machines under load are slow). async fn eventually(mut f: impl FnMut() -> bool) { - for _ in 0..500 { - if f() { - return; - } + let deadline = std::time::Instant::now() + Duration::from_secs(30); + while !f() { + assert!(std::time::Instant::now() < deadline, "condition not reached in time"); tokio::time::sleep(Duration::from_millis(10)).await; } - panic!("condition not reached in time"); +} + +/// Holds mock request handlers until the test opens it. +#[derive(Default)] +struct Gate { + entered: std::sync::atomic::AtomicBool, + open: std::sync::atomic::AtomicBool, +} + +impl Gate { + /// Called from a handler: blocks until [`Gate::open`] (or a generous bound). + fn hold(&self) { + self.entered.store(true, Ordering::SeqCst); + let deadline = std::time::Instant::now() + Duration::from_secs(30); + while !self.open.load(Ordering::SeqCst) && std::time::Instant::now() < deadline { + std::thread::sleep(Duration::from_millis(1)); + } + } + fn entered(&self) -> bool { + self.entered.load(Ordering::SeqCst) + } + fn open(&self) { + self.open.store(true, Ordering::SeqCst); + } } fn uri_of(path: &Path) -> String { @@ -313,7 +338,7 @@ async fn initialize_and_incremental_sync() { assert_eq!(init["capabilities"]["textDocument"]["completion"]["completionItem"]["snippetSupport"], true); assert_eq!(init["initializationOptions"], json!({ "flag": true, "nested": { "deep": 1 } })); assert_eq!(init["rootUri"], json!(uri_of(&h.root))); - assert_eq!(s.notified("initialized").len(), 1); + eventually(|| s.notified("initialized").len() == 1).await; assert_eq!(h.statuses("mock"), vec![ServerStatus::Starting, ServerStatus::Ready]); let uri = uri_of(&path); @@ -707,11 +732,13 @@ async fn gives_up_after_repeated_crashes_and_manual_restart_works() { let h = harness(&[("mock", incremental_caps())]); h.open("a.mk", "x"); let s = h.launcher.state("mock"); + let count = |status| h.statuses("mock").iter().filter(|s| **s == status).count(); for n in 1..=4 { - eventually(|| s.launches.load(Ordering::SeqCst) == n && s.peer.lock().is_some()).await; - h.wait_status("mock", ServerStatus::Ready).await; + // Counted, not the last status: a restart may already have replaced it. + eventually(|| count(ServerStatus::Ready) == n).await; + assert_eq!(s.launches.load(Ordering::SeqCst), n); s.peer().kill(); - eventually(|| h.statuses("mock").last() == Some(&ServerStatus::Crashed)).await; + eventually(|| count(ServerStatus::Crashed) == n).await; } tokio::time::sleep(Duration::from_millis(300)).await; assert_eq!(s.launches.load(Ordering::SeqCst), 4, "initial start + 3 restarts"); @@ -1526,23 +1553,27 @@ async fn cancelling_a_references_search_stops_the_candidate_search_and_closes_fi grep_candidates(&h); let slow = uri_of(&b); let inner = open_files_only(vec![]); + let gate = Arc::new(Gate::default()); + let held = gate.clone(); *s.responder.lock() = Some(Arc::new(move |s, method, params| { if s.docs.lock().contains_key(&slow) { - std::thread::sleep(Duration::from_millis(600)); + held.hold(); } inner(s, method, params) })); let cancel = ide_core::CancellationToken::new(); - let later = cancel.clone(); - tokio::spawn(async move { - tokio::time::sleep(Duration::from_millis(200)).await; - later.cancel(); + let mgr = h.mgr.clone(); + let search = tokio::spawn({ + let cancel = cancel.clone(); + async move { mgr.references(doc, Pos::new(0, 5), true, true, cancel).await } }); - let started = std::time::Instant::now(); - let err = h.mgr.references(doc, Pos::new(0, 5), true, true, cancel).await.unwrap_err(); + eventually(|| gate.entered()).await; + cancel.cancel(); + // Returns while the server still holds the answer. + let err = search.await.unwrap().unwrap_err(); assert_eq!(err.kind, ide_core::error::ErrorKind::Cancelled); - assert!(started.elapsed() < Duration::from_millis(550)); eventually(|| uris(&s.notified("textDocument/didClose")) == vec![uri_of(&b)]).await; + gate.open(); } #[tokio::test(flavor = "multi_thread")] @@ -1586,9 +1617,11 @@ async fn opening_a_file_during_a_search_takes_it_over() { let (doc, _) = h.open("a.mk", "def target\n"); let slow = uri_of(&b); let inner = open_files_only(vec![]); + let gate = Arc::new(Gate::default()); + let held = gate.clone(); *s.responder.lock() = Some(Arc::new(move |s, method, params| { if s.docs.lock().get(&slow).is_some_and(|t| t.starts_with("call")) { - std::thread::sleep(Duration::from_millis(400)); + held.hold(); } inner(s, method, params) })); @@ -1597,9 +1630,12 @@ async fn opening_a_file_during_a_search_takes_it_over() { let mgr = h.mgr.clone(); let search = tokio::spawn(async move { mgr.references(doc, Pos::new(0, 5), true, true, Default::default()).await }); - eventually(|| s.docs.lock().contains_key(&uri_of(&b))).await; + // The server answers with the copy opened for the search. + eventually(|| gate.entered()).await; let user_doc = DocId::next(); h.mgr.did_open(user_doc, &b, &h.lang, 7, "edited target\n"); + eventually(|| s.docs.lock().get(&uri_of(&b)).map(String::as_str) == Some("edited target\n")).await; + gate.open(); search.await.unwrap().unwrap(); // The temporary copy was closed before the user's didOpen, and the user's stays open. @@ -1633,10 +1669,15 @@ async fn an_empty_answer_while_indexing_is_retried_when_indexing_ends() { let peer = s.peer(); peer.notify("$/progress", json!({ "token": "idx", "value": { "kind": "begin", "title": "Indexing" } })).unwrap(); h.wait_status("mock", ServerStatus::Busy).await; - let (state, uri) = (s.clone(), uri_of(&path)); + // Nothing found while indexing; indexing ends once that answer is out. + s.respond("textDocument/references", Value::Array(occurrences(&uri_of(&path), "def target\n"))); + let answered = Arc::new(std::sync::atomic::AtomicBool::new(false)); + let first = answered.clone(); + *s.responder.lock() = Some(Arc::new(move |_, method, _| { + (method == "textDocument/references" && !first.swap(true, Ordering::SeqCst)).then(|| json!([])) + })); tokio::spawn(async move { - tokio::time::sleep(Duration::from_millis(200)).await; - state.respond("textDocument/references", Value::Array(occurrences(&uri, "def target\n"))); + eventually(|| answered.load(Ordering::SeqCst)).await; peer.notify("$/progress", json!({ "token": "idx", "value": { "kind": "end" } })).unwrap(); }); let refs = h.mgr.references(doc, Pos::new(0, 5), true, true, Default::default()).await.unwrap(); diff --git a/crates/ide-lsp/tests/tcp_server.rs b/crates/ide-lsp/tests/tcp_server.rs index 226e0e86..b965f50d 100644 --- a/crates/ide-lsp/tests/tcp_server.rs +++ b/crates/ide-lsp/tests/tcp_server.rs @@ -115,6 +115,14 @@ impl Harness { } } +/// A loopback port where nothing listens, held (bound, not listening) by the +/// returned socket so that no other test is handed it meanwhile. +fn idle_port() -> (u16, tokio::net::TcpSocket) { + let socket = tokio::net::TcpSocket::new_v4().unwrap(); + socket.bind(([127, 0, 0, 1], 0).into()).unwrap(); + (socket.local_addr().unwrap().port(), socket) +} + async fn eventually(mut f: impl FnMut() -> bool) { for _ in 0..1000 { if f() { @@ -157,7 +165,7 @@ async fn connects_to_a_running_server_requests_and_reconnects() { #[tokio::test(flavor = "multi_thread")] async fn nothing_listening_and_no_executable_reports_how_to_fix_it() { - let port = std::net::TcpListener::bind(("127.0.0.1", 0)).unwrap().local_addr().unwrap().port(); + let (port, _reserved) = idle_port(); let h = harness(godot_cfg(port, "openidea-no-such-godot")); h.open("player.gd", "extends Node\n"); h.wait_status(ServerStatus::Missing, 1).await; @@ -170,16 +178,18 @@ async fn nothing_listening_and_no_executable_reports_how_to_fix_it() { #[cfg(unix)] #[tokio::test(flavor = "multi_thread")] async fn starts_the_command_and_waits_until_it_listens() { - let port = std::net::TcpListener::bind(("127.0.0.1", 0)).unwrap().local_addr().unwrap().port(); - let cfg = LanguageServerConfig { args: vec!["30".into()], ..godot_cfg(port, "sleep") }; + let (port, socket) = idle_port(); + let args = ["-c", "touch started; exec sleep 30"].map(String::from).to_vec(); + let cfg = LanguageServerConfig { args, ..godot_cfg(port, "sh") }; let root = tempfile::tempdir().unwrap(); - // The "server" (`sleep`) never listens itself; a listener comes up a moment later. + // The "server" never listens itself; a listener comes up once it has been started. let log = Arc::new(ServerLog::default()); let server = { let log = log.clone(); + let started = root.path().join("started"); tokio::spawn(async move { - tokio::time::sleep(Duration::from_millis(400)).await; - let listener = tokio::net::TcpListener::bind(("127.0.0.1", port)).await.unwrap(); + eventually(|| started.exists()).await; + let listener = socket.listen(16).unwrap(); let (stream, _) = listener.accept().await.unwrap(); let (r, w) = stream.into_split(); log.peers.lock().push(Peer::connect(r, w, Arc::new(MockHandler(log.clone())))); @@ -205,7 +215,7 @@ async fn starts_the_command_and_waits_until_it_listens() { #[cfg(unix)] #[tokio::test(flavor = "multi_thread")] async fn a_command_that_exits_without_listening_fails_and_cancel_stops_waiting() { - let port = std::net::TcpListener::bind(("127.0.0.1", 0)).unwrap().local_addr().unwrap().port(); + let (port, _reserved) = idle_port(); let root = tempfile::tempdir().unwrap(); let path = root.path().to_path_buf(); let cfg = godot_cfg(port, "false");