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)