From 60be5e3c5d4aa42cf5920d1d372f83662b8bf965 Mon Sep 17 00:00:00 2001 From: Phelipe Lorran <112786693+httpsphl@users.noreply.github.com> Date: Wed, 30 Sep 2026 19:46:27 -0300 Subject: [PATCH] perf(store): stop waiting for the disk and scanning whole tables - synchronous=NORMAL: in WAL mode it never corrupts the database, and a commit no longer waits for FlushFileBuffers. With FULL every chat item, read receipt and delivery paid a flush while holding the store lock. - temp_store=MEMORY, a 16 MB page cache, a WAL size limit and a bounded PRAGMA optimize on open. - Statement cache of 128; statements run on every chat item, delivery, message and session lookup go through it instead of being parsed and planned each time. - Migration 0012 indexes every foreign key that had none (deleting a bot or crew scanned the child tables once per row), the tool lookup that each tool result and permission prompt runs, the files a bot wrote, a routine's runs in rowid order and the routines that can come due. - delivery_backlog counts through the state index; the routine run queries use EXISTS so they walk only the open runs. - A test checks the query plans of the hot queries and the foreign keys. --- .../migrations/0012_perf_indexes.sql | 29 +++++ crates/botloft-store/src/approvals.rs | 4 +- crates/botloft-store/src/bots.rs | 24 ++--- crates/botloft-store/src/chat.rs | 49 +++++---- crates/botloft-store/src/crews.rs | 2 +- crates/botloft-store/src/delete.rs | 2 +- crates/botloft-store/src/deliveries.rs | 74 ++++++------- crates/botloft-store/src/lib.rs | 35 ++++++ crates/botloft-store/src/messages.rs | 31 +++--- crates/botloft-store/src/migrate.rs | 1 + crates/botloft-store/src/plans.rs | 100 ++++++++++++++++++ crates/botloft-store/src/routine_runs.rs | 14 +-- crates/botloft-store/src/routines.rs | 6 +- crates/botloft-store/src/tasks.rs | 4 +- crates/botloft-store/src/usage.rs | 2 +- docs/spec.md | 2 +- 16 files changed, 272 insertions(+), 107 deletions(-) create mode 100644 crates/botloft-store/migrations/0012_perf_indexes.sql create mode 100644 crates/botloft-store/src/plans.rs diff --git a/crates/botloft-store/migrations/0012_perf_indexes.sql b/crates/botloft-store/migrations/0012_perf_indexes.sql new file mode 100644 index 00000000..ff836a4f --- /dev/null +++ b/crates/botloft-store/migrations/0012_perf_indexes.sql @@ -0,0 +1,29 @@ +-- Indexes for lookups that scanned whole tables. Rows are never deleted, so +-- each of those scans grew with every message, delivery and tool call. + +-- Foreign keys: deleting a parent row looks for its children (spec 7.6). +CREATE INDEX messages_task ON messages (task_id); +CREATE INDEX messages_routine ON messages (routine_id); +CREATE INDEX deliveries_message ON deliveries (message_id); +CREATE INDEX tasks_requester_status ON tasks (requester_bot_id, status); +CREATE INDEX tasks_crew ON tasks (crew_id); +CREATE INDEX tasks_origin ON tasks (origin_task_id); +CREATE INDEX approvals_bot ON approvals (bot_id); +CREATE INDEX approvals_chat_item ON approvals (chat_item_id); + +-- A tool result finds its call, and the files panel the files a bot wrote. +-- The expressions must match the queries in chat.rs word for word. +CREATE INDEX chat_items_tool_use ON chat_items (bot_id, json_extract(data, '$.toolUseId')) + WHERE kind = 'tool'; +CREATE INDEX chat_items_tool_file ON chat_items (bot_id, json_extract(data, '$.file')) + WHERE kind = 'tool' AND json_extract(data, '$.file') IS NOT NULL; + +-- A routine's runs are read newest first by rowid; `id` in the key forced a +-- sort. Open runs get their own index. +DROP INDEX routine_runs_routine; +CREATE INDEX routine_runs_routine ON routine_runs (routine_id); +CREATE INDEX routine_runs_queued ON routine_runs (routine_id) WHERE status = 'queued'; + +-- Only enabled, active routines ever come due. +DROP INDEX routines_due; +CREATE INDEX routines_due ON routines (next_run_at) WHERE enabled AND archived_at IS NULL; diff --git a/crates/botloft-store/src/approvals.rs b/crates/botloft-store/src/approvals.rs index b0c150c8..b7e69d14 100644 --- a/crates/botloft-store/src/approvals.rs +++ b/crates/botloft-store/src/approvals.rs @@ -99,7 +99,7 @@ impl Store { pub fn pending_approvals(&self, bot: &BotId) -> Result> { let mut stmt = self .conn - .prepare("SELECT id FROM approvals WHERE bot_id = ?1 AND status = 'pending'")?; + .prepare_cached("SELECT id FROM approvals WHERE bot_id = ?1 AND status = 'pending'")?; let rows = stmt.query_map([bot.as_str()], |row| parse_column(row, 0))?; Ok(rows.collect::>()?) } @@ -107,7 +107,7 @@ impl Store { /// Expires every pending approval, of one bot or of all: their process /// is gone, so nobody waits for the answer. pub fn expire_approvals(&self, bot: Option<&BotId>, now: i64) -> Result> { - let mut stmt = self.conn.prepare(&format!( + let mut stmt = self.conn.prepare_cached(&format!( "UPDATE approvals SET status = 'expired', answered_at = ?2 \ WHERE status = 'pending' AND (?1 IS NULL OR bot_id = ?1) \ RETURNING {COLUMNS}, tool_use_id" diff --git a/crates/botloft-store/src/bots.rs b/crates/botloft-store/src/bots.rs index 53ffb9f3..e63a62ef 100644 --- a/crates/botloft-store/src/bots.rs +++ b/crates/botloft-store/src/bots.rs @@ -4,7 +4,7 @@ use botloft_core::ids::{BotId, CrewId}; use botloft_core::protocol::{BotEffort, BotModel, ModelEffort, PermissionMode}; use rusqlite::{OptionalExtension, Row, params}; -use crate::{Result, Store, parse_column, unique_as_duplicate}; +use crate::{Result, Store, cached_execute, cached_row, parse_column, unique_as_duplicate}; /// A bot as stored. The protocol `Bot` adds runtime state and the workspace /// path, which the daemon derives. @@ -103,7 +103,7 @@ impl Store { /// Bots in creation order, optionally limited to one crew. pub fn bots(&self, crew: Option<&CrewId>, include_archived: bool) -> Result> { - let mut stmt = self.conn.prepare(&format!( + let mut stmt = self.conn.prepare_cached(&format!( "SELECT {COLUMNS} FROM bots \ WHERE (?1 IS NULL OR crew_id = ?1) AND (?2 OR archived_at IS NULL) \ ORDER BY created_at, id" @@ -192,20 +192,20 @@ impl Store { /// The Claude Code conversation the bot resumes (spec 7.3). pub fn session_id(&self, id: &BotId) -> Result> { - Ok(self - .conn - .query_row( - "SELECT session_id FROM bots WHERE id = ?1", - [id.as_str()], - |row| row.get(0), - ) - .optional()? - .flatten()) + Ok(cached_row( + &self.conn, + "SELECT session_id FROM bots WHERE id = ?1", + [id.as_str()], + |row| row.get(0), + ) + .optional()? + .flatten()) } /// `None` makes the next start a new conversation. pub fn set_session_id(&self, id: &BotId, session: Option<&str>) -> Result<()> { - self.conn.execute( + cached_execute( + &self.conn, "UPDATE bots SET session_id = ?2 WHERE id = ?1", params![id.as_str(), session], )?; diff --git a/crates/botloft-store/src/chat.rs b/crates/botloft-store/src/chat.rs index 15143bd8..67c767d0 100644 --- a/crates/botloft-store/src/chat.rs +++ b/crates/botloft-store/src/chat.rs @@ -7,7 +7,7 @@ use botloft_core::protocol::{Activity, ChatBody, ChatItem}; use rusqlite::types::Type; use rusqlite::{Connection, OptionalExtension, Row, params}; -use crate::{Result, Store, parse_column}; +use crate::{Result, Store, cached_execute, cached_row, parse_column}; const COLUMNS: &str = "id, bot_id, data, created_at, updated_at"; @@ -47,7 +47,8 @@ impl Store { } pub(crate) fn insert_chat_item_in(conn: &Connection, item: &ChatItem) -> Result<()> { - conn.execute( + cached_execute( + conn, "INSERT INTO chat_items (id, bot_id, kind, data, created_at, updated_at) \ VALUES (?1, ?2, ?3, ?4, ?5, ?6)", params![ @@ -69,17 +70,16 @@ impl Store { body: &ChatBody, now: i64, ) -> Result> { - Ok(self - .conn - .query_row( - &format!( - "UPDATE chat_items SET kind = ?2, data = ?3, updated_at = ?4 \ + Ok(cached_row( + &self.conn, + &format!( + "UPDATE chat_items SET kind = ?2, data = ?3, updated_at = ?4 \ WHERE id = ?1 RETURNING {COLUMNS}" - ), - params![id.as_str(), kind(body), json(body)?, now], - from_row, - ) - .optional()?) + ), + params![id.as_str(), kind(body), json(body)?, now], + from_row, + ) + .optional()?) } pub fn chat_item(&self, id: &ChatItemId) -> Result> { @@ -100,7 +100,7 @@ impl Store { before: Option<&ChatItemId>, limit: u32, ) -> Result> { - let mut stmt = self.conn.prepare(&format!( + let mut stmt = self.conn.prepare_cached(&format!( "SELECT {COLUMNS} FROM chat_items \ WHERE bot_id = ?1 \ AND (?2 IS NULL OR rowid < (SELECT rowid FROM chat_items WHERE id = ?2)) \ @@ -116,25 +116,24 @@ impl Store { /// The bot's newest tool item for `tool_use_id`, to update it when the /// tool returns. pub fn tool_item(&self, bot: &BotId, tool_use_id: &str) -> Result> { - Ok(self - .conn - .query_row( - &format!( - "SELECT {COLUMNS} FROM chat_items \ + Ok(cached_row( + &self.conn, + &format!( + "SELECT {COLUMNS} FROM chat_items \ WHERE bot_id = ?1 AND kind = 'tool' \ AND json_extract(data, '$.toolUseId') = ?2 \ ORDER BY rowid DESC LIMIT 1" - ), - params![bot.as_str(), tool_use_id], - from_row, - ) - .optional()?) + ), + params![bot.as_str(), tool_use_id], + from_row, + ) + .optional()?) } /// Paths of the files the bot's `Write`/`Edit` calls changed, newest /// first and without repeats; failed calls do not count. pub fn written_files(&self, bot: &BotId, limit: u32) -> Result> { - let mut stmt = self.conn.prepare( + let mut stmt = self.conn.prepare_cached( "SELECT json_extract(data, '$.file') FROM chat_items \ WHERE bot_id = ?1 AND kind = 'tool' \ AND json_extract(data, '$.file') IS NOT NULL \ @@ -148,7 +147,7 @@ impl Store { /// The conversation-list line: the newest item that has one. pub fn last_activity(&self, bot: &BotId) -> Result> { - let mut stmt = self.conn.prepare(&format!( + let mut stmt = self.conn.prepare_cached(&format!( "SELECT {COLUMNS} FROM chat_items WHERE bot_id = ?1 AND kind != 'turn' \ ORDER BY rowid DESC LIMIT 1" ))?; diff --git a/crates/botloft-store/src/crews.rs b/crates/botloft-store/src/crews.rs index 8ad96163..10954d04 100644 --- a/crates/botloft-store/src/crews.rs +++ b/crates/botloft-store/src/crews.rs @@ -64,7 +64,7 @@ impl Store { /// Crews in creation order. pub fn crews(&self, include_archived: bool) -> Result> { - let mut stmt = self.conn.prepare(&format!( + let mut stmt = self.conn.prepare_cached(&format!( "SELECT {COLUMNS} FROM crews WHERE ?1 OR archived_at IS NULL ORDER BY created_at, id" ))?; let rows = stmt.query_map([include_archived], from_row)?; diff --git a/crates/botloft-store/src/delete.rs b/crates/botloft-store/src/delete.rs index 1d586340..a6d21b5a 100644 --- a/crates/botloft-store/src/delete.rs +++ b/crates/botloft-store/src/delete.rs @@ -58,7 +58,7 @@ impl Store { pub fn delete_crew(&self, id: &CrewId) -> Result { let tx = self.conn.unchecked_transaction()?; let bots: Vec = { - let mut stmt = tx.prepare("SELECT id FROM bots WHERE crew_id = ?1")?; + let mut stmt = tx.prepare_cached("SELECT id FROM bots WHERE crew_id = ?1")?; let rows = stmt.query_map([id.as_str()], |row| row.get(0))?; rows.collect::>()? }; diff --git a/crates/botloft-store/src/deliveries.rs b/crates/botloft-store/src/deliveries.rs index 513b837e..aa80db25 100644 --- a/crates/botloft-store/src/deliveries.rs +++ b/crates/botloft-store/src/deliveries.rs @@ -6,7 +6,7 @@ use botloft_core::ids::{BotId, DeliveryId}; use botloft_core::protocol::{Delivery, DeliveryBacklog, DeliveryState}; use rusqlite::{Connection, OptionalExtension, Row, params}; -use crate::{Result, Store, parse_column, to_sql_int}; +use crate::{Result, Store, cached_execute, cached_row, parse_column, to_sql_int}; const COLUMNS: &str = "id, message_id, bot_id, state, attempts, next_attempt_at, last_error, read_at, updated_at"; @@ -46,7 +46,8 @@ pub enum DeliveryOutcome<'a> { impl Store { pub(crate) fn insert_delivery_in(conn: &Connection, delivery: &Delivery) -> Result<()> { - conn.execute( + cached_execute( + conn, &format!( "INSERT INTO deliveries ({COLUMNS}) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9)" ), @@ -82,7 +83,7 @@ impl Store { state: Option, bot: Option<&BotId>, ) -> Result> { - let mut stmt = self.conn.prepare(&format!( + let mut stmt = self.conn.prepare_cached(&format!( "SELECT {COLUMNS} FROM deliveries \ WHERE (?1 IS NULL OR state = ?1) AND (?2 IS NULL OR bot_id = ?2) \ ORDER BY updated_at DESC, rowid DESC LIMIT ?3" @@ -102,7 +103,7 @@ impl Store { /// due and nothing else of that bot is being sent. Keeps each bot's /// messages in order, one at a time. pub fn due_deliveries(&self, now: i64) -> Result> { - let mut stmt = self.conn.prepare(&format!( + let mut stmt = self.conn.prepare_cached(&format!( "SELECT {COLUMNS} FROM deliveries AS d \ WHERE d.state = 'pending' AND d.next_attempt_at <= ?1 \ AND d.rowid = (SELECT MIN(rowid) FROM deliveries \ @@ -123,17 +124,16 @@ impl Store { now: i64, lease_until: i64, ) -> Result> { - Ok(self - .conn - .query_row( - &format!( - "UPDATE deliveries SET state = 'sending', lease_until = ?2, updated_at = ?3 \ + Ok(cached_row( + &self.conn, + &format!( + "UPDATE deliveries SET state = 'sending', lease_until = ?2, updated_at = ?3 \ WHERE id = ?1 AND state = 'pending' RETURNING {COLUMNS}" - ), - params![id.as_str(), lease_until, now], - from_row, - ) - .optional()?) + ), + params![id.as_str(), lease_until, now], + from_row, + ) + .optional()?) } /// Records how a pending or sending delivery ended. `None` if it was in @@ -145,16 +145,16 @@ impl Store { now: i64, ) -> Result> { let update = |set: &str, values: &[&dyn rusqlite::ToSql]| { - self.conn - .query_row( - &format!( - "UPDATE deliveries SET {set}, lease_until = NULL \ + cached_row( + &self.conn, + &format!( + "UPDATE deliveries SET {set}, lease_until = NULL \ WHERE id = ?1 AND state IN ('pending', 'sending') RETURNING {COLUMNS}" - ), - values, - from_row, - ) - .optional() + ), + values, + from_row, + ) + .optional() }; let id = id.as_str(); let row = match outcome { @@ -197,24 +197,23 @@ impl Store { /// The bot began the turn for the delivery sent with `turn_uuid` /// (spec 9.1). `None` if no unread delivery went with that uuid. pub fn mark_read(&self, turn_uuid: &str, now: i64) -> Result> { - Ok(self - .conn - .query_row( - &format!( - "UPDATE deliveries SET read_at = ?2, updated_at = ?2 \ + Ok(cached_row( + &self.conn, + &format!( + "UPDATE deliveries SET read_at = ?2, updated_at = ?2 \ WHERE turn_uuid = ?1 AND state = 'sent' AND read_at IS NULL \ RETURNING {COLUMNS}" - ), - params![turn_uuid, now], - from_row, - ) - .optional()?) + ), + params![turn_uuid, now], + from_row, + ) + .optional()?) } /// Deliveries written to the process of `generation` that it never /// began, oldest first. pub fn unread_deliveries(&self, bot: &BotId, generation: u64) -> Result> { - let mut stmt = self.conn.prepare(&format!( + let mut stmt = self.conn.prepare_cached(&format!( "SELECT {COLUMNS} FROM deliveries \ WHERE bot_id = ?1 AND sent_generation = ?2 AND state = 'sent' AND read_at IS NULL \ ORDER BY rowid" @@ -253,7 +252,7 @@ impl Store { /// Returns deliveries whose sender died mid-send to `pending`. pub fn recover_leases(&self, now: i64) -> Result> { - let mut stmt = self.conn.prepare(&format!( + let mut stmt = self.conn.prepare_cached(&format!( "UPDATE deliveries SET state = 'pending', lease_until = NULL, updated_at = ?1 \ WHERE state = 'sending' AND lease_until <= ?1 RETURNING {COLUMNS}" ))?; @@ -280,8 +279,9 @@ impl Store { pub fn delivery_backlog(&self) -> Result { Ok(self.conn.query_row( - "SELECT COALESCE(SUM(state IN ('pending', 'sending')), 0), \ - COALESCE(SUM(state = 'dead'), 0) FROM deliveries", + // Counted through the state index, not by reading every row. + "SELECT (SELECT COUNT(*) FROM deliveries WHERE state IN ('pending', 'sending')), \ + (SELECT COUNT(*) FROM deliveries WHERE state = 'dead')", [], |row| { Ok(DeliveryBacklog { diff --git a/crates/botloft-store/src/lib.rs b/crates/botloft-store/src/lib.rs index d5bca92e..ecfab963 100644 --- a/crates/botloft-store/src/lib.rs +++ b/crates/botloft-store/src/lib.rs @@ -13,6 +13,8 @@ mod delete; mod deliveries; mod messages; mod migrate; +#[cfg(test)] +mod plans; mod routine_runs; mod routines; mod tasks; @@ -61,9 +63,18 @@ impl Store { fn init(mut conn: Connection) -> Result { conn.pragma_update(None, "journal_mode", "WAL")?; + // In WAL mode NORMAL still never corrupts the database; a power cut + // can lose the last commits, but no commit waits for the disk. + conn.pragma_update(None, "synchronous", "NORMAL")?; conn.pragma_update(None, "foreign_keys", "ON")?; + conn.pragma_update(None, "temp_store", "MEMORY")?; + conn.pragma_update(None, "cache_size", -16_000)?; + conn.pragma_update(None, "journal_size_limit", 64 << 20)?; conn.busy_timeout(Duration::from_millis(5000))?; + conn.set_prepared_statement_cache_capacity(128); migrate::run(&mut conn)?; + // Statistics for the planner, bounded so opening stays fast. + conn.execute_batch("PRAGMA optimize = 0x10002")?; Ok(Self { conn }) } @@ -86,6 +97,25 @@ where .map_err(|err| rusqlite::Error::FromSqlConversionFailure(idx, Type::Text, Box::new(err))) } +/// `query_row` through the statement cache, for statements that run on every +/// chat item, delivery or message. +fn cached_row(conn: &Connection, sql: &str, params: P, f: F) -> rusqlite::Result +where + P: rusqlite::Params, + F: FnOnce(&Row<'_>) -> rusqlite::Result, +{ + conn.prepare_cached(sql)?.query_row(params, f) +} + +/// `execute` through the statement cache; see [`cached_row`]. +fn cached_execute( + conn: &Connection, + sql: &str, + params: P, +) -> rusqlite::Result { + conn.prepare_cached(sql)?.execute(params) +} + /// Generations and sizes are `u64` in the protocol; SQLite stores `i64`. /// Neither comes near `i64::MAX`. fn to_sql_int(value: u64) -> i64 { @@ -205,8 +235,13 @@ pub(crate) mod tests { .conn .pragma_query_value(None, "foreign_keys", |row| row.get(0)) .expect("foreign_keys"); + let synchronous: i64 = store + .conn + .pragma_query_value(None, "synchronous", |row| row.get(0)) + .expect("synchronous"); assert_eq!(mode, "wal"); assert_eq!(fk, 1); + assert_eq!(synchronous, 1, "NORMAL"); assert_eq!(store.schema_version().expect("version"), LATEST_VERSION); } diff --git a/crates/botloft-store/src/messages.rs b/crates/botloft-store/src/messages.rs index 317ffc08..7cab2964 100644 --- a/crates/botloft-store/src/messages.rs +++ b/crates/botloft-store/src/messages.rs @@ -8,7 +8,7 @@ use botloft_core::protocol::{ }; use rusqlite::{Connection, OptionalExtension, Row, params}; -use crate::{Result, Store, parse_column, to_sql_int}; +use crate::{Result, Store, cached_execute, cached_row, parse_column, to_sql_int}; const COLUMNS: &str = "id, crew_id, from_kind, from_bot_id, to_bot_id, kind, body, task_id, created_at, routine_id"; @@ -41,7 +41,7 @@ fn attachment_from_row(row: &Row<'_>) -> rusqlite::Result { /// Fills in each message's attachments. fn with_attachments(conn: &Connection, mut messages: Vec) -> Result> { - let mut stmt = conn.prepare( + let mut stmt = conn.prepare_cached( "SELECT id, name, media_type, size, path FROM attachments \ WHERE message_id = ?1 ORDER BY rowid", )?; @@ -100,7 +100,8 @@ impl Store { message: &Message, delivery: &Delivery, ) -> Result { - conn.execute( + cached_execute( + conn, &format!( "INSERT INTO messages ({COLUMNS}) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10)" ), @@ -118,7 +119,8 @@ impl Store { ], )?; for attachment in &message.attachments { - conn.execute( + cached_execute( + conn, "INSERT INTO attachments (id, message_id, name, media_type, size, path, created_at) \ VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7)", params![ @@ -148,17 +150,16 @@ impl Store { /// The message that asked for `task`. pub fn task_request(&self, task: &TaskId) -> Result> { - Ok(self - .conn - .query_row( - &format!( - "SELECT {COLUMNS} FROM messages WHERE task_id = ?1 AND kind = 'task' \ + Ok(cached_row( + &self.conn, + &format!( + "SELECT {COLUMNS} FROM messages WHERE task_id = ?1 AND kind = 'task' \ ORDER BY rowid LIMIT 1" - ), - [task.as_str()], - from_row, - ) - .optional()?) + ), + [task.as_str()], + from_row, + ) + .optional()?) } pub fn message(&self, id: &MessageId) -> Result> { @@ -189,7 +190,7 @@ impl Store { /// Newest first, in the order they were stored. pub fn messages(&self, filter: MessageFilter<'_>) -> Result> { - let mut stmt = self.conn.prepare(&format!( + let mut stmt = self.conn.prepare_cached(&format!( "SELECT {COLUMNS} FROM messages \ WHERE (?1 IS NULL OR crew_id = ?1) \ AND (?2 IS NULL OR to_bot_id = ?2 OR from_bot_id = ?2) \ diff --git a/crates/botloft-store/src/migrate.rs b/crates/botloft-store/src/migrate.rs index 73cb0cda..3b74786b 100644 --- a/crates/botloft-store/src/migrate.rs +++ b/crates/botloft-store/src/migrate.rs @@ -18,6 +18,7 @@ const MIGRATIONS: &[(u32, &str)] = &[ (9, include_str!("../migrations/0009_browser_sites.sql")), (10, include_str!("../migrations/0010_bot_effort.sql")), (11, include_str!("../migrations/0011_turn_tokens.sql")), + (12, include_str!("../migrations/0012_perf_indexes.sql")), ]; /// Schema version after every migration has run. diff --git a/crates/botloft-store/src/plans.rs b/crates/botloft-store/src/plans.rs new file mode 100644 index 00000000..2f13c453 --- /dev/null +++ b/crates/botloft-store/src/plans.rs @@ -0,0 +1,100 @@ +//! The hot queries use an index instead of reading whole tables, which +//! never shrink. Each statement repeats the WHERE and ORDER BY of the query +//! it stands for; keep them in step. + +use rusqlite::Connection; + +use crate::Store; + +/// SQLite's plan for `sql`, one step per line. +fn plan(conn: &Connection, sql: &str) -> String { + let mut stmt = conn + .prepare(&format!("EXPLAIN QUERY PLAN {sql}")) + .expect("explain"); + // Parameters stay unbound: the plan does not depend on their values. + let mut rows = stmt.raw_query(); + let mut steps = Vec::new(); + while let Some(row) = rows.next().expect("plan") { + steps.push(row.get::<_, String>(3).expect("step")); + } + steps.join("\n") +} + +fn assert_indexed(conn: &Connection, sql: &str) { + let plan = plan(conn, sql); + let scans = plan + .lines() + .filter(|step| { + step.starts_with("SCAN") && !step.contains("USING") && step != &"SCAN CONSTANT ROW" + }) + .count(); + assert_eq!(scans, 0, "full scan in:\n{sql}\n{plan}"); + assert!(!plan.contains("TEMP B-TREE"), "sort in:\n{sql}\n{plan}"); +} + +#[test] +fn hot_queries_use_an_index() { + let store = Store::open_in_memory().expect("store"); + let conn = &store.conn; + // chat.rs: tool_item, written_files. + assert_indexed( + conn, + "SELECT id FROM chat_items WHERE bot_id = ?1 AND kind = 'tool' \ + AND json_extract(data, '$.toolUseId') = ?2 ORDER BY rowid DESC LIMIT 1", + ); + let written = plan( + conn, + "SELECT json_extract(data, '$.file') FROM chat_items \ + WHERE bot_id = ?1 AND kind = 'tool' \ + AND json_extract(data, '$.file') IS NOT NULL \ + AND json_extract(data, '$.status') != 'failed' \ + GROUP BY json_extract(data, '$.file') ORDER BY MAX(rowid) DESC LIMIT ?2", + ); + assert!(written.contains("chat_items_tool_file"), "{written}"); + // messages.rs: task_request. + assert_indexed( + conn, + "SELECT id FROM messages WHERE task_id = ?1 AND kind = 'task' ORDER BY rowid LIMIT 1", + ); + // deliveries.rs: delivery_backlog. + assert_indexed( + conn, + "SELECT (SELECT COUNT(*) FROM deliveries WHERE state IN ('pending', 'sending')), \ + (SELECT COUNT(*) FROM deliveries WHERE state = 'dead')", + ); + // routines.rs: due_routines; routine_runs.rs: runs, runs_with_dead_delivery. + // Only the ties among routines already due get sorted. + let due = plan( + conn, + "SELECT id FROM routines WHERE enabled AND archived_at IS NULL AND next_run_at <= ?1 \ + ORDER BY next_run_at, id", + ); + assert!(due.contains("USING INDEX routines_due"), "{due}"); + assert_indexed( + conn, + "SELECT id FROM routine_runs WHERE routine_id = ?1 ORDER BY rowid DESC LIMIT ?2", + ); + assert_indexed( + conn, + "SELECT r.id FROM routine_runs r WHERE r.status = 'queued' AND EXISTS \ + (SELECT 1 FROM deliveries d WHERE d.message_id = r.message_id AND d.state = 'dead')", + ); +} + +#[test] +fn deleting_a_parent_finds_its_children_by_index() { + let store = Store::open_in_memory().expect("store"); + let conn = &store.conn; + for sql in [ + "SELECT 1 FROM messages WHERE task_id = ?1", + "SELECT 1 FROM messages WHERE routine_id = ?1", + "SELECT 1 FROM deliveries WHERE message_id = ?1", + "SELECT 1 FROM tasks WHERE requester_bot_id = ?1", + "SELECT 1 FROM tasks WHERE crew_id = ?1", + "SELECT 1 FROM tasks WHERE origin_task_id = ?1", + "SELECT 1 FROM approvals WHERE bot_id = ?1", + "SELECT 1 FROM approvals WHERE chat_item_id = ?1", + ] { + assert_indexed(conn, sql); + } +} diff --git a/crates/botloft-store/src/routine_runs.rs b/crates/botloft-store/src/routine_runs.rs index 3dcb3d36..60791c15 100644 --- a/crates/botloft-store/src/routine_runs.rs +++ b/crates/botloft-store/src/routine_runs.rs @@ -114,10 +114,10 @@ impl Store { .map(|column| format!("r.{column}")) .collect::>() .join(", "); - let mut stmt = self.conn.prepare(&format!( + let mut stmt = self.conn.prepare_cached(&format!( "SELECT {columns} FROM routine_runs r \ - JOIN deliveries d ON d.message_id = r.message_id \ - WHERE r.status = 'queued' AND d.state = 'dead'" + WHERE r.status = 'queued' AND EXISTS (SELECT 1 FROM deliveries d \ + WHERE d.message_id = r.message_id AND d.state = 'dead')" ))?; let rows = stmt.query_map([], run_from_row)?; Ok(rows.collect::>()?) @@ -131,10 +131,10 @@ impl Store { .map(|column| format!("r.{column}")) .collect::>() .join(", "); - let mut stmt = self.conn.prepare(&format!( + let mut stmt = self.conn.prepare_cached(&format!( "SELECT {columns} FROM routine_runs r \ - JOIN deliveries d ON d.message_id = r.message_id \ - WHERE r.status = 'queued' AND d.read_at IS NOT NULL" + WHERE r.status = 'queued' AND EXISTS (SELECT 1 FROM deliveries d \ + WHERE d.message_id = r.message_id AND d.read_at IS NOT NULL)" ))?; let rows = stmt.query_map([], run_from_row)?; Ok(rows.collect::>()?) @@ -147,7 +147,7 @@ impl Store { before: Option<&RoutineRunId>, limit: u32, ) -> Result> { - let mut stmt = self.conn.prepare(&format!( + let mut stmt = self.conn.prepare_cached(&format!( "SELECT {RUN_COLUMNS} FROM routine_runs \ WHERE routine_id = ?1 \ AND (?2 IS NULL OR rowid < (SELECT rowid FROM routine_runs WHERE id = ?2)) \ diff --git a/crates/botloft-store/src/routines.rs b/crates/botloft-store/src/routines.rs index 96498073..8d6fa356 100644 --- a/crates/botloft-store/src/routines.rs +++ b/crates/botloft-store/src/routines.rs @@ -99,7 +99,7 @@ impl Store { /// Routines in creation order, optionally of one bot, with their latest run. pub fn routines(&self, bot: Option<&BotId>, include_archived: bool) -> Result> { - let mut stmt = self.conn.prepare(&format!( + let mut stmt = self.conn.prepare_cached(&format!( "SELECT {COLUMNS} FROM routines \ WHERE (?1 IS NULL OR bot_id = ?1) AND (?2 OR archived_at IS NULL) \ ORDER BY created_at, id" @@ -114,7 +114,7 @@ impl Store { /// Enabled routines whose time came, soonest first. pub fn due_routines(&self, now: i64) -> Result> { - let mut stmt = self.conn.prepare(&format!( + let mut stmt = self.conn.prepare_cached(&format!( "SELECT {COLUMNS} FROM routines \ WHERE enabled AND archived_at IS NULL AND next_run_at <= ?1 \ ORDER BY next_run_at, id" @@ -134,7 +134,7 @@ impl Store { /// Archives the routines of a bot that was archived. Returns their ids. pub fn archive_routines_of(&self, bot: &BotId, at: i64) -> Result> { - let mut stmt = self.conn.prepare( + let mut stmt = self.conn.prepare_cached( "UPDATE routines SET archived_at = ?2, updated_at = ?2, next_run_at = NULL \ WHERE bot_id = ?1 AND archived_at IS NULL RETURNING id", )?; diff --git a/crates/botloft-store/src/tasks.rs b/crates/botloft-store/src/tasks.rs index fee43285..2c96af8c 100644 --- a/crates/botloft-store/src/tasks.rs +++ b/crates/botloft-store/src/tasks.rs @@ -86,7 +86,7 @@ impl Store { /// Newest first, at most 500. pub fn tasks(&self, filter: TaskFilter<'_>) -> Result> { - let mut stmt = self.conn.prepare(&format!( + let mut stmt = self.conn.prepare_cached(&format!( "SELECT {COLUMNS} FROM tasks \ WHERE (?1 IS NULL OR crew_id = ?1) \ AND (?2 IS NULL OR assignee_bot_id = ?2) \ @@ -109,7 +109,7 @@ impl Store { /// Open tasks whose deadline has passed, oldest deadline first. pub fn overdue_tasks(&self, now: i64) -> Result> { - let mut stmt = self.conn.prepare(&format!( + let mut stmt = self.conn.prepare_cached(&format!( "SELECT {COLUMNS} FROM tasks WHERE status = 'open' AND deadline_at <= ?1 \ ORDER BY deadline_at, rowid" ))?; diff --git a/crates/botloft-store/src/usage.rs b/crates/botloft-store/src/usage.rs index 1c65a93b..cc019d66 100644 --- a/crates/botloft-store/src/usage.rs +++ b/crates/botloft-store/src/usage.rs @@ -14,7 +14,7 @@ impl Store { /// Each bot's tokens in turns that ended at or after `since`, the bot /// that used the most first. Bots with no such turn are left out. pub fn turn_tokens(&self, since: i64) -> Result> { - let mut stmt = self.conn.prepare( + let mut stmt = self.conn.prepare_cached( "SELECT b.id, b.name, b.color, b.archived_at IS NOT NULL, COUNT(*), \ SUM(json_extract(c.data, '$.tokens.input')), \ SUM(json_extract(c.data, '$.tokens.cacheWrite')), \ diff --git a/docs/spec.md b/docs/spec.md index 084d4bd6..f4a83a32 100644 --- a/docs/spec.md +++ b/docs/spec.md @@ -644,7 +644,7 @@ Além desses, os códigos padrão do JSON-RPC: `-32700` (JSON inválido), `-3260 ## 12. Dados (SQLite) -Pragmas: `journal_mode=WAL`, `foreign_keys=ON`, `busy_timeout=5000`. Migrations numeradas em `botloft-store/migrations/NNNN_nome.sql`, versão em `PRAGMA user_version`. Tempo em milissegundos Unix (`INTEGER`). Arquivamento é lógico (`archived_at`). Exclusão (7.6) apaga as linhas: as chaves estrangeiras não têm `ON DELETE`, então o daemon apaga ou solta, na ordem e numa transação só, tudo que aponta para o bot (`approvals`, `chat_items`, `browser_sites`, `routine_runs`, `attachments`, `deliveries`, as `messages` recebidas, `routines`, `tasks`), e zera `messages.from_bot_id`, `messages.task_id`, `tasks.origin_task_id` e `crews.lead_bot_id` onde apontavam para o que saiu. Uma crew sai depois dos bots dela. +Pragmas: `journal_mode=WAL`, `synchronous=NORMAL`, `foreign_keys=ON`, `busy_timeout=5000`, `temp_store=MEMORY`, cache de 16 MB e `PRAGMA optimize` ao abrir. Com WAL, `NORMAL` não corrompe o banco; uma queda de energia pode perder os últimos commits, mas nenhum commit espera o disco (com `FULL`, cada item de chat pagava um flush no Windows). As consultas quentes passam pelo cache de statements do rusqlite, e toda coluna que aponta para outra tabela tem índice: as tabelas nunca encolhem, e sem índice a exclusão varre a tabela filha uma vez por linha apagada. Migrations numeradas em `botloft-store/migrations/NNNN_nome.sql`, versão em `PRAGMA user_version`. Tempo em milissegundos Unix (`INTEGER`). Arquivamento é lógico (`archived_at`). Exclusão (7.6) apaga as linhas: as chaves estrangeiras não têm `ON DELETE`, então o daemon apaga ou solta, na ordem e numa transação só, tudo que aponta para o bot (`approvals`, `chat_items`, `browser_sites`, `routine_runs`, `attachments`, `deliveries`, as `messages` recebidas, `routines`, `tasks`), e zera `messages.from_bot_id`, `messages.task_id`, `tasks.origin_task_id` e `crews.lead_bot_id` onde apontavam para o que saiu. Uma crew sai depois dos bots dela. | Tabela | Colunas principais | |---|---|