diff --git a/lib/apps/fabro-server/src/run_files.rs b/lib/apps/fabro-server/src/run_files.rs index e82f9b861..ff7b9c201 100644 --- a/lib/apps/fabro-server/src/run_files.rs +++ b/lib/apps/fabro-server/src/run_files.rs @@ -1195,7 +1195,7 @@ async fn load_projection( state: &Arc, run_id: &RunId, ) -> std::result::Result, ApiError> { - state.cached_run_projection(run_id).await + state.load_run_projection(run_id).await } async fn reconnect_run_sandbox( diff --git a/lib/apps/fabro-server/src/serve.rs b/lib/apps/fabro-server/src/serve.rs index fe44e8a98..7cbe5ac7f 100644 --- a/lib/apps/fabro-server/src/serve.rs +++ b/lib/apps/fabro-server/src/serve.rs @@ -833,12 +833,6 @@ where #[cfg(any(test, feature = "test-support"))] automation_materializer_override: None, })?; - state - .stores - .runs - .warm_projection_cache() - .await - .context("warming run projection cache from SQLite")?; let reconciled = reconcile_incomplete_runs_on_startup(&state).await?; if reconciled > 0 { info!( diff --git a/lib/apps/fabro-server/src/server.rs b/lib/apps/fabro-server/src/server.rs index cfc671909..be2769a4a 100644 --- a/lib/apps/fabro-server/src/server.rs +++ b/lib/apps/fabro-server/src/server.rs @@ -794,7 +794,7 @@ impl SlackService { return; }; let event_name = event.body.event_name(); - let projection = match state.stores.runs.get_cached_projection(&event.run_id).await { + let projection = match state.stores.runs.load_run_projection(&event.run_id).await { Ok(Some(projection)) => projection, Ok(None) => { warn!( @@ -1508,16 +1508,16 @@ impl AppState { &self.stores.runs } - /// Current cached projection for `run_id`, with the standard HTTP error + /// Loads the current projection for `run_id`, with the standard HTTP error /// mapping: storage failures become 500s and a missing run becomes the /// canonical 404. - pub(crate) async fn cached_run_projection( + pub(crate) async fn load_run_projection( &self, run_id: &RunId, ) -> Result, ApiError> { self.stores .runs - .get_cached_projection(run_id) + .load_run_projection(run_id) .await .map_err(|err| ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string()))? .ok_or_else(|| ApiError::not_found("Run not found.")) @@ -3744,7 +3744,7 @@ async fn load_pending_interview( qid: &str, ) -> Result { let projection = state - .cached_run_projection(&run_id) + .load_run_projection(&run_id) .await .map_err(IntoResponse::into_response)?; let Some(record) = projection.pending_interviews.get(qid) else { @@ -4534,7 +4534,7 @@ async fn append_control_request( /// run is currently archived. Returns `None` otherwise (including when the run /// doesn't exist — the caller's own not-found handling will surface that). async fn reject_if_archived(state: &AppState, run_id: &RunId) -> Option { - let projection = state.cached_run_projection(run_id).await.ok()?; + let projection = state.load_run_projection(run_id).await.ok()?; projection.archived_at.is_some().then(|| { ApiError::new( StatusCode::CONFLICT, diff --git a/lib/apps/fabro-server/src/server/handler/artifacts.rs b/lib/apps/fabro-server/src/server/handler/artifacts.rs index 7773b7a3c..6f175660a 100644 --- a/lib/apps/fabro-server/src/server/handler/artifacts.rs +++ b/lib/apps/fabro-server/src/server/handler/artifacts.rs @@ -86,7 +86,7 @@ async fn get_checkpoint( Ok(id) => id, Err(response) => return response, }; - match state.cached_run_projection(&id).await { + match state.load_run_projection(&id).await { Ok(projection) => match projection.current_checkpoint() { Some(cp) => (StatusCode::OK, Json(cp.clone())).into_response(), None => (StatusCode::OK, Json(serde_json::json!(null))).into_response(), @@ -132,7 +132,7 @@ async fn read_run_blob( async fn ensure_run_exists(state: &AppState, run_id: &RunId) -> Result<(), Response> { state - .cached_run_projection(run_id) + .load_run_projection(run_id) .await .map(|_| ()) .map_err(IntoResponse::into_response) @@ -330,7 +330,7 @@ async fn download_run_artifacts( Ok(id) => id, Err(response) => return response, }; - let projection = match state.cached_run_projection(&id).await { + let projection = match state.load_run_projection(&id).await { Ok(projection) => projection, Err(error) => return error.into_response(), }; diff --git a/lib/apps/fabro-server/src/server/handler/billing.rs b/lib/apps/fabro-server/src/server/handler/billing.rs index c0839170a..cb0783088 100644 --- a/lib/apps/fabro-server/src/server/handler/billing.rs +++ b/lib/apps/fabro-server/src/server/handler/billing.rs @@ -69,7 +69,7 @@ async fn list_run_stages( Err(response) => return response, }; - let projection = match state.cached_run_projection(&id).await { + let projection = match state.load_run_projection(&id).await { Ok(projection) => projection, Err(err) => return err.into_response(), }; @@ -90,7 +90,7 @@ async fn get_run_billing( State(state): State>, Path(id): Path, ) -> Response { - let projection = match state.cached_run_projection(&id).await { + let projection = match state.load_run_projection(&id).await { Ok(projection) => projection, Err(err) => return err.into_response(), }; diff --git a/lib/apps/fabro-server/src/server/handler/graph.rs b/lib/apps/fabro-server/src/server/handler/graph.rs index ec4f649a4..17e8b903b 100644 --- a/lib/apps/fabro-server/src/server/handler/graph.rs +++ b/lib/apps/fabro-server/src/server/handler/graph.rs @@ -243,7 +243,7 @@ async fn load_run_dot_source(state: &AppState, id: &RunId) -> Result, id: &RunId, ) -> Result { - let projection = state.cached_run_projection(id).await?; + let projection = state.load_run_projection(id).await?; projection.pull_request.clone().ok_or_else(|| { ApiError::with_code( StatusCode::NOT_FOUND, @@ -319,7 +319,7 @@ async fn create_run_pull_request( let Ok(run_store) = state.stores.runs.open_run(&id).await else { return ApiError::not_found("Run not found.").into_response(); }; - let run_state = match state.cached_run_projection(&id).await { + let run_state = match state.load_run_projection(&id).await { Ok(run_state) => run_state, Err(err) => return err.into_response(), }; @@ -372,7 +372,7 @@ async fn create_run_pull_request( } }; - let run_state = match state.cached_run_projection(&id).await { + let run_state = match state.load_run_projection(&id).await { Ok(run_state) => run_state, Err(err) => return err.into_response(), }; @@ -427,7 +427,7 @@ async fn get_run_pull_request_creation( RequireRunScoped(id): RequireRunScoped, State(state): State>, ) -> Response { - let run_state = match state.cached_run_projection(&id).await { + let run_state = match state.load_run_projection(&id).await { Ok(run_state) => run_state, Err(err) => return err.into_response(), }; @@ -473,7 +473,7 @@ async fn unlink_run_pull_request( let Ok(run_store) = state.stores.runs.open_run(&id).await else { return ApiError::not_found("Run not found.").into_response(); }; - let run_state = match state.cached_run_projection(&id).await { + let run_state = match state.load_run_projection(&id).await { Ok(run_state) => run_state, Err(err) => return err.into_response(), }; diff --git a/lib/apps/fabro-server/src/server/handler/runs.rs b/lib/apps/fabro-server/src/server/handler/runs.rs index f10eeef7d..2864c4bcf 100644 --- a/lib/apps/fabro-server/src/server/handler/runs.rs +++ b/lib/apps/fabro-server/src/server/handler/runs.rs @@ -330,7 +330,7 @@ async fn run_summary_at( return Ok(None); }; if summary.timestamps.completed_at.is_none() { - let projection = state.stores.runs.get_cached_projection(run_id).await?; + let projection = state.stores.runs.load_run_projection(run_id).await?; if let Some(timing) = projection.and_then(|projection| projection.live_run_timing(now)) { summary.timing = Some(timing); } @@ -1608,7 +1608,7 @@ async fn get_run_settings( Ok(id) => id, Err(response) => return response, }; - let projection = match state.cached_run_projection(&id).await { + let projection = match state.load_run_projection(&id).await { Ok(projection) => projection, Err(err) => return err.into_response(), }; @@ -1619,7 +1619,7 @@ async fn get_questions( RequireRunManagementTarget(id, _actor): RequireRunManagementTarget, State(state): State>, ) -> Response { - match state.cached_run_projection(&id).await { + match state.load_run_projection(&id).await { Ok(projection) => { let questions = projection .pending_interviews @@ -1660,7 +1660,7 @@ async fn get_run_state( RequireRunManagementTarget(id, _actor): RequireRunManagementTarget, State(state): State>, ) -> Response { - match state.cached_run_projection(&id).await { + match state.load_run_projection(&id).await { Ok(projection) => Json(&*projection).into_response(), Err(err) => err.into_response(), } @@ -1697,7 +1697,7 @@ async fn get_run_stage_context_window( Ok(stage_id) => stage_id, Err(response) => return response, }; - let projection = match state.cached_run_projection(&id).await { + let projection = match state.load_run_projection(&id).await { Ok(projection) => projection, Err(err) => return err.into_response(), }; @@ -1753,7 +1753,7 @@ async fn get_run_stage_command_log( return ApiError::bad_request("limit must be greater than 0").into_response(); } let limit = query.limit.min(MAX_COMMAND_LOG_LIMIT); - let projection = match state.cached_run_projection(&id).await { + let projection = match state.load_run_projection(&id).await { Ok(projection) => projection, Err(err) => return err.into_response(), }; diff --git a/lib/apps/fabro-server/src/server/handler/sandbox.rs b/lib/apps/fabro-server/src/server/handler/sandbox.rs index 091730258..91ac2a9d8 100644 --- a/lib/apps/fabro-server/src/server/handler/sandbox.rs +++ b/lib/apps/fabro-server/src/server/handler/sandbox.rs @@ -952,7 +952,7 @@ async fn load_run_sandbox_instance( run_id: &RunId, ) -> Result { let projection = state - .cached_run_projection(run_id) + .load_run_projection(run_id) .await .map_err(IntoResponse::into_response)?; projection diff --git a/lib/apps/fabro-server/src/server/handler/worker_control.rs b/lib/apps/fabro-server/src/server/handler/worker_control.rs index 7d68e179c..dd4d91532 100644 --- a/lib/apps/fabro-server/src/server/handler/worker_control.rs +++ b/lib/apps/fabro-server/src/server/handler/worker_control.rs @@ -35,7 +35,7 @@ async fn worker_control_stream( Query(query): Query, ws: WebSocketUpgrade, ) -> Response { - let projection = match state.cached_run_projection(&id).await { + let projection = match state.load_run_projection(&id).await { Ok(projection) => projection, Err(err) => return err.into_response(), }; diff --git a/lib/apps/fabro-server/src/server/pull_request_supervisor.rs b/lib/apps/fabro-server/src/server/pull_request_supervisor.rs index 60fec9a6e..a84371054 100644 --- a/lib/apps/fabro-server/src/server/pull_request_supervisor.rs +++ b/lib/apps/fabro-server/src/server/pull_request_supervisor.rs @@ -139,7 +139,7 @@ pub(in crate::server) async fn process_pull_request_creation( ) -> anyhow::Result<()> { let _create_guard = state.pull_request_create_locks.lock(run_id).await; let run_store = state.stores.runs.open_run(&run_id).await?; - let Some(run_state) = state.stores.runs.get_cached_projection(&run_id).await? else { + let Some(run_state) = state.stores.runs.load_run_projection(&run_id).await? else { return Ok(()); }; let Some(creation) = run_state @@ -257,7 +257,7 @@ pub(super) async fn recover_pending_pull_request_creations( if !can_dispatch(&run_id, active, failures) { continue; } - let projection = match state.stores.runs.get_cached_projection(&run_id).await { + let projection = match state.stores.runs.load_run_projection(&run_id).await { Ok(Some(projection)) => projection, Ok(None) => continue, Err(error) => { diff --git a/lib/apps/fabro-server/src/server/tests.rs b/lib/apps/fabro-server/src/server/tests.rs index 25bb99803..69cec239c 100644 --- a/lib/apps/fabro-server/src/server/tests.rs +++ b/lib/apps/fabro-server/src/server/tests.rs @@ -11630,7 +11630,109 @@ async fn get_run_state_exposes_pending_interviews() { } #[tokio::test] -async fn cache_backed_run_endpoints_reflect_events_appended_after_warmup() { +async fn restarted_run_state_details_load_from_sql_and_preserve_error_statuses() { + let object_store: Arc = + Arc::new(object_store::memory::InMemory::new()); + let summaries = fabro_store::test_support::test_run_summary_store(); + let blobs = fabro_store::test_support::test_blob_store(); + let first_store = Arc::new(fabro_store::test_support::test_database_with_stores( + Arc::clone(&object_store), + "runs", + std::time::Duration::from_millis(1), + None, + Arc::clone(&blobs), + Arc::clone(&summaries), + )); + let first_state = test_app_state_with_store( + default_test_server_settings(), + RunLayer::default(), + 5, + first_store, + ArtifactStore::new(Arc::clone(&object_store), "artifacts"), + ); + let healthy_id = fixtures::RUN_1; + let broken_id = fixtures::RUN_2; + create_durable_run_with_events(&first_state, healthy_id, &[ + workflow_event::Event::RunSubmitted { + definition_blob: None, + }, + ]) + .await; + create_succeeded_run(&first_state, broken_id).await; + first_state + .stores + .run_summaries + .test_delete_run_events(&broken_id) + .await + .unwrap(); + drop(first_state); + + let reopened_store = Arc::new(fabro_store::test_support::test_database_with_stores( + Arc::clone(&object_store), + "runs", + std::time::Duration::from_millis(1), + None, + blobs, + summaries, + )); + let reopened_state = test_app_state_with_store( + default_test_server_settings(), + RunLayer::default(), + 5, + reopened_store, + ArtifactStore::new(object_store, "artifacts"), + ); + assert_eq!( + reconcile_incomplete_runs_on_startup(&reopened_state) + .await + .unwrap(), + 0, + "startup reconciliation must not replay terminal histories" + ); + let app = crate::test_support::build_test_router(reopened_state); + + let healthy = app + .clone() + .oneshot( + Request::builder() + .method("GET") + .uri(api(&format!("/runs/{healthy_id}/state"))) + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + let healthy_body = response_json!(healthy, StatusCode::OK).await; + assert_eq!(healthy_body["spec"]["run_id"], healthy_id.to_string()); + + let missing = app + .clone() + .oneshot( + Request::builder() + .method("GET") + .uri(api(&format!("/runs/{}/state", fixtures::RUN_3))) + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + assert_status!(missing, StatusCode::NOT_FOUND).await; + + let broken = app + .oneshot( + Request::builder() + .method("GET") + .uri(api(&format!("/runs/{broken_id}/state"))) + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + assert_status!(broken, StatusCode::INTERNAL_SERVER_ERROR).await; +} + +#[tokio::test] +async fn run_projection_endpoints_reflect_events_appended_to_an_open_run() { let state = test_app_state(); let app = crate::test_support::build_test_router(Arc::clone(&state)); let run_id = create_run(&app, MINIMAL_DOT) @@ -11638,8 +11740,6 @@ async fn cache_backed_run_endpoints_reflect_events_appended_after_warmup() { .parse::() .unwrap(); - state.stores.runs.warm_projection_cache().await.unwrap(); - let run_store = state.stores.runs.open_run(&run_id).await.unwrap(); workflow_event::append_event(&run_store, &run_id, &workflow_event::Event::RunRunnable { source: fabro_types::RunRunnableSource::StartRequested, @@ -12976,14 +13076,14 @@ async fn run_tool_worker_token_can_use_client_backend_routes_across_runs() { assert_ne!(response.status(), StatusCode::FORBIDDEN); let created_child = create_run_with_bearer(&app, &run_tool_worker_token).await; - let cached = state + let projection = state .stores .runs - .get_cached_projection(&created_child) + .load_run_projection(&created_child) .await .unwrap() - .expect("created run should be cached"); - assert_eq!(cached.spec.provenance.subject, Principal::Worker { + .expect("created run should have a projection"); + assert_eq!(projection.spec.provenance.subject, Principal::Worker { run_id: parent_run_id, },); diff --git a/lib/components/fabro-store/src/slate/mod.rs b/lib/components/fabro-store/src/slate/mod.rs index b5906013d..aeb9b5973 100644 --- a/lib/components/fabro-store/src/slate/mod.rs +++ b/lib/components/fabro-store/src/slate/mod.rs @@ -1,4 +1,3 @@ -mod projection_cache; mod run_store; use std::collections::HashMap; @@ -9,13 +8,11 @@ use std::time::Duration; use chrono::{DateTime, Utc}; use fabro_types::{RunId, SessionId}; use object_store::ObjectStore; -pub(crate) use projection_cache::CachedRunProjection; -use projection_cache::RunProjectionCache; +pub(crate) use run_store::CachedRunProjection; pub use run_store::RunDatabase; use run_store::RunDatabaseInner; use slatedb::config::{CompressionCodec, Settings}; use tokio::sync::{Mutex, MutexGuard, OnceCell}; -use tracing::warn; use crate::{BlobStore, Error, EventPayload, Result, RunProjection, RunSummaryStore, keys}; @@ -28,15 +25,13 @@ pub struct UnreadableRun { #[derive(Clone)] pub struct Database { - object_store: Arc, - base_prefix: String, - flush_interval: Duration, - cache_path: Option, - db: Arc>, - active_runs: Arc>>>, - blobs: Arc, - projection_cache: Arc, - projection_cache_warmed: Arc>, + object_store: Arc, + base_prefix: String, + flush_interval: Duration, + cache_path: Option, + db: Arc>, + active_runs: Arc>>>, + blobs: Arc, run_summary_store: Arc, } @@ -67,8 +62,6 @@ impl Database { db: Arc::new(OnceCell::new()), active_runs: Arc::new(Mutex::new(HashMap::new())), blobs, - projection_cache: Arc::new(RunProjectionCache::default()), - projection_cache_warmed: Arc::new(OnceCell::new()), run_summary_store, } } @@ -126,14 +119,7 @@ impl Database { /// Builds a run handle wired to the Database-owned shared stores. async fn open_run_database(&self, run_id: &RunId, read_only: bool) -> Result { - RunDatabase::build( - *run_id, - read_only, - self.blobs(), - Arc::clone(&self.projection_cache), - self.run_summary_store(), - ) - .await + RunDatabase::build(*run_id, read_only, self.blobs(), self.run_summary_store()).await } pub async fn create_run_with_first_event( @@ -168,22 +154,15 @@ impl Database { MutexGuard<'_, HashMap>>, RunDatabase, )> { - self.warm_projection_cache().await?; let active_runs = self.active_runs.lock().await; if active_runs.contains_key(run_id) || self.run_summary_store.contains(run_id).await? { return Err(Error::RunAlreadyExists(run_id.to_string())); } - let run_store = RunDatabase::build_empty( - *run_id, - self.blobs(), - Arc::clone(&self.projection_cache), - self.run_summary_store(), - ); + let run_store = RunDatabase::build_empty(*run_id, self.blobs(), self.run_summary_store()); Ok((active_runs, run_store)) } pub async fn open_run(&self, run_id: &RunId) -> Result { - self.warm_projection_cache().await?; // Keep the active-writer miss and insert atomic. Otherwise concurrent // callers can create independent writers with the same recovered seq. let mut active_runs = self.active_runs.lock().await; @@ -219,38 +198,11 @@ impl Database { self.open_run_database(run_id, true).await } - pub async fn warm_projection_cache(&self) -> Result<()> { - self.projection_cache_warmed - .get_or_try_init(|| async { - let run_ids = self.run_summary_store.list_run_ids().await?; - let mut entries = Vec::new(); - for run_id in run_ids { - match RunDatabase::build_cached_projection(&self.run_summary_store, &run_id) - .await - { - Ok(Some(entry)) => entries.push(entry), - Ok(None) => {} - Err(err) => { - warn!( - run_id = %run_id, - error = %err, - "Skipping run during projection cache warmup" - ); - } - } - } - self.projection_cache.replace_all(entries); - Ok::<_, Error>(()) - }) - .await?; - Ok(()) - } - pub async fn list_unreadable_runs(&self) -> Result> { 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_cached_projection(&self.run_summary_store, &run_id).await { + match RunDatabase::build_projection(&self.run_summary_store, &run_id).await { Ok(Some(_)) => {} Ok(None) => unreadable.push(UnreadableRun { run_id, @@ -284,7 +236,6 @@ impl Database { .test_insert_unvalidated_event(run_id, seq, payload) .await?; self.active_runs.lock().await.remove(run_id); - self.projection_cache.remove(run_id); Ok(()) } @@ -304,15 +255,20 @@ impl Database { Ok(()) } - pub async fn get_cached_projection( - &self, - run_id: &RunId, - ) -> Result>> { - self.warm_projection_cache().await?; - Ok(self - .projection_cache - .projection_snapshot(run_id) - .map(|(projection, _)| projection)) + pub async fn load_run_projection(&self, run_id: &RunId) -> Result>> { + if let Some(active) = self.get_active_run(run_id).await { + if !active.matches_run(run_id) { + return Err(Error::Other(format!( + "active run cache mismatch for run_id {run_id:?}" + ))); + } + return active.projection_snapshot().await.map(Some); + } + Ok( + RunDatabase::build_projection(&self.run_summary_store, run_id) + .await? + .map(|entry| entry.projection), + ) } /// Resolves the run that owns `session_id` from the canonical typed @@ -321,10 +277,6 @@ impl Database { self.run_summary_store.find_session_owner(session_id).await } - pub(crate) fn remove_cached_run(&self, run_id: &RunId) { - self.projection_cache.remove(run_id); - } - pub async fn delete_run(&self, run_id: &RunId) -> Result<()> { let mut active_runs = self.active_runs.lock().await; let active = active_runs.get(run_id).cloned(); @@ -336,7 +288,6 @@ impl Database { .delete_canonical(run_id, Utc::now().timestamp_millis()) .await?; active_runs.remove(run_id); - self.remove_cached_run(run_id); Ok(()) } @@ -1017,7 +968,7 @@ mod tests { } #[tokio::test] - async fn rejected_transition_writes_nothing_and_preserves_projection_cache() { + async fn rejected_transition_writes_nothing_and_preserves_projection_state() { let (_object_store, store) = make_store(); let run_id = test_run_id("run-1"); let run = store.create_run(&run_id).await.unwrap(); @@ -1043,8 +994,8 @@ mod tests { )); assert_eq!(run.list_events().await.unwrap(), events_before); assert_eq!(run.state().await.unwrap().status, RunStatus::Runnable); - let (projection, last_seq) = store.projection_cache.projection_snapshot(&run_id).unwrap(); - assert_eq!(last_seq, 4); + let projection = store.load_run_projection(&run_id).await.unwrap().unwrap(); + assert_eq!(run.last_event_seq().await.unwrap(), Some(4)); assert_eq!(projection.status, RunStatus::Runnable); } @@ -1062,7 +1013,8 @@ mod tests { .unwrap_err(); assert!(matches!(err, Error::EventRejected { .. })); - let (projection, last_seq) = store.projection_cache.projection_snapshot(&run_id).unwrap(); + 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( run_id, Arc::unwrap_or_clone(projection), @@ -1080,7 +1032,7 @@ mod tests { 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; - let cached_before = store.projection_cache.projection_snapshot(&run_id).unwrap(); + let projection_before = store.load_run_projection(&run_id).await.unwrap().unwrap(); summaries.close_pool().await; let result = run @@ -1096,9 +1048,8 @@ mod tests { result, Err(Error::Sqlite(sqlx::Error::PoolClosed)) )); - let cached = store.projection_cache.projection_snapshot(&run_id).unwrap(); - assert_eq!(cached.1, cached_before.1); - assert_eq!(cached.0.title, cached_before.0.title); + let projection_after = store.load_run_projection(&run_id).await.unwrap().unwrap(); + assert!(Arc::ptr_eq(&projection_before, &projection_after)); } #[tokio::test] @@ -1463,13 +1414,15 @@ mod tests { } #[tokio::test] - async fn projection_cache_warmup_rebuilds_full_projections() { + async fn inactive_projection_loads_on_demand_without_registering_run() { let (_directory, summaries) = make_run_summary_store().await; let (object_store, store) = make_store_with_run_summaries(Arc::clone(&summaries)); let run_1 = store.create_run(&test_run_id("run-1")).await.unwrap(); let run_2 = store.create_run(&test_run_id("run-2")).await.unwrap(); + let run_3 = store.create_run(&test_run_id("run-3")).await.unwrap(); append_completed(&run_1, "run-1", dt("2026-03-27T12:00:00Z")).await; - append_running(&run_2, "run-2", dt("2026-03-27T12:00:10Z")).await; + append_completed(&run_2, "run-2", dt("2026-03-27T12:00:10Z")).await; + append_completed(&run_3, "run-3", dt("2026-03-27T12:00:20Z")).await; let reopened = store_test_support::test_database_with_stores( object_store, @@ -1479,32 +1432,141 @@ mod tests { store_test_support::test_blob_store(), summaries, ); - reopened.warm_projection_cache().await.unwrap(); + assert!(reopened.active_runs.lock().await.is_empty()); - let (run_1_projection, run_1_last_seq) = reopened - .projection_cache - .projection_snapshot(&test_run_id("run-1")) + let projection = reopened + .load_run_projection(&test_run_id("run-2")) + .await + .unwrap() .unwrap(); - assert_eq!(run_1_projection.status, RunStatus::Succeeded { + + assert_eq!(projection.spec.run_id, test_run_id("run-2")); + assert_eq!(projection.status, RunStatus::Succeeded { reason: SuccessReason::Completed, }); - assert_eq!(run_1_last_seq, 5); - let (run_2_projection, run_2_last_seq) = reopened - .projection_cache - .projection_snapshot(&test_run_id("run-2")) - .unwrap(); - assert_eq!(run_2_projection.status, RunStatus::Running); - assert_eq!(run_2_last_seq, 4); + assert!( + reopened.active_runs.lock().await.is_empty(), + "inactive detail reads must stay request-local" + ); } #[tokio::test] - async fn required_run_summary_append_refreshes_cache_and_delete_removes_rows() { + async fn active_projection_reads_are_coherent_immutable_snapshots() { + let (_object_store, store) = make_store(); + 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; + + let first = store.load_run_projection(&run_id).await.unwrap().unwrap(); + let second = store.load_run_projection(&run_id).await.unwrap().unwrap(); + assert!(Arc::ptr_eq(&first, &second)); + let original_title = first.title.clone(); + + run.append_event(&event_payload( + "run-1", + "2026-03-27T12:00:01Z", + "run.title.updated", + &serde_json::json!({ "title": "New title" }), + )) + .await + .unwrap(); + + let updated = store.load_run_projection(&run_id).await.unwrap().unwrap(); + assert!(!Arc::ptr_eq(&first, &updated)); + assert_eq!(first.title, original_title); + assert_eq!(updated.title, "New title"); + } + + #[tokio::test] + async fn inactive_projection_replay_accepts_sequence_gaps_and_failures_do_not_poison_reads() { + let (_directory, summaries) = make_run_summary_store().await; + let (object_store, store) = make_store_with_run_summaries(Arc::clone(&summaries)); + let healthy_id = test_run_id("run-1"); + let broken_id = test_run_id("run-2"); + let healthy = store.create_run(&healthy_id).await.unwrap(); + let broken = store.create_run(&broken_id).await.unwrap(); + append_created(&healthy, "run-1", dt("2026-03-27T12:00:00Z")).await; + append_created(&broken, "run-2", dt("2026-03-27T12:00:10Z")).await; + store.remove_active_run(&healthy_id).await; + store.remove_active_run(&broken_id).await; + store + .put_unvalidated_run_event( + &healthy_id, + 3, + event_payload( + "run-1", + "2026-03-27T12:00:02Z", + "run.title.updated", + &serde_json::json!({ "title": "Imported gap" }), + ) + .as_value(), + ) + .await + .unwrap(); + summaries.test_delete_run_events(&broken_id).await.unwrap(); + + let reopened = store_test_support::test_database_with_stores( + object_store, + "runs", + Duration::from_millis(1), + None, + store_test_support::test_blob_store(), + summaries, + ); + assert!(matches!( + reopened.load_run_projection(&broken_id).await, + Err(Error::RunHeadMismatch { .. }) + )); + + let projection = reopened + .load_run_projection(&healthy_id) + .await + .unwrap() + .unwrap(); + assert_eq!(projection.title, "Imported gap"); + assert!(reopened.active_runs.lock().await.is_empty()); + } + + #[tokio::test] + async fn inactive_projection_read_concurrent_with_append_is_one_coherent_snapshot() { + let (_directory, summaries) = make_run_summary_store().await; + let (object_store, writer_store) = make_store_with_run_summaries(Arc::clone(&summaries)); + let run_id = test_run_id("run-1"); + let writer = writer_store.create_run(&run_id).await.unwrap(); + append_created(&writer, "run-1", dt("2026-03-27T12:00:00Z")).await; + let reader_store = store_test_support::test_database_with_stores( + object_store, + "runs", + Duration::from_millis(1), + None, + store_test_support::test_blob_store(), + summaries, + ); + let original_title = writer.state().await.unwrap().title; + let update = event_payload( + "run-1", + "2026-03-27T12:00:01Z", + "run.title.updated", + &serde_json::json!({ "title": "Concurrent title" }), + ); + + let (loaded, appended) = tokio::join!( + reader_store.load_run_projection(&run_id), + writer.append_event(&update) + ); + + appended.unwrap(); + let loaded = loaded.unwrap().unwrap(); + assert!(loaded.title == original_title || loaded.title == "Concurrent title"); + assert!(reader_store.active_runs.lock().await.is_empty()); + } + + #[tokio::test] + async fn required_run_summary_append_refreshes_active_projection_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(); - run.append_event(&event_payload( "run-1", "2026-03-27T12:00:01Z", @@ -1583,12 +1645,13 @@ mod tests { .await .unwrap(); - let (projection, last_seq) = store - .projection_cache - .projection_snapshot(&test_run_id("run-1")) + let projection = store + .load_run_projection(&test_run_id("run-1")) + .await + .unwrap() .unwrap(); assert_eq!(projection.status, RunStatus::Running); - assert_eq!(last_seq, 7); + assert_eq!(run.last_event_seq().await.unwrap(), Some(7)); assert_eq!( projection .stage(&StageId::new("review", 1)) @@ -1620,8 +1683,9 @@ mod tests { store.delete_run(&test_run_id("run-1")).await.unwrap(); assert!( store - .projection_cache - .projection_snapshot(&test_run_id("run-1")) + .load_run_projection(&test_run_id("run-1")) + .await + .unwrap() .is_none() ); assert!( @@ -1649,11 +1713,8 @@ mod tests { store_test_support::test_blob_store(), summaries, ); - reopened.warm_projection_cache().await.unwrap(); - - // If opening or projecting the run starts at the beginning, this - // unreadable old key makes the operation fail. A hydrated run starts - // after the shared projection's last sequence instead. + // Opening and projecting from canonical SQLite must not inspect an + // unreadable key in the retained legacy event history. let mut unreadable_old_key = keys::run_event_seq_prefix(&run_id, 2).as_ref().to_vec(); unreadable_old_key.push(0xff); reopened @@ -1744,7 +1805,7 @@ mod tests { }); let cached = reopened - .get_cached_projection(&run_id) + .load_run_projection(&run_id) .await .unwrap() .unwrap(); diff --git a/lib/components/fabro-store/src/slate/projection_cache.rs b/lib/components/fabro-store/src/slate/projection_cache.rs deleted file mode 100644 index 33a0842a8..000000000 --- a/lib/components/fabro-store/src/slate/projection_cache.rs +++ /dev/null @@ -1,78 +0,0 @@ -use std::collections::HashMap; -use std::sync::{Arc, Mutex, MutexGuard}; - -use fabro_types::{RunId, RunProjection}; - -#[derive(Debug, Clone)] -pub(crate) struct CachedRunProjection { - 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 { - Self { - run_id, - projection: Arc::new(projection), - last_seq, - } - } -} - -#[derive(Debug, Default)] -pub(crate) struct RunProjectionCache { - // Cache operations are bounded in-memory work and never await. Keeping - // this lock synchronous lets a committed event update both projection - // caches without introducing a cancellation point. - state: Mutex, -} - -#[derive(Debug, Default)] -struct RunProjectionCacheState { - entries: HashMap, -} - -impl RunProjectionCacheState { - fn replace_all(&mut self, entries: Vec) { - self.entries.clear(); - for entry in entries { - self.insert(entry); - } - } - - fn insert(&mut self, entry: CachedRunProjection) { - self.entries.insert(entry.run_id, entry); - } - - fn remove(&mut self, run_id: &RunId) { - self.entries.remove(run_id); - } -} - -impl RunProjectionCache { - fn lock(&self) -> MutexGuard<'_, RunProjectionCacheState> { - self.state.lock().expect( - "run projection cache mutex is never poisoned: no code panics while holding this lock", - ) - } - - pub(crate) fn replace_all(&self, entries: Vec) { - self.lock().replace_all(entries); - } - - pub(crate) fn replace(&self, entry: CachedRunProjection) { - self.lock().insert(entry); - } - - pub(crate) fn projection_snapshot(&self, run_id: &RunId) -> Option<(Arc, u32)> { - self.lock() - .entries - .get(run_id) - .map(|entry| (Arc::clone(&entry.projection), entry.last_seq)) - } - - pub(crate) fn remove(&self, run_id: &RunId) { - self.lock().remove(run_id); - } -} diff --git a/lib/components/fabro-store/src/slate/run_store.rs b/lib/components/fabro-store/src/slate/run_store.rs index cdbf92cc2..20daa63ca 100644 --- a/lib/components/fabro-store/src/slate/run_store.rs +++ b/lib/components/fabro-store/src/slate/run_store.rs @@ -6,7 +6,6 @@ use futures::Stream; use tokio::sync::{Mutex as AsyncMutex, broadcast, mpsc}; use tokio_stream::wrappers::UnboundedReceiverStream; -use super::projection_cache::{CachedRunProjection, RunProjectionCache}; use crate::run_state::{EventProjectionCache, RunProjectionReducer}; use crate::{ BlobStore, Error, EventEnvelope, EventPayload, Result, RunProjection, RunSummaryStore, StageId, @@ -17,6 +16,23 @@ use crate::{ /// from SQLite. const EVENT_BROADCAST_CAPACITY: usize = 1024; +#[derive(Debug, Clone)] +pub(crate) struct CachedRunProjection { + 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 { + Self { + run_id, + projection: Arc::new(projection), + last_seq, + } + } +} + #[derive(Clone)] pub struct RunDatabase { inner: Arc, @@ -34,13 +50,12 @@ impl std::fmt::Debug for RunDatabase { } pub(crate) struct RunDatabaseInner { - pub(crate) run_id: RunId, - blob_store: Arc, - pub(crate) state_lock: AsyncMutex<()>, - projection_cache: StdMutex, - shared_projection_cache: Arc, - run_summary_store: Arc, - event_tx: broadcast::Sender, + pub(crate) run_id: RunId, + blob_store: Arc, + pub(crate) state_lock: AsyncMutex<()>, + projection_cache: StdMutex, + run_summary_store: Arc, + event_tx: broadcast::Sender, } impl RunDatabaseInner { @@ -56,30 +71,19 @@ impl RunDatabase { run_id: RunId, read_only: bool, blob_store: Arc, - shared_projection_cache: Arc, run_summary_store: Arc, ) -> Result { - let projection_cache = if let Some((projection, last_seq)) = - shared_projection_cache.projection_snapshot(&run_id) - { - EventProjectionCache { - last_seq, - state: Some(projection), - } - } else { - let cached = Self::build_cached_projection(&run_summary_store, &run_id) - .await? - .ok_or_else(|| Error::RunNotFound(run_id.to_string()))?; - EventProjectionCache { - last_seq: cached.last_seq, - state: Some(cached.projection), - } + let cached = Self::build_projection(&run_summary_store, &run_id) + .await? + .ok_or_else(|| Error::RunNotFound(run_id.to_string()))?; + let projection_cache = EventProjectionCache { + last_seq: cached.last_seq, + state: Some(cached.projection), }; - Ok(Self::from_projection_cache( + Ok(Self::from_event_projection_cache( run_id, read_only, blob_store, - shared_projection_cache, run_summary_store, projection_cache, )) @@ -88,24 +92,21 @@ impl RunDatabase { pub(crate) fn build_empty( run_id: RunId, blob_store: Arc, - shared_projection_cache: Arc, run_summary_store: Arc, ) -> Self { - Self::from_projection_cache( + Self::from_event_projection_cache( run_id, false, blob_store, - shared_projection_cache, run_summary_store, EventProjectionCache::default(), ) } - fn from_projection_cache( + fn from_event_projection_cache( run_id: RunId, read_only: bool, blob_store: Arc, - shared_projection_cache: Arc, run_summary_store: Arc, projection_cache: EventProjectionCache, ) -> Self { @@ -116,7 +117,6 @@ impl RunDatabase { blob_store, state_lock: AsyncMutex::new(()), projection_cache: StdMutex::new(projection_cache), - shared_projection_cache, run_summary_store, event_tx, }), @@ -154,7 +154,7 @@ impl RunDatabase { self.inner.run_id == *run_id } - pub(crate) async fn build_cached_projection( + pub(crate) async fn build_projection( store: &RunSummaryStore, run_id: &RunId, ) -> Result> { @@ -173,7 +173,7 @@ impl RunDatabase { ))) } - async fn projected_state(&self) -> Result> { + pub(super) async fn projection_snapshot(&self) -> Result> { let _state_guard = self.inner.state_lock.lock().await; self.projected_state_locked() } @@ -201,7 +201,6 @@ impl RunDatabase { projection_cache.state = Some(Arc::clone(&cached.projection)); projection_cache.last_seq = event.seq; } - self.inner.shared_projection_cache.replace(cached.clone()); } pub(crate) fn publish(&self, event: &EventEnvelope) { @@ -274,7 +273,7 @@ impl RunDatabase { ) -> Result { let (envelope, cached) = self.commit_event_locked(payload, event).await?; // Keep post-commit propagation await-free: cancellation after SQLite - // commits must not leave either cache stale or omit the broadcast. + // commits must not leave in-memory state stale or omit the broadcast. self.install_in_memory_state(&envelope, &cached); self.publish(&envelope); Ok(envelope) @@ -434,7 +433,7 @@ impl RunDatabase { } pub async fn state(&self) -> Result { - Ok(Arc::unwrap_or_clone(self.projected_state().await?)) + Ok(Arc::unwrap_or_clone(self.projection_snapshot().await?)) } }