scriptoriumadmin/fabro
1
0
Fork 0
mirror of https://github.com/fabro-sh/fabro.git synced 2026-08-28 05:27:41 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 4

Add inactive SQL run storage foundation

Browse source
This commit is contained in:
Scott Werner 2026-08-27 13:32:01 -04:00
parent 039517a6a5
commit b62e458289
16 changed files with 2470 additions and 1013 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
lib/apps/fabro-server/migrations/2026082301_sqlite_blob_activation.rs
View file

@ -125,12 +125,14 @@ pub(crate) async fn activate_blob_storage(
);
let blob_store = Arc::new(fabro_store::BlobStore::new(database.clone_pool()));
let run_record_store = Arc::new(fabro_store::RunRecordStore::new(database.clone_pool()));
let store = Arc::new(fabro_store::Database::new(
object_store,
slatedb_prefix,
flush_interval,
cache_path,
Arc::clone(&blob_store),
run_record_store,
));
let inventory = store

2
lib/apps/fabro-server/src/serve.rs
View file

@ -829,7 +829,7 @@ where
.runs
.warm_projection_cache()
.await
.context("warming run projection cache and reconciling run summaries")?;
.context("warming run projection cache and reconciling run records")?;
let reconciled = reconcile_incomplete_runs_on_startup(&state).await?;
if reconciled > 0 {
info!(

9
lib/apps/fabro-server/src/server.rs
View file

@ -86,7 +86,7 @@ use fabro_slack::{blocks as slack_blocks, connection as slack_connection};
use fabro_static::EnvVars;
use fabro_store::{
ArtifactKey, ArtifactStore, AuthCodeStore, AuthSessionStore, CachedRunProjection, Database,
EventEnvelope, EventPayload, KeyedMutex, NodeArtifact, PendingInterviewRecord, RunSummaryStore,
EventEnvelope, EventPayload, KeyedMutex, NodeArtifact, PendingInterviewRecord, RunRecordStore,
StageArtifactEntry, StageId,
};
#[cfg(test)]
@ -1153,7 +1153,7 @@ pub struct AppState {
pub(crate) struct AppStores {
pub(crate) runs: Arc<Database>,
pub(crate) run_summaries: Arc<RunSummaryStore>,
pub(crate) run_records: Arc<RunRecordStore>,
pub(crate) auth_codes: Arc<AuthCodeStore>,
pub(crate) auth_sessions: Arc<AuthSessionStore>,
pub(crate) automations: Arc<AutomationStore>,
@ -2447,8 +2447,7 @@ pub(crate) fn build_app_state(config: AppStateConfig) -> anyhow::Result<Arc<AppS
})
.context("load environments")?,
);
let run_summaries =
store.attach_run_summary_store(Arc::new(RunSummaryStore::new(db_pool.clone())));
let run_records = store.run_record_store();
let auth_codes = Arc::new(AuthCodeStore::new(db_pool.clone()));
let auth_sessions = Arc::new(AuthSessionStore::new(db_pool.clone()));
let mcp_server_dir = mcp_server_dir_for_active_config(&active_config_path);
@ -2556,7 +2555,7 @@ pub(crate) fn build_app_state(config: AppStateConfig) -> anyhow::Result<Arc<AppS
aggregate_billing: Mutex::new(BillingAccumulator::default()),
stores: AppStores {
runs: store,
run_summaries,
run_records,
auth_codes,
auth_sessions,
automations: automation_store,

6
lib/apps/fabro-server/src/server/handler/runs.rs
View file

@ -335,7 +335,7 @@ async fn run_summary_at(
run_id: &RunId,
now: DateTime<Utc>,
) -> fabro_store::Result<Option<Run>> {
let Some(mut summary) = state.stores.run_summaries.get(run_id, now).await? else {
let Some(mut summary) = state.stores.run_records.get(run_id, now).await? else {
return Ok(None);
};
if summary.timestamps.completed_at.is_none() {
@ -361,7 +361,7 @@ pub(super) async fn run_summary_page_response(
state: &AppState,
query: &RunSummaryListQuery,
) -> Response {
match state.stores.run_summaries.list(query, Utc::now()).await {
match state.stores.run_records.list(query, Utc::now()).await {
Ok(page) => {
let data = state.decorate_run_summaries(page.data).await;
(
@ -415,7 +415,7 @@ async fn resolve_run(
State(state): State<Arc<AppState>>,
Query(query): Query<ResolveRunQuery>,
) -> Response {
let identities = match state.stores.run_summaries.list_identities().await {
let identities = match state.stores.run_records.list_identities().await {
Ok(identities) => identities,
Err(err) => {
return ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string())

19
lib/apps/fabro-server/src/server/tests.rs
View file

@ -210,6 +210,15 @@ methods = ["dev-token"]
)
}
#[test]
fn run_record_store_is_database_owned() {
let state = test_app_state();
assert!(Arc::ptr_eq(
&state.stores.run_records,
&state.stores.runs.run_record_store(),
));
}
async fn body_json(body: Body) -> serde_json::Value {
let bytes = to_bytes(body, usize::MAX).await.unwrap();
serde_json::from_slice(&bytes).unwrap()
@ -3915,7 +3924,7 @@ async fn post_runs_run_intent_rejects_invalid_folder_paths_before_persistence()
assert!(
state
.stores
.run_summaries
.run_records
.list_identities()
.await
.unwrap()
@ -3947,7 +3956,7 @@ async fn post_runs_run_intent_applies_the_folder_target_environment_matrix() {
assert!(
state
.stores
.run_summaries
.run_records
.list_identities()
.await
.unwrap()
@ -3982,7 +3991,7 @@ enabled = false
assert!(
disabled_state
.stores
.run_summaries
.run_records
.list_identities()
.await
.unwrap()
@ -4224,7 +4233,7 @@ async fn post_runs_run_intent_rejects_none_target_with_local_environment_before_
assert!(
state
.stores
.run_summaries
.run_records
.list_identities()
.await
.unwrap()
@ -4269,7 +4278,7 @@ async fn assert_run_intent_targets_unavailable(state: &Arc<AppState>) {
assert!(
state
.stores
.run_summaries
.run_records
.list_identities()
.await
.unwrap()

12
lib/components/fabro-store/src/error.rs
View file

@ -56,6 +56,18 @@ pub enum Error {
run_id: String,
field: &'static str,
},
#[error("run {run_id} head mismatch: expected {expected_last_seq}, stored {actual_last_seq:?}")]
RunHeadMismatch {
run_id: String,
expected_last_seq: u32,
actual_last_seq: Option<u32>,
},
#[error("stored run event {run_id} sequence {seq} has inconsistent field {field}")]
RunEventMismatch {
run_id: String,
seq: u32,
field: &'static str,
},
#[error(transparent)]
InvalidTransition(#[from] fabro_types::InvalidTransition),
#[error("{0}")]

4
lib/components/fabro-store/src/legacy_blob_import.rs
View file

@ -1128,7 +1128,7 @@ mod tests {
PASSIVE_CHECKPOINT_BYTES, set_automatic_checkpoint,
};
use crate::keys::SlateKey;
use crate::{BlobStore, Database};
use crate::{BlobStore, Database, test_support as store_test_support};
type TestResult<T> = std::result::Result<T, Box<dyn std::error::Error>>;
@ -1153,6 +1153,7 @@ mod tests {
Duration::from_millis(1),
None,
Arc::clone(&target),
store_test_support::test_run_record_store(),
);
let source_db = source.open_db().await?;
Ok(Self {
@ -1853,6 +1854,7 @@ mod tests {
Duration::from_millis(1),
None,
Arc::clone(&target),
store_test_support::test_run_record_store(),
);
let mut connection = pool.acquire().await?;

10
lib/components/fabro-store/src/lib.rs
View file

@ -9,9 +9,9 @@ mod keyed_mutex;
mod keys;
mod legacy_blob_import;
mod record;
mod run_record_store;
mod run_sessions;
mod run_state;
mod run_summary_store;
mod serializable_projection;
mod slate;
mod sqlite_row;
@ -37,15 +37,15 @@ pub use legacy_blob_import::{
LegacyBlobImportError, LegacyBlobImportReport, LegacyBlobInventory, LegacyBlobInventoryError,
LegacyBlobVerificationError, LegacyBlobVerificationReport,
};
pub use run_record_store::{
RunRecordStore, RunSummaryIdentity, RunSummaryListQuery, RunSummaryPage, RunSummarySort,
RunSummarySortDirection, RunSummaryVisibility,
};
pub use run_sessions::{
ProjectedRunSession, project_run_session, project_run_session_with_context,
project_run_sessions,
};
pub use run_state::RunProjectionReducer;
pub use run_summary_store::{
RunSummaryIdentity, RunSummaryListQuery, RunSummaryPage, RunSummarySort,
RunSummarySortDirection, RunSummaryStore, RunSummaryVisibility,
};
pub use serializable_projection::SerializableProjection;
pub use slate::{CachedRunProjection, Database, RunCatalogIndex, RunDatabase, Runs, UnreadableRun};
pub use types::EventPayload;

1978
lib/components/fabro-store/src/run_record_store.rs Normal file
View file

File diff suppressed because it is too large Load diff

932
lib/components/fabro-store/src/run_summary_store.rs
View file

@ -1,932 +0,0 @@
use std::collections::{HashMap, HashSet};
use std::fmt::Write as _;
use std::sync::LazyLock;
use chrono::{DateTime, Utc};
use fabro_types::{BilledTokenCounts, Run, RunId, RunSize, RunStatusKind, RunTiming, timing};
use sqlx::sqlite::{SqliteConnection, SqliteRow};
use sqlx::{QueryBuilder, Row as _, Sqlite, SqlitePool};
use strum::VariantArray as _;
use crate::run_state::projected_billing;
use crate::slate::CachedRunProjection;
use crate::{Error, Result};
const UPSERT_RUN_SQL: &str = r"
INSERT INTO runs (
id, source_last_seq, created_at_ms, started_at_ms, last_event_at_ms, completed_at_ms,
status, archived_at_ms, parent_id, title, workflow_slug, workflow_name,
repository_name, automation_id, diff_files_changed, diff_additions, diff_deletions,
input_tokens, output_tokens, reasoning_tokens, cache_read_tokens, cache_write_tokens,
total_usd_micros, summary_json
) VALUES (
?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?
)
ON CONFLICT(id) DO UPDATE SET
source_last_seq = excluded.source_last_seq,
created_at_ms = excluded.created_at_ms,
started_at_ms = excluded.started_at_ms,
last_event_at_ms = excluded.last_event_at_ms,
completed_at_ms = excluded.completed_at_ms,
status = excluded.status,
archived_at_ms = excluded.archived_at_ms,
parent_id = excluded.parent_id,
title = excluded.title,
workflow_slug = excluded.workflow_slug,
workflow_name = excluded.workflow_name,
repository_name = excluded.repository_name,
automation_id = excluded.automation_id,
diff_files_changed = excluded.diff_files_changed,
diff_additions = excluded.diff_additions,
diff_deletions = excluded.diff_deletions,
input_tokens = excluded.input_tokens,
output_tokens = excluded.output_tokens,
reasoning_tokens = excluded.reasoning_tokens,
cache_read_tokens = excluded.cache_read_tokens,
cache_write_tokens = excluded.cache_write_tokens,
total_usd_micros = excluded.total_usd_micros,
summary_json = excluded.summary_json
WHERE excluded.source_last_seq > runs.source_last_seq
";
const SELECT_RUN_SUMMARIES_SQL: &str = r"
SELECT runs.id, runs.summary_json,
(SELECT COUNT(*) FROM runs AS child WHERE child.parent_id = runs.id) AS children_count
FROM runs";
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, serde::Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum RunSummarySort {
#[default]
CreatedAt,
UpdatedAt,
Status,
Elapsed,
#[serde(rename = "repo")]
Repository,
Title,
Workflow,
Changes,
Size,
}
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, serde::Deserialize)]
#[serde(rename_all = "lowercase")]
pub enum RunSummarySortDirection {
Asc,
#[default]
Desc,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum RunSummaryVisibility {
All,
Default {
include_archived: bool,
},
Selected {
statuses: Vec<RunStatusKind>,
archived: bool,
},
}
impl Default for RunSummaryVisibility {
fn default() -> Self {
Self::Default {
include_archived: false,
}
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct RunSummaryListQuery {
pub parent_id: Option<RunId>,
pub automation_id: Option<String>,
pub visibility: RunSummaryVisibility,
pub sort: RunSummarySort,
pub direction: RunSummarySortDirection,
pub limit: u32,
pub offset: u32,
}
impl Default for RunSummaryListQuery {
fn default() -> Self {
Self {
parent_id: None,
automation_id: None,
visibility: RunSummaryVisibility::default(),
sort: RunSummarySort::default(),
direction: RunSummarySortDirection::default(),
limit: 100,
offset: 0,
}
}
}
#[derive(Debug, Clone, PartialEq)]
pub struct RunSummaryPage {
pub data: Vec<Run>,
pub total: u64,
pub has_more: bool,
}
#[derive(Clone)]
pub struct RunSummaryStore {
pool: SqlitePool,
}
impl std::fmt::Debug for RunSummaryStore {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("RunSummaryStore").finish_non_exhaustive()
}
}
impl RunSummaryStore {
#[must_use]
pub fn new(pool: SqlitePool) -> Self {
Self { pool }
}
pub(crate) async fn upsert_projection(&self, entry: &CachedRunProjection) -> Result<()> {
let record = ProjectedRunSummary::from_entry(entry);
let mut connection = self.pool.acquire().await?;
upsert_run(&mut connection, &record).await?;
Ok(())
}
#[cfg(test)]
pub(crate) async fn close_pool(&self) {
self.pool.close().await;
}
pub(crate) async fn reconcile(&self, entries: &[CachedRunProjection]) -> Result<()> {
let mut transaction = self.pool.begin().await?;
let stored_seqs: HashMap<String, i64> =
sqlx::query_as::<_, (String, i64)>("SELECT id, source_last_seq FROM runs")
.fetch_all(&mut *transaction)
.await?
.into_iter()
.collect();
let mut authoritative_ids = HashSet::new();
for entry in entries {
let run_id = entry.run_id.to_string();
let up_to_date = stored_seqs
.get(&run_id)
.is_some_and(|stored_seq| *stored_seq >= i64::from(entry.last_seq));
authoritative_ids.insert(run_id);
if up_to_date {
continue;
}
upsert_run(&mut transaction, &ProjectedRunSummary::from_entry(entry)).await?;
}
let stale_ids = stored_seqs
.keys()
.filter(|stored_id| !authoritative_ids.contains(stored_id.as_str()))
.collect::<Vec<_>>();
for chunk in stale_ids.chunks(500) {
let mut delete = QueryBuilder::<Sqlite>::new("DELETE FROM runs WHERE id IN (");
let mut separated = delete.separated(", ");
for stale_id in chunk {
separated.push_bind(stale_id.as_str());
}
delete.push(")");
delete.build().execute(&mut *transaction).await?;
}
transaction.commit().await?;
Ok(())
}
pub async fn get(&self, run_id: &RunId, now: DateTime<Utc>) -> Result<Option<Run>> {
let mut query = QueryBuilder::<Sqlite>::new(SELECT_RUN_SUMMARIES_SQL);
query
.push(" WHERE runs.id = ")
.push_bind(run_id.to_string());
let row = query.build().fetch_optional(&self.pool).await?;
row.map(|row| decode_run_row(&row, now)).transpose()
}
/// Identity fields for every stored run, for selector resolution without
/// decoding full summaries.
pub async fn list_identities(&self) -> Result<Vec<RunSummaryIdentity>> {
let rows = sqlx::query(
r"
SELECT id, workflow_slug,
json_extract(summary_json, '$.workflow.name') AS workflow_name,
json_extract(summary_json, '$.repository.origin_url') AS repository_origin_url
FROM runs",
)
.fetch_all(&self.pool)
.await?;
rows.iter()
.map(|row| {
let stored_id: String = row.try_get("id")?;
let id = stored_id
.parse::<RunId>()
.map_err(|_| Error::RunSummaryMismatch {
run_id: stored_id,
field: "id",
})?;
Ok(RunSummaryIdentity {
id,
workflow_slug: row.try_get("workflow_slug")?,
workflow_name: row.try_get("workflow_name")?,
repository_origin_url: row.try_get("repository_origin_url")?,
})
})
.collect()
}
pub async fn list(
&self,
query: &RunSummaryListQuery,
now: DateTime<Utc>,
) -> Result<RunSummaryPage> {
let mut transaction = self.pool.begin().await?;
let mut count_query = QueryBuilder::<Sqlite>::new("SELECT COUNT(*) FROM runs");
push_filters(&mut count_query, query);
let total: i64 = count_query
.build_query_scalar()
.fetch_one(&mut *transaction)
.await?;
let mut rows_query = QueryBuilder::<Sqlite>::new(SELECT_RUN_SUMMARIES_SQL);
push_filters(&mut rows_query, query);
push_order(&mut rows_query, query.sort, query.direction, now);
rows_query.push(" LIMIT ").push_bind(i64::from(query.limit));
rows_query
.push(" OFFSET ")
.push_bind(i64::from(query.offset));
let rows = rows_query.build().fetch_all(&mut *transaction).await?;
transaction.commit().await?;
let data = rows
.iter()
.map(|row| decode_run_row(row, now))
.collect::<Result<Vec<_>>>()?;
let total = u64::try_from(total).expect("COUNT(*) is non-negative");
let consumed = u64::from(query.offset).saturating_add(data.len() as u64);
Ok(RunSummaryPage {
data,
total,
has_more: consumed < total,
})
}
pub async fn delete(&self, run_id: &RunId) -> Result<()> {
sqlx::query("DELETE FROM runs WHERE id = ?")
.bind(run_id.to_string())
.execute(&self.pool)
.await?;
Ok(())
}
}
/// Identity fields of a stored run summary, cheap to list for selector
/// resolution.
#[derive(Debug, Clone)]
pub struct RunSummaryIdentity {
pub id: RunId,
pub workflow_slug: Option<String>,
pub workflow_name: Option<String>,
pub repository_origin_url: Option<String>,
}
#[derive(Debug)]
struct ProjectedRunSummary {
run: Run,
last_seq: u32,
workflow_name: Option<String>,
repository_name: Option<String>,
input_tokens: i64,
output_tokens: i64,
reasoning_tokens: i64,
cache_read_tokens: i64,
cache_write_tokens: i64,
total_usd_micros: Option<i64>,
}
impl ProjectedRunSummary {
fn from_entry(entry: &CachedRunProjection) -> Self {
let mut run = entry.summary.clone();
if run.timing.is_none() {
let at = run
.timestamps
.last_event_at
.unwrap_or(run.timestamps.created_at);
run.timing = entry.projection.live_run_timing(at);
}
let billing = normalize_billing_for_read_model(projected_billing(&entry.projection));
let workflow_name = run.workflow.display_name().map(str::to_string);
let repository_name = run
.repository
.as_ref()
.map(|repository| repository.name.clone());
Self {
run,
last_seq: entry.last_seq,
workflow_name,
repository_name,
input_tokens: billing.input_tokens,
output_tokens: billing.output_tokens,
reasoning_tokens: billing.reasoning_tokens,
cache_read_tokens: billing.cache_read_tokens,
cache_write_tokens: billing.cache_write_tokens,
total_usd_micros: billing.total_usd_micros,
}
}
}
/// Older provider codecs could persist a negative disjoint bucket when a
/// detail count exceeded its inclusive parent total. The SQLite summary is a
/// rebuildable, nonnegative read model, so normalize those legacy values here
/// without rewriting the authoritative run events.
fn normalize_billing_for_read_model(mut billing: BilledTokenCounts) -> BilledTokenCounts {
let input_total = billing
.input_tokens
.saturating_add(billing.cache_read_tokens)
.saturating_add(billing.cache_write_tokens)
.max(0);
billing.cache_read_tokens = billing.cache_read_tokens.clamp(0, input_total);
billing.cache_write_tokens = billing
.cache_write_tokens
.clamp(0, input_total - billing.cache_read_tokens);
billing.input_tokens = input_total - billing.cache_read_tokens - billing.cache_write_tokens;
let output_total = billing
.output_tokens
.saturating_add(billing.reasoning_tokens)
.max(0);
billing.reasoning_tokens = billing.reasoning_tokens.clamp(0, output_total);
billing.output_tokens = output_total - billing.reasoning_tokens;
billing.total_tokens = input_total.saturating_add(output_total);
billing.total_usd_micros = billing.total_usd_micros.map(|value| value.max(0));
billing
}
async fn upsert_run(connection: &mut SqliteConnection, record: &ProjectedRunSummary) -> Result<()> {
let run = &record.run;
let diff = run.diff.unwrap_or_default();
let summary_json = serde_json::to_string(run)?;
sqlx::query(UPSERT_RUN_SQL)
.bind(run.id.to_string())
.bind(i64::from(record.last_seq))
.bind(run.timestamps.created_at.timestamp_millis())
.bind(
run.timestamps
.started_at
.map(|value| value.timestamp_millis()),
)
.bind(
run.timestamps
.last_event_at
.unwrap_or(run.timestamps.created_at)
.timestamp_millis(),
)
.bind(
run.timestamps
.completed_at
.map(|value| value.timestamp_millis()),
)
.bind(run.lifecycle.status.kind().to_string())
.bind(
run.lifecycle
.archived_at
.map(|value| value.timestamp_millis()),
)
.bind(run.parent_id.map(|value| value.to_string()))
.bind(&run.title)
.bind(&run.workflow.slug)
.bind(&record.workflow_name)
.bind(&record.repository_name)
.bind(run.automation.as_ref().map(|automation| &automation.id))
.bind(diff.files_changed)
.bind(diff.additions)
.bind(diff.deletions)
.bind(record.input_tokens)
.bind(record.output_tokens)
.bind(record.reasoning_tokens)
.bind(record.cache_read_tokens)
.bind(record.cache_write_tokens)
.bind(record.total_usd_micros)
.bind(summary_json)
.execute(connection)
.await?;
Ok(())
}
fn push_filters(builder: &mut QueryBuilder<Sqlite>, query: &RunSummaryListQuery) {
builder.push(" WHERE 1 = 1");
if let Some(parent_id) = query.parent_id {
builder
.push(" AND parent_id = ")
.push_bind(parent_id.to_string());
}
if let Some(automation_id) = &query.automation_id {
builder
.push(" AND automation_id = ")
.push_bind(automation_id.clone());
}
match &query.visibility {
RunSummaryVisibility::All => {}
RunSummaryVisibility::Default { include_archived } => {
let not_removing = format!("status <> '{}'", RunStatusKind::Removing);
if *include_archived {
builder.push(format!(
" AND (archived_at_ms IS NOT NULL OR {not_removing})"
));
} else {
builder.push(format!(" AND archived_at_ms IS NULL AND {not_removing}"));
}
}
RunSummaryVisibility::Selected { statuses, archived } => {
builder.push(" AND (");
let mut has_condition = false;
if *archived {
builder.push("archived_at_ms IS NOT NULL");
has_condition = true;
}
if !statuses.is_empty() {
if has_condition {
builder.push(" OR ");
}
builder.push("(archived_at_ms IS NULL AND status IN (");
let mut separated = builder.separated(", ");
for status in statuses {
separated.push_bind(status.to_string());
}
separated.push_unseparated("))");
has_condition = true;
}
if !has_condition {
builder.push("0");
}
builder.push(")");
}
}
}
/// Status sort rank derived from [`RunStatusKind::board_rank`], so the SQL
/// order and the board column order share one source. Archived runs rank 7,
/// matching the `archived` board column.
static STATUS_RANK_CASE_SQL: LazyLock<String> = LazyLock::new(|| {
let mut case = String::from("CASE WHEN archived_at_ms IS NOT NULL THEN 7");
for kind in RunStatusKind::VARIANTS {
let _ = write!(case, " WHEN status = '{kind}' THEN {}", kind.board_rank());
}
case.push_str(" ELSE 9 END");
case
});
/// Size sort rank derived from [`RunSize::BUCKET_MAX_USD_MICROS`], so the SQL
/// order and the displayed size buckets share one source.
static SIZE_RANK_CASE_SQL: LazyLock<String> = LazyLock::new(|| {
let mut case = String::from("CASE");
for (rank, (_, max_usd_micros)) in RunSize::BUCKET_MAX_USD_MICROS.iter().enumerate() {
let _ = write!(
case,
" WHEN COALESCE(total_usd_micros, 0) <= {max_usd_micros} THEN {rank}"
);
}
let _ = write!(case, " ELSE {} END", RunSize::BUCKET_MAX_USD_MICROS.len());
case
});
fn push_order(
builder: &mut QueryBuilder<Sqlite>,
sort: RunSummarySort,
direction: RunSummarySortDirection,
now: DateTime<Utc>,
) {
builder.push(" ORDER BY ");
match sort {
RunSummarySort::CreatedAt => builder.push("created_at_ms"),
RunSummarySort::UpdatedAt => builder.push("last_event_at_ms"),
RunSummarySort::Status => builder.push(STATUS_RANK_CASE_SQL.as_str()),
RunSummarySort::Elapsed => builder
.push("(COALESCE(completed_at_ms, ")
.push_bind(now.timestamp_millis())
.push(") - COALESCE(started_at_ms, created_at_ms))"),
RunSummarySort::Repository => builder.push("COALESCE(repository_name, '') COLLATE NOCASE"),
RunSummarySort::Title => builder.push("TRIM(title) COLLATE NOCASE"),
RunSummarySort::Workflow => builder.push("COALESCE(workflow_name, '') COLLATE NOCASE"),
RunSummarySort::Changes => builder.push("(diff_additions + diff_deletions)"),
RunSummarySort::Size => builder.push(SIZE_RANK_CASE_SQL.as_str()),
};
match direction {
RunSummarySortDirection::Asc => builder.push(" ASC"),
RunSummarySortDirection::Desc => builder.push(" DESC"),
};
builder.push(", id DESC");
}
fn decode_run_row(row: &SqliteRow, now: DateTime<Utc>) -> Result<Run> {
let stored_id: String = row.try_get("id")?;
let summary_json: String = row.try_get("summary_json")?;
let children_count: i64 = row.try_get("children_count")?;
let mut run: Run = serde_json::from_str(&summary_json)?;
if stored_id != run.id.to_string() {
return Err(Error::RunSummaryMismatch {
run_id: stored_id,
field: "id",
});
}
run.children_count = u64::try_from(children_count).map_err(|_| Error::RunSummaryMismatch {
run_id: run.id.to_string(),
field: "children_count",
})?;
overlay_live_wall_time(&mut run, now);
Ok(run)
}
fn overlay_live_wall_time(run: &mut Run, now: DateTime<Utc>) {
if run.timestamps.completed_at.is_some() {
return;
}
let Some(started_at) = run.timestamps.started_at else {
return;
};
let wall_time_ms = timing::elapsed_ms(started_at, now);
run.timing = Some(
run.timing
.unwrap_or_else(|| RunTiming::wall_only(wall_time_ms))
.with_wall_time(wall_time_ms),
);
}
#[cfg(test)]
mod tests {
use std::collections::HashMap;
use chrono::{DateTime, Utc};
use fabro_types::{
AutomationRef, BilledTokenCounts, BlockedReason, Conclusion, DiffSummary, FailureReason,
Graph, PendingReason, RunDiff, RunId, RunProjection, RunSize, RunSpec, RunStatus,
RunStatusKind, RunTiming, StageOutcome, SuccessReason, WorkflowSettings, test_support,
};
use strum::VariantArray as _;
use ulid::Ulid;
use super::{
RunSummaryListQuery, RunSummarySort, RunSummarySortDirection, RunSummaryStore,
RunSummaryVisibility,
};
use crate::slate::CachedRunProjection;
use crate::test_support as store_test_support;
fn dt(value: &str) -> DateTime<Utc> {
value.parse().unwrap()
}
fn run_id(timestamp_ms: u64, random: u128) -> RunId {
RunId::from(Ulid::from_parts(timestamp_ms, random))
}
fn projection(run_id: RunId, title: &str, created_at: DateTime<Utc>) -> RunProjection {
RunProjection::new(
title.to_string(),
RunSpec {
run_id,
settings: WorkflowSettings::default(),
graph: Graph::new("test"),
graph_source: None,
workflow_slug: Some("test-workflow".to_string()),
workflow_version_id: None,
target: None,
automation: None,
source_directory: None,
labels: HashMap::new(),
provenance: test_support::test_run_provenance(),
manifest_blob: None,
definition_blob: None,
spec_blob: None,
git: None,
fork_source_ref: None,
},
created_at,
)
}
fn entry(projection: RunProjection, last_seq: u32) -> CachedRunProjection {
CachedRunProjection::from_projection(projection.spec.run_id, projection, last_seq)
}
async fn store() -> (tempfile::TempDir, RunSummaryStore) {
store_test_support::sqlite_summary_store().await
}
fn sample_status(kind: RunStatusKind) -> RunStatus {
match kind {
RunStatusKind::Submitted => RunStatus::Submitted,
RunStatusKind::Pending => RunStatus::Pending {
reason: PendingReason::ApprovalRequired,
},
RunStatusKind::Runnable => RunStatus::Runnable,
RunStatusKind::Starting => RunStatus::Starting,
RunStatusKind::Running => RunStatus::Running,
RunStatusKind::Blocked => RunStatus::Blocked {
blocked_reason: BlockedReason::HumanInputRequired,
},
RunStatusKind::Paused => RunStatus::Paused { prior_block: None },
RunStatusKind::Removing => RunStatus::Removing,
RunStatusKind::Succeeded => RunStatus::Succeeded {
reason: SuccessReason::Completed,
},
RunStatusKind::Failed => RunStatus::Failed {
reason: FailureReason::WorkflowError,
},
RunStatusKind::Dead => RunStatus::Dead,
}
}
/// The migration's `CHECK (status IN (...))` freezes the status strings;
/// prove every `RunStatusKind` variant passes it so an enum change that
/// forgets a follow-up migration fails in CI instead of at runtime.
#[tokio::test]
async fn every_status_kind_upserts_within_schema_check() {
let (_directory, store) = store().await;
let created_at = dt("2026-07-11T12:00:00Z");
for (index, kind) in RunStatusKind::VARIANTS.iter().enumerate() {
let id = run_id(
created_at.timestamp_millis().cast_unsigned(),
u128::try_from(index).unwrap() + 1,
);
let mut projected = projection(id, "status", created_at);
projected.status = sample_status(*kind);
store.upsert_projection(&entry(projected, 1)).await.unwrap();
}
}
#[tokio::test]
async fn upsert_is_monotonic_and_get_applies_children_count() {
let (_directory, store) = store().await;
let created_at = dt("2026-07-11T12:00:00Z");
let parent_id = run_id(created_at.timestamp_millis().cast_unsigned(), 1);
let child_id = run_id(created_at.timestamp_millis().cast_unsigned() + 1, 2);
let parent = entry(projection(parent_id, "parent", created_at), 1);
store.upsert_projection(&parent).await.unwrap();
let mut child_projection = projection(child_id, "new title", created_at);
child_projection.parent_id = Some(parent_id);
child_projection.last_event_at = created_at + chrono::Duration::seconds(2);
store
.upsert_projection(&entry(child_projection, 2))
.await
.unwrap();
let mut stale = projection(child_id, "stale title", created_at);
stale.parent_id = Some(parent_id);
store.upsert_projection(&entry(stale, 1)).await.unwrap();
let parent = store.get(&parent_id, created_at).await.unwrap().unwrap();
let child = store.get(&child_id, created_at).await.unwrap().unwrap();
assert_eq!(parent.children_count, 1);
assert_eq!(child.title, "new title");
}
#[tokio::test]
async fn list_filters_sorts_and_paginates_in_sqlite() {
let (_directory, store) = store().await;
let created_at = dt("2026-07-11T12:00:00Z");
let first_id = run_id(created_at.timestamp_millis().cast_unsigned(), 1);
let second_id = run_id(created_at.timestamp_millis().cast_unsigned() + 1, 2);
let archived_id = run_id(created_at.timestamp_millis().cast_unsigned() + 2, 3);
let mut first = projection(first_id, "bravo", created_at);
first.spec.automation = Some(AutomationRef {
id: "nightly".to_string(),
name: None,
trigger_id: None,
});
let mut second = projection(second_id, "alpha", created_at);
second.spec.automation = Some(AutomationRef {
id: "nightly".to_string(),
name: None,
trigger_id: None,
});
let mut archived = projection(archived_id, "charlie", created_at);
archived.archived_at = Some(created_at);
for projected in [first, second, archived] {
store.upsert_projection(&entry(projected, 1)).await.unwrap();
}
let page = store
.list(
&RunSummaryListQuery {
automation_id: Some("nightly".to_string()),
sort: RunSummarySort::Title,
direction: RunSummarySortDirection::Asc,
limit: 1,
..RunSummaryListQuery::default()
},
created_at,
)
.await
.unwrap();
assert_eq!(page.total, 2);
assert!(page.has_more);
assert_eq!(page.data[0].title, "alpha");
let archived = store
.list(
&RunSummaryListQuery {
visibility: RunSummaryVisibility::Selected {
statuses: Vec::new(),
archived: true,
},
..RunSummaryListQuery::default()
},
created_at,
)
.await
.unwrap();
assert_eq!(archived.data.len(), 1);
assert_eq!(archived.data[0].id, archived_id);
}
#[tokio::test]
async fn projection_persists_billing_diff_and_derived_size() {
let (_directory, store) = store().await;
let created_at = dt("2026-07-11T12:00:00Z");
let run_id = run_id(created_at.timestamp_millis().cast_unsigned(), 1);
let mut projection = projection(run_id, "billed", created_at);
projection.spec.automation = Some(AutomationRef {
id: "nightly".to_string(),
name: None,
trigger_id: None,
});
projection.status = RunStatus::Succeeded {
reason: SuccessReason::Completed,
};
projection.last_event_at = created_at + chrono::Duration::minutes(1);
projection.conclusion = Some(Conclusion {
timestamp: projection.last_event_at,
status: StageOutcome::Succeeded,
timing: RunTiming::wall_only(60_000),
failure: None,
final_git_commit_sha: None,
stages: Vec::new(),
billing: Some(BilledTokenCounts {
input_tokens: 100,
output_tokens: 20,
total_tokens: 135,
reasoning_tokens: 5,
cache_read_tokens: 10,
cache_write_tokens: 0,
total_usd_micros: Some(21_000_000),
}),
total_retries: 0,
diff: RunDiff {
patch: None,
summary: Some(DiffSummary {
files_changed: 2,
additions: 10,
deletions: 3,
}),
},
});
store
.upsert_projection(&entry(projection, 4))
.await
.unwrap();
let row = sqlx::query(
"SELECT source_last_seq, created_at_ms, last_event_at_ms, status, title, workflow_slug, \
automation_id, input_tokens, reasoning_tokens, cache_read_tokens, total_usd_micros, \
diff_files_changed, diff_additions, diff_deletions FROM runs WHERE id = ?",
)
.bind(run_id.to_string())
.fetch_one(&store.pool)
.await
.unwrap();
assert_eq!(sqlx::Row::get::<i64, _>(&row, "source_last_seq"), 4);
assert_eq!(
sqlx::Row::get::<i64, _>(&row, "created_at_ms"),
created_at.timestamp_millis()
);
assert_eq!(
sqlx::Row::get::<i64, _>(&row, "last_event_at_ms"),
(created_at + chrono::Duration::minutes(1)).timestamp_millis()
);
assert_eq!(sqlx::Row::get::<String, _>(&row, "status"), "succeeded");
assert_eq!(sqlx::Row::get::<String, _>(&row, "title"), "billed");
assert_eq!(
sqlx::Row::get::<String, _>(&row, "workflow_slug"),
"test-workflow"
);
assert_eq!(
sqlx::Row::get::<String, _>(&row, "automation_id"),
"nightly"
);
assert_eq!(sqlx::Row::get::<i64, _>(&row, "input_tokens"), 100);
assert_eq!(sqlx::Row::get::<i64, _>(&row, "reasoning_tokens"), 5);
assert_eq!(sqlx::Row::get::<i64, _>(&row, "cache_read_tokens"), 10);
assert_eq!(
sqlx::Row::get::<i64, _>(&row, "total_usd_micros"),
21_000_000
);
assert_eq!(sqlx::Row::get::<i64, _>(&row, "diff_files_changed"), 2);
assert_eq!(sqlx::Row::get::<i64, _>(&row, "diff_additions"), 10);
assert_eq!(sqlx::Row::get::<i64, _>(&row, "diff_deletions"), 3);
let run = store.get(&run_id, created_at).await.unwrap().unwrap();
assert_eq!(run.size, RunSize::S);
}
#[tokio::test]
async fn projection_normalizes_legacy_overlapping_reasoning_tokens() {
let (_directory, store) = store().await;
let created_at = dt("2026-07-11T12:00:00Z");
let run_id = run_id(created_at.timestamp_millis().cast_unsigned(), 1);
let mut projection = projection(run_id, "legacy billing", created_at);
projection.conclusion = Some(Conclusion {
timestamp: created_at,
status: StageOutcome::Succeeded,
timing: RunTiming::default(),
failure: None,
final_git_commit_sha: None,
stages: Vec::new(),
billing: Some(BilledTokenCounts {
input_tokens: 53,
output_tokens: -7,
total_tokens: 112,
reasoning_tokens: 66,
..BilledTokenCounts::default()
}),
total_retries: 0,
diff: RunDiff::default(),
});
store
.upsert_projection(&entry(projection, 1))
.await
.unwrap();
let row = sqlx::query(
"SELECT input_tokens, output_tokens, reasoning_tokens FROM runs WHERE id = ?",
)
.bind(run_id.to_string())
.fetch_one(&store.pool)
.await
.unwrap();
assert_eq!(sqlx::Row::get::<i64, _>(&row, "input_tokens"), 53);
assert_eq!(sqlx::Row::get::<i64, _>(&row, "output_tokens"), 0);
assert_eq!(sqlx::Row::get::<i64, _>(&row, "reasoning_tokens"), 59);
}
#[tokio::test]
async fn reconcile_removes_rows_absent_from_authoritative_entries() {
let (_directory, store) = store().await;
let created_at = dt("2026-07-11T12:00:00Z");
let kept_id = run_id(created_at.timestamp_millis().cast_unsigned(), 1);
let removed_id = run_id(created_at.timestamp_millis().cast_unsigned() + 1, 2);
let kept = entry(projection(kept_id, "kept", created_at), 1);
let removed = entry(projection(removed_id, "removed", created_at), 1);
store.upsert_projection(&kept).await.unwrap();
store.upsert_projection(&removed).await.unwrap();
store.reconcile(std::slice::from_ref(&kept)).await.unwrap();
assert!(store.get(&kept_id, created_at).await.unwrap().is_some());
assert!(store.get(&removed_id, created_at).await.unwrap().is_none());
}
#[tokio::test]
async fn failed_reconcile_rolls_back_and_can_be_retried() {
let (_directory, store) = store().await;
let created_at = dt("2026-07-11T12:00:00Z");
let stale_id = run_id(created_at.timestamp_millis().cast_unsigned(), 1);
let good_id = run_id(created_at.timestamp_millis().cast_unsigned() + 1, 2);
let recovered_id = run_id(created_at.timestamp_millis().cast_unsigned() + 2, 3);
store
.upsert_projection(&entry(projection(stale_id, "stale", created_at), 1))
.await
.unwrap();
let good = entry(projection(good_id, "good", created_at), 1);
let recovered_projection = projection(recovered_id, "recovered", created_at);
let invalid = entry(recovered_projection.clone(), 0);
assert!(store.reconcile(&[good.clone(), invalid]).await.is_err());
assert!(store.get(&stale_id, created_at).await.unwrap().is_some());
assert!(store.get(&good_id, created_at).await.unwrap().is_none());
let recovered = entry(recovered_projection, 1);
store.reconcile(&[good, recovered]).await.unwrap();
assert!(store.get(&stale_id, created_at).await.unwrap().is_none());
assert!(store.get(&good_id, created_at).await.unwrap().is_some());
assert!(
store
.get(&recovered_id, created_at)
.await
.unwrap()
.is_some()
);
}
}

98
lib/components/fabro-store/src/slate/mod.rs
View file

@ -4,7 +4,7 @@ mod run_store;
use std::collections::HashMap;
use std::path::PathBuf;
use std::sync::{Arc, OnceLock};
use std::sync::Arc;
use std::time::Duration;
use chrono::{DateTime, Utc};
@ -19,7 +19,7 @@ use slatedb::config::{CompressionCodec, Settings};
use tokio::sync::{Mutex, OnceCell};
use tracing::warn;
use crate::{BlobStore, Error, ListRunsQuery, Result, RunProjection, RunSummaryStore, keys};
use crate::{BlobStore, Error, ListRunsQuery, Result, RunProjection, RunRecordStore, keys};
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct UnreadableRun {
@ -45,7 +45,7 @@ pub struct Database {
catalog_index: Arc<OnceCell<Arc<RunCatalogIndex>>>,
projection_cache: Arc<RunProjectionCache>,
projection_cache_warmed: Arc<OnceCell<()>>,
run_summary_store: Arc<OnceLock<Arc<RunSummaryStore>>>,
run_record_store: Arc<RunRecordStore>,
}
impl std::fmt::Debug for Database {
@ -65,6 +65,7 @@ impl Database {
flush_interval: Duration,
cache_path: Option<PathBuf>,
blobs: Arc<BlobStore>,
run_record_store: Arc<RunRecordStore>,
) -> Self {
Self {
object_store,
@ -77,16 +78,13 @@ impl Database {
catalog_index: Arc::new(OnceCell::new()),
projection_cache: Arc::new(RunProjectionCache::default()),
projection_cache_warmed: Arc::new(OnceCell::new()),
run_summary_store: Arc::new(OnceLock::new()),
run_record_store,
}
}
pub fn attach_run_summary_store(&self, store: Arc<RunSummaryStore>) -> Arc<RunSummaryStore> {
Arc::clone(self.run_summary_store.get_or_init(|| store))
}
fn run_summary_store(&self) -> Option<Arc<RunSummaryStore>> {
self.run_summary_store.get().cloned()
#[must_use]
pub fn run_record_store(&self) -> Arc<RunRecordStore> {
Arc::clone(&self.run_record_store)
}
fn shared_db_prefix(&self) -> String {
@ -142,7 +140,7 @@ impl Database {
read_only,
self.blobs(),
Arc::clone(&self.projection_cache),
Arc::clone(&self.run_summary_store),
self.run_record_store(),
)
.await
}
@ -240,9 +238,7 @@ impl Database {
}
}
}
if let Some(store) = self.run_summary_store() {
store.reconcile(&entries).await?;
}
self.run_record_store.reconcile(&entries).await?;
self.projection_cache.replace_all(entries).await;
Ok::<_, Error>(())
})
@ -389,9 +385,7 @@ impl Database {
self.delete_session_indexes_for_run(run_id).await?;
self.catalog_index().await?.remove(run_id).await?;
self.remove_cached_run(run_id).await;
if let Some(store) = self.run_summary_store() {
store.delete(run_id).await?;
}
self.run_record_store.delete(run_id).await?;
Ok(())
}
@ -558,6 +552,31 @@ mod tests {
(object_store, store)
}
fn make_store_with_run_records(
run_records: Arc<RunRecordStore>,
) -> (Arc<dyn ObjectStore>, Database) {
let object_store: Arc<dyn ObjectStore> = Arc::new(InMemory::new());
let store = store_test_support::test_database_with_stores(
object_store.clone(),
"runs/",
Duration::from_millis(1),
None,
store_test_support::test_blob_store(),
run_records,
);
(object_store, store)
}
#[tokio::test]
async fn required_run_record_store_is_shared_with_run_handles() {
let (_object_store, store) = make_store();
let records = store.run_record_store();
let run = store.create_run(&test_run_id("run-1")).await.unwrap();
assert!(run.shares_run_record_store(&records));
assert!(Arc::ptr_eq(&records, &store.clone().run_record_store()));
}
#[tokio::test]
async fn retire_refresh_token_keyspace_clears_the_prefix_and_is_idempotent() {
let (_object_store, store) = make_store();
@ -596,8 +615,8 @@ mod tests {
);
}
async fn make_summary_store() -> (tempfile::TempDir, Arc<RunSummaryStore>) {
let (directory, store) = store_test_support::sqlite_summary_store().await;
async fn make_run_record_store() -> (tempfile::TempDir, Arc<RunRecordStore>) {
let (directory, store) = store_test_support::sqlite_run_record_store().await;
(directory, Arc::new(store))
}
@ -972,9 +991,8 @@ mod tests {
#[tokio::test]
async fn rejected_transition_leaves_reconciled_summary_present() {
let (_object_store, store) = make_store();
let (_directory, summaries) = make_summary_store().await;
store.attach_run_summary_store(Arc::clone(&summaries));
let (_directory, summaries) = make_run_record_store().await;
let (_object_store, store) = make_store_with_run_records(Arc::clone(&summaries));
let run_id = test_run_id("run-1");
let run = store.create_run(&run_id).await.unwrap();
append_runnable(&run, "run-1", dt("2026-03-27T12:00:00Z")).await;
@ -995,10 +1013,9 @@ mod tests {
}
#[tokio::test]
async fn committed_append_succeeds_when_summary_update_fails_and_is_repairable() {
let (object_store, store) = make_store();
let (directory, summaries) = make_summary_store().await;
store.attach_run_summary_store(Arc::clone(&summaries));
async fn best_effort_run_record_update_failure_keeps_slate_append_repairable() {
let (directory, summaries) = make_run_record_store().await;
let (object_store, store) = make_store_with_run_records(Arc::clone(&summaries));
let run_id = test_run_id("run-1");
let run = store.create_run(&run_id).await.unwrap();
append_created(&run, "run-1", dt("2026-03-27T12:00:00Z")).await;
@ -1022,7 +1039,7 @@ mod tests {
assert_eq!(stored.event, result.unwrap().event);
let repaired_summaries =
Arc::new(store_test_support::sqlite_summary_store_at(directory.path()).await);
Arc::new(store_test_support::sqlite_run_record_store_at(directory.path()).await);
let stale = repaired_summaries
.get(&run_id, Utc::now())
.await
@ -1030,13 +1047,14 @@ mod tests {
.unwrap();
assert_ne!(stale.title, "Committed title");
let reopened = store_test_support::test_database(
let reopened = store_test_support::test_database_with_stores(
object_store,
"runs/",
Duration::from_millis(1),
None,
store_test_support::test_blob_store(),
Arc::clone(&repaired_summaries),
);
reopened.attach_run_summary_store(Arc::clone(&repaired_summaries));
reopened.warm_projection_cache().await.unwrap();
let repaired = repaired_summaries
.get(&run_id, Utc::now())
@ -1657,10 +1675,9 @@ mod tests {
}
#[tokio::test]
async fn append_event_refreshes_projection_cache_and_delete_removes_it() {
let (_object_store, store) = make_store();
let (_directory, summaries) = make_summary_store().await;
store.attach_run_summary_store(Arc::clone(&summaries));
async fn required_run_record_append_refreshes_cache_and_delete_removes_rows() {
let (_directory, summaries) = make_run_record_store().await;
let (_object_store, store) = make_store_with_run_records(Arc::clone(&summaries));
let run = store.create_run(&test_run_id("run-1")).await.unwrap();
append_created(&run, "run-1", dt("2026-03-27T12:00:00Z")).await;
store.warm_projection_cache().await.unwrap();
@ -1834,15 +1851,20 @@ mod tests {
}
#[tokio::test]
async fn projection_cache_warmup_backfills_sqlite_run_summaries() {
async fn required_run_record_warmup_backfills_sqlite_run_records() {
let (object_store, store) = make_store();
let run = store.create_run(&test_run_id("run-1")).await.unwrap();
append_completed(&run, "run-1", dt("2026-03-27T12:00:00Z")).await;
let reopened =
store_test_support::test_database(object_store, "runs", Duration::from_millis(1), None);
let (_directory, summaries) = make_summary_store().await;
reopened.attach_run_summary_store(Arc::clone(&summaries));
let (_directory, summaries) = make_run_record_store().await;
let reopened = store_test_support::test_database_with_stores(
object_store,
"runs",
Duration::from_millis(1),
None,
store_test_support::test_blob_store(),
Arc::clone(&summaries),
);
reopened.warm_projection_cache().await.unwrap();
let summary = summaries

33
lib/components/fabro-store/src/slate/run_store.rs
View file

@ -1,6 +1,6 @@
use std::collections::VecDeque;
use std::sync::Arc;
use std::sync::atomic::{AtomicU32, Ordering};
use std::sync::{Arc, OnceLock};
use bytes::Bytes;
use chrono::Utc;
@ -14,7 +14,7 @@ use tracing::warn;
use super::projection_cache::{CachedRunProjection, RunProjectionCache};
use crate::run_state::{EventProjectionCache, RunProjectionReducer};
use crate::{
BlobStore, Error, EventEnvelope, EventPayload, Result, RunProjection, RunSummaryStore, StageId,
BlobStore, Error, EventEnvelope, EventPayload, Result, RunProjection, RunRecordStore, StageId,
keys,
};
@ -45,9 +45,7 @@ pub(crate) struct RunDatabaseInner {
state_lock: Mutex<()>,
projection_cache: Mutex<EventProjectionCache>,
shared_projection_cache: Arc<RunProjectionCache>,
// Shared cell rather than a snapshot so a summary store attached after
// this writer opened is still picked up by later appends.
run_summary_store: Arc<OnceLock<Arc<RunSummaryStore>>>,
run_record_store: Arc<RunRecordStore>,
recent_events: Mutex<VecDeque<EventEnvelope>>,
recent_event_limit: usize,
event_tx: broadcast::Sender<EventEnvelope>,
@ -60,7 +58,7 @@ impl RunDatabase {
read_only: bool,
blob_store: Arc<BlobStore>,
shared_projection_cache: Arc<RunProjectionCache>,
run_summary_store: Arc<OnceLock<Arc<RunSummaryStore>>>,
run_record_store: Arc<RunRecordStore>,
) -> Result<Self> {
let cached_projection = shared_projection_cache.projection_snapshot(&run_id).await;
let projection_cache = cached_projection.as_ref().map_or_else(
@ -92,7 +90,7 @@ impl RunDatabase {
state_lock: Mutex::new(()),
projection_cache: Mutex::new(projection_cache),
shared_projection_cache,
run_summary_store,
run_record_store,
recent_events: Mutex::new(VecDeque::with_capacity(DEFAULT_EVENT_TAIL_LIMIT)),
recent_event_limit: DEFAULT_EVENT_TAIL_LIMIT,
event_tx,
@ -123,6 +121,11 @@ impl RunDatabase {
self.inner.run_id
}
#[cfg(test)]
pub(crate) fn shares_run_record_store(&self, store: &Arc<RunRecordStore>) -> bool {
Arc::ptr_eq(&self.inner.run_record_store, store)
}
pub fn subscribe(&self) -> broadcast::Receiver<EventEnvelope> {
self.inner.event_tx.subscribe()
}
@ -231,15 +234,13 @@ impl RunDatabase {
}
async fn update_summary_after_committed_append(&self, cached: &CachedRunProjection) {
if let Some(store) = self.inner.run_summary_store.get() {
if let Err(err) = store.upsert_projection(cached).await {
warn!(
run_id = %self.inner.run_id,
source_last_seq = cached.last_seq,
error = ?err,
"failed to update SQLite run summary after committed append"
);
}
if let Err(err) = self.inner.run_record_store.upsert_projection(cached).await {
warn!(
run_id = %self.inner.run_id,
source_last_seq = cached.last_seq,
error = ?err,
"failed to update SQLite run record after committed append"
);
}
}

70
lib/components/fabro-store/src/test_support/mod.rs
View file

@ -8,8 +8,8 @@ use sqlx::sqlite::{SqliteConnectOptions, SqlitePoolOptions};
use crate::keys::SlateKey;
#[cfg(test)]
use crate::{AuthCodeStore, AuthSessionStore, RunSummaryStore};
use crate::{BlobStore, Database, Result};
use crate::{AuthCodeStore, AuthSessionStore};
use crate::{BlobStore, Database, Result, RunRecordStore};
/// Returns an isolated SQLite blob authority backed by its own in-memory
/// database.
@ -46,6 +46,32 @@ pub fn test_blob_store() -> Arc<BlobStore> {
Arc::new(BlobStore::new(pool))
}
/// Returns an isolated SQLite run-record store backed by its own in-memory
/// database and the production `runs` and `run_events` schemas.
#[must_use]
pub fn test_run_record_store() -> Arc<RunRecordStore> {
let options = SqliteConnectOptions::new()
.filename(":memory:")
.foreign_keys(true);
let pool = SqlitePoolOptions::new()
.max_connections(1)
.max_lifetime(None)
.idle_timeout(None)
.after_connect(|connection, _metadata| {
Box::pin(async move {
sqlx::raw_sql(fabro_db::RUNS_MIGRATION_SQL)
.execute(&mut *connection)
.await?;
sqlx::raw_sql(fabro_db::RUN_EVENTS_MIGRATION_SQL)
.execute(&mut *connection)
.await?;
Ok(())
})
})
.connect_lazy_with(options);
Arc::new(RunRecordStore::new(pool))
}
/// Returns the SQLite file backing [`test_blob_store_at`] for `store_dir`.
#[must_use]
pub fn test_blob_store_path(store_dir: &Path) -> PathBuf {
@ -119,7 +145,37 @@ pub fn test_database_with_blobs(
cache_path: Option<PathBuf>,
blobs: Arc<BlobStore>,
) -> Database {
Database::new(object_store, base_prefix, flush_interval, cache_path, blobs)
test_database_with_stores(
object_store,
base_prefix,
flush_interval,
cache_path,
blobs,
test_run_record_store(),
)
}
/// Builds a Slate-backed run database with explicit shared SQLite stores.
///
/// Use this only when a test needs a failing, persistent, or shared store;
/// ordinary fixtures should use [`test_database`].
#[must_use]
pub fn test_database_with_stores(
object_store: Arc<dyn ObjectStore>,
base_prefix: impl Into<String>,
flush_interval: Duration,
cache_path: Option<PathBuf>,
blobs: Arc<BlobStore>,
run_records: Arc<RunRecordStore>,
) -> Database {
Database::new(
object_store,
base_prefix,
flush_interval,
cache_path,
blobs,
run_records,
)
}
/// Seeds one canonical row in the legacy SlateDB blob keyspace.
@ -171,13 +227,13 @@ pub(crate) async fn sqlite_auth_code_store() -> (tempfile::TempDir, AuthCodeStor
}
#[cfg(test)]
pub(crate) async fn sqlite_summary_store() -> (tempfile::TempDir, RunSummaryStore) {
pub(crate) async fn sqlite_run_record_store() -> (tempfile::TempDir, RunRecordStore) {
let directory = tempfile::tempdir().unwrap();
let store = sqlite_summary_store_at(directory.path()).await;
let store = sqlite_run_record_store_at(directory.path()).await;
(directory, store)
}
#[cfg(test)]
pub(crate) async fn sqlite_summary_store_at(directory: &Path) -> RunSummaryStore {
RunSummaryStore::new(sqlite_test_pool(directory).await)
pub(crate) async fn sqlite_run_record_store_at(directory: &Path) -> RunRecordStore {
RunRecordStore::new(sqlite_test_pool(directory).await)
}

30
lib/foundation/fabro-db/migrations/2026082701_run_events.sql Normal file
View file

@ -0,0 +1,30 @@
CREATE TABLE run_events (
run_id TEXT NOT NULL,
seq INTEGER NOT NULL,
event_name TEXT NOT NULL,
node_id TEXT,
stage_id TEXT,
session_id TEXT,
event_json TEXT NOT NULL,
PRIMARY KEY (run_id, seq),
FOREIGN KEY (run_id) REFERENCES runs(id) ON DELETE CASCADE,
CHECK (seq BETWEEN 1 AND 999999),
CHECK (json_valid(event_json))
);
CREATE INDEX run_events_by_stage
ON run_events(run_id, stage_id, seq)
WHERE stage_id IS NOT NULL;
CREATE INDEX run_events_by_legacy_node
ON run_events(run_id, node_id, seq)
WHERE stage_id IS NULL AND node_id IS NOT NULL;
CREATE INDEX run_events_by_session
ON run_events(run_id, session_id, seq)
WHERE session_id IS NOT NULL
AND event_name GLOB 'run.session.*';
CREATE INDEX run_events_by_pull_request_creation_request
ON run_events(run_id, seq)
WHERE event_name = 'pull_request.creation_requested';

8
lib/foundation/fabro-db/src/lib.rs
View file

@ -20,6 +20,14 @@ static MIGRATOR: Migrator = sqlx::migrate!("./migrations");
/// the production blob schema without a filesystem path into this crate.
pub const BLOBS_MIGRATION_SQL: &str = include_str!("../migrations/2026081301_blobs.sql");
/// The run-record migration, exposed so fixtures in other crates can install
/// the production schema without a filesystem path into this crate.
pub const RUNS_MIGRATION_SQL: &str = include_str!("../migrations/2026071104_runs.sql");
/// The run-event migration, exposed so fixtures in other crates can install
/// the production schema without a filesystem path into this crate.
pub const RUN_EVENTS_MIGRATION_SQL: &str = include_str!("../migrations/2026082701_run_events.sql");
#[derive(Clone)]
pub struct Database {
pool: DbPool,

270
lib/foundation/fabro-db/tests/sqlite.rs
View file

@ -475,6 +475,276 @@ async fn runs_schema_creates_indexes_and_rejects_invalid_rows() -> anyhow::Resul
Ok(())
}
#[tokio::test]
async fn run_events_schema_has_final_shape_constraints_and_indexes() -> anyhow::Result<()> {
let dir = tempfile::tempdir()?;
let database = fabro_db::Database::connect(dir.path().join("fabro.sqlite3")).await?;
database.migrate().await?;
let run_columns = sqlx::query("PRAGMA table_info(runs)")
.fetch_all(database.pool())
.await?;
assert_eq!(
run_columns.len(),
24,
"the existing runs row must stay unchanged"
);
let event_columns = sqlx::query("PRAGMA table_info(run_events)")
.fetch_all(database.pool())
.await?;
let event_column_contract = event_columns
.iter()
.map(|column| {
(
column.get::<String, _>("name"),
column.get::<String, _>("type"),
column.get::<i64, _>("notnull"),
column.get::<i64, _>("pk"),
)
})
.collect::<Vec<_>>();
assert_eq!(event_column_contract, vec![
("run_id".to_string(), "TEXT".to_string(), 1, 1),
("seq".to_string(), "INTEGER".to_string(), 1, 2),
("event_name".to_string(), "TEXT".to_string(), 1, 0),
("node_id".to_string(), "TEXT".to_string(), 0, 0),
("stage_id".to_string(), "TEXT".to_string(), 0, 0),
("session_id".to_string(), "TEXT".to_string(), 0, 0),
("event_json".to_string(), "TEXT".to_string(), 1, 0),
]);
let foreign_keys = sqlx::query("PRAGMA foreign_key_list(run_events)")
.fetch_all(database.pool())
.await?;
assert_eq!(foreign_keys.len(), 1);
assert_eq!(foreign_keys[0].get::<String, _>("table"), "runs");
assert_eq!(foreign_keys[0].get::<String, _>("from"), "run_id");
assert_eq!(foreign_keys[0].get::<String, _>("to"), "id");
assert_eq!(foreign_keys[0].get::<String, _>("on_delete"), "CASCADE");
let indexes = sqlx::query("PRAGMA index_list(run_events)")
.fetch_all(database.pool())
.await?;
let named_indexes = indexes
.iter()
.filter_map(|index| {
let name = index.get::<String, _>("name");
name.starts_with("run_events_by_").then_some((
name,
index.get::<i64, _>("unique"),
index.get::<i64, _>("partial"),
))
})
.collect::<Vec<_>>();
assert_eq!(named_indexes, vec![
(
"run_events_by_pull_request_creation_request".to_string(),
0,
1,
),
("run_events_by_session".to_string(), 0, 1),
("run_events_by_legacy_node".to_string(), 0, 1),
("run_events_by_stage".to_string(), 0, 1),
]);
assert!(indexes.iter().all(|index| {
index.get::<i64, _>("unique") == 0
|| index.get::<String, _>("name") == "sqlite_autoindex_run_events_1"
}));
insert_run_with_id(database.pool(), "parent", None).await?;
insert_run_with_id(database.pool(), "child", Some("parent")).await?;
insert_run_event(
database.pool(),
"parent",
1,
"run.created",
None,
None,
None,
)
.await?;
for invalid in [
insert_run_event(
database.pool(),
"parent",
1,
"run.created",
None,
None,
None,
)
.await,
insert_run_event(
database.pool(),
"missing",
1,
"run.created",
None,
None,
None,
)
.await,
insert_run_event(
database.pool(),
"parent",
0,
"run.created",
None,
None,
None,
)
.await,
insert_run_event(
database.pool(),
"parent",
1_000_000,
"run.created",
None,
None,
None,
)
.await,
] {
assert!(invalid.is_err());
}
let invalid_json = sqlx::query(
"INSERT INTO run_events (run_id, seq, event_name, event_json) VALUES (?, ?, ?, ?)",
)
.bind("parent")
.bind(2_i64)
.bind("run.started")
.bind("not-json")
.execute(database.pool())
.await;
assert!(invalid_json.is_err());
sqlx::query("INSERT INTO blobs (hash, data) VALUES (?, ?)")
.bind("a".repeat(64))
.bind(vec![1_u8])
.execute(database.pool())
.await?;
sqlx::query("DELETE FROM runs WHERE id = ?")
.bind("parent")
.execute(database.pool())
.await?;
let event_count: i64 =
sqlx::query_scalar("SELECT COUNT(*) FROM run_events WHERE run_id = 'parent'")
.fetch_one(database.pool())
.await?;
let child_parent: Option<String> =
sqlx::query_scalar("SELECT parent_id FROM runs WHERE id = 'child'")
.fetch_one(database.pool())
.await?;
let blob_count: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM blobs")
.fetch_one(database.pool())
.await?;
assert_eq!(event_count, 0);
assert_eq!(child_parent.as_deref(), Some("parent"));
assert_eq!(blob_count, 1);
Ok(())
}
#[tokio::test]
async fn run_events_schema_query_plans_use_candidate_indexes() -> anyhow::Result<()> {
let dir = tempfile::tempdir()?;
let database = fabro_db::Database::connect(dir.path().join("fabro.sqlite3")).await?;
database.migrate().await?;
for (sql, expected_index) in [
(
"EXPLAIN QUERY PLAN SELECT * FROM run_events WHERE run_id = ? AND seq > ? ORDER BY seq ASC LIMIT ?",
"sqlite_autoindex_run_events_1",
),
(
"EXPLAIN QUERY PLAN SELECT * FROM run_events WHERE run_id = ? AND seq = ?",
"sqlite_autoindex_run_events_1",
),
(
"EXPLAIN QUERY PLAN SELECT * FROM run_events WHERE run_id = ? AND stage_id = ? ORDER BY seq ASC LIMIT ?",
"run_events_by_stage",
),
(
"EXPLAIN QUERY PLAN SELECT * FROM run_events WHERE run_id = ? AND stage_id IS NULL AND node_id = ? ORDER BY seq ASC LIMIT ?",
"run_events_by_legacy_node",
),
(
"EXPLAIN QUERY PLAN SELECT * FROM run_events WHERE run_id = ? AND session_id = ? AND event_name GLOB 'run.session.*' ORDER BY seq ASC LIMIT ?",
"run_events_by_session",
),
(
"EXPLAIN QUERY PLAN SELECT * FROM run_events WHERE event_name = 'pull_request.creation_requested' ORDER BY run_id, seq",
"run_events_by_pull_request_creation_request",
),
] {
let details = sqlx::query(sql)
.bind("run")
.bind("value")
.bind(10_i64)
.fetch_all(database.pool())
.await?
.into_iter()
.map(|row| row.get::<String, _>("detail"))
.collect::<Vec<_>>()
.join("; ");
assert!(
details.contains(expected_index),
"expected {expected_index} in query plan: {details}"
);
}
Ok(())
}
async fn insert_run_with_id(
pool: &fabro_db::DbPool,
id: &str,
parent_id: Option<&str>,
) -> Result<(), sqlx::Error> {
sqlx::query(
r"
INSERT INTO runs (
id, source_last_seq, created_at_ms, last_event_at_ms, status, parent_id, title,
input_tokens, summary_json
) VALUES (?, 1, 0, 0, 'submitted', ?, 'title', 0, ?)
",
)
.bind(id)
.bind(parent_id)
.bind(format!(r#"{{"id":"{id}"}}"#))
.execute(pool)
.await?;
Ok(())
}
async fn insert_run_event(
pool: &fabro_db::DbPool,
run_id: &str,
seq: i64,
event_name: &str,
node_id: Option<&str>,
stage_id: Option<&str>,
session_id: Option<&str>,
) -> Result<(), sqlx::Error> {
sqlx::query(
r"
INSERT INTO run_events (run_id, seq, event_name, node_id, stage_id, session_id, event_json)
VALUES (?, ?, ?, ?, ?, ?, '{}')
",
)
.bind(run_id)
.bind(seq)
.bind(event_name)
.bind(node_id)
.bind(stage_id)
.bind(session_id)
.execute(pool)
.await?;
Ok(())
}
async fn insert_minimal_run(
pool: &fabro_db::DbPool,
status: &str,