From 2360e8046ba7f980ec6f3ee1d1b3650c96a70e55 Mon Sep 17 00:00:00 2001 From: Scott Werner Date: Wed, 2 Sep 2026 13:32:43 -0400 Subject: [PATCH] Move run projection replay into the summary store --- .../src/legacy_run_history_import.rs | 19 ++- .../fabro-store/src/run_summary_store.rs | 136 +++++++++++++++--- lib/components/fabro-store/src/slate/mod.rs | 41 +++--- .../fabro-store/src/slate/run_store.rs | 60 +++----- 4 files changed, 174 insertions(+), 82 deletions(-) diff --git a/lib/components/fabro-store/src/legacy_run_history_import.rs b/lib/components/fabro-store/src/legacy_run_history_import.rs index dfad82b55..edef01d77 100644 --- a/lib/components/fabro-store/src/legacy_run_history_import.rs +++ b/lib/components/fabro-store/src/legacy_run_history_import.rs @@ -16,7 +16,7 @@ use tokio::sync::Barrier; use tracing::debug; use crate::keys::SlateKey; -use crate::slate::CachedRunProjection; +use crate::slate::ProjectedRun; use crate::{Database, EventPayload, RunProjectionReducer, RunSummaryStore, keys}; /// Count-only observations about the legacy catalog and session indexes. @@ -458,7 +458,7 @@ struct ValidatedLegacyRunEvent { struct ValidatedLegacyRunHistory { run_id: RunId, events: Vec, - current: CachedRunProjection, + current: ProjectedRun, } struct LegacyRunHistorySource { @@ -534,7 +534,7 @@ impl LegacyRunHistorySource { .expect("a validated history contains at least one event") .envelope .seq; - let current = CachedRunProjection::from_projection(run_id, projection, last_seq); + let current = ProjectedRun::new(run_id, projection, last_seq); Ok(Some(ValidatedLegacyRunHistory { run_id, events, @@ -1159,7 +1159,7 @@ fn require_exact_prefix( fn replay_destination( run_id: &RunId, events: &[(EventEnvelope, String)], -) -> crate::Result { +) -> crate::Result { let Some((first, _event_json)) = events.first() else { return Err(crate::Error::InvalidEvent( "run projection requires an event".to_owned(), @@ -1181,9 +1181,7 @@ fn replay_destination( .last() .expect("a destination history validated as nonempty") .seq; - Ok(CachedRunProjection::from_projection( - *run_id, projection, last_seq, - )) + Ok(ProjectedRun::new(*run_id, projection, last_seq)) } fn usize_to_import_count(value: usize) -> Result { @@ -1273,7 +1271,7 @@ mod tests { parse_source_event, }; use crate::keys::SlateKey; - use crate::slate::CachedRunProjection; + use crate::slate::ProjectedRun; use crate::{ Database, EventEnvelope, EventPayload, RunProjectionReducer, RunSummaryStore, keys, test_support as store_test_support, @@ -1479,8 +1477,7 @@ mod tests { .map(|(_payload, envelope)| envelope.clone()) .collect::>(); let projection = RunProjection::apply_events(&envelopes)?; - let current = - CachedRunProjection::from_projection(*run_id, projection, events.last().unwrap().0); + let current = ProjectedRun::new(*run_id, projection, events.last().unwrap().0); let mut transaction = pool.begin().await?; RunSummaryStore::insert_imported_run_on_connection(&mut transaction, ¤t).await?; for ((_, event_json), (payload, envelope)) in events.iter().zip(&decoded) { @@ -1514,7 +1511,7 @@ mod tests { .map(|(_payload, envelope)| envelope.clone()) .collect::>(); let projection = RunProjection::apply_events(&envelopes)?; - let current = CachedRunProjection::from_projection(*run_id, projection, next.0); + let current = ProjectedRun::new(*run_id, projection, next.0); let mut transaction = pool.begin().await?; RunSummaryStore::append_event_on_connection( &mut transaction, diff --git a/lib/components/fabro-store/src/run_summary_store.rs b/lib/components/fabro-store/src/run_summary_store.rs index 97e07eee0..66839f2d4 100644 --- a/lib/components/fabro-store/src/run_summary_store.rs +++ b/lib/components/fabro-store/src/run_summary_store.rs @@ -3,8 +3,8 @@ use std::sync::LazyLock; use chrono::{DateTime, Utc}; use fabro_types::{ - BilledTokenCounts, EventEnvelope, Run, RunEvent, RunId, RunSize, RunStatusKind, RunTiming, - SessionId, StageId, timing, + BilledTokenCounts, EventEnvelope, Run, RunEvent, RunId, RunProjection, RunSize, RunStatusKind, + RunTiming, SessionId, StageId, timing, }; use sqlx::pool::PoolConnection; use sqlx::query::Query; @@ -12,8 +12,8 @@ use sqlx::sqlite::{SqliteArguments, SqliteConnection, SqliteRow}; use sqlx::{Connection as _, QueryBuilder, Row as _, Sqlite, SqlitePool, Transaction}; use strum::VariantArray as _; -use crate::run_state::{build_summary, projected_billing}; -use crate::slate::CachedRunProjection; +use crate::run_state::{RunProjectionReducer, build_summary, projected_billing}; +use crate::slate::ProjectedRun; use crate::{Error, EventPayload, Result, keys}; const INSERT_RUN_SQL: &str = r" @@ -288,14 +288,14 @@ ON CONFLICT(singleton) DO NOTHING } #[cfg(test)] - pub(crate) async fn upsert_projection(&self, entry: &CachedRunProjection) -> Result<()> { + pub(crate) async fn upsert_projection(&self, entry: &ProjectedRun) -> Result<()> { let record = PreparedRunSummary::from_entry(entry); let mut connection = self.pool.acquire().await?; upsert_run_on_connection(&mut connection, &record).await } #[cfg(test)] - pub(crate) async fn reconcile(&self, entries: &[CachedRunProjection]) -> Result<()> { + pub(crate) async fn reconcile(&self, entries: &[ProjectedRun]) -> Result<()> { use std::collections::{HashMap, HashSet}; let mut transaction = self.pool.begin().await?; @@ -351,6 +351,22 @@ ON CONFLICT(singleton) DO NOTHING select_run_head(&mut connection, run_id).await } + /// Replays one run's canonical history from one validated SQLite snapshot, + /// returning `None` when the run does not exist. + pub(crate) async fn load_projection(&self, run_id: &RunId) -> Result> { + let events = match self.list_events_for_run(run_id).await { + Ok(events) => events, + Err(Error::RunNotFound(_)) => return Ok(None), + Err(error) => return Err(error), + }; + let last_seq = events + .last() + .map(|event| event.seq) + .ok_or_else(|| Error::InvalidEvent(format!("run {run_id} has no run.created event")))?; + let projection = RunProjection::apply_events(&events)?; + Ok(Some(ProjectedRun::new(*run_id, projection, last_seq))) + } + pub(crate) async fn list_events_for_run(&self, run_id: &RunId) -> Result> { let mut connection = self.acquire().await?; Self::list_events_on_connection(&mut connection, run_id).await @@ -638,7 +654,7 @@ FROM runs", impl RunSummaryStore { pub(crate) async fn insert_first_event_on_connection( connection: &mut SqliteConnection, - entry: &CachedRunProjection, + entry: &ProjectedRun, payload: &EventPayload, ) -> Result { let record = PreparedRunSummary::from_entry(entry); @@ -657,7 +673,7 @@ impl RunSummaryStore { pub(crate) async fn append_event_on_connection( connection: &mut SqliteConnection, expected_last_seq: u32, - entry: &CachedRunProjection, + entry: &ProjectedRun, payload: &EventPayload, ) -> Result { let next_seq = next_event_seq_after(expected_last_seq)?; @@ -734,7 +750,7 @@ impl RunSummaryStore { pub(crate) async fn insert_imported_run_on_connection( connection: &mut SqliteConnection, - entry: &CachedRunProjection, + entry: &ProjectedRun, ) -> Result<()> { let record = PreparedRunSummary::from_entry(entry); ensure_entry_identity(entry, &record, entry.last_seq)?; @@ -768,7 +784,7 @@ impl RunSummaryStore { pub(crate) async fn verify_current_run_on_connection( connection: &mut SqliteConnection, - entry: &CachedRunProjection, + entry: &ProjectedRun, ) -> Result<()> { let record = PreparedRunSummary::from_entry(entry); ensure_entry_identity(entry, &record, entry.last_seq)?; @@ -1004,7 +1020,7 @@ struct PreparedRunSummary { } impl PreparedRunSummary { - fn from_entry(entry: &CachedRunProjection) -> Self { + fn from_entry(entry: &ProjectedRun) -> Self { let mut run = build_summary(&entry.projection, &entry.run_id); if run.timing.is_none() { let at = run @@ -1036,7 +1052,7 @@ impl PreparedRunSummary { } fn ensure_entry_identity( - entry: &CachedRunProjection, + entry: &ProjectedRun, record: &PreparedRunSummary, seq: u32, ) -> Result<()> { @@ -1544,8 +1560,8 @@ mod tests { INSERT_EVENT_SQL, RunSummaryListQuery, RunSummarySort, RunSummarySortDirection, RunSummaryStore, RunSummaryVisibility, decode_event_row, }; - use crate::slate::CachedRunProjection; - use crate::{Error, EventPayload, test_support as store_test_support}; + use crate::slate::ProjectedRun; + use crate::{Error, EventPayload, RunProjectionReducer, test_support as store_test_support}; fn dt(value: &str) -> DateTime { value.parse().unwrap() @@ -1580,8 +1596,8 @@ mod tests { ) } - fn entry(projection: RunProjection, last_seq: u32) -> CachedRunProjection { - CachedRunProjection::from_projection(projection.spec.run_id, projection, last_seq) + fn entry(projection: RunProjection, last_seq: u32) -> ProjectedRun { + ProjectedRun::new(projection.spec.run_id, projection, last_seq) } async fn store() -> (tempfile::TempDir, RunSummaryStore) { @@ -1920,6 +1936,94 @@ mod tests { assert_eq!(sequences, vec![1, 2, 3]); } + #[tokio::test] + async fn load_projection_replays_committed_events_and_reports_missing_run() { + let (_directory, store) = store().await; + let created_at = dt("2026-08-27T12:00:00Z"); + let id = run_id(created_at.timestamp_millis().cast_unsigned(), 31); + 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(); + + let second = entry(projection(id, "updated", created_at), 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(); + let second_envelope = RunSummaryStore::append_event_on_connection( + &mut transaction, + 1, + &second, + &second_payload, + ) + .await + .unwrap(); + transaction.commit().await.unwrap(); + + let expected = RunProjection::apply_events(&[first_envelope, second_envelope]).unwrap(); + + let loaded = store.load_projection(&id).await.unwrap().unwrap(); + assert_eq!(loaded.run_id, id); + assert_eq!(loaded.last_seq, 2); + assert_eq!( + serde_json::to_value(loaded.projection.as_ref()).unwrap(), + serde_json::to_value(expected).unwrap() + ); + + let missing = run_id(created_at.timestamp_millis().cast_unsigned() + 1, 32); + assert!(store.load_projection(&missing).await.unwrap().is_none()); + } + + #[tokio::test] + async fn load_projection_reports_removed_events_without_poisoning_following_reads() { + let (_directory, store) = store().await; + let created_at = dt("2026-08-27T12:00:00Z"); + let broken_id = run_id(created_at.timestamp_millis().cast_unsigned(), 33); + let healthy_id = run_id(created_at.timestamp_millis().cast_unsigned() + 1, 34); + + for id in [broken_id, healthy_id] { + let current = entry(projection(id, "created", created_at), 1); + let mut transaction = store.pool.begin().await.unwrap(); + RunSummaryStore::insert_first_event_on_connection( + &mut transaction, + ¤t, + &created_payload(&id), + ) + .await + .unwrap(); + transaction.commit().await.unwrap(); + } + store.test_delete_run_events(&broken_id).await.unwrap(); + + assert!(matches!( + store.load_projection(&broken_id).await, + Err(Error::RunHeadMismatch { + expected_last_seq: 1, + actual_last_seq: None, + .. + }) + )); + + let loaded = store.load_projection(&healthy_id).await.unwrap().unwrap(); + assert_eq!(loaded.run_id, healthy_id); + assert_eq!(loaded.last_seq, 1); + assert_eq!(loaded.projection.title, "created"); + } + #[tokio::test] async fn session_owner_lookup_resolves_only_typed_creation_events() { let (_directory, store) = store().await; diff --git a/lib/components/fabro-store/src/slate/mod.rs b/lib/components/fabro-store/src/slate/mod.rs index 5678ee954..9a30f8d19 100644 --- a/lib/components/fabro-store/src/slate/mod.rs +++ b/lib/components/fabro-store/src/slate/mod.rs @@ -8,7 +8,7 @@ use std::time::Duration; use chrono::{DateTime, Utc}; use fabro_types::{RunId, SessionId}; use object_store::ObjectStore; -pub(crate) use run_store::CachedRunProjection; +pub(crate) use run_store::ProjectedRun; pub use run_store::RunDatabase; use run_store::RunDatabaseInner; use slatedb::config::{CompressionCodec, Settings}; @@ -128,8 +128,8 @@ impl Database { payload: &EventPayload, ) -> Result { let (mut active_runs, run_store) = self.reserve_new_run(run_id).await?; - let (envelope, cached) = run_store.commit_first_event(payload).await?; - run_store.install_in_memory_state(&cached); + let (envelope, projected) = run_store.commit_first_event(payload).await?; + run_store.install_in_memory_state(&projected); Self::cache_active_run(&mut active_runs, &run_store); run_store.publish(&envelope); Ok(run_store) @@ -170,9 +170,6 @@ impl Database { if let Some(active) = active_run_from(&active_runs, run_id) { return Ok(active); } - if !self.run_summary_store.contains(run_id).await? { - return Err(Error::RunNotFound(run_id.to_string())); - } let run_store = self.open_run_database(run_id, false).await?; Self::cache_active_run(&mut active_runs, &run_store); Ok(run_store) @@ -182,9 +179,6 @@ impl Database { if let Some(active) = self.get_active_run(run_id).await { return Ok(active.read_only_clone()); } - if !self.run_summary_store.contains(run_id).await? { - return Err(Error::RunNotFound(run_id.to_string())); - } self.open_run_database(run_id, true).await } @@ -192,7 +186,7 @@ impl Database { let run_ids = self.run_summary_store.list_run_ids().await?; let mut unreadable = Vec::new(); for run_id in run_ids { - match RunDatabase::build_projection(&self.run_summary_store, &run_id).await { + match self.run_summary_store.load_projection(&run_id).await { Ok(Some(_)) => {} Ok(None) => unreadable.push(UnreadableRun { run_id, @@ -249,11 +243,11 @@ impl Database { if let Some(active) = self.get_active_run(run_id).await { return active.projection_snapshot().await.map(Some); } - Ok( - RunDatabase::build_projection(&self.run_summary_store, run_id) - .await? - .map(|entry| entry.projection), - ) + Ok(self + .run_summary_store + .load_projection(run_id) + .await? + .map(|projected| projected.projection)) } /// Resolves the run that owns `session_id` from the canonical typed @@ -902,6 +896,21 @@ mod tests { ); } + #[tokio::test] + async fn missing_run_open_paths_return_run_not_found() { + let (_object_store, store) = make_store(); + let run_id = test_run_id("run-4"); + + assert!(matches!( + store.open_run(&run_id).await, + Err(Error::RunNotFound(id)) if id == run_id.to_string() + )); + assert!(matches!( + store.open_run_reader(&run_id).await, + Err(Error::RunNotFound(id)) if id == run_id.to_string() + )); + } + #[tokio::test] async fn open_run_reader_is_read_only() { let (_object_store, store) = make_store(); @@ -1016,7 +1025,7 @@ mod tests { let projection = store.load_run_projection(&run_id).await.unwrap().unwrap(); let last_seq = run.last_event_seq().await.unwrap().unwrap(); - let entries = [CachedRunProjection::from_projection( + let entries = [ProjectedRun::new( run_id, Arc::unwrap_or_clone(projection), last_seq, diff --git a/lib/components/fabro-store/src/slate/run_store.rs b/lib/components/fabro-store/src/slate/run_store.rs index e64fe6fc1..af8cdc92e 100644 --- a/lib/components/fabro-store/src/slate/run_store.rs +++ b/lib/components/fabro-store/src/slate/run_store.rs @@ -20,14 +20,14 @@ const EVENT_BROADCAST_CAPACITY: usize = 1024; /// SQLite history or by applying a newly committed event, and consumed by /// `RunSummaryStore` writes that must stay in step with the event log. #[derive(Debug, Clone)] -pub(crate) struct CachedRunProjection { +pub(crate) struct ProjectedRun { pub(crate) run_id: RunId, pub(crate) projection: Arc, pub(crate) last_seq: u32, } -impl CachedRunProjection { - pub(crate) fn from_projection(run_id: RunId, projection: RunProjection, last_seq: u32) -> Self { +impl ProjectedRun { + pub(crate) fn new(run_id: RunId, projection: RunProjection, last_seq: u32) -> Self { Self { run_id, projection: Arc::new(projection), @@ -36,11 +36,11 @@ impl CachedRunProjection { } } -impl From for EventProjectionCache { - fn from(cached: CachedRunProjection) -> Self { +impl From for EventProjectionCache { + fn from(projected: ProjectedRun) -> Self { Self { - last_seq: cached.last_seq, - state: Some(cached.projection), + last_seq: projected.last_seq, + state: Some(projected.projection), } } } @@ -85,7 +85,8 @@ impl RunDatabase { blob_store: Arc, run_summary_store: Arc, ) -> Result { - let cached = Self::build_projection(&run_summary_store, &run_id) + let projected = run_summary_store + .load_projection(&run_id) .await? .ok_or_else(|| Error::RunNotFound(run_id.to_string()))?; Ok(Self::from_event_projection_cache( @@ -93,7 +94,7 @@ impl RunDatabase { read_only, blob_store, run_summary_store, - cached.into(), + projected.into(), )) } @@ -158,25 +159,6 @@ impl RunDatabase { self.inner.event_tx.subscribe() } - pub(crate) async fn build_projection( - store: &RunSummaryStore, - run_id: &RunId, - ) -> Result> { - let events = match store.list_events_for_run(run_id).await { - Ok(events) => events, - Err(Error::RunNotFound(_)) => return Ok(None), - Err(error) => return Err(error), - }; - let last_seq = events - .last() - .map(|event| event.seq) - .ok_or_else(|| Error::InvalidEvent(format!("run {run_id} has no run.created event")))?; - let state = RunProjection::apply_events(&events)?; - Ok(Some(CachedRunProjection::from_projection( - *run_id, state, last_seq, - ))) - } - pub(super) async fn projection_snapshot(&self) -> Result> { let _state_guard = self.inner.state_lock.lock().await; self.projection_snapshot_locked() @@ -195,10 +177,10 @@ impl RunDatabase { }) } - pub(crate) fn install_in_memory_state(&self, cached: &CachedRunProjection) { + pub(crate) fn install_in_memory_state(&self, projected: &ProjectedRun) { let mut projection_cache = self.inner.lock_projection_cache(); - projection_cache.state = Some(Arc::clone(&cached.projection)); - projection_cache.last_seq = cached.last_seq; + projection_cache.state = Some(Arc::clone(&projected.projection)); + projection_cache.last_seq = projected.last_seq; } pub(crate) fn publish(&self, event: &EventEnvelope) { @@ -208,7 +190,7 @@ impl RunDatabase { pub(crate) async fn commit_first_event( &self, payload: &EventPayload, - ) -> Result<(EventEnvelope, CachedRunProjection)> { + ) -> Result<(EventEnvelope, ProjectedRun)> { payload.validate(&self.inner.run_id)?; let event = RunEvent::try_from(payload)?; let _state_guard = self.inner.state_lock.lock().await; @@ -269,10 +251,10 @@ impl RunDatabase { payload: &EventPayload, event: RunEvent, ) -> Result { - let (envelope, cached) = self.commit_event_locked(payload, event).await?; + let (envelope, projected) = self.commit_event_locked(payload, event).await?; // Keep post-commit propagation await-free: cancellation after SQLite // commits must not leave in-memory state stale or omit the broadcast. - self.install_in_memory_state(&cached); + self.install_in_memory_state(&projected); self.publish(&envelope); Ok(envelope) } @@ -281,7 +263,7 @@ impl RunDatabase { &self, payload: &EventPayload, event: RunEvent, - ) -> Result<(EventEnvelope, CachedRunProjection)> { + ) -> Result<(EventEnvelope, ProjectedRun)> { let (expected_last_seq, mut next_state) = { let cache = self.inner.lock_projection_cache(); (cache.last_seq, cache.state.clone()) @@ -291,7 +273,7 @@ impl RunDatabase { apply_cached_projection_event(&mut next_state, &prospective).map_err(event_rejected)?; let next_projection = next_state.expect("applying a valid event should always produce a projection"); - let cached = CachedRunProjection::from_projection( + let projected = ProjectedRun::new( self.inner.run_id, Arc::unwrap_or_clone(next_projection), seq, @@ -299,19 +281,19 @@ impl RunDatabase { let mut transaction = self.inner.run_summary_store.begin().await?; let envelope = if expected_last_seq == 0 { - RunSummaryStore::insert_first_event_on_connection(&mut transaction, &cached, payload) + RunSummaryStore::insert_first_event_on_connection(&mut transaction, &projected, payload) .await? } else { RunSummaryStore::append_event_on_connection( &mut transaction, expected_last_seq, - &cached, + &projected, payload, ) .await? }; transaction.commit().await?; - Ok((envelope, cached)) + Ok((envelope, projected)) } pub async fn list_events(&self) -> Result> {