From 00a647daa273df59dd84d282e2af2b4d7cfb40df Mon Sep 17 00:00:00 2001 From: l0ng-ai <24760907+l0ng-ai@users.noreply.github.com> Date: Tue, 27 Oct 2026 21:57:42 +0800 Subject: [PATCH] fix(lsp): Start JSON-RPC requests after the notifications before them ide-jsonrpc handled incoming notifications in order on a dedicated thread but started each incoming request on its own blocking task right away, so a request could be handled before a notification that arrived ahead of it. The mock language server in the ide-lsp tests answered `textDocument/references` before applying the `didOpen` of a file opened for a project-wide search (missing results, a search finishing before the test could cancel it or open the file), and on the client side an ACP `session/request_permission` could overtake the `session/update` that announced its tool call (ide-agent polled for up to 200 ms to cover it). A request now starts once every notification received before it has been handled; requests still run concurrently. The ide-agent polling is gone. Tests no longer depend on timing: - the references tests hold the mock's answer with a gate and cancel or open the file while it is held, instead of sleeping and measuring; - the indexing retry ends indexing once the first answer is out; - the crash/restart test counts Ready/Crashed statuses instead of reading the last one, which a restart can already have replaced; - the mock records a notification under the same lock as its effect, and `eventually` waits up to 30 s; - the TCP tests hold their port bound (not listening) instead of freeing it for another test to take, and listen only after the command started. Claude-Session: https://claude.ai/code/session_01VUYsdXA2nkF5v6zsC93dpn --- crates/ide-agent/src/lib.rs | 9 --- crates/ide-agent/src/timeline.rs | 4 -- crates/ide-jsonrpc/src/lib.rs | 34 +++++++---- crates/ide-jsonrpc/src/tests.rs | 21 +++++++ crates/ide-lsp/tests/mock_server.rs | 87 +++++++++++++++++++++-------- crates/ide-lsp/tests/tcp_server.rs | 24 +++++--- 6 files changed, 126 insertions(+), 53 deletions(-) diff --git a/crates/ide-agent/src/lib.rs b/crates/ide-agent/src/lib.rs index c62073bf..7f78a7ef 100644 --- a/crates/ide-agent/src/lib.rs +++ b/crates/ide-agent/src/lib.rs @@ -923,15 +923,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 638fd75f..0d628656 100644 --- a/crates/ide-jsonrpc/src/lib.rs +++ b/crates/ide-jsonrpc/src/lib.rs @@ -202,7 +202,8 @@ 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. pub trait Handler: Send + Sync + 'static { fn on_notification(&self, method: &str, params: Value); /// Respond to a server→client request (e.g. `workspace/configuration`). @@ -219,6 +220,13 @@ impl Handler for NoopHandler { } } +/// Work for the in-order notification thread. +enum Inbound { + Notification(String, Value), + /// A request may start: the notifications before it are handled. + Release(oneshot::Sender<()>), +} + type Pending = HashMap>>; struct Inner { @@ -499,13 +507,20 @@ impl Peer { } // Notifications are delivered in order on a dedicated thread. - let (notif_tx, notif_rx) = std::sync::mpsc::channel::<(String, Value)>(); + let (notif_tx, notif_rx) = std::sync::mpsc::channel::(); { let handler = handler.clone(); std::thread::Builder::new() .name("jsonrpc-notify".into()) .spawn(move || { - while let Ok((method, params)) = notif_rx.recv() { + while let Ok(inbound) = notif_rx.recv() { + let (method, params) = match inbound { + Inbound::Notification(method, params) => (method, params), + Inbound::Release(start) => { + let _ = start.send(()); + continue; + } + }; let h = handler.clone(); let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| { h.on_notification(&method, params) @@ -680,12 +695,7 @@ fn parse_id(id: &Value) -> Option { } } -fn dispatch( - inner: &Arc, - handler: &Arc, - notif_tx: &std::sync::mpsc::Sender<(String, Value)>, - body: &[u8], -) { +fn dispatch(inner: &Arc, handler: &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, @@ -706,7 +716,11 @@ fn dispatch( Some(id) if !id.is_null() => { let handler = handler.clone(); let inner = inner.clone(); + let (start, started) = oneshot::channel(); + let _ = notif_tx.send(Inbound::Release(start)); tokio::spawn(async move { + // E.g. a `didOpen` before a request must be applied when the request is answered. + let _ = started.await; let m = method.clone(); let result = tokio::task::spawn_blocking(move || handler.on_request(&m, params)).await; let reply = match result { @@ -723,7 +737,7 @@ fn dispatch( }); } _ => { - let _ = notif_tx.send((method, params)); + let _ = notif_tx.send(Inbound::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");