diff --git a/lib/apps/fabro-server/migrations/2026082301_sqlite_blob_activation.rs b/lib/apps/fabro-server/migrations/2026082301_sqlite_blob_activation.rs index afd616500..494da96ac 100644 --- a/lib/apps/fabro-server/migrations/2026082301_sqlite_blob_activation.rs +++ b/lib/apps/fabro-server/migrations/2026082301_sqlite_blob_activation.rs @@ -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_summary_store = Arc::new(fabro_store::RunSummaryStore::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_summary_store, )); let inventory = store diff --git a/lib/apps/fabro-server/src/server.rs b/lib/apps/fabro-server/src/server.rs index 9d60b694c..95d9e95e1 100644 --- a/lib/apps/fabro-server/src/server.rs +++ b/lib/apps/fabro-server/src/server.rs @@ -2447,8 +2447,7 @@ pub(crate) fn build_app_state(config: AppStateConfig) -> anyhow::Result, + }, + #[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}")] diff --git a/lib/components/fabro-store/src/legacy_blob_import.rs b/lib/components/fabro-store/src/legacy_blob_import.rs index cc4739db7..2894f12de 100644 --- a/lib/components/fabro-store/src/legacy_blob_import.rs +++ b/lib/components/fabro-store/src/legacy_blob_import.rs @@ -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 = std::result::Result>; @@ -1153,6 +1153,7 @@ mod tests { Duration::from_millis(1), None, Arc::clone(&target), + store_test_support::test_run_summary_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_summary_store(), ); let mut connection = pool.acquire().await?; diff --git a/lib/components/fabro-store/src/run_summary_store.rs b/lib/components/fabro-store/src/run_summary_store.rs index a84283f83..56ea142a4 100644 --- a/lib/components/fabro-store/src/run_summary_store.rs +++ b/lib/components/fabro-store/src/run_summary_store.rs @@ -3,14 +3,30 @@ 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 fabro_types::{ + BilledTokenCounts, EventEnvelope, Run, RunEvent, RunId, RunSize, RunStatusKind, RunTiming, + SessionId, StageId, timing, +}; +use sqlx::query::Query; +use sqlx::sqlite::{SqliteArguments, 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}; +use crate::{Error, EventPayload, Result, keys}; + +const INSERT_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 ( + ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ? +) +"; const UPSERT_RUN_SQL: &str = r" INSERT INTO runs ( @@ -49,6 +65,42 @@ ON CONFLICT(id) DO UPDATE SET WHERE excluded.source_last_seq > runs.source_last_seq "; +const UPDATE_RUN_SQL: &str = r" +UPDATE runs SET + 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 = ? +WHERE id = ? AND source_last_seq = ? +"; + +const SELECT_EVENT_COLUMNS: &str = + "SELECT run_id, seq, event_name, node_id, stage_id, session_id, event_json FROM run_events"; + +const INSERT_EVENT_SQL: &str = r" +INSERT INTO run_events (run_id, seq, event_name, node_id, stage_id, session_id, event_json) +VALUES (?, ?, ?, ?, ?, ?, ?) +"; + 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 @@ -148,9 +200,9 @@ impl RunSummaryStore { } pub(crate) async fn upsert_projection(&self, entry: &CachedRunProjection) -> Result<()> { - let record = ProjectedRunSummary::from_entry(entry); + let record = PreparedRunSummary::from_entry(entry); let mut connection = self.pool.acquire().await?; - upsert_run(&mut connection, &record).await?; + upsert_run_on_connection(&mut connection, &record).await?; Ok(()) } @@ -178,7 +230,8 @@ impl RunSummaryStore { if up_to_date { continue; } - upsert_run(&mut transaction, &ProjectedRunSummary::from_entry(entry)).await?; + upsert_run_on_connection(&mut transaction, &PreparedRunSummary::from_entry(entry)) + .await?; } let stale_ids = stored_seqs @@ -284,6 +337,188 @@ FROM runs", } } +#[cfg_attr( + not(test), + expect( + dead_code, + reason = "the atomic SQL run path stays inactive until the authority cutover" + ) +)] +impl RunSummaryStore { + pub(crate) async fn insert_first_event_on_connection( + connection: &mut SqliteConnection, + entry: &CachedRunProjection, + payload: &EventPayload, + ) -> Result { + let record = PreparedRunSummary::from_entry(entry); + ensure_entry_identity(entry, &record, 1)?; + ensure_prepared_head(&record, 1)?; + let envelope = validate_event_for_record(&record, payload, 1)?; + if envelope.event.event_name() != "run.created" { + return Err(run_event_mismatch(&record.run.id, 1, "event_name")); + } + + insert_run_on_connection(connection, &record).await?; + insert_event_on_connection(connection, &record, payload, &envelope).await?; + Ok(envelope) + } + + pub(crate) async fn append_event_on_connection( + connection: &mut SqliteConnection, + expected_last_seq: u32, + entry: &CachedRunProjection, + payload: &EventPayload, + ) -> Result { + let next_seq = next_event_seq_after(expected_last_seq)?; + let record = PreparedRunSummary::from_entry(entry); + ensure_entry_identity(entry, &record, next_seq)?; + ensure_prepared_head(&record, next_seq)?; + let envelope = validate_event_for_record(&record, payload, next_seq)?; + + update_run_on_connection(connection, &record, expected_last_seq).await?; + insert_event_on_connection(connection, &record, payload, &envelope).await?; + Ok(envelope) + } + + pub(crate) async fn list_events_on_connection( + connection: &mut SqliteConnection, + run_id: &RunId, + ) -> Result> { + let mut query = QueryBuilder::::new(SELECT_EVENT_COLUMNS); + query + .push(" WHERE run_id = ") + .push_bind(run_id.to_string()) + .push(" ORDER BY seq ASC"); + let expected_last_seq = select_run_head(&mut *connection, run_id) + .await? + .ok_or_else(|| Error::RunNotFound(run_id.to_string()))?; + let rows = query.build().fetch_all(&mut *connection).await?; + let actual_last_seq = rows + .last() + .map(|row| row.try_get::("seq")) + .transpose()? + .and_then(stored_seq); + if actual_last_seq != Some(expected_last_seq) { + return Err(Error::RunHeadMismatch { + run_id: run_id.to_string(), + expected_last_seq, + actual_last_seq, + }); + } + decode_event_rows(&rows, run_id) + } + + pub(crate) async fn list_events_from_with_limit_on_connection( + connection: &mut SqliteConnection, + run_id: &RunId, + start_seq: u32, + limit: usize, + ) -> Result> { + let mut query = QueryBuilder::::new(SELECT_EVENT_COLUMNS); + query + .push(" WHERE run_id = ") + .push_bind(run_id.to_string()) + .push(" AND seq >= ") + .push_bind(i64::from(start_seq)) + .push(" ORDER BY seq ASC LIMIT ") + .push_bind(sql_limit(limit)); + let rows = query.build().fetch_all(&mut *connection).await?; + decode_event_rows(&rows, run_id) + } + + pub(crate) async fn list_events_before_with_limit_on_connection( + connection: &mut SqliteConnection, + run_id: &RunId, + before_seq: Option, + limit: usize, + ) -> Result> { + let mut query = QueryBuilder::::new(SELECT_EVENT_COLUMNS); + query.push(" WHERE run_id = ").push_bind(run_id.to_string()); + if let Some(before_seq) = before_seq { + query.push(" AND seq < ").push_bind(i64::from(before_seq)); + } + query + .push(" ORDER BY seq DESC LIMIT ") + .push_bind(sql_limit(limit)); + let rows = query.build().fetch_all(&mut *connection).await?; + decode_event_rows(&rows, run_id) + } + + pub(crate) async fn get_event_on_connection( + connection: &mut SqliteConnection, + run_id: &RunId, + seq: u32, + ) -> Result> { + let mut query = QueryBuilder::::new(SELECT_EVENT_COLUMNS); + query + .push(" WHERE run_id = ") + .push_bind(run_id.to_string()) + .push(" AND seq = ") + .push_bind(i64::from(seq)); + let row = query.build().fetch_optional(&mut *connection).await?; + row.as_ref() + .map(|row| decode_event_row(row, run_id, &run_id.to_string())) + .transpose() + } + + pub(crate) async fn list_events_for_stage_from_with_limit_on_connection( + connection: &mut SqliteConnection, + run_id: &RunId, + stage_id: &StageId, + start_seq: u32, + limit: usize, + ) -> Result> { + // Legacy rows for a first visit carry only `node_id`. Query them as a + // second `UNION ALL` arm instead of an `OR` so each arm can use its + // own partial index rather than scanning the run's primary key range. + let mut query = QueryBuilder::::new(SELECT_EVENT_COLUMNS); + query + .push(" WHERE run_id = ") + .push_bind(run_id.to_string()) + .push(" AND seq >= ") + .push_bind(i64::from(start_seq)) + .push(" AND stage_id = ") + .push_bind(stage_id.to_string()); + if stage_id.visit() == 1 { + query + .push(" UNION ALL ") + .push(SELECT_EVENT_COLUMNS) + .push(" WHERE run_id = ") + .push_bind(run_id.to_string()) + .push(" AND seq >= ") + .push_bind(i64::from(start_seq)) + .push(" AND stage_id IS NULL AND node_id = ") + .push_bind(stage_id.node_id().to_string()); + } + query + .push(" ORDER BY seq ASC LIMIT ") + .push_bind(sql_limit(limit)); + let rows = query.build().fetch_all(&mut *connection).await?; + decode_event_rows(&rows, run_id) + } + + pub(crate) async fn list_events_for_session_from_with_limit_on_connection( + connection: &mut SqliteConnection, + run_id: &RunId, + session_id: &SessionId, + start_seq: u32, + limit: usize, + ) -> Result> { + let mut query = QueryBuilder::::new(SELECT_EVENT_COLUMNS); + query + .push(" WHERE run_id = ") + .push_bind(run_id.to_string()) + .push(" AND seq >= ") + .push_bind(i64::from(start_seq)) + .push(" AND session_id = ") + .push_bind(session_id.to_string()) + .push(" AND event_name GLOB 'run.session.*' ORDER BY seq ASC LIMIT ") + .push_bind(sql_limit(limit)); + let rows = query.build().fetch_all(&mut *connection).await?; + decode_event_rows(&rows, run_id) + } +} + /// Identity fields of a stored run summary, cheap to list for selector /// resolution. #[derive(Debug, Clone)] @@ -295,7 +530,7 @@ pub struct RunSummaryIdentity { } #[derive(Debug)] -struct ProjectedRunSummary { +struct PreparedRunSummary { run: Run, last_seq: u32, workflow_name: Option, @@ -308,7 +543,7 @@ struct ProjectedRunSummary { total_usd_micros: Option, } -impl ProjectedRunSummary { +impl PreparedRunSummary { fn from_entry(entry: &CachedRunProjection) -> Self { let mut run = entry.summary.clone(); if run.timing.is_none() { @@ -340,6 +575,149 @@ impl ProjectedRunSummary { } } +fn ensure_entry_identity( + entry: &CachedRunProjection, + record: &PreparedRunSummary, + seq: u32, +) -> Result<()> { + if entry.run_id != record.run.id || entry.projection.spec.run_id != entry.run_id { + return Err(run_event_mismatch(&entry.run_id, seq, "run_id")); + } + Ok(()) +} + +fn ensure_prepared_head(record: &PreparedRunSummary, expected_last_seq: u32) -> Result<()> { + if record.last_seq != expected_last_seq { + return Err(Error::RunHeadMismatch { + run_id: record.run.id.to_string(), + expected_last_seq, + actual_last_seq: Some(record.last_seq), + }); + } + Ok(()) +} + +fn validate_event_for_record( + record: &PreparedRunSummary, + payload: &EventPayload, + seq: u32, +) -> Result { + payload.validate(&record.run.id)?; + let event = RunEvent::try_from(payload)?; + if event.run_id != record.run.id { + return Err(run_event_mismatch(&record.run.id, seq, "run_id")); + } + Ok(EventEnvelope { seq, event }) +} + +fn run_event_mismatch(run_id: &RunId, seq: u32, field: &'static str) -> Error { + Error::RunEventMismatch { + run_id: run_id.to_string(), + seq, + field, + } +} + +async fn insert_event_on_connection( + connection: &mut SqliteConnection, + record: &PreparedRunSummary, + payload: &EventPayload, + envelope: &EventEnvelope, +) -> Result<()> { + let event_json = serde_json::to_string(payload)?; + sqlx::query(INSERT_EVENT_SQL) + .bind(record.run.id.to_string()) + .bind(i64::from(envelope.seq)) + .bind(envelope.event.event_name()) + .bind(envelope.event.node_id.as_deref()) + .bind(envelope.event.stage_id.as_ref().map(ToString::to_string)) + .bind(envelope.event.session_id.as_deref()) + .bind(event_json) + .execute(connection) + .await?; + Ok(()) +} + +fn sql_limit(limit: usize) -> i64 { + i64::try_from(limit.saturating_add(1)).unwrap_or(i64::MAX) +} + +fn next_event_seq_after(last_seq: u32) -> Result { + last_seq + .checked_add(1) + .filter(|seq| *seq <= keys::MAX_EVENT_SEQ) + .ok_or(Error::EventSequenceExhausted { + max_seq: keys::MAX_EVENT_SEQ, + }) +} + +/// Decodes a stored sequence column, rejecting anything outside the valid +/// `1..=MAX_EVENT_SEQ` range. +fn stored_seq(value: i64) -> Option { + u32::try_from(value) + .ok() + .filter(|seq| (1..=keys::MAX_EVENT_SEQ).contains(seq)) +} + +fn decode_event_rows(rows: &[SqliteRow], run_id: &RunId) -> Result> { + let run_id_text = run_id.to_string(); + rows.iter() + .map(|row| decode_event_row(row, run_id, &run_id_text)) + .collect() +} + +fn decode_event_row( + row: &SqliteRow, + expected_run_id: &RunId, + expected_run_id_text: &str, +) -> Result { + let stored_run_id: String = row.try_get("run_id")?; + let raw_seq: i64 = row.try_get("seq")?; + let seq = stored_seq(raw_seq).ok_or_else(|| run_event_mismatch(expected_run_id, 0, "seq"))?; + if stored_run_id != expected_run_id_text { + return Err(run_event_mismatch(expected_run_id, seq, "run_id")); + } + + let event_json: String = row.try_get("event_json")?; + let payload: EventPayload = serde_json::from_str(&event_json)?; + let event = RunEvent::try_from(&payload)?; + if event.run_id != *expected_run_id { + return Err(run_event_mismatch(expected_run_id, seq, "run_id")); + } + + let stored_event_name: String = row.try_get("event_name")?; + if stored_event_name != event.event_name() { + return Err(run_event_mismatch(expected_run_id, seq, "event_name")); + } + let stored_node_id: Option = row.try_get("node_id")?; + if stored_node_id.as_deref() != event.node_id.as_deref() { + return Err(run_event_mismatch(expected_run_id, seq, "node_id")); + } + let stored_stage_id: Option = row.try_get("stage_id")?; + let decoded_stage_id = event.stage_id.as_ref().map(ToString::to_string); + if stored_stage_id != decoded_stage_id { + return Err(run_event_mismatch(expected_run_id, seq, "stage_id")); + } + let stored_session_id: Option = row.try_get("session_id")?; + if stored_session_id.as_deref() != event.session_id.as_deref() { + return Err(run_event_mismatch(expected_run_id, seq, "session_id")); + } + + Ok(EventEnvelope { seq, event }) +} + +async fn select_run_head(connection: &mut SqliteConnection, run_id: &RunId) -> Result> { + let stored: Option = sqlx::query_scalar("SELECT source_last_seq FROM runs WHERE id = ?") + .bind(run_id.to_string()) + .fetch_optional(connection) + .await?; + stored + .map(|value| { + stored_seq(value).ok_or_else(|| run_event_mismatch(run_id, 0, "source_last_seq")) + }) + .transpose() +} + /// 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 @@ -367,12 +745,42 @@ fn normalize_billing_for_read_model(mut billing: BilledTokenCounts) -> BilledTok billing } -async fn upsert_run(connection: &mut SqliteConnection, record: &ProjectedRunSummary) -> Result<()> { +async fn upsert_run_on_connection( + connection: &mut SqliteConnection, + record: &PreparedRunSummary, +) -> Result<()> { + write_insert_shaped_run(connection, record, UPSERT_RUN_SQL).await +} + +async fn insert_run_on_connection( + connection: &mut SqliteConnection, + record: &PreparedRunSummary, +) -> Result<()> { + write_insert_shaped_run(connection, record, INSERT_RUN_SQL).await +} + +async fn write_insert_shaped_run( + connection: &mut SqliteConnection, + record: &PreparedRunSummary, + sql: &'static str, +) -> Result<()> { + bind_run_columns(sqlx::query(sql).bind(record.run.id.to_string()), record)? + .execute(connection) + .await?; + Ok(()) +} + +/// Binds the `runs` columns shared by the insert, upsert, and update +/// statements, in the positional order those statements declare them +/// (`source_last_seq` through `summary_json`). +fn bind_run_columns<'q>( + query: Query<'q, Sqlite, SqliteArguments>, + record: &'q PreparedRunSummary, +) -> 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()) + Ok(query .bind(i64::from(record.last_seq)) .bind(run.timestamps.created_at.timestamp_millis()) .bind( @@ -412,9 +820,27 @@ async fn upsert_run(connection: &mut SqliteConnection, record: &ProjectedRunSumm .bind(record.cache_read_tokens) .bind(record.cache_write_tokens) .bind(record.total_usd_micros) - .bind(summary_json) - .execute(connection) + .bind(summary_json)) +} + +async fn update_run_on_connection( + connection: &mut SqliteConnection, + record: &PreparedRunSummary, + expected_last_seq: u32, +) -> Result<()> { + let run = &record.run; + let result = bind_run_columns(sqlx::query(UPDATE_RUN_SQL), record)? + .bind(run.id.to_string()) + .bind(i64::from(expected_last_seq)) + .execute(&mut *connection) .await?; + if result.rows_affected() == 0 { + return Err(Error::RunHeadMismatch { + run_id: run.id.to_string(), + expected_last_seq, + actual_last_seq: select_run_head(connection, &run.id).await?, + }); + } Ok(()) } @@ -564,19 +990,20 @@ mod tests { 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, + AutomationRef, BilledTokenCounts, BlockedReason, Conclusion, DiffSummary, EventEnvelope, + FailureReason, Graph, PendingReason, RunDiff, RunId, RunProjection, RunSize, RunSpec, + RunStatus, RunStatusKind, RunTiming, SessionId, StageId, StageOutcome, SuccessReason, + WorkflowSettings, test_support, }; use strum::VariantArray as _; use ulid::Ulid; use super::{ - RunSummaryListQuery, RunSummarySort, RunSummarySortDirection, RunSummaryStore, - RunSummaryVisibility, + INSERT_EVENT_SQL, RunSummaryListQuery, RunSummarySort, RunSummarySortDirection, + RunSummaryStore, RunSummaryVisibility, decode_event_row, }; use crate::slate::CachedRunProjection; - use crate::test_support as store_test_support; + use crate::{Error, EventPayload, test_support as store_test_support}; fn dt(value: &str) -> DateTime { value.parse().unwrap() @@ -616,7 +1043,77 @@ mod tests { } async fn store() -> (tempfile::TempDir, RunSummaryStore) { - store_test_support::sqlite_summary_store().await + store_test_support::sqlite_run_summary_store().await + } + + fn sql_event_payload( + run_id: &RunId, + event: &str, + node_id: Option<&str>, + stage_id: Option<&StageId>, + session_id: Option<&SessionId>, + properties: serde_json::Value, + ) -> EventPayload { + let mut value = serde_json::json!({ + "id": format!("evt-{event}"), + "ts": "2026-08-27T12:00:00Z", + "run_id": run_id.to_string(), + "event": event, + }); + let object = value.as_object_mut().unwrap(); + object.insert("properties".to_string(), properties); + if let Some(node_id) = node_id { + object.insert("node_id".to_string(), node_id.into()); + } + if let Some(stage_id) = stage_id { + object.insert("stage_id".to_string(), stage_id.to_string().into()); + } + if let Some(session_id) = session_id { + object.insert("session_id".to_string(), session_id.to_string().into()); + } + EventPayload::new(value, run_id).unwrap() + } + + fn seqs(events: &[EventEnvelope]) -> Vec { + events.iter().map(|event| event.seq).collect() + } + + fn created_payload(run_id: &RunId) -> EventPayload { + sql_event_payload( + run_id, + "run.created", + None, + None, + None, + serde_json::json!({ + "title": "created", + "settings": WorkflowSettings::default(), + "graph": Graph::new("test"), + "workflow_slug": "test-workflow", + "labels": {}, + "provenance": test_support::test_run_provenance(), + }), + ) + } + + async fn seed_sql_event( + store: &RunSummaryStore, + run_id: &RunId, + seq: u32, + payload: &EventPayload, + ) { + let event = fabro_types::RunEvent::try_from(payload).unwrap(); + sqlx::query(INSERT_EVENT_SQL) + .bind(run_id.to_string()) + .bind(i64::from(seq)) + .bind(event.event_name()) + .bind(event.node_id) + .bind(event.stage_id.map(|stage_id| stage_id.to_string())) + .bind(event.session_id) + .bind(serde_json::to_string(payload).unwrap()) + .execute(&store.pool) + .await + .unwrap(); } fn sample_status(kind: RunStatusKind) -> RunStatus { @@ -643,6 +1140,500 @@ mod tests { } } + #[tokio::test] + async fn sql_run_transitions_are_atomic_and_guard_the_current_head() { + let (_directory, store) = store().await; + let created_at = dt("2026-08-27T12:00:00Z"); + let id = run_id(created_at.timestamp_millis().cast_unsigned(), 1); + let first = entry(projection(id, "created", created_at), 1); + let first_payload = created_payload(&id); + + let mut transaction = store.pool.begin().await.unwrap(); + let first_envelope = RunSummaryStore::insert_first_event_on_connection( + &mut transaction, + &first, + &first_payload, + ) + .await + .unwrap(); + transaction.commit().await.unwrap(); + assert_eq!(first_envelope.seq, 1); + + let stored_first: (i64, String) = + sqlx::query_as("SELECT source_last_seq, event_json FROM runs JOIN run_events ON run_events.run_id = runs.id WHERE runs.id = ? AND run_events.seq = 1") + .bind(id.to_string()) + .fetch_one(&store.pool) + .await + .unwrap(); + assert_eq!(stored_first.0, 1); + assert_eq!( + stored_first.1, + serde_json::to_string(&first_payload).unwrap() + ); + + let mut duplicate_first = store.pool.begin().await.unwrap(); + assert!( + RunSummaryStore::insert_first_event_on_connection( + &mut duplicate_first, + &first, + &first_payload, + ) + .await + .is_err() + ); + duplicate_first.rollback().await.unwrap(); + + let rolled_back_id = run_id(created_at.timestamp_millis().cast_unsigned() + 1, 2); + let rolled_back = entry(projection(rolled_back_id, "rollback", created_at), 1); + let mut transaction = store.pool.begin().await.unwrap(); + RunSummaryStore::insert_first_event_on_connection( + &mut transaction, + &rolled_back, + &created_payload(&rolled_back_id), + ) + .await + .unwrap(); + transaction.rollback().await.unwrap(); + let rolled_back_rows: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM runs WHERE id = ?") + .bind(rolled_back_id.to_string()) + .fetch_one(&store.pool) + .await + .unwrap(); + assert_eq!(rolled_back_rows, 0); + + let invalid_id = run_id(created_at.timestamp_millis().cast_unsigned() + 2, 3); + let invalid = entry(projection(invalid_id, "invalid", created_at), 1); + let invalid_payload = sql_event_payload( + &invalid_id, + "run.title.updated", + None, + None, + None, + serde_json::json!({ "title": "too early" }), + ); + let mut transaction = store.pool.begin().await.unwrap(); + let error = RunSummaryStore::insert_first_event_on_connection( + &mut transaction, + &invalid, + &invalid_payload, + ) + .await + .unwrap_err(); + assert!(matches!(error, Error::RunEventMismatch { + field: "event_name", + .. + })); + transaction.rollback().await.unwrap(); + + let rejected_id = run_id(created_at.timestamp_millis().cast_unsigned() + 3, 4); + sqlx::query( + "CREATE TRIGGER reject_test_event BEFORE INSERT ON run_events BEGIN SELECT RAISE(ABORT, 'rejected'); END", + ) + .execute(&store.pool) + .await + .unwrap(); + let rejected = entry(projection(rejected_id, "rejected", created_at), 1); + let mut transaction = store.pool.begin().await.unwrap(); + assert!( + RunSummaryStore::insert_first_event_on_connection( + &mut transaction, + &rejected, + &created_payload(&rejected_id), + ) + .await + .is_err() + ); + transaction.rollback().await.unwrap(); + let rejected_rows: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM runs WHERE id = ?") + .bind(rejected_id.to_string()) + .fetch_one(&store.pool) + .await + .unwrap(); + assert_eq!(rejected_rows, 0); + sqlx::query("DROP TRIGGER reject_test_event") + .execute(&store.pool) + .await + .unwrap(); + + let mut updated_projection = projection(id, "updated", created_at); + updated_projection.last_event_at = created_at + chrono::Duration::seconds(1); + updated_projection.status = RunStatus::Running; + let second = entry(updated_projection, 2); + let second_payload = sql_event_payload( + &id, + "run.title.updated", + None, + None, + None, + serde_json::json!({ "title": "updated" }), + ); + let mut transaction = store.pool.begin().await.unwrap(); + RunSummaryStore::append_event_on_connection(&mut transaction, 1, &second, &second_payload) + .await + .unwrap(); + transaction.commit().await.unwrap(); + let updated_row: (i64, String, String, i64) = sqlx::query_as( + "SELECT source_last_seq, status, title, last_event_at_ms FROM runs WHERE id = ?", + ) + .bind(id.to_string()) + .fetch_one(&store.pool) + .await + .unwrap(); + assert_eq!( + updated_row, + ( + 2, + "running".to_string(), + "updated".to_string(), + (created_at + chrono::Duration::seconds(1)).timestamp_millis(), + ) + ); + + let mut stale = store.pool.begin().await.unwrap(); + let stale_error = + RunSummaryStore::append_event_on_connection(&mut stale, 1, &second, &second_payload) + .await + .unwrap_err(); + assert!(matches!(stale_error, Error::RunHeadMismatch { + expected_last_seq: 1, + actual_last_seq: Some(2), + .. + })); + stale.rollback().await.unwrap(); + + let third = entry(projection(id, "third", created_at), 3); + let third_payload = sql_event_payload( + &id, + "future.event", + None, + None, + None, + serde_json::json!({ "preserved": true }), + ); + let mut mismatched_value = third_payload.as_value().clone(); + mismatched_value["run_id"] = rolled_back_id.to_string().into(); + let mismatched_payload: EventPayload = serde_json::from_value(mismatched_value).unwrap(); + let mut invalid = store.pool.begin().await.unwrap(); + assert!( + RunSummaryStore::append_event_on_connection( + &mut invalid, + 2, + &third, + &mismatched_payload, + ) + .await + .is_err() + ); + invalid.rollback().await.unwrap(); + let mut rollback = store.pool.begin().await.unwrap(); + RunSummaryStore::append_event_on_connection(&mut rollback, 2, &third, &third_payload) + .await + .unwrap(); + rollback.rollback().await.unwrap(); + + seed_sql_event(&store, &id, 3, &third_payload).await; + let mut duplicate = store.pool.begin().await.unwrap(); + assert!( + RunSummaryStore::append_event_on_connection(&mut duplicate, 2, &third, &third_payload,) + .await + .is_err() + ); + duplicate.rollback().await.unwrap(); + let head_after_failures: i64 = + sqlx::query_scalar("SELECT source_last_seq FROM runs WHERE id = ?") + .bind(id.to_string()) + .fetch_one(&store.pool) + .await + .unwrap(); + assert_eq!(head_after_failures, 2); + + sqlx::query("DELETE FROM run_events WHERE run_id = ? AND seq = 3") + .bind(id.to_string()) + .execute(&store.pool) + .await + .unwrap(); + let mut transaction = store.pool.begin().await.unwrap(); + RunSummaryStore::append_event_on_connection(&mut transaction, 2, &third, &third_payload) + .await + .unwrap(); + transaction.commit().await.unwrap(); + let sequences = sqlx::query_scalar::<_, i64>( + "SELECT seq FROM run_events WHERE run_id = ? ORDER BY seq", + ) + .bind(id.to_string()) + .fetch_all(&store.pool) + .await + .unwrap(); + assert_eq!(sequences, vec![1, 2, 3]); + } + + #[tokio::test] + async fn sql_run_reads_preserve_paging_filters_json_and_legacy_gaps() { + let (_directory, store) = store().await; + let created_at = dt("2026-08-27T12:00:00Z"); + let id = run_id(created_at.timestamp_millis().cast_unsigned(), 11); + let first = entry(projection(id, "created", created_at), 1); + let mut transaction = store.pool.begin().await.unwrap(); + RunSummaryStore::insert_first_event_on_connection( + &mut transaction, + &first, + &created_payload(&id), + ) + .await + .unwrap(); + transaction.commit().await.unwrap(); + + let visit_one = StageId::new("work", 1); + let visit_two = StageId::new("work", 2); + let session_id = SessionId::new(); + let payloads = [ + sql_event_payload( + &id, + "future.stage", + Some("work"), + Some(&visit_one), + None, + serde_json::json!({ "kind": "visit-one" }), + ), + sql_event_payload( + &id, + "future.stage", + Some("work"), + Some(&visit_two), + None, + serde_json::json!({ "kind": "visit-two" }), + ), + sql_event_payload( + &id, + "future.legacy", + Some("work"), + None, + None, + serde_json::json!({ "kind": "legacy" }), + ), + sql_event_payload( + &id, + "run.session.future", + None, + None, + Some(&session_id), + serde_json::json!({ "redacted": "[REDACTED]" }), + ), + sql_event_payload( + &id, + "future.non_session", + None, + None, + Some(&session_id), + serde_json::json!({ "same_session": true }), + ), + ]; + for (index, payload) in payloads.iter().enumerate() { + seed_sql_event(&store, &id, u32::try_from(index).unwrap() + 2, payload).await; + } + sqlx::query("UPDATE runs SET source_last_seq = 6 WHERE id = ?") + .bind(id.to_string()) + .execute(&store.pool) + .await + .unwrap(); + + let mut connection = store.pool.acquire().await.unwrap(); + let all = RunSummaryStore::list_events_on_connection(&mut connection, &id) + .await + .unwrap(); + assert_eq!(seqs(&all), vec![1, 2, 3, 4, 5, 6]); + let forward = + RunSummaryStore::list_events_from_with_limit_on_connection(&mut connection, &id, 2, 2) + .await + .unwrap(); + assert_eq!(seqs(&forward), vec![2, 3, 4]); + let reverse = RunSummaryStore::list_events_before_with_limit_on_connection( + &mut connection, + &id, + Some(5), + 2, + ) + .await + .unwrap(); + assert_eq!(seqs(&reverse), vec![4, 3, 2]); + let exact = RunSummaryStore::get_event_on_connection(&mut connection, &id, 5) + .await + .unwrap() + .unwrap(); + assert_eq!(exact.event.event_name(), "run.session.future"); + assert_eq!( + serde_json::to_value(&exact.event).unwrap(), + payloads[3].as_value().clone() + ); + + let visit_one_events = + RunSummaryStore::list_events_for_stage_from_with_limit_on_connection( + &mut connection, + &id, + &visit_one, + 1, + 10, + ) + .await + .unwrap(); + assert_eq!(seqs(&visit_one_events), vec![2, 4]); + let visit_two_events = + RunSummaryStore::list_events_for_stage_from_with_limit_on_connection( + &mut connection, + &id, + &visit_two, + 1, + 10, + ) + .await + .unwrap(); + assert_eq!(seqs(&visit_two_events), vec![3]); + let session_events = + RunSummaryStore::list_events_for_session_from_with_limit_on_connection( + &mut connection, + &id, + &session_id, + 1, + 10, + ) + .await + .unwrap(); + assert_eq!(seqs(&session_events), vec![5]); + drop(connection); + + sqlx::query("DELETE FROM run_events WHERE run_id = ? AND seq = 3") + .bind(id.to_string()) + .execute(&store.pool) + .await + .unwrap(); + let mut connection = store.pool.acquire().await.unwrap(); + let gapped = RunSummaryStore::list_events_on_connection(&mut connection, &id) + .await + .unwrap(); + assert_eq!(gapped.last().unwrap().seq, 6); + assert_eq!(gapped.len(), 5); + } + + #[tokio::test] + async fn sql_run_reads_reject_extracted_column_and_identity_corruption() { + let (_directory, store) = store().await; + let created_at = dt("2026-08-27T12:00:00Z"); + let id = run_id(created_at.timestamp_millis().cast_unsigned(), 21); + let other_id = run_id(created_at.timestamp_millis().cast_unsigned() + 1, 22); + store + .upsert_projection(&entry(projection(id, "one", created_at), 2)) + .await + .unwrap(); + store + .upsert_projection(&entry(projection(other_id, "two", created_at), 1)) + .await + .unwrap(); + let stage_id = StageId::new("work", 1); + let session_id = SessionId::new(); + let payload = sql_event_payload( + &id, + "run.session.future", + Some("work"), + Some(&stage_id), + Some(&session_id), + serde_json::json!({ "safe": true }), + ); + seed_sql_event(&store, &id, 2, &payload).await; + + for (sql, field) in [ + ( + "UPDATE run_events SET event_name = 'wrong' WHERE run_id = ? AND seq = 2", + "event_name", + ), + ( + "UPDATE run_events SET node_id = 'wrong' WHERE run_id = ? AND seq = 2", + "node_id", + ), + ( + "UPDATE run_events SET stage_id = 'wrong@1' WHERE run_id = ? AND seq = 2", + "stage_id", + ), + ( + "UPDATE run_events SET session_id = 'wrong' WHERE run_id = ? AND seq = 2", + "session_id", + ), + ] { + sqlx::query(sql) + .bind(id.to_string()) + .execute(&store.pool) + .await + .unwrap(); + let mut connection = store.pool.acquire().await.unwrap(); + let error = RunSummaryStore::get_event_on_connection(&mut connection, &id, 2) + .await + .unwrap_err(); + assert!(matches!( + &error, + Error::RunEventMismatch { + field: mismatch_field, + .. + } if *mismatch_field == field + )); + assert!(!error.to_string().contains(&session_id.to_string())); + drop(connection); + seed_sql_event_restore(&store, &id, 2, &payload).await; + } + + let mut wrong_json = payload.as_value().clone(); + wrong_json["run_id"] = other_id.to_string().into(); + sqlx::query("UPDATE run_events SET event_json = ? WHERE run_id = ? AND seq = 2") + .bind(serde_json::to_string(&wrong_json).unwrap()) + .bind(id.to_string()) + .execute(&store.pool) + .await + .unwrap(); + let mut connection = store.pool.acquire().await.unwrap(); + assert!(matches!( + RunSummaryStore::get_event_on_connection(&mut connection, &id, 2) + .await + .unwrap_err(), + Error::RunEventMismatch { + field: "run_id", + .. + } + )); + drop(connection); + + seed_sql_event_restore(&store, &id, 2, &payload).await; + sqlx::query("UPDATE run_events SET run_id = ? WHERE run_id = ? AND seq = 2") + .bind(other_id.to_string()) + .bind(id.to_string()) + .execute(&store.pool) + .await + .unwrap(); + let row = sqlx::query(super::SELECT_EVENT_COLUMNS) + .fetch_one(&store.pool) + .await + .unwrap(); + assert!(matches!( + decode_event_row(&row, &id, &id.to_string()).unwrap_err(), + Error::RunEventMismatch { + field: "run_id", + .. + } + )); + } + + async fn seed_sql_event_restore( + store: &RunSummaryStore, + run_id: &RunId, + seq: u32, + payload: &EventPayload, + ) { + sqlx::query("DELETE FROM run_events WHERE run_id = ? AND seq = ?") + .bind(run_id.to_string()) + .bind(i64::from(seq)) + .execute(&store.pool) + .await + .unwrap(); + seed_sql_event(store, run_id, seq, payload).await; + } + /// 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. diff --git a/lib/components/fabro-store/src/slate/mod.rs b/lib/components/fabro-store/src/slate/mod.rs index 0e7d48d90..39a2129b3 100644 --- a/lib/components/fabro-store/src/slate/mod.rs +++ b/lib/components/fabro-store/src/slate/mod.rs @@ -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}; @@ -45,7 +45,7 @@ pub struct Database { catalog_index: Arc>>, projection_cache: Arc, projection_cache_warmed: Arc>, - run_summary_store: Arc>>, + run_summary_store: Arc, } impl std::fmt::Debug for Database { @@ -65,6 +65,7 @@ impl Database { flush_interval: Duration, cache_path: Option, blobs: Arc, + run_summary_store: Arc, ) -> 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_summary_store, } } - pub fn attach_run_summary_store(&self, store: Arc) -> Arc { - Arc::clone(self.run_summary_store.get_or_init(|| store)) - } - - fn run_summary_store(&self) -> Option> { - self.run_summary_store.get().cloned() + #[must_use] + pub fn run_summary_store(&self) -> Arc { + Arc::clone(&self.run_summary_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_summary_store(), ) .await } @@ -240,9 +238,7 @@ impl Database { } } } - if let Some(store) = self.run_summary_store() { - store.reconcile(&entries).await?; - } + self.run_summary_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_summary_store.delete(run_id).await?; Ok(()) } @@ -558,6 +552,21 @@ mod tests { (object_store, store) } + fn make_store_with_run_summaries( + run_summaries: Arc, + ) -> (Arc, Database) { + let object_store: Arc = 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_summaries, + ); + (object_store, store) + } + #[tokio::test] async fn retire_refresh_token_keyspace_clears_the_prefix_and_is_idempotent() { let (_object_store, store) = make_store(); @@ -596,8 +605,8 @@ mod tests { ); } - async fn make_summary_store() -> (tempfile::TempDir, Arc) { - let (directory, store) = store_test_support::sqlite_summary_store().await; + async fn make_run_summary_store() -> (tempfile::TempDir, Arc) { + let (directory, store) = store_test_support::sqlite_run_summary_store().await; (directory, Arc::new(store)) } @@ -972,9 +981,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_summary_store().await; + let (_object_store, store) = make_store_with_run_summaries(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 +1003,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_summary_update_failure_keeps_slate_append_repairable() { + let (directory, summaries) = make_run_summary_store().await; + let (object_store, store) = make_store_with_run_summaries(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 +1029,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_summary_store_at(directory.path()).await); let stale = repaired_summaries .get(&run_id, Utc::now()) .await @@ -1030,13 +1037,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 +1665,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_summary_append_refreshes_cache_and_delete_removes_rows() { + let (_directory, summaries) = make_run_summary_store().await; + let (_object_store, store) = make_store_with_run_summaries(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 +1841,20 @@ mod tests { } #[tokio::test] - async fn projection_cache_warmup_backfills_sqlite_run_summaries() { + async fn required_run_summary_warmup_backfills_sqlite_run_summaries() { 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_summary_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 diff --git a/lib/components/fabro-store/src/slate/run_store.rs b/lib/components/fabro-store/src/slate/run_store.rs index c01ab0904..6e60b9fc4 100644 --- a/lib/components/fabro-store/src/slate/run_store.rs +++ b/lib/components/fabro-store/src/slate/run_store.rs @@ -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; @@ -45,9 +45,7 @@ pub(crate) struct RunDatabaseInner { state_lock: Mutex<()>, projection_cache: Mutex, shared_projection_cache: Arc, - // 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>>, + run_summary_store: Arc, recent_events: Mutex>, recent_event_limit: usize, event_tx: broadcast::Sender, @@ -60,7 +58,7 @@ impl RunDatabase { read_only: bool, blob_store: Arc, shared_projection_cache: Arc, - run_summary_store: Arc>>, + run_summary_store: Arc, ) -> Result { let cached_projection = shared_projection_cache.projection_snapshot(&run_id).await; let projection_cache = cached_projection.as_ref().map_or_else( @@ -231,15 +229,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_summary_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" + ); } } diff --git a/lib/components/fabro-store/src/test_support/mod.rs b/lib/components/fabro-store/src/test_support/mod.rs index d1a7f4848..83f8e4656 100644 --- a/lib/components/fabro-store/src/test_support/mod.rs +++ b/lib/components/fabro-store/src/test_support/mod.rs @@ -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, RunSummaryStore}; /// Returns an isolated SQLite blob authority backed by its own in-memory /// database. @@ -18,32 +18,48 @@ use crate::{BlobStore, Database, Result}; /// by other tests in the same process. Reopen-style tests that model one /// process-wide blob authority across several store handles should call this /// once and share the result through [`test_database_with_blobs`]. -/// -/// The pool connects lazily so synchronous fixture builders can remain -/// synchronous. Its single connection installs the production blob schema on -/// first use. #[must_use] pub fn test_blob_store() -> Arc { + Arc::new(BlobStore::new(lazy_in_memory_pool(&[ + fabro_db::BLOBS_MIGRATION_SQL, + ]))) +} + +/// Returns an isolated SQLite run-summary store backed by its own in-memory +/// database and the production `runs` and `run_events` schemas. +#[must_use] +pub fn test_run_summary_store() -> Arc { + Arc::new(RunSummaryStore::new(lazy_in_memory_pool(&[ + fabro_db::RUNS_MIGRATION_SQL, + fabro_db::RUN_EVENTS_MIGRATION_SQL, + ]))) +} + +/// Builds a single-connection in-memory SQLite pool that installs +/// `migrations` on first use. +/// +/// The pool connects lazily so synchronous fixture builders can remain +/// synchronous. +fn lazy_in_memory_pool(migrations: &'static [&'static str]) -> sqlx::SqlitePool { let options = SqliteConnectOptions::new() .filename(":memory:") .foreign_keys(true); - let pool = SqlitePoolOptions::new() + SqlitePoolOptions::new() .max_connections(1) // A single in-memory test connection never needs reaping. Disabling // both timers also keeps this lazy fixture constructible from sync // tests, where SQLx has no Tokio runtime for maintenance tasks. .max_lifetime(None) .idle_timeout(None) - .after_connect(|connection, _metadata| { + .after_connect(move |connection, _metadata| { Box::pin(async move { - sqlx::query(fabro_db::BLOBS_MIGRATION_SQL) - .execute(&mut *connection) - .await?; + for migration in migrations { + sqlx::raw_sql(*migration).execute(&mut *connection).await?; + } Ok(()) }) }) - .connect_lazy_with(options); - Arc::new(BlobStore::new(pool)) + .connect_lazy_with(options) } /// Returns the SQLite file backing [`test_blob_store_at`] for `store_dir`. @@ -119,7 +135,37 @@ pub fn test_database_with_blobs( cache_path: Option, blobs: Arc, ) -> 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_summary_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, + base_prefix: impl Into, + flush_interval: Duration, + cache_path: Option, + blobs: Arc, + run_summaries: Arc, +) -> Database { + Database::new( + object_store, + base_prefix, + flush_interval, + cache_path, + blobs, + run_summaries, + ) } /// Seeds one canonical row in the legacy SlateDB blob keyspace. @@ -171,13 +217,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_summary_store() -> (tempfile::TempDir, RunSummaryStore) { let directory = tempfile::tempdir().unwrap(); - let store = sqlite_summary_store_at(directory.path()).await; + let store = sqlite_run_summary_store_at(directory.path()).await; (directory, store) } #[cfg(test)] -pub(crate) async fn sqlite_summary_store_at(directory: &Path) -> RunSummaryStore { +pub(crate) async fn sqlite_run_summary_store_at(directory: &Path) -> RunSummaryStore { RunSummaryStore::new(sqlite_test_pool(directory).await) } diff --git a/lib/foundation/fabro-db/migrations/2026082701_run_events.sql b/lib/foundation/fabro-db/migrations/2026082701_run_events.sql new file mode 100644 index 000000000..1fca0563c --- /dev/null +++ b/lib/foundation/fabro-db/migrations/2026082701_run_events.sql @@ -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'; diff --git a/lib/foundation/fabro-db/src/lib.rs b/lib/foundation/fabro-db/src/lib.rs index 26567d2c9..4f3687437 100644 --- a/lib/foundation/fabro-db/src/lib.rs +++ b/lib/foundation/fabro-db/src/lib.rs @@ -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 summary 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, diff --git a/lib/foundation/fabro-db/tests/sqlite.rs b/lib/foundation/fabro-db/tests/sqlite.rs index 4d8d7328c..824630b58 100644 --- a/lib/foundation/fabro-db/tests/sqlite.rs +++ b/lib/foundation/fabro-db/tests/sqlite.rs @@ -639,22 +639,283 @@ 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::("name"), + column.get::("type"), + column.get::("notnull"), + column.get::("pk"), + ) + }) + .collect::>(); + 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::("table"), "runs"); + assert_eq!(foreign_keys[0].get::("from"), "run_id"); + assert_eq!(foreign_keys[0].get::("to"), "id"); + assert_eq!(foreign_keys[0].get::("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::("name"); + name.starts_with("run_events_by_").then_some(( + name, + index.get::("unique"), + index.get::("partial"), + )) + }) + .collect::>(); + 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::("unique") == 0 + || index.get::("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").await?; + + for invalid in [ + insert_run_event(database.pool(), "parent", 1, "run.created").await, + insert_run_event(database.pool(), "missing", 1, "run.created").await, + insert_run_event(database.pool(), "parent", 0, "run.created").await, + insert_run_event(database.pool(), "parent", 1_000_000, "run.created").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 = + 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::("detail")) + .collect::>() + .join("; "); + assert!( + details.contains(expected_index), + "expected {expected_index} in query plan: {details}" + ); + } + + // The first-visit stage listing unions both shapes so each arm keeps its + // own partial index instead of scanning the run's primary key range. + let details = sqlx::query( + "EXPLAIN QUERY PLAN SELECT * FROM run_events WHERE run_id = ? AND seq >= ? AND stage_id = ? \ + UNION ALL SELECT * FROM run_events WHERE run_id = ? AND seq >= ? AND stage_id IS NULL AND node_id = ? \ + ORDER BY seq ASC LIMIT ?", + ) + .bind("run") + .bind(1_i64) + .bind("stage") + .bind("run") + .bind(1_i64) + .bind("node") + .bind(10_i64) + .fetch_all(database.pool()) + .await? + .into_iter() + .map(|row| row.get::("detail")) + .collect::>() + .join("; "); + for expected_index in ["run_events_by_stage", "run_events_by_legacy_node"] { + assert!( + details.contains(expected_index), + "expected {expected_index} in query plan: {details}" + ); + } + + Ok(()) +} + +async fn insert_run_event( + pool: &fabro_db::DbPool, + run_id: &str, + seq: i64, + event_name: &str, +) -> Result<(), sqlx::Error> { + sqlx::query( + r" +INSERT INTO run_events (run_id, seq, event_name, event_json) +VALUES (?, ?, ?, '{}') +", + ) + .bind(run_id) + .bind(seq) + .bind(event_name) + .execute(pool) + .await?; + Ok(()) +} + async fn insert_minimal_run( pool: &fabro_db::DbPool, status: &str, input_tokens: i64, summary_json: &str, +) -> Result<(), sqlx::Error> { + insert_run_row( + pool, + &format!("run-{status}-{input_tokens}"), + None, + status, + input_tokens, + summary_json, + ) + .await +} + +async fn insert_run_with_id( + pool: &fabro_db::DbPool, + id: &str, + parent_id: Option<&str>, +) -> Result<(), sqlx::Error> { + insert_run_row( + pool, + id, + parent_id, + "submitted", + 0, + &format!(r#"{{"id":"{id}"}}"#), + ) + .await +} + +async fn insert_run_row( + pool: &fabro_db::DbPool, + id: &str, + parent_id: Option<&str>, + status: &str, + input_tokens: i64, + summary_json: &str, ) -> Result<(), sqlx::Error> { sqlx::query( r" INSERT INTO runs ( - id, source_last_seq, created_at_ms, last_event_at_ms, status, title, + id, source_last_seq, created_at_ms, last_event_at_ms, status, parent_id, title, input_tokens, summary_json -) VALUES (?, 1, 0, 0, ?, 'title', ?, ?) +) VALUES (?, 1, 0, 0, ?, ?, 'title', ?, ?) ", ) - .bind(format!("run-{status}-{input_tokens}")) + .bind(id) .bind(status) + .bind(parent_id) .bind(input_tokens) .bind(summary_json) .execute(pool)