fix(daemon): wait for a busy store off the runtime threads

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<T>(wait: impl FnOnce() -> T) -> T {
+    match Handle::try_current() {
+        Ok(handle) if handle.runtime_flavor() == RuntimeFlavor::MultiThread => {
+            tokio::task::block_in_place(wait)
+        }
+        _ => wait(),
+    }
+}

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"
+    );
+}

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)