Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
29 changes: 29 additions & 0 deletions crates/botloft-store/migrations/0012_perf_indexes.sql
Original file line number Diff line number Diff line change
@@ -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;
4 changes: 2 additions & 2 deletions crates/botloft-store/src/approvals.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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"
Expand Down
24 changes: 12 additions & 12 deletions crates/botloft-store/src/bots.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand Down Expand Up @@ -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"
Expand Down Expand Up @@ -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],
)?;
Expand Down
49 changes: 24 additions & 25 deletions crates/botloft-store/src/chat.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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";

Expand Down Expand Up @@ -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![
Expand All @@ -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>> {
Expand All @@ -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)) \
Expand All @@ -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 \
Expand All @@ -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"
))?;
Expand Down
2 changes: 1 addition & 1 deletion crates/botloft-store/src/crews.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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)?;
Expand Down
2 changes: 1 addition & 1 deletion crates/botloft-store/src/delete.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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<_>>()?
};
Expand Down
74 changes: 37 additions & 37 deletions crates/botloft-store/src/deliveries.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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";
Expand Down Expand Up @@ -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)"
),
Expand Down Expand Up @@ -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"
Expand All @@ -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 \
Expand All @@ -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
Expand All @@ -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 {
Expand Down Expand Up @@ -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"
Expand Down Expand Up @@ -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}"
))?;
Expand All @@ -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 {
Expand Down
Loading
Loading