From 13f3ff867bd6e0c1d1fb2a2f7ea2d5d9d3288e57 Mon Sep 17 00:00:00 2001 From: Scott Werner Date: Wed, 2 Sep 2026 11:43:17 -0400 Subject: [PATCH] Simplify on-demand run projection loading Drop the unreachable active-run mismatch guard that was copied into load_run_projection: the active-runs map is only ever inserted under the handle's own run ID, so the check could never fire. Remove it from the two pre-existing sites too and delete matches_run. Trim install_in_memory_state to take only the committed projection, since the event envelope duplicated last_seq and the inner scope only existed to release the lock before the now-removed shared cache update. Add a From impl so RunDatabase::build no longer hand-builds EventProjectionCache, and rename projected_state_locked to match its projection_snapshot sibling. In fabro-server, have reject_if_archived and ensure_run_exists read the run summary row instead of replaying the full event history for inactive runs; the summary is written in the same transaction as the event. Fold the repeated store-reopen fixtures in fabro-store and fabro-server tests into helpers, and fix a stale comment about the deleted shared projection cache. Co-Authored-By: Claude Fable 5.1 --- lib/apps/fabro-server/src/server.rs | 10 ++- .../src/server/handler/artifacts.rs | 15 +++- lib/apps/fabro-server/src/server/tests.rs | 56 ++++++------- lib/components/fabro-store/src/run_state.rs | 4 +- lib/components/fabro-store/src/slate/mod.rs | 84 ++++++------------- .../fabro-store/src/slate/run_store.rs | 44 +++++----- 6 files changed, 93 insertions(+), 120 deletions(-) diff --git a/lib/apps/fabro-server/src/server.rs b/lib/apps/fabro-server/src/server.rs index be2769a4a..bb69cc765 100644 --- a/lib/apps/fabro-server/src/server.rs +++ b/lib/apps/fabro-server/src/server.rs @@ -4534,8 +4534,14 @@ 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.load_run_projection(run_id).await.ok()?; - projection.archived_at.is_some().then(|| { + let summary = state + .stores + .run_summaries + .get(run_id, Utc::now()) + .await + .ok() + .flatten()?; + summary.lifecycle.archived_at.is_some().then(|| { ApiError::new( StatusCode::CONFLICT, operations::archived_rejection_message(run_id), diff --git a/lib/apps/fabro-server/src/server/handler/artifacts.rs b/lib/apps/fabro-server/src/server/handler/artifacts.rs index 6f175660a..d597e05be 100644 --- a/lib/apps/fabro-server/src/server/handler/artifacts.rs +++ b/lib/apps/fabro-server/src/server/handler/artifacts.rs @@ -131,11 +131,18 @@ async fn read_run_blob( } async fn ensure_run_exists(state: &AppState, run_id: &RunId) -> Result<(), Response> { - state - .load_run_projection(run_id) + match state + .stores + .run_summaries + .get(run_id, chrono::Utc::now()) .await - .map(|_| ()) - .map_err(IntoResponse::into_response) + { + Ok(Some(_)) => Ok(()), + Ok(None) => Err(ApiError::not_found("Run not found.").into_response()), + Err(err) => { + Err(ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string()).into_response()) + } + } } async fn list_run_artifacts( diff --git a/lib/apps/fabro-server/src/server/tests.rs b/lib/apps/fabro-server/src/server/tests.rs index 69cec239c..9533e0d2f 100644 --- a/lib/apps/fabro-server/src/server/tests.rs +++ b/lib/apps/fabro-server/src/server/tests.rs @@ -11629,27 +11629,37 @@ async fn get_run_state_exposes_pending_interviews() { ); } +/// Builds an app state over shared object, blob, and summary stores so a test +/// can drop it and open a second state that sees the same durable data. +fn test_app_state_over_shared_stores( + object_store: &Arc, + blobs: &Arc, + summaries: &Arc, +) -> Arc { + let 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), + )); + test_app_state_with_store( + default_test_server_settings(), + RunLayer::default(), + 5, + store, + ArtifactStore::new(Arc::clone(object_store), "artifacts"), + ) +} + #[tokio::test] 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 first_state = test_app_state_over_shared_stores(&object_store, &blobs, &summaries); let healthy_id = fixtures::RUN_1; let broken_id = fixtures::RUN_2; create_durable_run_with_events(&first_state, healthy_id, &[ @@ -11667,21 +11677,7 @@ async fn restarted_run_state_details_load_from_sql_and_preserve_error_statuses() .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"), - ); + let reopened_state = test_app_state_over_shared_stores(&object_store, &blobs, &summaries); assert_eq!( reconcile_incomplete_runs_on_startup(&reopened_state) .await diff --git a/lib/components/fabro-store/src/run_state.rs b/lib/components/fabro-store/src/run_state.rs index 37068764b..28780c1a5 100644 --- a/lib/components/fabro-store/src/run_state.rs +++ b/lib/components/fabro-store/src/run_state.rs @@ -28,8 +28,8 @@ use crate::{Error, EventEnvelope, Result}; #[derive(Debug, Clone, Default)] pub(crate) struct EventProjectionCache { pub last_seq: u32, - // Arc-shared with the shared projection cache so opening a run does not - // deep-copy the projection; mutated copy-on-write via `Arc::make_mut`. + // Arc-shared with readers so a snapshot never deep-copies the projection; + // mutated copy-on-write via `Arc::make_mut`. pub state: Option>, } diff --git a/lib/components/fabro-store/src/slate/mod.rs b/lib/components/fabro-store/src/slate/mod.rs index aeb9b5973..5678ee954 100644 --- a/lib/components/fabro-store/src/slate/mod.rs +++ b/lib/components/fabro-store/src/slate/mod.rs @@ -129,7 +129,7 @@ impl Database { ) -> 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(&envelope, &cached); + run_store.install_in_memory_state(&cached); Self::cache_active_run(&mut active_runs, &run_store); run_store.publish(&envelope); Ok(run_store) @@ -168,11 +168,6 @@ impl Database { let mut active_runs = self.active_runs.lock().await; if let Some(active) = active_run_from(&active_runs, run_id) { - if !active.matches_run(run_id) { - return Err(Error::Other(format!( - "active run cache mismatch for run_id {run_id:?}" - ))); - } return Ok(active); } if !self.run_summary_store.contains(run_id).await? { @@ -185,11 +180,6 @@ impl Database { pub async fn open_run_reader(&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 Ok(active.read_only_clone()); } if !self.run_summary_store.contains(run_id).await? { @@ -257,11 +247,6 @@ impl Database { 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( @@ -416,6 +401,22 @@ mod tests { (object_store, store) } + /// Reopens a `Database` over an existing object store and SQLite summary + /// store, simulating a process restart with no active run handles. + fn reopen_store( + object_store: Arc, + run_summaries: Arc, + ) -> Database { + store_test_support::test_database_with_stores( + object_store, + "runs", + Duration::from_millis(1), + None, + store_test_support::test_blob_store(), + run_summaries, + ) + } + #[tokio::test] async fn retire_refresh_token_keyspace_clears_the_prefix_and_is_idempotent() { let (_object_store, store) = make_store(); @@ -1393,14 +1394,7 @@ mod tests { 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_with_stores( - object_store, - "runs", - Duration::from_millis(1), - None, - store_test_support::test_blob_store(), - summaries, - ); + let reopened = reopen_store(object_store, summaries); let summary = reopened .run_summary_store() .list_all(Utc::now()) @@ -1424,14 +1418,7 @@ mod tests { 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, - "runs", - Duration::from_millis(1), - None, - store_test_support::test_blob_store(), - summaries, - ); + let reopened = reopen_store(object_store, summaries); assert!(reopened.active_runs.lock().await.is_empty()); let projection = reopened @@ -1505,14 +1492,7 @@ mod tests { .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, - ); + let reopened = reopen_store(object_store, summaries); assert!(matches!( reopened.load_run_projection(&broken_id).await, Err(Error::RunHeadMismatch { .. }) @@ -1534,14 +1514,7 @@ mod tests { 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 reader_store = reopen_store(object_store, summaries); let original_title = writer.state().await.unwrap().title; let update = event_payload( "run-1", @@ -1705,14 +1678,7 @@ mod tests { let run = store.create_run(&run_id).await.unwrap(); append_completed(&run, "run-1", dt("2026-03-27T12:00:00Z")).await; - let reopened = store_test_support::test_database_with_stores( - object_store, - "runs", - Duration::from_millis(1), - None, - store_test_support::test_blob_store(), - summaries, - ); + let reopened = reopen_store(object_store, summaries); // 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(); @@ -1804,13 +1770,13 @@ mod tests { reason: FailureReason::WorkflowError, }); - let cached = reopened + let projection = reopened .load_run_projection(&run_id) .await .unwrap() .unwrap(); - assert_eq!(cached.title, "Renamed failed run"); - assert_eq!(cached.status, RunStatus::Failed { + assert_eq!(projection.title, "Renamed failed run"); + assert_eq!(projection.status, RunStatus::Failed { reason: FailureReason::WorkflowError, }); } diff --git a/lib/components/fabro-store/src/slate/run_store.rs b/lib/components/fabro-store/src/slate/run_store.rs index 20daa63ca..e64fe6fc1 100644 --- a/lib/components/fabro-store/src/slate/run_store.rs +++ b/lib/components/fabro-store/src/slate/run_store.rs @@ -16,6 +16,9 @@ use crate::{ /// from SQLite. const EVENT_BROADCAST_CAPACITY: usize = 1024; +/// A run's projection as of its last committed event. Produced by replaying +/// 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) run_id: RunId, @@ -33,6 +36,15 @@ impl CachedRunProjection { } } +impl From for EventProjectionCache { + fn from(cached: CachedRunProjection) -> Self { + Self { + last_seq: cached.last_seq, + state: Some(cached.projection), + } + } +} + #[derive(Clone)] pub struct RunDatabase { inner: Arc, @@ -76,16 +88,12 @@ impl RunDatabase { 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_event_projection_cache( run_id, read_only, blob_store, run_summary_store, - projection_cache, + cached.into(), )) } @@ -150,10 +158,6 @@ impl RunDatabase { self.inner.event_tx.subscribe() } - pub(crate) fn matches_run(&self, run_id: &RunId) -> bool { - self.inner.run_id == *run_id - } - pub(crate) async fn build_projection( store: &RunSummaryStore, run_id: &RunId, @@ -175,10 +179,10 @@ impl RunDatabase { pub(super) async fn projection_snapshot(&self) -> Result> { let _state_guard = self.inner.state_lock.lock().await; - self.projected_state_locked() + self.projection_snapshot_locked() } - fn projected_state_locked(&self) -> Result> { + fn projection_snapshot_locked(&self) -> Result> { self.inner .lock_projection_cache() .state @@ -191,16 +195,10 @@ impl RunDatabase { }) } - pub(crate) fn install_in_memory_state( - &self, - event: &EventEnvelope, - cached: &CachedRunProjection, - ) { - { - let mut projection_cache = self.inner.lock_projection_cache(); - projection_cache.state = Some(Arc::clone(&cached.projection)); - projection_cache.last_seq = event.seq; - } + pub(crate) fn install_in_memory_state(&self, cached: &CachedRunProjection) { + let mut projection_cache = self.inner.lock_projection_cache(); + projection_cache.state = Some(Arc::clone(&cached.projection)); + projection_cache.last_seq = cached.last_seq; } pub(crate) fn publish(&self, event: &EventEnvelope) { @@ -244,7 +242,7 @@ impl RunDatabase { payload.validate(&self.inner.run_id)?; let event = RunEvent::try_from(payload)?; let _state_guard = self.inner.state_lock.lock().await; - let projection = self.projected_state_locked()?; + let projection = self.projection_snapshot_locked()?; if !predicate(&projection) { return Ok(None); } @@ -274,7 +272,7 @@ impl RunDatabase { let (envelope, cached) = 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(&envelope, &cached); + self.install_in_memory_state(&cached); self.publish(&envelope); Ok(envelope) }