diff --git a/lib/apps/fabro-server/src/run_files.rs b/lib/apps/fabro-server/src/run_files.rs index ebe3f7f53..1f07d09af 100644 --- a/lib/apps/fabro-server/src/run_files.rs +++ b/lib/apps/fabro-server/src/run_files.rs @@ -1195,15 +1195,13 @@ async fn load_projection( state: &Arc, run_id: &RunId, ) -> std::result::Result { - let reader = state + let cached = state .store_ref() - .open_run_reader(run_id) + .get_cached_run(run_id) .await - .map_err(|_| ApiError::not_found("Run not found."))?; - reader - .state() - .await - .map_err(|err| ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string())) + .map_err(|err| ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string()))? + .ok_or_else(|| ApiError::not_found("Run not found."))?; + Ok((*cached.projection).clone()) } async fn reconnect_run_sandbox( diff --git a/lib/apps/fabro-server/src/server.rs b/lib/apps/fabro-server/src/server.rs index fac2dd51a..ed67c1898 100644 --- a/lib/apps/fabro-server/src/server.rs +++ b/lib/apps/fabro-server/src/server.rs @@ -2686,9 +2686,8 @@ async fn delete_run_internal( } async fn load_durable_run_status(state: &AppState, id: &RunId) -> Option { - let run_store = state.stores.runs.open_run(id).await.ok()?; - let projection = run_store.state().await.ok()?; - Some(projection.status) + let cached = state.stores.runs.get_cached_run(id).await.ok()??; + Some(cached.projection.status) } async fn delete_run_sandbox_resource( @@ -4521,9 +4520,8 @@ 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 run_store = state.stores.runs.open_run_reader(run_id).await.ok()?; - let projection = run_store.state().await.ok()?; - projection.archived_at.is_some().then(|| { + let cached = state.stores.runs.get_cached_run(run_id).await.ok()??; + cached.projection.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 277bc1e59..0d7403e0e 100644 --- a/lib/apps/fabro-server/src/server/handler/artifacts.rs +++ b/lib/apps/fabro-server/src/server/handler/artifacts.rs @@ -119,16 +119,16 @@ async fn read_run_blob( } async fn load_run_spec(state: &AppState, run_id: &RunId) -> Result { - let run_store = state + let cached = state .stores .runs - .open_run_reader(run_id) + .get_cached_run(run_id) .await - .map_err(|_| ApiError::not_found("Run not found.").into_response())?; - let run_state = run_store.state().await.map_err(|err| { - ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string()).into_response() - })?; - Ok(run_state.spec) + .map_err(|err| { + ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string()).into_response() + })? + .ok_or_else(|| ApiError::not_found("Run not found.").into_response())?; + Ok(cached.projection.spec.clone()) } async fn list_run_artifacts( diff --git a/lib/apps/fabro-server/src/server/handler/events.rs b/lib/apps/fabro-server/src/server/handler/events.rs index b8508bd5c..131c3f956 100644 --- a/lib/apps/fabro-server/src/server/handler/events.rs +++ b/lib/apps/fabro-server/src/server/handler/events.rs @@ -390,8 +390,8 @@ async fn attach_run_events( let start_seq = match params.since_seq { Some(seq) if seq >= 1 => seq, Some(_) => 1, - None => match run_store.list_events().await { - Ok(events) => events.last().map_or(1, |event| event.seq.saturating_add(1)), + None => match run_store.last_event_seq().await { + Ok(last_seq) => last_seq.map_or(1, |seq| seq.saturating_add(1)), Err(err) => { return ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string()) .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 5b7b9bc4b..e2982abed 100644 --- a/lib/apps/fabro-server/src/server/handler/graph.rs +++ b/lib/apps/fabro-server/src/server/handler/graph.rs @@ -242,17 +242,17 @@ async fn load_run_dot_source(state: &AppState, id: &RunId) -> Result match run_store.state().await { - Ok(run_state) => run_state.spec.graph_source, - Err(err) => { - return Err( - ApiError::new(StatusCode::BAD_GATEWAY, err.to_string()).into_response() - ); - } - }, - Err(_) => return Err(ApiError::not_found("Run not found.").into_response()), - } + state + .stores + .runs + .get_cached_run(id) + .await + .map_err(|err| ApiError::new(StatusCode::BAD_GATEWAY, err.to_string()).into_response())? + .ok_or_else(|| ApiError::not_found("Run not found.").into_response())? + .projection + .spec + .graph_source + .clone() }; dot_source diff --git a/lib/apps/fabro-server/src/server/handler/pull_requests.rs b/lib/apps/fabro-server/src/server/handler/pull_requests.rs index 7acd8fb9d..1e3753668 100644 --- a/lib/apps/fabro-server/src/server/handler/pull_requests.rs +++ b/lib/apps/fabro-server/src/server/handler/pull_requests.rs @@ -131,17 +131,14 @@ async fn load_pull_request_record( state: &Arc, id: &RunId, ) -> Result { - let run_store = state + let cached = state .stores .runs - .open_run_reader(id) + .get_cached_run(id) .await - .map_err(|_| ApiError::not_found("Run not found."))?; - let run_state = run_store - .state() - .await - .map_err(|err| ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string()))?; - run_state.pull_request.ok_or_else(|| { + .map_err(|err| ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string()))? + .ok_or_else(|| ApiError::not_found("Run not found."))?; + cached.projection.pull_request.clone().ok_or_else(|| { ApiError::with_code( StatusCode::NOT_FOUND, format!("No pull request found in store. Create one first with: fabro pr create {id}"), @@ -297,14 +294,22 @@ 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 run_store.state().await { - Ok(run_state) => run_state, + let cached = match state.stores.runs.get_cached_run(&id).await { + Ok(Some(cached)) => cached, + Ok(None) => { + return ApiError::new( + StatusCode::INTERNAL_SERVER_ERROR, + "Run projection unavailable.", + ) + .into_response(); + } Err(err) => { return ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string()) .into_response(); } }; - let inputs = match RunPrInputs::extract(&run_state, body.force) { + let run_state = cached.projection.as_ref(); + let inputs = match RunPrInputs::extract(run_state, body.force) { Ok(inputs) => inputs, Err(err) => return err.into_response(), }; @@ -343,7 +348,7 @@ async fn create_run_pull_request( llm_source: state.llm_source.as_ref(), catalog, conclusion: Some(inputs.conclusion), - run_state: Some(&run_state), + run_state: Some(run_state), }; let created_pull_request = match pull_request::maybe_open_pull_request(request).await { Ok(Some(created)) => created, @@ -401,14 +406,21 @@ 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 run_store.state().await { - Ok(run_state) => run_state, + let cached = match state.stores.runs.get_cached_run(&id).await { + Ok(Some(cached)) => cached, + Ok(None) => { + return ApiError::new( + StatusCode::INTERNAL_SERVER_ERROR, + "Run projection unavailable.", + ) + .into_response(); + } Err(err) => { return ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string()) .into_response(); } }; - let Some(pull_request) = run_state.pull_request else { + let Some(pull_request) = cached.projection.pull_request.clone() else { return ApiError::with_code( StatusCode::NOT_FOUND, format!("No pull request found in store. Create one first with: fabro pr create {id}"), diff --git a/lib/apps/fabro-server/src/server/handler/runs.rs b/lib/apps/fabro-server/src/server/handler/runs.rs index 614f2d719..66e1f10be 100644 --- a/lib/apps/fabro-server/src/server/handler/runs.rs +++ b/lib/apps/fabro-server/src/server/handler/runs.rs @@ -1109,14 +1109,15 @@ async fn get_run_stage_command_log( let Ok(run_store) = state.stores.runs.open_run_reader(&id).await else { return ApiError::not_found("Run not found.").into_response(); }; - let run_state = match run_store.state().await { - Ok(run_state) => run_state, + let cached = match state.stores.runs.get_cached_run(&id).await { + Ok(Some(cached)) => cached, + Ok(None) => return ApiError::not_found("Run not found.").into_response(), Err(err) => { return ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string()) .into_response(); } }; - let Some(node) = run_state.stage(&stage_id) else { + let Some(node) = cached.projection.stage(&stage_id) else { return ApiError::not_found("Stage not found.").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 ab12a27a1..96fab6b16 100644 --- a/lib/apps/fabro-server/src/server/handler/sandbox.rs +++ b/lib/apps/fabro-server/src/server/handler/sandbox.rs @@ -951,18 +951,21 @@ async fn load_run_sandbox_instance( state: &Arc, run_id: &RunId, ) -> Result { - match state.stores.runs.open_run_reader(run_id).await { - Ok(run_store) => match run_store.state().await { - Ok(run_state) => run_state - .sandbox - .and_then(fabro_types::RunSandbox::into_instance) - .ok_or_else(|| ApiError::not_found("Run sandbox was not created.").into_response()), - Err(err) => Err( - ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string()).into_response(), - ), - }, - Err(_) => Err(ApiError::not_found("Run not found.").into_response()), - } + let cached = state + .stores + .runs + .get_cached_run(run_id) + .await + .map_err(|err| { + ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string()).into_response() + })? + .ok_or_else(|| ApiError::not_found("Run not found.").into_response())?; + cached + .projection + .sandbox + .clone() + .and_then(fabro_types::RunSandbox::into_instance) + .ok_or_else(|| ApiError::not_found("Run sandbox was not created.").into_response()) } #[cfg(test)] diff --git a/lib/apps/fabro-server/src/server/handler/sessions.rs b/lib/apps/fabro-server/src/server/handler/sessions.rs index f053207f6..c37092b83 100644 --- a/lib/apps/fabro-server/src/server/handler/sessions.rs +++ b/lib/apps/fabro-server/src/server/handler/sessions.rs @@ -305,8 +305,8 @@ async fn attach_session_events( }; let start_seq = match params.since_seq { Some(seq) => seq.max(1), - None => match run_store.list_events().await { - Ok(events) => events.last().map_or(1, |event| event.seq.saturating_add(1)), + None => match run_store.last_event_seq().await { + Ok(last_seq) => last_seq.map_or(1, |seq| seq.saturating_add(1)), Err(err) => return store_error(&err).into_response(), }, }; diff --git a/lib/components/fabro-store/src/slate/mod.rs b/lib/components/fabro-store/src/slate/mod.rs index 7546a741f..617d56565 100644 --- a/lib/components/fabro-store/src/slate/mod.rs +++ b/lib/components/fabro-store/src/slate/mod.rs @@ -1591,6 +1591,48 @@ mod tests { }); } + #[tokio::test] + async fn opening_cached_run_does_not_read_older_event_history() { + let (object_store, store) = make_store(); + let run_id = test_run_id("run-1"); + let run = store.create_run(&run_id).await.unwrap(); + append_completed(&run, "run-1", dt("2026-03-27T12:00:00Z")).await; + + let reopened = Database::new(object_store, "runs", Duration::from_millis(1), None); + 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. + let mut unreadable_old_key = keys::run_event_seq_prefix(&run_id, 2).as_ref().to_vec(); + unreadable_old_key.push(0xff); + reopened + .open_db() + .await + .unwrap() + .put(unreadable_old_key, b"invalid json") + .await + .unwrap(); + + let fresh_writer = reopened.open_run(&run_id).await.unwrap(); + assert_eq!(fresh_writer.last_event_seq().await.unwrap(), Some(5)); + let state = fresh_writer.state().await.unwrap(); + assert_eq!(state.status, RunStatus::Succeeded { + reason: SuccessReason::Completed, + }); + + let seq = fresh_writer + .append_event(&event_payload( + "run-1", + "2026-03-27T12:00:05Z", + "run.title.updated", + &serde_json::json!({ "title": "Renamed completed run" }), + )) + .await + .unwrap(); + assert_eq!(seq, 6); + } + #[tokio::test] async fn append_event_hydrates_local_projection_cache_for_fresh_writer() { let (object_store, store) = make_store(); diff --git a/lib/components/fabro-store/src/slate/projection_cache.rs b/lib/components/fabro-store/src/slate/projection_cache.rs index 73f8aaf2e..c5bf4b2e3 100644 --- a/lib/components/fabro-store/src/slate/projection_cache.rs +++ b/lib/components/fabro-store/src/slate/projection_cache.rs @@ -171,6 +171,15 @@ impl RunProjectionCache { .map(|entry| state.with_children_count(entry)) } + pub(crate) async fn last_seq(&self, run_id: &RunId) -> Option { + self.state + .lock() + .await + .entries + .get(run_id) + .map(|entry| entry.last_seq) + } + pub(crate) async fn get_summary(&self, run_id: &RunId, now: DateTime) -> Option { let mut entry = { let state = self.state.lock().await; diff --git a/lib/components/fabro-store/src/slate/run_store.rs b/lib/components/fabro-store/src/slate/run_store.rs index 83b0a087a..2b982f0df 100644 --- a/lib/components/fabro-store/src/slate/run_store.rs +++ b/lib/components/fabro-store/src/slate/run_store.rs @@ -84,12 +84,27 @@ impl RunDatabase { shared_projection_cache: Arc, run_summary_store: Arc>>, ) -> Result { - let event_seq = if read_only { - // Readers never append, so they do not need to scan the full event - // history to recover the next write sequence. - 1 - } else { - recover_next_seq(&db, keys::run_events_prefix(&run_id), keys::parse_event_seq).await? + let cached_projection = shared_projection_cache.get(&run_id).await; + let projection_cache = + cached_projection + .as_ref() + .map_or_else(EventProjectionCache::default, |cached| { + EventProjectionCache { + last_seq: cached.last_seq, + state: Some((*cached.projection).clone()), + } + }); + let event_seq = match (&cached_projection, read_only) { + (Some(cached), _) => cached.last_seq.saturating_add(1), + (None, true) => { + // Readers never append, so they do not need to scan the full event + // history to recover the next write sequence. + 1 + } + (None, false) => { + recover_next_seq(&db, keys::run_events_prefix(&run_id), keys::parse_event_seq) + .await? + } }; let (event_tx, _) = broadcast::channel(DEFAULT_EVENT_TAIL_LIMIT.max(16)); let blob_store = BlobStore::new(Arc::new(db.clone())); @@ -101,7 +116,7 @@ impl RunDatabase { event_seq: AtomicU32::new(event_seq), close_lock: Mutex::new(()), state_lock: Mutex::new(()), - projection_cache: Mutex::new(EventProjectionCache::default()), + projection_cache: Mutex::new(projection_cache), shared_projection_cache, run_summary_store, recent_events: Mutex::new(VecDeque::with_capacity(DEFAULT_EVENT_TAIL_LIMIT)), @@ -368,6 +383,31 @@ impl RunDatabase { self.list_events_from_with_limit(1, usize::MAX / 2).await } + /// Returns the newest stored event sequence without reading event bodies + /// when a current local or shared projection is available. + pub async fn last_event_seq(&self) -> Result> { + let local_last_seq = self.inner.projection_cache.lock().await.last_seq; + if local_last_seq > 0 { + return Ok(Some(local_last_seq)); + } + if let Some(last_seq) = self + .inner + .shared_projection_cache + .last_seq(&self.inner.run_id) + .await + { + return Ok(Some(last_seq)); + } + + let next_seq = recover_next_seq( + &self.inner.db, + keys::run_events_prefix(&self.inner.run_id), + keys::parse_event_seq, + ) + .await?; + Ok(next_seq.checked_sub(1).filter(|seq| *seq > 0)) + } + pub async fn list_events_from_with_limit( &self, start_seq: u32, @@ -540,9 +580,15 @@ async fn list_events_from(db: &R, run_id: &RunId, start_seq: u32) -> Result( where R: DbRead + Sync, { - // Unbounded scan first: filtering by stage identity with a generic - // limit-bounded scan would silently drop matches whenever the stage's - // events are sparse late in the event log. + // Scan without a storage-level item limit from the requested cursor: + // filtering by stage identity with a generic limit-bounded scan would + // silently drop matches whenever the stage's events are sparse late in + // the event log. // // We probe just the stage identity fields with a small partial deserialize and // only run the full `RunEvent` parse on matches. Most events in a run @@ -642,9 +689,15 @@ where let stage_id_string = stage_id.to_string(); let max_events = limit.saturating_add(1); - let mut iter = db.scan_prefix(keys::run_events_prefix(run_id)).await?; + let event_prefix = keys::run_events_prefix(run_id); + let mut iter = db + .scan(keys::run_event_seq_prefix(run_id, start_seq)..) + .await?; let mut events: Vec = Vec::new(); while let Some(entry) = iter.next().await? { + if !entry.key.starts_with(event_prefix.as_ref()) { + break; + } let key = key_to_string(&entry.key)?; let Some(seq) = keys::parse_event_seq(&key) else { continue; @@ -702,9 +755,15 @@ where let session_id_string = session_id.to_string(); let max_events = limit.saturating_add(1); - let mut iter = db.scan_prefix(keys::run_events_prefix(run_id)).await?; + let event_prefix = keys::run_events_prefix(run_id); + let mut iter = db + .scan(keys::run_event_seq_prefix(run_id, start_seq)..) + .await?; let mut events = Vec::new(); while let Some(entry) = iter.next().await? { + if !entry.key.starts_with(event_prefix.as_ref()) { + break; + } let key = key_to_string(&entry.key)?; let Some(seq) = keys::parse_event_seq(&key) else { continue; @@ -980,6 +1039,36 @@ mod tests { assert_eq!(seqs, vec![4, 6]); } + #[tokio::test] + async fn list_events_for_stage_seeks_to_start_sequence() { + let run = fresh_run().await; + let run_id = run.run_id(); + run.append_event(&stage_prompt_payload(&run_id, 1, Some("alpha"))) + .await + .unwrap(); + run.append_event(&stage_prompt_payload(&run_id, 2, Some("beta"))) + .await + .unwrap(); + run.append_event(&stage_prompt_payload(&run_id, 3, Some("alpha"))) + .await + .unwrap(); + let mut unreadable_earlier_key = keys::run_event_seq_prefix(&run_id, 2).as_ref().to_vec(); + unreadable_earlier_key.push(0xff); + run.inner + .db + .put(unreadable_earlier_key, b"invalid json") + .await + .unwrap(); + + let events = run + .list_events_for_stage_from_with_limit(&StageId::new("alpha", 1), 3, 100) + .await + .unwrap(); + + let seqs: Vec = events.iter().map(|event| event.seq).collect(); + assert_eq!(seqs, vec![4]); + } + #[tokio::test] async fn list_events_for_stage_walks_past_unrelated_events_for_sparse_matches() { let run = fresh_run().await; @@ -1088,6 +1177,38 @@ mod tests { assert_eq!(seqs, vec![3, 5]); } + #[tokio::test] + async fn list_events_for_session_seeks_to_start_sequence() { + let run = fresh_run().await; + let run_id = run.run_id(); + let session_id = SessionId::new(); + let other_session_id = SessionId::new(); + run.append_event(&session_message_payload(&run_id, 1, session_id)) + .await + .unwrap(); + run.append_event(&session_message_payload(&run_id, 2, other_session_id)) + .await + .unwrap(); + run.append_event(&session_message_payload(&run_id, 3, session_id)) + .await + .unwrap(); + let mut unreadable_earlier_key = keys::run_event_seq_prefix(&run_id, 2).as_ref().to_vec(); + unreadable_earlier_key.push(0xff); + run.inner + .db + .put(unreadable_earlier_key, b"invalid json") + .await + .unwrap(); + + let events = run + .list_events_for_session_from_with_limit(session_id, 3, 100) + .await + .unwrap(); + + let seqs: Vec = events.iter().map(|event| event.seq).collect(); + assert_eq!(seqs, vec![4]); + } + #[tokio::test] async fn list_events_for_session_returns_limit_plus_one_for_has_more_signal() { let run = fresh_run().await;