diff --git a/crates/botloftd/src/state.rs b/crates/botloftd/src/state.rs index 6a443d2e..a9fcaafa 100644 --- a/crates/botloftd/src/state.rs +++ b/crates/botloftd/src/state.rs @@ -1,7 +1,7 @@ //! State shared by every connection: the store, the supervisor, the //! courier, the event bus and what the daemon knows about itself. -use std::sync::{Arc, Mutex, MutexGuard}; +use std::sync::{Arc, Mutex, MutexGuard, TryLockError}; use std::time::Instant; use botloft_core::protocol::{ @@ -10,6 +10,7 @@ use botloft_core::protocol::{ RoutineRun, ScreenDraft, Task, }; use botloft_store::Store; +use tokio::runtime::{Handle, RuntimeFlavor}; use tokio::sync::broadcast; use crate::approvals::Approvals; @@ -159,9 +160,13 @@ impl Daemon { pub fn store(&self) -> MutexGuard<'_, Store> { // A panic while holding the lock cannot leave SQLite half-written // (every write is its own statement or transaction), so keep going. - self.store - .lock() - .unwrap_or_else(|poisoned| poisoned.into_inner()) + match self.store.try_lock() { + Ok(store) => store, + Err(TryLockError::Poisoned(poisoned)) => poisoned.into_inner(), + Err(TryLockError::WouldBlock) => { + off_runtime(|| self.store.lock()).unwrap_or_else(|poisoned| poisoned.into_inner()) + } + } } pub fn workspace_env(&self) -> WorkspaceEnv<'_> { @@ -209,3 +214,16 @@ impl Daemon { i64::try_from(self.started.elapsed().as_millis()).unwrap_or(i64::MAX) } } + +/// Runs `wait` so that it holds up no other task. A runtime thread stuck in +/// it would also stop the timers and the sockets until another thread woke +/// up (spec 11.1), so it hands its work to a new one first. Elsewhere, and +/// on the single-threaded runtime of some tests, it just waits. +fn off_runtime(wait: impl FnOnce() -> T) -> T { + match Handle::try_current() { + Ok(handle) if handle.runtime_flavor() == RuntimeFlavor::MultiThread => { + tokio::task::block_in_place(wait) + } + _ => wait(), + } +} diff --git a/crates/botloftd/tests/store_wait.rs b/crates/botloftd/tests/store_wait.rs new file mode 100644 index 00000000..8bb60180 --- /dev/null +++ b/crates/botloftd/tests/store_wait.rs @@ -0,0 +1,88 @@ +//! A task that waits for the store holds up no other task: timers fire and +//! connections are served while a slow request has the store. + +mod common; + +use std::io::{Read, Write}; +use std::net::{SocketAddr, TcpStream}; +use std::sync::Arc; +use std::time::{Duration, Instant}; + +use botloftd::state::Event; +use common::TestDaemon; +use common::bots::two_bots; +use common::stream; + +/// How long the store stays held. +const HELD: Duration = Duration::from_millis(1000); + +/// Holds the store on a plain thread, like a slow request on the blocking +/// pool, and returns once it is held. The courier wakes on its own timer +/// meanwhile (every 10 ms here) and waits for the store on a runtime +/// thread. +fn hold_store(t: &TestDaemon) -> std::thread::JoinHandle<()> { + let (locked, is_locked) = std::sync::mpsc::channel(); + let holder = std::thread::spawn({ + let daemon = Arc::clone(&t.daemon); + move || { + let _store = daemon.store(); + locked.send(()).expect("locked"); + std::thread::sleep(HELD); + } + }); + is_locked.recv().expect("store held"); + // Long enough for the courier to be waiting. A tokio sleep here would + // itself depend on the timers. + std::thread::sleep(Duration::from_millis(50)); + holder +} + +/// `GET /health` from a plain thread: how long the answer took. +fn health(addr: SocketAddr) -> Duration { + let asked = Instant::now(); + let mut socket = TcpStream::connect(addr).expect("connect"); + socket + .write_all(b"GET /health HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n") + .expect("ask"); + let mut answer = String::new(); + socket.read_to_string(&mut answer).expect("answer"); + assert!(answer.starts_with("HTTP/1.1 200"), "{answer}"); + asked.elapsed() +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn the_daemon_answers_while_the_store_is_held() { + let c = two_bots().await; + let addr = c.t.addr; + let holder = hold_store(&c.t); + // The runtime thread this test runs on stays free meanwhile. + let waited = tokio::task::spawn_blocking(move || health(addr)) + .await + .expect("health"); + holder.join().expect("holder"); + assert!(waited < HELD / 2, "/health waited {waited:?} for the store"); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn live_text_goes_out_while_the_store_is_held() { + let c = two_bots().await; + let mut events = c.t.daemon.subscribe(); + let holder = hold_store(&c.t); + let sent = Instant::now(); + // Live text needs no store: it goes out on the 40 ms timer (spec 8.3). + c.lead_process + .output(format!("{}\n", stream::delta("Still here.")).as_bytes()) + .await; + loop { + if let Ok(Event::ChatDelta(delta)) = events.recv().await { + assert_eq!(delta.text, "Still here."); + break; + } + } + let waited = sent.elapsed(); + holder.join().expect("holder"); + assert!( + waited < HELD / 2, + "live text waited {waited:?} for the store" + ); +} diff --git a/docs/spec.md b/docs/spec.md index 98d1305b..09ec413e 100644 --- a/docs/spec.md +++ b/docs/spec.md @@ -582,7 +582,7 @@ Endpoint: `ws://127.0.0.1:45710/rpc`. Mensagens seguem JSON-RPC 2.0: requests co - 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 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. +- Os requests rodam numa thread de bloqueio, nunca numa do runtime. Quem espera o banco numa thread do runtime (o courier, o supervisor, as tools) espera com `block_in_place`, que passa o resto do trabalho dessa thread para outra: parada ali, ela deixaria de rodar os timers e os sockets de todo o daemon até outra thread acordar, e um request lento seguraria até o `/health`. 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)