fix(jsonrpc): Start incoming requests after the notifications that preceded them

Requests from the other side were handed to the blocking pool as soon as
they were read, while notifications run in order on their own thread, so a
request could be handled before an earlier didOpen. Requests now start from
the same ordered queue.
Цей коміт міститься в:
l0ng-ai committed 2026-10-27 21:38:42 +08:00
1 джерело cf421e4fcb
коміт b5bc1501a0
1 file changed
+44 -34
+44 -34
Переглянути файл
@@ -498,20 +498,30 @@ impl Peer {
});
}
// Notifications are delivered in order on a dedicated thread.
let (notif_tx, notif_rx) = std::sync::mpsc::channel::<(String, Value)>();
// 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.
let (notif_tx, notif_rx) = std::sync::mpsc::channel::<Incoming>();
{
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((method, params)) = notif_rx.recv() {
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}");
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));
}
}
}
})
@@ -543,7 +553,7 @@ impl Peer {
match msg {
Ok(Some(body)) => {
seen_message.store(true, Ordering::Relaxed);
dispatch(&inner, &handler, &notif_tx, &body);
dispatch(&inner, &notif_tx, &body);
}
Ok(None) => break,
Err(e) => {
@@ -680,12 +690,28 @@ fn parse_id(id: &Value) -> Option<i64> {
}
}
fn dispatch(
inner: &Arc<Inner>,
handler: &Arc<dyn Handler>,
notif_tx: &std::sync::mpsc::Sender<(String, Value)>,
body: &[u8],
) {
enum Incoming {
Notification(String, Value),
Request(Value, String, Value),
}
async fn handle_request(inner: Arc<Inner>, handler: Arc<dyn Handler>, 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<Inner>, notif_tx: &std::sync::mpsc::Sender<Incoming>, 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,
@@ -704,26 +730,10 @@ fn dispatch(
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((method, params));
let _ = notif_tx.send(Incoming::Notification(method, params));
}
}
}