scriptoriumadmin/fabro
1
0
Fork 0
mirror of https://github.com/fabro-sh/fabro.git synced 2026-09-12 23:02:41 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 4

Simplify runs read model: single-source mappings, leaner queries

Browse source 
Consolidate duplicated logic from the SQLite runs read model review:

- Derive the status sort CASE and board-column filter from a new
  RunStatusKind::board_rank(), replacing three hand-maintained copies
  of the status/column mapping; add a test upserting every status
  variant so the migration CHECK can't silently drift
- Share RunSize bucket thresholds between from_total_usd_micros and
  the generated size-sort CASE via RunSize::BUCKET_MAX_USD_MICROS
- Resolve run selectors from a lean identity query instead of
  decoding every stored summary per request
- Delete the RunsSortKey/RunsSortDirection adapter enums; the store
  sort enums now carry the wire serde names
- Consolidate the workflow display-name fallback chain into
  WorkflowRef::display_name() (store, CLI, run lookup)
- Share pagination clamping and the paginated list envelope across
  handlers
- Reconcile now skips rows whose source seq is unchanged and
  batch-deletes stale rows; drop the two indexes no query can use
- Hold the summary store OnceLock cell in RunDatabaseInner instead of
  a snapshot so late attachment reaches already-open writers
- Misc: expect() on COUNT(*) sign, %err logging, shared wall-time
  helper, shared SQLite test fixture, dead billing fallback removed

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
Bryan Helmkamp 2026-07-11 15:46:51 -04:00 • committed by Scott Werner
parent 22844300af
commit 2843b33d92
19 changed files with 376 additions and 326 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

1
Cargo.lock generated
View file

@ -3126,6 +3126,7 @@ dependencies = [
"serde_json",
"slatedb",
"sqlx",
"strum 0.28.0",
"tempfile",
"thiserror 2.0.18",
"tokio",

6
lib/crates/fabro-cli/src/server_runs.rs
View file

@ -38,11 +38,7 @@ impl ServerRunInfo {
}
pub(crate) fn workflow_display_name(&self) -> String {
self.workflow_name()
.or_else(|| self.workflow_graph_name())
.or_else(|| self.workflow_slug())
.unwrap_or("-")
.to_string()
self.run.workflow.display_name().unwrap_or("-").to_string()
}
pub(crate) fn workflow_matches(&self, pattern: &str) -> bool {

2
lib/crates/fabro-db/migrations/2026071104_runs.sql
View file

@ -53,6 +53,4 @@ CREATE INDEX runs_by_created_at ON runs(created_at_ms DESC, id DESC);
CREATE INDEX runs_by_updated_at ON runs(last_event_at_ms DESC, id DESC);
CREATE INDEX runs_by_status ON runs(archived_at_ms, status, last_event_at_ms DESC, id DESC);
CREATE INDEX runs_by_parent ON runs(parent_id, created_at_ms DESC, id DESC);
CREATE INDEX runs_by_workflow ON runs(workflow_slug, created_at_ms DESC, id DESC);
CREATE INDEX runs_by_repository ON runs(repository_name, created_at_ms DESC, id DESC);
CREATE INDEX runs_by_automation ON runs(automation_id, created_at_ms DESC, id DESC);

2
lib/crates/fabro-db/tests/sqlite.rs
View file

@ -349,7 +349,7 @@ async fn runs_schema_creates_indexes_and_rejects_invalid_rows() -> anyhow::Resul
)
.fetch_one(database.pool())
.await?;
assert_eq!(index_count, 7);
assert_eq!(index_count, 5);
insert_minimal_run(database.pool(), "submitted", 0, r#"{"id":"run"}"#).await?;
for (status, input_tokens, summary_json) in [

25
lib/crates/fabro-server/src/server.rs
View file

@ -204,9 +204,17 @@ pub struct PaginationParams {
pub offset: u32,
}
pub(crate) fn clamp_page_limit(limit: u32) -> u32 {
limit.clamp(1, 100)
}
pub(crate) fn clamp_page_offset(offset: u32) -> u32 {
offset.min(MAX_PAGE_OFFSET)
}
pub(crate) fn paginate_items<T>(items: Vec<T>, pagination: &PaginationParams) -> (Vec<T>, bool) {
let limit = pagination.limit.clamp(1, 100) as usize;
let offset = pagination.offset.min(MAX_PAGE_OFFSET) as usize;
let limit = clamp_page_limit(pagination.limit) as usize;
let offset = clamp_page_offset(pagination.offset) as usize;
let mut data: Vec<_> = items.into_iter().skip(offset).take(limit + 1).collect();
let has_more = data.len() > limit;
data.truncate(limit);
@ -219,7 +227,7 @@ pub(crate) struct DfParams {
pub(crate) verbose: bool,
}
/// Non-paginated list response wrapper with `has_more: false`.
/// List response envelope with pagination metadata.
#[derive(serde::Serialize)]
pub struct ListResponse<T: serde::Serialize> {
data: T,
@ -227,6 +235,7 @@ pub struct ListResponse<T: serde::Serialize> {
}
impl<T: serde::Serialize> ListResponse<T> {
/// Non-paginated response with `has_more: false`.
pub fn new(data: T) -> Self {
Self {
data,
@ -236,6 +245,16 @@ impl<T: serde::Serialize> ListResponse<T> {
},
}
}
pub fn paginated(data: T, has_more: bool, total: u64) -> Self {
Self {
data,
meta: PaginationMeta {
has_more,
total: i64::try_from(total).ok(),
},
}
}
}
/// Snapshot of a managed run.

28
lib/crates/fabro-server/src/server/handler/automations.rs
View file

@ -2,7 +2,6 @@ use std::sync::Arc;
use axum::http::HeaderMap;
use axum_extra::extract::Query as ExtraQuery;
use chrono::Utc;
use fabro_automation::{
Automation, AutomationDraft, AutomationId, AutomationReplace, AutomationStoreError,
};
@ -11,8 +10,8 @@ use fabro_types::{AutomationRef, RunId};
use serde::Serialize;
use super::super::{
ApiError, AppState, IntoResponse, Json, MAX_PAGE_OFFSET, PaginationParams, Path, RequiredUser,
Response, Router, State, StatusCode, get,
ApiError, AppState, IntoResponse, Json, PaginationParams, Path, RequiredUser, Response, Router,
State, StatusCode, clamp_page_limit, clamp_page_offset, get,
};
use super::{json_with_etag_response, lifecycle, parse_required_if_match, runs};
use crate::automation_materializer::AutomationRunMaterializeInput;
@ -84,28 +83,11 @@ async fn list_automation_runs(
let query = RunSummaryListQuery {
automation_id: Some(id.to_string()),
visibility: RunSummaryVisibility::All,
limit: pagination.limit.clamp(1, 100),
offset: pagination.offset.min(MAX_PAGE_OFFSET),
limit: clamp_page_limit(pagination.limit),
offset: clamp_page_offset(pagination.offset),
..RunSummaryListQuery::default()
};
let page = match state.stores.run_summaries.list(&query, Utc::now()).await {
Ok(page) => page,
Err(err) => {
return ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string())
.into_response();
}
};
let data = state.decorate_run_summaries(page.data).await;
(
StatusCode::OK,
Json(serde_json::json!({
"data": data,
"meta": { "has_more": page.has_more, "total": page.total }
})),
)
.into_response()
runs::run_summary_page_response(&state, &query).await
}
async fn create_automation_run(

198
lib/crates/fabro-server/src/server/handler/runs.rs
View file

@ -24,20 +24,21 @@ use fabro_store::{
use fabro_types::settings::ResolveError;
use fabro_types::{
AutomationRef, Principal, RunClientProvenance, RunId, RunProvenance, RunServerProvenance,
StageContextWindow, StageContextWindowStaleness, StageContextWindowUnavailableReason,
StageHandler, StageModelUsage, StageProjection, SystemActorKind, WorkflowSettings,
parse_blob_ref,
RunStatusKind, StageContextWindow, StageContextWindowStaleness,
StageContextWindowUnavailableReason, StageHandler, StageModelUsage, StageProjection,
SystemActorKind, WorkflowSettings, parse_blob_ref,
};
use fabro_util::version::FABRO_VERSION;
use fabro_workflow::command_log::{command_log_path, read_json_string_blob, read_log_slice};
use fabro_workflow::run_status::RunStatus;
use fabro_workflow::{Error as WorkflowError, operations};
use strum::VariantArray as _;
use tokio::fs;
use tracing::info;
use super::super::{
AppState, DeleteRunOutcome, ListResponse, MAX_PAGE_OFFSET, RunExecutionMode, VariableError,
answer_from_request, api_question_from_pending_interview, default_page_limit,
AppState, DeleteRunOutcome, ListResponse, RunExecutionMode, VariableError, answer_from_request,
api_question_from_pending_interview, clamp_page_limit, clamp_page_offset, default_page_limit,
delete_run_internal, load_pending_interview, managed_run, parse_run_id_path,
parse_stage_id_path, reject_if_archived, submit_pending_interview_answer, workflow_event,
};
@ -88,29 +89,6 @@ pub(super) fn routes() -> Router<Arc<AppState>> {
.merge(manifest_routes())
}
#[derive(Debug, Clone, Copy, Default, serde::Deserialize)]
#[serde(rename_all = "snake_case")]
enum RunsSortKey {
#[default]
CreatedAt,
UpdatedAt,
Status,
Elapsed,
Repo,
Title,
Workflow,
Changes,
Size,
}
#[derive(Debug, Clone, Copy, Default, serde::Deserialize)]
#[serde(rename_all = "lowercase")]
enum RunsSortDirection {
Asc,
#[default]
Desc,
}
#[derive(serde::Deserialize)]
struct ListRunsParams {
#[serde(rename = "page[limit]", default = "default_page_limit")]
@ -124,9 +102,9 @@ struct ListRunsParams {
#[serde(default)]
status: Vec<BoardColumn>,
#[serde(default)]
sort: RunsSortKey,
sort: RunSummarySort,
#[serde(default)]
direction: RunsSortDirection,
direction: RunSummarySortDirection,
}
impl ListRunsParams {
@ -134,10 +112,10 @@ impl ListRunsParams {
RunSummaryListQuery {
parent_id: self.parent_id,
visibility: summary_visibility(&self.status, self.include_archived),
sort: summary_sort(self.sort),
direction: summary_sort_direction(self.direction),
limit: self.limit.clamp(1, 100),
offset: self.offset.min(MAX_PAGE_OFFSET),
sort: self.sort,
direction: self.direction,
limit: clamp_page_limit(self.limit),
offset: clamp_page_offset(self.offset),
..RunSummaryListQuery::default()
}
}
@ -151,35 +129,14 @@ fn summary_visibility(selected: &[BoardColumn], include_archived: bool) -> RunSu
let mut statuses = HashSet::new();
let mut archived = false;
for column in selected {
match column {
BoardColumn::Pending => {
statuses.insert(fabro_types::RunStatusKind::Submitted);
statuses.insert(fabro_types::RunStatusKind::Pending);
}
BoardColumn::Runnable => {
statuses.insert(fabro_types::RunStatusKind::Runnable);
}
BoardColumn::Initializing => {
statuses.insert(fabro_types::RunStatusKind::Starting);
}
BoardColumn::Running => {
statuses.insert(fabro_types::RunStatusKind::Running);
statuses.insert(fabro_types::RunStatusKind::Paused);
}
BoardColumn::Blocked => {
statuses.insert(fabro_types::RunStatusKind::Blocked);
}
BoardColumn::Succeeded => {
statuses.insert(fabro_types::RunStatusKind::Succeeded);
}
BoardColumn::Failed => {
statuses.insert(fabro_types::RunStatusKind::Failed);
statuses.insert(fabro_types::RunStatusKind::Dead);
}
BoardColumn::Archived => archived = true,
BoardColumn::Removing => {
statuses.insert(fabro_types::RunStatusKind::Removing);
}
match board_column_rank(*column) {
None => archived = true,
Some(rank) => statuses.extend(
RunStatusKind::VARIANTS
.iter()
.copied()
.filter(|kind| kind.board_rank() == rank),
),
}
}
RunSummaryVisibility::Selected {
@ -188,24 +145,20 @@ fn summary_visibility(selected: &[BoardColumn], include_archived: bool) -> RunSu
}
}
fn summary_sort(sort: RunsSortKey) -> RunSummarySort {
match sort {
RunsSortKey::CreatedAt => RunSummarySort::CreatedAt,
RunsSortKey::UpdatedAt => RunSummarySort::UpdatedAt,
RunsSortKey::Status => RunSummarySort::Status,
RunsSortKey::Elapsed => RunSummarySort::Elapsed,
RunsSortKey::Repo => RunSummarySort::Repository,
RunsSortKey::Title => RunSummarySort::Title,
RunsSortKey::Workflow => RunSummarySort::Workflow,
RunsSortKey::Changes => RunSummarySort::Changes,
RunsSortKey::Size => RunSummarySort::Size,
}
}
fn summary_sort_direction(direction: RunsSortDirection) -> RunSummarySortDirection {
match direction {
RunsSortDirection::Asc => RunSummarySortDirection::Asc,
RunsSortDirection::Desc => RunSummarySortDirection::Desc,
/// Rank of each board column, mirroring the `BoardColumn` enum order.
/// Statuses map to columns through [`RunStatusKind::board_rank`]; `archived`
/// has no rank because it selects on the archival overlay, not a status.
fn board_column_rank(column: BoardColumn) -> Option<u8> {
match column {
BoardColumn::Pending => Some(0),
BoardColumn::Runnable => Some(1),
BoardColumn::Initializing => Some(2),
BoardColumn::Running => Some(3),
BoardColumn::Blocked => Some(4),
BoardColumn::Succeeded => Some(5),
BoardColumn::Failed => Some(6),
BoardColumn::Archived => None,
BoardColumn::Removing => Some(8),
}
}
@ -361,29 +314,28 @@ async fn list_runs(
State(state): State<Arc<AppState>>,
ExtraQuery(params): ExtraQuery<ListRunsParams>,
) -> Response {
let page = match state
.stores
.run_summaries
.list(&params.summary_query(), Utc::now())
.await
{
Ok(page) => page,
Err(err) => {
return ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string())
.into_response();
run_summary_page_response(&state, &params.summary_query()).await
}
/// List run summaries matching `query`, decorate them, and wrap them in the
/// paginated list envelope. Shared by the runs and automation-runs lists.
pub(super) async fn run_summary_page_response(
state: &AppState,
query: &RunSummaryListQuery,
) -> Response {
match state.stores.run_summaries.list(query, Utc::now()).await {
Ok(page) => {
let data = state.decorate_run_summaries(page.data).await;
(
StatusCode::OK,
Json(ListResponse::paginated(data, page.has_more, page.total)),
)
.into_response()
}
};
let data = state.decorate_run_summaries(page.data).await;
(
StatusCode::OK,
Json(serde_json::json!({
"data": data,
"meta": { "has_more": page.has_more, "total": page.total }
})),
)
.into_response()
Err(err) => {
ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string()).into_response()
}
}
}
#[derive(Debug, serde::Deserialize)]
@ -425,49 +377,33 @@ async fn resolve_run(
State(state): State<Arc<AppState>>,
Query(query): Query<ResolveRunQuery>,
) -> Response {
let summary_query = RunSummaryListQuery {
visibility: RunSummaryVisibility::All,
limit: u32::MAX,
..RunSummaryListQuery::default()
};
let runs = match state
.stores
.run_summaries
.list(&summary_query, Utc::now())
.await
{
Ok(page) => page.data,
let identities = match state.stores.run_summaries.list_identities().await {
Ok(identities) => identities,
Err(err) => {
return ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string())
.into_response();
}
};
match resolve_run_by_selector(
&runs,
let resolved_id = match resolve_run_by_selector(
&identities,
&query.selector,
|run| run.id.to_string(),
|run| run.workflow.slug.clone(),
|run| run.workflow.name.clone(),
|run| run.workflow_slug.clone(),
|run| run.workflow_name.clone(),
|run| run.id.created_at(),
|run| run.id.created_at().to_rfc3339(),
|run| {
run.repository
.as_ref()
.and_then(|repository| repository.origin_url.clone())
},
|run| run.repository_origin_url.clone(),
) {
Ok(run) => {
let run = state.decorate_run_summary(run.clone()).await;
(StatusCode::OK, Json(run)).into_response()
}
Ok(identity) => identity.id,
Err(err @ (ResolveRunError::InvalidSelector | ResolveRunError::AmbiguousPrefix { .. })) => {
ApiError::bad_request(err.to_string()).into_response()
return ApiError::bad_request(err.to_string()).into_response();
}
Err(err @ ResolveRunError::NotFound { .. }) => {
ApiError::not_found(err.to_string()).into_response()
return ApiError::not_found(err.to_string()).into_response();
}
}
};
updated_run_response(&state, &resolved_id).await
}
async fn delete_run(

1
lib/crates/fabro-store/Cargo.toml
View file

@ -25,6 +25,7 @@ dashmap.workspace = true
serde.workspace = true
serde_json.workspace = true
sqlx.workspace = true
strum.workspace = true
chrono = { workspace = true, features = ["serde"] }
bytes.workspace = true
thiserror.workspace = true

6
lib/crates/fabro-store/src/lib.rs
View file

@ -10,6 +10,8 @@ mod run_state;
mod run_summary_store;
mod serializable_projection;
mod slate;
#[cfg(test)]
mod test_util;
mod types;
pub use artifact_store::{
@ -27,8 +29,8 @@ pub use run_sessions::{
};
pub use run_state::RunProjectionReducer;
pub use run_summary_store::{
RunSummaryListQuery, RunSummaryPage, RunSummarySort, RunSummarySortDirection, RunSummaryStore,
RunSummaryVisibility,
RunSummaryIdentity, RunSummaryListQuery, RunSummaryPage, RunSummarySort,
RunSummarySortDirection, RunSummaryStore, RunSummaryVisibility,
};
pub use serializable_projection::SerializableProjection;
pub use slate::{

6
lib/crates/fabro-store/src/run_state.rs
View file

@ -935,7 +935,7 @@ pub(crate) fn build_summary(state: &RunProjection, run_id: &RunId) -> Run {
.as_ref()
.map(|conclusion| conclusion.timing);
let terminal_total = terminal_total_usd_micros(state);
let current_total = terminal_total.or_else(|| projected_total_usd_micros(state));
let current_total = projected_billing(state).total_usd_micros;
Run {
id: *run_id,
@ -1003,10 +1003,6 @@ fn terminal_total_usd_micros(state: &RunProjection) -> Option<i64> {
.and_then(|billing| billing.total_usd_micros)
}
fn projected_total_usd_micros(state: &RunProjection) -> Option<i64> {
projected_billing(state).total_usd_micros
}
pub(crate) fn projected_billing(state: &RunProjection) -> BilledTokenCounts {
if let Some(billing) = state
.conclusion

266
lib/crates/fabro-store/src/run_summary_store.rs
View file

@ -1,9 +1,12 @@
use std::collections::HashSet;
use std::collections::{HashMap, HashSet};
use std::fmt::Write as _;
use std::sync::LazyLock;
use chrono::{DateTime, Utc};
use fabro_types::{Run, RunId, RunStatusKind, RunTiming};
use fabro_types::{Run, RunId, RunSize, RunStatusKind, RunTiming};
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;
@ -46,13 +49,20 @@ ON CONFLICT(id) DO UPDATE SET
WHERE excluded.source_last_seq > runs.source_last_seq
";
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
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,
@ -60,7 +70,8 @@ pub enum RunSummarySort {
Size,
}
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, serde::Deserialize)]
#[serde(rename_all = "lowercase")]
pub enum RunSummarySortDirection {
Asc,
#[default]
@ -144,46 +155,84 @@ impl RunSummaryStore {
}
pub(crate) async fn reconcile(&self, entries: &[CachedRunProjection]) -> Result<()> {
let authoritative_ids = entries
.iter()
.map(|entry| entry.run_id.to_string())
.collect::<HashSet<_>>();
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 record = ProjectedRunSummary::from_entry(entry);
upsert_run(&mut transaction, &record).await?;
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 stored_ids = sqlx::query_scalar::<_, String>("SELECT id FROM runs")
.fetch_all(&mut *transaction)
.await?;
for stored_id in stored_ids {
if !authoritative_ids.contains(&stored_id) {
sqlx::query("DELETE FROM runs WHERE id = ?")
.bind(stored_id)
.execute(&mut *transaction)
.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 row = sqlx::query(
r"
SELECT runs.id, runs.summary_json,
(SELECT COUNT(*) FROM runs AS child WHERE child.parent_id = runs.id) AS children_count
FROM runs
WHERE runs.id = ?
",
)
.bind(run_id.to_string())
.fetch_optional(&self.pool)
.await?;
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,
@ -198,12 +247,7 @@ WHERE runs.id = ?
.fetch_one(&mut *transaction)
.await?;
let mut rows_query = QueryBuilder::<Sqlite>::new(
r"
SELECT runs.id, runs.summary_json,
(SELECT COUNT(*) FROM runs AS child WHERE child.parent_id = runs.id) AS children_count
FROM runs",
);
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));
@ -217,10 +261,7 @@ FROM runs",
.iter()
.map(|row| decode_run_row(row, now))
.collect::<Result<Vec<_>>>()?;
let total = u64::try_from(total).map_err(|_| Error::RunSummaryMismatch {
run_id: "<list>".to_string(),
field: "negative total",
})?;
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,
@ -238,6 +279,16 @@ FROM runs",
}
}
/// 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,
@ -263,12 +314,7 @@ impl ProjectedRunSummary {
run.timing = entry.projection.live_run_timing(at);
}
let billing = projected_billing(&entry.projection);
let workflow_name = run
.workflow
.name
.clone()
.or_else(|| run.workflow.graph_name.clone())
.or_else(|| run.workflow.slug.clone());
let workflow_name = run.workflow.display_name().map(str::to_string);
let repository_name = run
.repository
.as_ref()
@ -356,12 +402,13 @@ fn push_filters(builder: &mut QueryBuilder<Sqlite>, query: &RunSummaryListQuery)
match &query.visibility {
RunSummaryVisibility::All => {}
RunSummaryVisibility::Default { include_archived } => {
let not_removing = format!("status <> '{}'", RunStatusKind::Removing);
if *include_archived {
builder.push(
" AND (archived_at_ms IS NOT NULL OR (archived_at_ms IS NULL AND status <> 'removing'))",
);
builder.push(format!(
" AND (archived_at_ms IS NOT NULL OR {not_removing})"
));
} else {
builder.push(" AND archived_at_ms IS NULL AND status <> 'removing'");
builder.push(format!(" AND archived_at_ms IS NULL AND {not_removing}"));
}
}
RunSummaryVisibility::Selected { statuses, archived } => {
@ -391,6 +438,32 @@ fn push_filters(builder: &mut QueryBuilder<Sqlite>, query: &RunSummaryListQuery)
}
}
/// 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,
@ -401,20 +474,7 @@ fn push_order(
match sort {
RunSummarySort::CreatedAt => builder.push("created_at_ms"),
RunSummarySort::UpdatedAt => builder.push("last_event_at_ms"),
RunSummarySort::Status => builder.push(
r"CASE
WHEN archived_at_ms IS NOT NULL THEN 7
WHEN status IN ('submitted', 'pending') THEN 0
WHEN status = 'runnable' THEN 1
WHEN status = 'starting' THEN 2
WHEN status IN ('running', 'paused') THEN 3
WHEN status = 'blocked' THEN 4
WHEN status = 'succeeded' THEN 5
WHEN status IN ('failed', 'dead') THEN 6
WHEN status = 'removing' THEN 8
ELSE 9
END",
),
RunSummarySort::Status => builder.push(STATUS_RANK_CASE_SQL.as_str()),
RunSummarySort::Elapsed => builder
.push("(COALESCE(completed_at_ms, ")
.push_bind(now.timestamp_millis())
@ -423,15 +483,7 @@ fn push_order(
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(
r"CASE
WHEN COALESCE(total_usd_micros, 0) <= 20000000 THEN 0
WHEN total_usd_micros <= 50000000 THEN 1
WHEN total_usd_micros <= 100000000 THEN 2
WHEN total_usd_micros <= 200000000 THEN 3
ELSE 4
END",
),
RunSummarySort::Size => builder.push(SIZE_RANK_CASE_SQL.as_str()),
};
match direction {
RunSummarySortDirection::Asc => builder.push(" ASC"),
@ -455,23 +507,18 @@ fn decode_run_row(row: &SqliteRow, now: DateTime<Utc>) -> Result<Run> {
run_id: run.id.to_string(),
field: "children_count",
})?;
apply_read_overlays(&mut run, now);
overlay_live_wall_time(&mut run, now);
Ok(run)
}
fn apply_read_overlays(run: &mut Run, now: DateTime<Utc>) {
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 = u64::try_from(
now.signed_duration_since(started_at)
.num_milliseconds()
.max(0),
)
.expect("non-negative milliseconds fit in u64");
let wall_time_ms = RunTiming::wall_time_ms_since(started_at, now);
run.timing = Some(
run.timing
.unwrap_or_else(|| RunTiming::wall_only(wall_time_ms))
@ -485,10 +532,11 @@ mod tests {
use chrono::{DateTime, Utc};
use fabro_types::{
AutomationRef, BilledTokenCounts, Conclusion, DiffSummary, Graph, RunDiff, RunId,
RunProjection, RunSize, RunSpec, RunStatus, RunTiming, StageOutcome, SuccessReason,
WorkflowSettings, test_support,
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::{
@ -496,6 +544,7 @@ mod tests {
RunSummaryVisibility,
};
use crate::slate::CachedRunProjection;
use crate::test_util;
fn dt(value: &str) -> DateTime<Utc> {
value.parse().unwrap()
@ -532,12 +581,49 @@ mod tests {
}
async fn store() -> (tempfile::TempDir, RunSummaryStore) {
let directory = tempfile::tempdir().unwrap();
let database = fabro_db::Database::connect(directory.path().join("fabro.sqlite3"))
.await
.unwrap();
database.migrate().await.unwrap();
(directory, RunSummaryStore::new(database.clone_pool()))
test_util::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]

19
lib/crates/fabro-store/src/slate/mod.rs
View file

@ -168,7 +168,7 @@ impl Database {
*run_id,
db,
Arc::clone(&self.projection_cache),
self.run_summary_store(),
Arc::clone(&self.run_summary_store),
)
.await?;
Self::cache_active_run(&mut active_runs, &run_store);
@ -197,7 +197,7 @@ impl Database {
*run_id,
db,
Arc::clone(&self.projection_cache),
self.run_summary_store(),
Arc::clone(&self.run_summary_store),
)
.await?;
Self::cache_active_run(&mut active_runs, &run_store);
@ -221,7 +221,7 @@ impl Database {
*run_id,
db,
Arc::clone(&self.projection_cache),
self.run_summary_store(),
Arc::clone(&self.run_summary_store),
)
.await
}
@ -511,7 +511,7 @@ mod tests {
use object_store::path::Path;
use super::*;
use crate::{EventPayload, keys};
use crate::{EventPayload, keys, test_util};
fn dt(value: &str) -> DateTime<Utc> {
value.parse().unwrap()
@ -560,15 +560,8 @@ mod tests {
}
async fn make_summary_store() -> (tempfile::TempDir, Arc<RunSummaryStore>) {
let directory = tempfile::tempdir().unwrap();
let database = fabro_db::Database::connect(directory.path().join("fabro.sqlite3"))
.await
.unwrap();
database.migrate().await.unwrap();
(
directory,
Arc::new(RunSummaryStore::new(database.clone_pool())),
)
let (directory, store) = test_util::sqlite_summary_store().await;
(directory, Arc::new(store))
}
fn sample_run_spec(label: &str) -> RunSpec {

51
lib/crates/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;
@ -43,7 +43,9 @@ pub(crate) struct RunDatabaseInner {
state_lock: Mutex<()>,
projection_cache: Mutex<EventProjectionCache>,
shared_projection_cache: Arc<RunProjectionCache>,
run_summary_store: Option<Arc<RunSummaryStore>>,
// 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>>>,
recent_events: Mutex<VecDeque<EventEnvelope>>,
recent_event_limit: usize,
event_tx: broadcast::Sender<EventEnvelope>,
@ -54,7 +56,7 @@ impl RunDatabase {
run_id: RunId,
db: Db,
shared_projection_cache: Arc<RunProjectionCache>,
run_summary_store: Option<Arc<RunSummaryStore>>,
run_summary_store: Arc<OnceLock<Arc<RunSummaryStore>>>,
) -> Result<Self> {
Self::build(
run_id,
@ -70,7 +72,7 @@ impl RunDatabase {
run_id: RunId,
db: Db,
shared_projection_cache: Arc<RunProjectionCache>,
run_summary_store: Option<Arc<RunSummaryStore>>,
run_summary_store: Arc<OnceLock<Arc<RunSummaryStore>>>,
) -> Result<Self> {
Self::build(run_id, db, true, shared_projection_cache, run_summary_store).await
}
@ -80,7 +82,7 @@ impl RunDatabase {
db: Db,
read_only: bool,
shared_projection_cache: Arc<RunProjectionCache>,
run_summary_store: Option<Arc<RunSummaryStore>>,
run_summary_store: Arc<OnceLock<Arc<RunSummaryStore>>>,
) -> Result<Self> {
let event_seq =
recover_next_seq(&db, keys::run_events_prefix(&run_id), keys::parse_event_seq).await?;
@ -271,6 +273,8 @@ impl RunDatabase {
)
.await?;
self.cache_event(&event).await?;
// Box::pin keeps append_event_envelope's future small enough for the
// clippy::large_futures budget of its many callers.
Box::pin(self.update_summary_projection_after_append(&event)).await?;
Ok(event)
}
@ -292,31 +296,21 @@ impl RunDatabase {
.await;
entry
}
Ok(None) => {
rebuild => {
self.inner
.shared_projection_cache
.remove(&self.inner.run_id)
.await;
if let Err(rebuild_err) = rebuild {
warn!(
run_id = %self.inner.run_id,
error = %rebuild_err,
"Failed to rebuild run projection cache after append"
);
}
warn!(
run_id = %self.inner.run_id,
error = ?err,
"Failed to update run projection cache after append"
);
return Err(err);
}
Err(rebuild_err) => {
self.inner
.shared_projection_cache
.remove(&self.inner.run_id)
.await;
warn!(
run_id = %self.inner.run_id,
error = %rebuild_err,
"Failed to rebuild run projection cache after append"
);
warn!(
run_id = %self.inner.run_id,
error = ?err,
error = %err,
"Failed to update run projection cache after append"
);
return Err(err);
@ -324,14 +318,11 @@ impl RunDatabase {
}
}
};
if let Some(store) = &self.inner.run_summary_store {
let source_last_seq = cached.last_seq;
let store = Arc::clone(store);
let upsert = Box::pin(async move { store.upsert_projection(&cached).await });
if let Err(err) = upsert.await {
if let Some(store) = self.inner.run_summary_store.get() {
if let Err(err) = store.upsert_projection(&cached).await {
error!(
run_id = %self.inner.run_id,
source_last_seq,
source_last_seq = cached.last_seq,
error = %err,
"Failed to update SQLite run summary after append"
);

10
lib/crates/fabro-store/src/test_util.rs Normal file
View file

@ -0,0 +1,10 @@
use crate::RunSummaryStore;
pub(crate) async fn sqlite_summary_store() -> (tempfile::TempDir, RunSummaryStore) {
let directory = tempfile::tempdir().unwrap();
let database = fabro_db::Database::connect(directory.path().join("fabro.sqlite3"))
.await
.unwrap();
database.migrate().await.unwrap();
(directory, RunSummaryStore::new(database.clone_pool()))
}

7
lib/crates/fabro-types/src/run_projection.rs
View file

@ -616,12 +616,7 @@ impl RunProjection {
#[must_use]
pub fn live_run_timing(&self, now: DateTime<Utc>) -> Option<RunTiming> {
let start = self.start.as_ref()?;
let wall_time_ms = u64::try_from(
now.signed_duration_since(start.start_time)
.num_milliseconds()
.max(0),
)
.expect("non-negative milliseconds fit in u64");
let wall_time_ms = RunTiming::wall_time_ms_since(start.start_time, now);
let active = self
.stages
.values()

34
lib/crates/fabro-types/src/run_summary.rs
View file

@ -101,6 +101,18 @@ pub struct WorkflowRef {
pub edge_count: i64,
}
impl WorkflowRef {
/// Best available human-facing workflow name: explicit name, then graph
/// name, then slug.
#[must_use]
pub fn display_name(&self) -> Option<&str> {
self.name
.as_deref()
.or(self.graph_name.as_deref())
.or(self.slug.as_deref())
}
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct AutomationRef {
pub id: String,
@ -233,15 +245,23 @@ pub enum RunSize {
}
impl RunSize {
/// Inclusive upper bounds in USD micros for each bucket below [`Self::Xl`],
/// ordered smallest to largest. Shared with the SQLite size sort so both
/// stay in step.
pub const BUCKET_MAX_USD_MICROS: [(Self, i64); 4] = [
(Self::Xs, 20_000_000),
(Self::S, 50_000_000),
(Self::M, 100_000_000),
(Self::L, 200_000_000),
];
#[must_use]
pub fn from_total_usd_micros(total_usd_micros: Option<i64>) -> Self {
match total_usd_micros.unwrap_or(0) {
..=20_000_000 => Self::Xs,
20_000_001..=50_000_000 => Self::S,
50_000_001..=100_000_000 => Self::M,
100_000_001..=200_000_000 => Self::L,
_ => Self::Xl,
}
let total = total_usd_micros.unwrap_or(0);
Self::BUCKET_MAX_USD_MICROS
.iter()
.find(|(_, max)| total <= *max)
.map_or(Self::Xl, |(size, _)| *size)
}
}

24
lib/crates/fabro-types/src/status.rs
View file

@ -1,7 +1,7 @@
use std::fmt;
use serde::{Deserialize, Serialize};
use strum::{Display, EnumString, IntoStaticStr};
use strum::{Display, EnumString, IntoStaticStr, VariantArray};
#[derive(
Debug,
@ -15,6 +15,7 @@ use strum::{Display, EnumString, IntoStaticStr};
Display,
EnumString,
IntoStaticStr,
VariantArray,
)]
#[serde(rename_all = "snake_case")]
#[strum(serialize_all = "snake_case")]
@ -32,6 +33,27 @@ pub enum RunStatusKind {
Dead,
}
impl RunStatusKind {
/// Position of this status in the run-board column order. Ranks mirror
/// the API `BoardColumn` enum: pending 0, runnable 1, initializing 2,
/// running 3, blocked 4, succeeded 5, failed 6, archived 7, removing 8.
/// Rank 7 is reserved for archived runs, which is an overlay flag rather
/// than a status.
#[must_use]
pub fn board_rank(self) -> u8 {
match self {
Self::Submitted | Self::Pending => 0,
Self::Runnable => 1,
Self::Starting => 2,
Self::Running | Self::Paused => 3,
Self::Blocked => 4,
Self::Succeeded => 5,
Self::Failed | Self::Dead => 6,
Self::Removing => 8,
}
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(tag = "kind", rename_all = "snake_case")]
pub enum RunStatus {

8
lib/crates/fabro-types/src/timing.rs
View file

@ -15,6 +15,7 @@
//! child branches carry their own work timing; run-level active time sums work
//! across stage visits and can exceed run wall time when work runs in parallel.
use chrono::{DateTime, Utc};
use serde::{Deserialize, Serialize};
/// Timing breakdown for one stage visit.
@ -135,6 +136,13 @@ impl RunTiming {
..self
}
}
/// Milliseconds elapsed from `start` to `now`, clamped at zero.
#[must_use]
pub fn wall_time_ms_since(start: DateTime<Utc>, now: DateTime<Utc>) -> u64 {
u64::try_from(now.signed_duration_since(start).num_milliseconds().max(0))
.expect("non-negative milliseconds fit in u64")
}
}
impl From<StageTiming> for RunTiming {

8
lib/crates/fabro-workflow/src/run_lookup.rs
View file

@ -83,13 +83,7 @@ impl RunInfo {
pub fn workflow_display_name(&self) -> String {
self.summary.as_ref().map_or_else(
|| "[no run spec]".to_string(),
|_| {
self.workflow_name()
.or_else(|| self.workflow_graph_name())
.or_else(|| self.workflow_slug())
.unwrap_or("-")
.to_string()
},
|summary| summary.workflow.display_name().unwrap_or("-").to_string(),
)
}