From 90b8e05a51fc196b1ac3950fd821853775c32489 Mon Sep 17 00:00:00 2001 From: Phelipe Lorran <112786693+httpsphl@users.noreply.github.com> Date: Wed, 30 Sep 2026 22:10:30 -0300 Subject: [PATCH] perf(rpc): answer requests on the blocking pool; slow reads aside Each request ran inline in the connection loop, on a runtime thread: a files.list walk over a large work folder held both that thread and every notification behind it. Requests now run on the blocking pool. They still answer in order and before the notifications after them, since the app stores a whole Bot or Routine from an answer and takes it as newer than what came before. Reads the app never sets against notifications (files, screens, attachments, token usage) answer on their own, so they hold back no live text. Notifications are written straight to text, without building a Value first. --- crates/botloftd/src/rpc/jsonrpc.rs | 22 ++++- crates/botloftd/src/rpc/mod.rs | 136 +++++++++++++++++++--------- crates/botloftd/src/rpc/watching.rs | 3 +- crates/botloftd/tests/rpc_queue.rs | 88 ++++++++++++++++++ docs/spec.md | 3 +- 5 files changed, 203 insertions(+), 49 deletions(-) create mode 100644 crates/botloftd/tests/rpc_queue.rs diff --git a/crates/botloftd/src/rpc/jsonrpc.rs b/crates/botloftd/src/rpc/jsonrpc.rs index 889c2a4d..5ec60e1b 100644 --- a/crates/botloftd/src/rpc/jsonrpc.rs +++ b/crates/botloftd/src/rpc/jsonrpc.rs @@ -2,6 +2,7 @@ //! notifications. use botloft_core::protocol::error_code; +use serde::Serialize; use serde_json::{Map, Value, json}; use crate::service::ApiError; @@ -104,8 +105,23 @@ pub fn failure(id: &Value, err: &RpcError) -> String { json!({ "jsonrpc": "2.0", "id": id, "error": error }).to_string() } -pub fn notification(method: &str, params: Value) -> String { - json!({ "jsonrpc": "2.0", "method": method, "params": params }).to_string() +/// A notification, written straight to text without building a `Value`. +pub fn notification(method: &str, params: &impl Serialize) -> String { + #[derive(Serialize)] + struct Notification<'a, P> { + jsonrpc: &'static str, + method: &'a str, + params: &'a P, + } + let note = Notification { + jsonrpc: "2.0", + method, + params, + }; + serde_json::to_string(¬e).unwrap_or_else(|err| { + tracing::warn!(method, "could not write a notification: {err}"); + json!({ "jsonrpc": "2.0", "method": method, "params": null }).to_string() + }) } /// Params for methods that take none. @@ -182,7 +198,7 @@ mod tests { let failed: Value = serde_json::from_str(&failure(&json!(1), &err)).expect("json"); assert_eq!(failed["error"], json!({"code":-32002,"message":"missing"})); let note: Value = - serde_json::from_str(¬ification("crew.changed", json!({}))).expect("json"); + serde_json::from_str(¬ification("crew.changed", &json!({}))).expect("json"); assert_eq!( note, json!({"jsonrpc":"2.0","method":"crew.changed","params":{}}) diff --git a/crates/botloftd/src/rpc/mod.rs b/crates/botloftd/src/rpc/mod.rs index 22c8f441..f6888b10 100644 --- a/crates/botloftd/src/rpc/mod.rs +++ b/crates/botloftd/src/rpc/mod.rs @@ -134,13 +134,20 @@ async fn run( outbox: &mpsc::Sender, frames: &watch::Sender>, mut events: broadcast::Receiver, - daemon: &Daemon, + daemon: &Arc, ) { let mut watch = Watch::default(); loop { let frame = tokio::select! { message = stream.next() => match message { - Some(Ok(Message::Text(text))) => handle(daemon, &text, &mut watch, frames), + Some(Ok(Message::Text(text))) => match route(daemon, &text, &mut watch, frames) { + Routed::Reply(frame) => frame, + Routed::InOrder(request) => answer(daemon, request).await, + Routed::Aside(request) => { + aside(daemon, request, outbox.clone()); + None + } + }, Some(Ok(Message::Close(_)) | Err(_)) | None => return, Some(Ok(_)) => None, }, @@ -166,18 +173,40 @@ async fn run( } } -fn handle( +/// Reads that the app never sets against notifications: their answer may +/// come after newer notifications, so they wait neither for the requests +/// before them nor hold back the notifications meanwhile. +const ASIDE: [&str; 5] = [ + method::FILES_LIST, + method::FILES_READ, + method::SCREENS_LIST, + method::ATTACHMENTS_READ, + method::USAGE_TOKENS, +]; + +enum Routed { + /// Answered here, or nothing to answer. + Reply(Option), + /// Answered before anything else goes out (see [`answer`]). + InOrder(jsonrpc::Request), + /// Answered whenever it is ready (see [`ASIDE`]). + Aside(jsonrpc::Request), +} + +fn route( daemon: &Daemon, text: &str, watch: &mut Watch, frames: &watch::Sender>, -) -> Option { +) -> Routed { let request = match jsonrpc::parse(text) { Ok(request) => request, - Err((id, err)) => return Some(jsonrpc::failure(&id, &err)), + Err((id, err)) => return Routed::Reply(Some(jsonrpc::failure(&id, &err))), }; debug!(method = %request.method, "rpc request"); - let id = request.id.clone()?; + let Some(id) = request.id.clone() else { + return Routed::Reply(None); + }; // Watching a browser belongs to this connection (spec 21.7). let result = match request.method.as_str() { method::BROWSER_WATCH | method::BROWSER_UNWATCH => { @@ -194,46 +223,67 @@ fn handle( | method::BROWSER_NEW_TAB | method::BROWSER_SWITCH_TAB | method::BROWSER_OPEN => watch.request(daemon, &request.method, request.params), - name => dispatch::dispatch(daemon, name, request.params), + name if ASIDE.contains(&name) => return Routed::Aside(request), + _ => return Routed::InOrder(request), }; - Some(match result { - Ok(value) => jsonrpc::success(&id, value), - Err(err) => jsonrpc::failure(&id, &err), + Routed::Reply(Some(reply(&id, result))) +} + +/// Answers on the blocking pool, so the request holds no runtime thread. +/// The connection waits for it: the app takes an answer as newer than +/// every notification before it (spec 11.1). +async fn answer(daemon: &Arc, request: jsonrpc::Request) -> Option { + let daemon = Arc::clone(daemon); + let id = request.id.clone().unwrap_or(Value::Null); + let answered = tokio::task::spawn_blocking(move || { + dispatch::dispatch(&daemon, &request.method, request.params) }) + .await; + let result = answered.unwrap_or_else(|_| { + Err(RpcError::new( + error_code::INTERNAL_ERROR, + "the request failed inside the daemon", + )) + }); + Some(reply(&id, result)) } -fn to_notification(event: &Event) -> String { - let (name, params) = match event { - Event::CrewChanged(crew) => (notification::CREW_CHANGED, serde_json::to_value(crew)), - Event::CrewDeleted(crew) => (notification::CREW_DELETED, serde_json::to_value(crew)), - Event::BotChanged(bot) => (notification::BOT_CHANGED, serde_json::to_value(bot)), - Event::BotDeleted(bot) => (notification::BOT_DELETED, serde_json::to_value(bot)), - Event::FolderRecycled(folder) => { - (notification::FOLDER_RECYCLED, serde_json::to_value(folder)) - } - Event::BotState(state) => (notification::BOT_STATE, serde_json::to_value(state)), - Event::BotContext(context) => (notification::BOT_CONTEXT, serde_json::to_value(context)), - Event::ChatItem(item) => (notification::CHAT_ITEM, serde_json::to_value(item)), - Event::ChatDelta(delta) => (notification::CHAT_DELTA, serde_json::to_value(delta)), - Event::MessageCreated(message) => { - (notification::MESSAGE_CREATED, serde_json::to_value(message)) - } - Event::DeliveryChanged(delivery) => ( - notification::DELIVERY_CHANGED, - serde_json::to_value(delivery), - ), - Event::TaskChanged(task) => (notification::TASK_CHANGED, serde_json::to_value(task)), - Event::RoutineChanged(routine) => { - (notification::ROUTINE_CHANGED, serde_json::to_value(routine)) +/// Answers an [`ASIDE`] read on its own. +fn aside(daemon: &Arc, request: jsonrpc::Request, outbox: mpsc::Sender) { + let daemon = Arc::clone(daemon); + tokio::spawn(async move { + if let Some(frame) = answer(&daemon, request).await { + let _ = outbox.send(frame).await; } - Event::RoutineRun(run) => (notification::ROUTINE_RUN, serde_json::to_value(run)), - Event::BrowserChanged(state) => { - (notification::BROWSER_CHANGED, serde_json::to_value(state)) - } - Event::BrowserAction(action) => { - (notification::BROWSER_ACTION, serde_json::to_value(action)) - } - Event::ScreenDraft(draft) => (notification::SCREEN_DRAFT, serde_json::to_value(draft)), - }; - jsonrpc::notification(name, params.unwrap_or(Value::Null)) + }); +} + +fn reply(id: &Value, result: Result) -> String { + match result { + Ok(value) => jsonrpc::success(id, value), + Err(err) => jsonrpc::failure(id, &err), + } +} + +fn to_notification(event: &Event) -> String { + use jsonrpc::notification as note; + match event { + Event::CrewChanged(crew) => note(notification::CREW_CHANGED, crew), + Event::CrewDeleted(crew) => note(notification::CREW_DELETED, crew), + Event::BotChanged(bot) => note(notification::BOT_CHANGED, bot), + Event::BotDeleted(bot) => note(notification::BOT_DELETED, bot), + Event::FolderRecycled(folder) => note(notification::FOLDER_RECYCLED, folder), + Event::BotState(state) => note(notification::BOT_STATE, state), + Event::BotContext(context) => note(notification::BOT_CONTEXT, context), + Event::ChatItem(item) => note(notification::CHAT_ITEM, item), + Event::ChatDelta(delta) => note(notification::CHAT_DELTA, delta), + Event::MessageCreated(message) => note(notification::MESSAGE_CREATED, message), + Event::DeliveryChanged(delivery) => note(notification::DELIVERY_CHANGED, delivery), + Event::TaskChanged(task) => note(notification::TASK_CHANGED, task), + Event::RoutineChanged(routine) => note(notification::ROUTINE_CHANGED, routine), + Event::RoutineRun(run) => note(notification::ROUTINE_RUN, run), + Event::BrowserChanged(state) => note(notification::BROWSER_CHANGED, state), + Event::BrowserAction(action) => note(notification::BROWSER_ACTION, action), + Event::ScreenDraft(draft) => note(notification::SCREEN_DRAFT, draft), + } } diff --git a/crates/botloftd/src/rpc/watching.rs b/crates/botloftd/src/rpc/watching.rs index 120e7fb7..e9e665f5 100644 --- a/crates/botloftd/src/rpc/watching.rs +++ b/crates/botloftd/src/rpc/watching.rs @@ -174,8 +174,7 @@ impl Watch { return None; } let frame = watching.frames.borrow_and_update().clone()?; - let params = serde_json::to_value(&*frame).unwrap_or(Value::Null); - Some(jsonrpc::notification(notification::BROWSER_FRAME, params)) + Some(jsonrpc::notification(notification::BROWSER_FRAME, &*frame)) } } diff --git a/crates/botloftd/tests/rpc_queue.rs b/crates/botloftd/tests/rpc_queue.rs new file mode 100644 index 00000000..906cd58b --- /dev/null +++ b/crates/botloftd/tests/rpc_queue.rs @@ -0,0 +1,88 @@ +//! Requests run off the runtime's threads (spec 11.1). An answer still +//! goes out before the notifications that follow it, except for the reads +//! the app never sets against notifications, which hold nothing back. + +mod common; + +use std::sync::Arc; +use std::sync::mpsc; +use std::time::Duration; + +use common::bots::{Crew, two_bots}; +use common::stream; +use serde_json::{Value, json}; + +/// Holds the store for a while on another thread, so the next request +/// waits for it, then sends `request` and has the lead bot write text. +/// Returns which went out first: the answer or the text. +async fn race(c: &mut Crew, request: Value) -> Vec<&'static str> { + // Both bots are ready and the courier has nothing to do. + for _ in 0..2 { + while c.app.notification("bot.state").await["state"] != "idle" {} + } + let (locked, is_locked) = mpsc::channel(); + let holder = std::thread::spawn({ + let daemon = Arc::clone(&c.t.daemon); + move || { + let _store = daemon.store(); + locked.send(()).expect("locked"); + std::thread::sleep(Duration::from_millis(1000)); + } + }); + is_locked.recv().expect("store held"); + c.app.send_raw(&request.to_string()).await; + // The block that ends sends the text at once, without the timer. + let stop = + json!({ "type": "stream_event", "event": { "type": "content_block_stop", "index": 0 } }); + c.lead_process + .output(format!("{}\n{stop}\n", stream::delta("Still here.")).as_bytes()) + .await; + + let mut order = Vec::new(); + while order.len() < 2 { + let frame = c.app.recv().await.expect("open"); + if frame["id"] == request["id"] { + assert!(frame.get("result").is_some(), "{frame}"); + order.push("answer"); + } else if frame["method"] == "chat.delta" { + assert_eq!(frame["params"]["text"], "Still here."); + order.push("text"); + } + } + holder.join().expect("holder"); + order +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 16)] +async fn a_slow_read_holds_back_no_notification() { + let mut c = two_bots().await; + let bots = c.app.call("bots.list", json!({})).await.expect("bots"); + let bot = bots[0]["id"].clone(); + let request = + json!({ "jsonrpc": "2.0", "id": 900, "method": "files.list", "params": { "botId": bot } }); + assert_eq!(race(&mut c, request).await, ["text", "answer"]); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 16)] +async fn an_answer_goes_out_before_the_notifications_after_it() { + let mut c = two_bots().await; + let request = json!({ "jsonrpc": "2.0", "id": 901, "method": "crews.list" }); + assert_eq!(race(&mut c, request).await, ["answer", "text"]); +} + +#[tokio::test] +async fn requests_are_answered_in_the_order_they_came() { + let mut c = two_bots().await; + for id in 1000..1010 { + let request = json!({ "jsonrpc": "2.0", "id": id, "method": "crews.list" }); + c.app.send_raw(&request.to_string()).await; + } + let mut ids = Vec::new(); + while ids.len() < 10 { + let frame = c.app.recv().await.expect("open"); + if let Some(id) = frame.get("id").and_then(Value::as_u64) { + ids.push(id); + } + } + assert_eq!(ids, (1000..1010).collect::>()); +} diff --git a/docs/spec.md b/docs/spec.md index 265817d7..f1e2f844 100644 --- a/docs/spec.md +++ b/docs/spec.md @@ -579,7 +579,8 @@ Endpoint: `ws://127.0.0.1:45710/rpc`. Mensagens seguem JSON-RPC 2.0: requests co - Primeiro request obrigatório: `session.hello {token, client: {name, version}, protocol: 2}` -> `{daemonVersion, protocol}`. - Qualquer outro método antes do hello: erro `-32001` e a conexão fecha. Token errado também dá `-32001`; `protocol` diferente dá `-32004`; nos dois casos a conexão fecha. O hello tem que chegar em até 10 s. - `Origin` aceito: `http://tauri.localhost`, `tauri://localhost` e `http://localhost:1420` (dev). Qualquer outro `Origin` recebe HTTP 403 antes do upgrade. Sem `Origin` (cliente nativo, testes) é aceito: navegadores sempre mandam o header, e o token continua obrigatório. -- Um cliente lento que deixa acumular mais de 1024 notificações é desconectado e recarrega o estado ao reconectar. +- Um cliente lento que deixa acumular mais de 4096 notificações é desconectado e recarrega o estado ao reconectar. +- Os requests rodam numa thread de bloqueio, nunca numa do runtime. Os de uma conexão são respondidos um de cada vez, na ordem em que chegaram, e a resposta sai antes de qualquer notificação que venha depois dela: o app trata uma resposta como mais nova que tudo o que recebeu antes e grava o objeto inteiro (um `Bot`, uma `Routine`, a lista de uma carga). A exceção são as leituras que o app nunca cruza com notificações (`files.list`, `files.read`, `screens.list`, `attachments.read`, `usage.tokens`). Elas correm por fora e respondem quando ficam prontas, de modo que varrer uma pasta grande ou ler um arquivo grande não segura o texto ao vivo. Os requests do navegador (21.7, 21.10) pertencem à conexão e são respondidos na hora. - A versão 2 do protocolo troca `terminal.*` pelo chat (ADR 0001). ### 11.2 Métodos (MVP)