From 05386447e7644ddc540dad7c4bd61046bfa9fac2 Mon Sep 17 00:00:00 2001 From: Phelipe Lorran <112786693+httpsphl@users.noreply.github.com> Date: Wed, 30 Sep 2026 22:16:12 -0300 Subject: [PATCH] perf(courier): read and write attachments with the store free The courier held the store for its whole cycle, including reading each inline image (up to 5 MB) and base64-encoding it, and messages.send decoded and wrote the owner's files (up to the attachment limit) under the store lock. Every chat item, tool call and request waited. The courier now locks per step and renders a delivery between them; claiming still checks the delivery is waiting. messages.send saves the files first and checks the bot again before storing the message. --- crates/botloftd/src/courier/mod.rs | 103 +++++++++++++++--------- crates/botloftd/src/service/messages.rs | 8 +- 2 files changed, 73 insertions(+), 38 deletions(-) diff --git a/crates/botloftd/src/courier/mod.rs b/crates/botloftd/src/courier/mod.rs index d3cf76c7..68f46124 100644 --- a/crates/botloftd/src/courier/mod.rs +++ b/crates/botloftd/src/courier/mod.rs @@ -6,11 +6,12 @@ mod render; mod settings; mod sleep; +use std::path::PathBuf; use std::sync::Arc; use std::time::Duration; use botloft_core::ids::BotId; -use botloft_core::protocol::{BotState, Delivery, Message, SenderKind}; +use botloft_core::protocol::{BotState, Delivery, Message, SenderKind, Task}; use botloft_store::{DeliveryOutcome, Store, StoreError}; use tokio::sync::Notify; use tracing::{debug, warn}; @@ -69,36 +70,47 @@ pub async fn run(daemon: Arc) { /// What to do with a due delivery. enum Step { - Send(Rendered), + Send(Box), /// The bot cannot take messages yet. Wait, /// It can never be delivered. Drop(&'static str), } +/// One pass over the queue. The store is locked step by step, never +/// while an attachment is read from disk. fn cycle(daemon: &Arc) { let now = daemon.clock.now_ms(); - let store = daemon.store(); - match store.recover_leases(now) { - Ok(recovered) => { - for delivery in recovered { - debug!(delivery = %delivery.id, "a send was cut off; delivering again"); - daemon.emit(Event::DeliveryChanged(delivery)); + let due = { + let store = daemon.store(); + match store.recover_leases(now) { + Ok(recovered) => { + for delivery in recovered { + debug!(delivery = %delivery.id, "a send was cut off; delivering again"); + daemon.emit(Event::DeliveryChanged(delivery)); + } } + Err(err) => warn!("courier could not recover leases: {err}"), } - Err(err) => warn!("courier could not recover leases: {err}"), - } - tasks::expire_overdue(daemon, &store, now); - let due = match store.due_deliveries(now) { - Ok(due) => due, - Err(err) => { - warn!("courier could not read the queue: {err}"); - return; + tasks::expire_overdue(daemon, &store, now); + match store.due_deliveries(now) { + Ok(due) => due, + Err(err) => { + warn!("courier could not read the queue: {err}"); + return; + } } }; for delivery in due { - let changed = match prepare(daemon, &store, &delivery, now) { - Ok(Step::Send(rendered)) => send(daemon, &store, &delivery, &rendered, now), + let store = daemon.store(); + let changed = match prepare(daemon, &store, &delivery) { + Ok(Step::Send(draft)) => { + drop(store); + // Inline images are read here, with the store free. + let rendered = draft.render(now); + // Claiming checks the delivery is still waiting. + send(daemon, &daemon.store(), &delivery, &rendered, now) + } Ok(Step::Wait) => { let until = clock::after(now, WAIT_FOR_BOT); finish(&store, &delivery, DeliveryOutcome::Defer { until }, now) @@ -170,13 +182,8 @@ fn send( finish(store, delivery, outcome, now) } -/// Decides whether the delivery can go now and renders it. -fn prepare( - daemon: &Daemon, - store: &Store, - delivery: &Delivery, - now: i64, -) -> Result { +/// Decides whether the delivery can go now and gathers what it needs. +fn prepare(daemon: &Daemon, store: &Store, delivery: &Delivery) -> Result { let Some(message) = store.message(&delivery.message_id)? else { return Ok(Step::Drop("the message no longer exists")); }; @@ -204,18 +211,42 @@ fn prepare( (SenderKind::Bot, Some(id)) => store.bot(id)?.map(|sender| sender.handle), _ => None, }; - let workspace = daemon.paths.bot_workspace(&crew.slug, &bot.slug); let routine = routine_context(store, &message)?; - let context = Context { - crew_name: &crew.name, - sender_handle: sender.as_deref(), - task: task.as_ref(), - workspace: &workspace, - routine: routine - .as_ref() - .map(|(name, scheduled, zone)| (name.as_str(), scheduled.as_str(), zone.as_str())), - }; - Ok(Step::Send(render(&message, &context, now))) + Ok(Step::Send(Box::new(Draft { + workspace: daemon.paths.bot_workspace(&crew.slug, &bot.slug), + message, + crew_name: crew.name, + sender_handle: sender, + task, + routine, + }))) +} + +/// A delivery about to go, read from the store. +struct Draft { + message: Message, + crew_name: String, + sender_handle: Option, + task: Option, + workspace: PathBuf, + routine: Option<(String, String, String)>, +} + +impl Draft { + /// The line for the bot's stdin; reads the images it inlines. + fn render(&self, now: i64) -> Rendered { + let context = Context { + crew_name: &self.crew_name, + sender_handle: self.sender_handle.as_deref(), + task: self.task.as_ref(), + workspace: &self.workspace, + routine: self + .routine + .as_ref() + .map(|(name, scheduled, zone)| (name.as_str(), scheduled.as_str(), zone.as_str())), + }; + render(&self.message, &context, now) + } } /// A routine's name, the local time the run is for and the zone. diff --git a/crates/botloftd/src/service/messages.rs b/crates/botloftd/src/service/messages.rs index dd158143..138630ea 100644 --- a/crates/botloftd/src/service/messages.rs +++ b/crates/botloftd/src/service/messages.rs @@ -26,12 +26,16 @@ pub fn send(daemon: &Daemon, params: MessagesSendParams) -> ApiResult { // Files alone are a message too. String::new() }; - let store = daemon.store(); - let (crew, bot) = bots::active(&store, ¶ms.bot_id)?; + let (crew, bot) = bots::active(&daemon.store(), ¶ms.bot_id)?; let now = daemon.clock.now_ms(); + // Decoding and writing the files takes a while: the store waits for + // nobody meanwhile. let workspace = daemon.paths.bot_workspace(&crew.slug, &bot.slug); let attachments = attachments::save(&workspace, uploads, daemon.bots.attachment_max_bytes, now)?; + let store = daemon.store(); + // Archived or deleted while the files were saved. + let (crew, bot) = bots::active(&store, ¶ms.bot_id)?; let message = Message { id: MessageId::generate(), crew_id: crew.id,