Merge pull request #79 from httpsphl/perf/courier

perf(courier): sleep until the next job, and deliver as a bot gets ready

Author: httpsphl

crates/botloft-store/src/deliveries.rs

@@ -293,5 +293,6 @@ impl Store {
     }
 }
 
+mod courier;
 #[cfg(test)]
 mod tests;

crates/botloft-store/src/deliveries/courier.rs

@@ -0,0 +1,43 @@
+//! What the courier asks to plan its sleep and to catch up with a bot that
+//! became ready (spec 9.1).
+
+use botloft_core::ids::BotId;
+use rusqlite::params;
+
+use crate::{Result, Store};
+
+impl Store {
+    /// When the courier has something to do next without being woken: the
+    /// first delivery of a bot's queue comes due, a lease runs out or an
+    /// open task passes its deadline. `None` when nothing waits for a time.
+    pub fn courier_next_at(&self) -> Result<Option<i64>> {
+        Ok(self
+            .conn
+            .prepare_cached(
+                "SELECT MIN(at) FROM ( \
+               SELECT MIN(d.next_attempt_at) AS at FROM deliveries AS d \
+               WHERE d.state = 'pending' \
+                 AND d.rowid = (SELECT MIN(rowid) FROM deliveries \
+                                WHERE bot_id = d.bot_id AND state = 'pending') \
+               UNION ALL \
+               SELECT MIN(lease_until) FROM deliveries WHERE state = 'sending' \
+               UNION ALL \
+               SELECT MIN(deadline_at) FROM tasks WHERE status = 'open')",
+            )?
+            .query_row([], |row| row.get(0))?)
+    }
+
+    /// A bot became ready: what waited only for that (never tried and
+    /// failed) goes now instead of at its next check (spec 9.1). Deliveries
+    /// that failed keep their backoff, so a message that makes the bot crash
+    /// does not come straight back.
+    pub fn hasten_deliveries(&self, bot: &BotId, now: i64) -> Result<usize> {
+        Ok(self
+            .conn
+            .prepare_cached(
+                "UPDATE deliveries SET next_attempt_at = ?2 \
+             WHERE bot_id = ?1 AND state = 'pending' AND attempts = 0 AND next_attempt_at > ?2",
+            )?
+            .execute(params![bot.as_str(), now])?)
+    }
+}

crates/botloftd/src/config.rs

@@ -75,7 +75,7 @@ pub struct CourierConfig {
 impl Default for CourierConfig {
     fn default() -> Self {
         Self {
-            poll_interval_ms: 500,
+            poll_interval_ms: 10_000,
             lease_ms: 15_000,
             max_attempts: 8,
             retry_backoff_initial_ms: 2_000,

crates/botloftd/src/courier/mod.rs

@@ -4,6 +4,7 @@
 
 mod render;
 mod settings;
+mod sleep;
 
 use std::sync::Arc;
 use std::time::Duration;
@@ -16,6 +17,7 @@ use tracing::{debug, warn};
 
 use self::render::{Context, Rendered, render};
 pub use self::settings::CourierSettings;
+use self::sleep::{became_ready, hasten, next_wait};
 use crate::service::tasks;
 use crate::state::{Daemon, Event};
 use crate::{clock, routines};
@@ -45,13 +47,22 @@ impl Courier {
 }
 
 /// Delivers until the daemon stops. Leases left by a previous run are
-/// recovered on the first cycle.
+/// recovered on the first cycle. Between cycles it sleeps until the next
+/// thing comes due, at most `poll_interval`; a new message, a finished
+/// send or a bot becoming ready wakes it sooner.
 pub async fn run(daemon: Arc<Daemon>) {
+    let mut states = daemon.subscribe();
     loop {
         cycle(&daemon);
+        let wait = next_wait(&daemon);
         tokio::select! {
-            () = tokio::time::sleep(daemon.courier.settings.poll_interval) => {}
+            () = tokio::time::sleep(wait) => {}
             () = daemon.courier.wake.notified() => {}
+            ready = became_ready(&mut states) => {
+                if let Some(bot) = ready {
+                    hasten(&daemon, &bot);
+                }
+            }
         }
     }
 }

crates/botloftd/src/courier/sleep.rs

@@ -0,0 +1,61 @@
+//! How long the courier sleeps between cycles, and catching up with a bot
+//! that became ready (spec 9.1).
+
+use std::time::Duration;
+
+use botloft_core::ids::BotId;
+use botloft_core::protocol::BotState;
+use tokio::sync::broadcast;
+use tracing::warn;
+
+use crate::state::{Daemon, Event};
+
+/// Shortest sleep between cycles, so a delivery that stays due (its
+/// prepare failed) cannot spin the courier.
+const MIN_WAIT: Duration = Duration::from_millis(250);
+
+/// How long to sleep before the next cycle.
+pub(super) fn next_wait(daemon: &Daemon) -> Duration {
+    let most = daemon.courier.settings.poll_interval;
+    let next = match daemon.store().courier_next_at() {
+        Ok(next) => next,
+        Err(err) => {
+            warn!("courier could not plan its next cycle: {err}");
+            None
+        }
+    };
+    let Some(at) = next else {
+        return most;
+    };
+    let until = at.saturating_sub(daemon.clock.now_ms()).max(0);
+    Duration::from_millis(u64::try_from(until).unwrap_or(0)).clamp(MIN_WAIT.min(most), most)
+}
+
+/// The next bot that can take messages again (spec 9.1). `None` when
+/// changes were missed: a cycle catches up with all of them.
+pub(super) async fn became_ready(states: &mut broadcast::Receiver<Event>) -> Option<BotId> {
+    loop {
+        match states.recv().await {
+            Ok(Event::BotState(change))
+                if matches!(
+                    change.state,
+                    BotState::Idle | BotState::Busy | BotState::NeedsApproval
+                ) =>
+            {
+                return Some(change.bot_id);
+            }
+            Ok(_) => {}
+            Err(broadcast::error::RecvError::Lagged(_)) => return None,
+            Err(broadcast::error::RecvError::Closed) => std::future::pending().await,
+        }
+    }
+}
+
+/// What waited only for `bot` to be ready goes in the next cycle, not at
+/// its next check.
+pub(super) fn hasten(daemon: &Daemon, bot: &BotId) {
+    let now = daemon.clock.now_ms();
+    if let Err(err) = daemon.store().hasten_deliveries(bot, now) {
+        warn!(bot = %bot, "courier could not bring deliveries forward: {err}");
+    }
+}

crates/botloftd/tests/courier.rs

@@ -75,8 +75,7 @@ async fn an_owner_message_reaches_the_bot_as_its_user_and_is_read() {
         (&json!("owner"), &json!("Olá! Tudo bem?"))
     );
 
-    t.until_state(&bot, BotState::Idle).await;
-    t.clock.advance(Duration::from_secs(5));
+    // It waited for the bot to be ready, and went as soon as it was.
     let line = process.wait_lines(1).await.remove(0);
     assert_eq!(line["type"], "user");
     assert_eq!(
@@ -115,8 +114,6 @@ async fn messages_to_one_bot_arrive_in_order() {
     for body in ["one", "two", "three"] {
         send(&mut app, &bot, body).await;
     }
-    t.until_state(&bot, BotState::Idle).await;
-    t.clock.advance(Duration::from_secs(5));
     let lines = process.wait_lines(3).await;
     let texts: Vec<_> = lines.iter().map(text_of).collect();
     assert_eq!(texts, ["one", "two", "three"]);
@@ -162,9 +159,7 @@ async fn what_a_dead_process_never_read_goes_back_until_it_gives_up() {
         (&retried["state"], &retried["attempts"]),
         (&json!("pending"), &json!(0))
     );
-    // The bot was still restarting at the retry: the courier waits 5 s.
-    t.until_state(&bot, BotState::Idle).await;
-    t.clock.advance(Duration::from_secs(5));
+    // The bot was still restarting at the retry: it goes once it is ready.
     let fresh = t.process_of(&bot).await;
     assert_eq!(text_of(&fresh.wait_lines(1).await[0]), "hello");
     let again = app
@@ -260,3 +255,25 @@ async fn invalid_messages_are_refused() {
         .expect_err("missing");
     assert_eq!(err.code, NOT_FOUND);
 }
+
+#[tokio::test]
+async fn a_message_that_waited_for_the_bot_goes_as_soon_as_it_is_ready() {
+    let t = TestDaemon::start_supervised().await;
+    let mut app = t.session().await;
+    let bot = crew_and_bot(&mut app).await;
+    t.until_state(&bot, BotState::Idle).await;
+    let paused = |paused: bool| json!({ "botId": bot["id"], "paused": paused });
+    app.call("bots.setPaused", paused(true))
+        .await
+        .expect("pause");
+    send(&mut app, &bot, "hello").await;
+    // The courier sees it and puts it off while the bot is paused.
+    tokio::time::sleep(Duration::from_millis(100)).await;
+
+    app.call("bots.setPaused", paused(false))
+        .await
+        .expect("resume");
+    // The manual clock never moves: the bot being ready is what it waited for.
+    let process = t.runtime.process(2).await;
+    assert_eq!(text_of(&process.wait_lines(1).await[0]), "hello");
+}

crates/botloftd/tests/delete_messages.rs

@@ -3,9 +3,6 @@
 
 mod common;
 
-use std::time::Duration;
-
-use botloft_core::protocol::BotState;
 use common::Client;
 use common::bots::{text_of, two_bots};
 use serde_json::{Value, json};
@@ -133,8 +130,6 @@ async fn a_message_still_waiting_arrives_from_a_deleted_bot() {
         .expect("resume");
     // The lead's, the writer's first and the one the writer resumes in.
     let process = c.t.runtime.process(3).await;
-    c.t.until_state(&writer, BotState::Idle).await;
-    c.t.clock.advance(Duration::from_secs(5));
     let lines = process.wait_lines(1).await;
     assert_eq!(
         text_of(&lines[0]),

docs/spec.md

@@ -154,7 +154,7 @@ restart_backoff_max_ms = 300000
 fresh_start_if_dies_within_s = 15
 
 [courier]
-poll_interval_ms = 500
+poll_interval_ms = 10000     # o maior intervalo entre ciclos do courier (9.1)
 lease_ms = 15000
 max_attempts = 8
 retry_backoff_initial_ms = 2000
@@ -422,9 +422,9 @@ O dono vê quantos tokens cada bot gasta: em cada turno, no chat, e somados por
 ### 9.1 Fluxo
 
 1. `messages.send` (owner) ou tool `send_message` (bot) grava `message`, `delivery` (`pending`) e o item `inbound` no chat do destinatário, numa transação. Anexos são gravados antes (9.5).
-2. O **courier** acorda a cada `poll_interval_ms` (e na hora, via notify, quando entra delivery nova ou termina um envio).
+2. O **courier** dorme até a próxima coisa com hora marcada: a primeira delivery da fila de algum bot vencer, um lease acabar ou uma task aberta passar do prazo; no mínimo 250 ms e no máximo `poll_interval_ms`. Acorda na hora, via notify, quando entra delivery nova ou termina um envio, e quando um bot passa a aceitar mensagens (`bot.state` em `idle`, `busy` ou `needs_approval`). Antes, acordava a cada 500 ms, parado ou não.
 3. Entrega em ordem, uma por vez por bot: de cada bot, só a delivery `pending` mais antiga pode sair, quando `next_attempt_at <= agora` e o bot não tem outra em `sending`. Ela vira `sending` com `lease_until`.
-4. Se o bot não está em `idle`/`busy`/`needs_approval`: volta para `pending` com `next_attempt_at` em 5 s, **sem** contar tentativa. Bot ou crew arquivados: a delivery vira `dead` na hora.
+4. Se o bot não está em `idle`/`busy`/`needs_approval`: volta para `pending` com `next_attempt_at` em 5 s, **sem** contar tentativa. Quando o bot passa a aceitar mensagens, as deliveries dele que nunca falharam (`attempts = 0`) vencem na hora, sem esperar esses 5 s; as que falharam mantêm o backoff, para uma mensagem que derruba o bot não voltar em seguida. Bot ou crew arquivados: a delivery vira `dead` na hora.
 5. Monta a mensagem (9.2) na hora do envio e escreve uma linha no stdin do processo, com timeout de 10 s.
 6. Escrita aceita: `sent`, com a generation do processo. Falha de escrita (processo saindo): volta para `pending` sem contar tentativa.
 7. Quando o Claude Code pega a mensagem, num turno novo ou no turno que já roda, ele a devolve no stdout com o mesmo `uuid` (`--replay-user-messages`), e a delivery ganha `read_at`. O app mostra isso como "lida", com os dois tiques em azul-claro (15.3).