Merge pull request #81 from httpsphl/perf/rpc-off-loop

perf(rpc): answer requests on the blocking pool; slow reads aside

Author: httpsphl

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(&note).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(&notification("crew.changed", json!({}))).expect("json");
+            serde_json::from_str(&notification("crew.changed", &json!({}))).expect("json");
         assert_eq!(
             note,
             json!({"jsonrpc":"2.0","method":"crew.changed","params":{}})

crates/botloftd/src/rpc/mod.rs

@@ -134,13 +134,20 @@ async fn run(
     outbox: &mpsc::Sender<String>,
     frames: &watch::Sender<Option<String>>,
     mut events: broadcast::Receiver<Event>,
-    daemon: &Daemon,
+    daemon: &Arc<Daemon>,
 ) {
     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<String>),
+    /// 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<String>>,
-) -> Option<String> {
+) -> 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<Daemon>, request: jsonrpc::Request) -> Option<String> {
+    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<Daemon>, request: jsonrpc::Request, outbox: mpsc::Sender<String>) {
+    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<Value, RpcError>) -> 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),
+    }
 }

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))
     }
 }
 

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::<Vec<_>>());
+}

docs/spec.md

@@ -580,7 +580,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)