Merge pull request #74 from httpsphl/perf/store

perf(store): stop waiting for the disk and scanning whole tables

Author: httpsphl

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;

crates/botloft-store/src/approvals.rs

@@ -99,15 +99,15 @@ impl Store {
     pub fn pending_approvals(&self, bot: &BotId) -> Result<Vec<ApprovalId>> {
         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::<rusqlite::Result<_>>()?)
     }
 
     /// 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<Vec<ApprovalRecord>> {
-        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"

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<Vec<BotRecord>> {
-        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<Option<String>> {
-        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],
         )?;

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<Option<ChatItem>> {
-        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<Option<ChatItem>> {
@@ -100,7 +100,7 @@ impl Store {
         before: Option<&ChatItemId>,
         limit: u32,
     ) -> Result<Vec<ChatItem>> {
-        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<Option<ChatItem>> {
-        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<Vec<String>> {
-        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<Option<Activity>> {
-        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"
         ))?;

crates/botloft-store/src/crews.rs

@@ -64,7 +64,7 @@ impl Store {
 
     /// Crews in creation order.
     pub fn crews(&self, include_archived: bool) -> Result<Vec<Crew>> {
-        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)?;

crates/botloft-store/src/delete.rs

@@ -58,7 +58,7 @@ impl Store {
     pub fn delete_crew(&self, id: &CrewId) -> Result<bool> {
         let tx = self.conn.unchecked_transaction()?;
         let bots: Vec<String> = {
-            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::<rusqlite::Result<_>>()?
         };

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<DeliveryState>,
         bot: Option<&BotId>,
     ) -> Result<Vec<Delivery>> {
-        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<Vec<Delivery>> {
-        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<Option<Delivery>> {
-        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<Option<Delivery>> {
         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<Option<Delivery>> {
-        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<Vec<Delivery>> {
-        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<Vec<Delivery>> {
-        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<DeliveryBacklog> {
         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 {