From 1a2d8a90561d5f8e09118b42369146b4cde2b4d9 Mon Sep 17 00:00:00 2001 From: Scott Werner Date: Mon, 31 Aug 2026 14:05:57 -0400 Subject: [PATCH 1/2] Move run queries and pull request recovery to SQLite --- lib/apps/fabro-server/src/run_files.rs | 2 +- lib/apps/fabro-server/src/server.rs | 82 ++-- .../src/server/automation_scheduler.rs | 32 +- .../src/server/handler/artifacts.rs | 12 +- .../src/server/handler/billing.rs | 10 +- .../fabro-server/src/server/handler/graph.rs | 3 +- .../src/server/handler/lifecycle.rs | 6 +- .../src/server/handler/pull_requests.rs | 5 +- .../fabro-server/src/server/handler/runs.rs | 59 ++- .../src/server/handler/sandbox.rs | 7 +- .../fabro-server/src/server/handler/system.rs | 14 +- .../src/server/handler/worker_control.rs | 8 +- .../src/server/pull_request_supervisor.rs | 246 ++++++++++-- .../src/server/resource_sampler.rs | 4 +- lib/apps/fabro-server/src/server/tests.rs | 159 +++++--- lib/components/fabro-store/src/lib.rs | 11 +- .../fabro-store/src/run_summary_store.rs | 239 +++++++++++- lib/components/fabro-store/src/slate/mod.rs | 366 +++--------------- .../fabro-store/src/slate/projection_cache.rs | 170 +------- .../fabro-workflow/src/run_lookup.rs | 3 +- .../tests/it/daytona_integration.rs | 42 +- .../fabro-workflow/tests/it/integration.rs | 46 +-- lib/foundation/fabro-db/tests/sqlite.rs | 2 +- 23 files changed, 771 insertions(+), 757 deletions(-) diff --git a/lib/apps/fabro-server/src/run_files.rs b/lib/apps/fabro-server/src/run_files.rs index 7fb9cc2a8..e82f9b861 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> { - Ok(state.cached_run(run_id).await?.projection) + state.cached_run_projection(run_id).await } async fn reconnect_run_sandbox( diff --git a/lib/apps/fabro-server/src/server.rs b/lib/apps/fabro-server/src/server.rs index ab06cf48e..eee8d3ec0 100644 --- a/lib/apps/fabro-server/src/server.rs +++ b/lib/apps/fabro-server/src/server.rs @@ -85,8 +85,8 @@ use fabro_slack::threads::ThreadRegistry; use fabro_slack::{blocks as slack_blocks, connection as slack_connection}; use fabro_static::EnvVars; use fabro_store::{ - ArtifactKey, ArtifactStore, AuthCodeStore, AuthSessionStore, CachedRunProjection, Database, - EventEnvelope, EventPayload, KeyedMutex, NodeArtifact, PendingInterviewRecord, RunSummaryStore, + ArtifactKey, ArtifactStore, AuthCodeStore, AuthSessionStore, Database, EventEnvelope, + EventPayload, KeyedMutex, NodeArtifact, PendingInterviewRecord, RunSummaryStore, StageArtifactEntry, StageId, }; #[cfg(test)] @@ -100,7 +100,7 @@ use fabro_types::{ AgentBackend, AskFabro, AskFabroUnavailableReason, BlobHash, EventBody, InterviewQuestionRecord, PairId, PairMessageId, PairTarget, PendingReason, Principal, PullRequestLink, QuestionType, RunControlAction, RunEvent, RunId, RunRunnableSource, - SandboxProviderKind, ServerSettings, SessionCapability, + RunStatusKind, SandboxProviderKind, ServerSettings, SessionCapability, }; use fabro_util::error::{ SharedError, collect_causes, render_compact_with_causes, render_with_causes, @@ -794,8 +794,8 @@ impl SlackService { return; }; let event_name = event.body.event_name(); - let projection = match state.stores.runs.get_cached_run(&event.run_id).await { - Ok(Some(cached)) => cached.projection, + let projection = match state.stores.runs.get_cached_projection(&event.run_id).await { + Ok(Some(projection)) => projection, Ok(None) => { warn!( run_id = %event.run_id, @@ -1123,6 +1123,7 @@ pub struct AppState { scheduler_notify: Notify, automation_scheduler_notify: Notify, pull_request_scheduler_notify: Notify, + pull_request_creation_queue: Mutex, global_event_tx: broadcast::Sender, /// Per-run coalescing registry for `GET /runs/{id}/files`. Concurrent /// callers for the same run share one materialization; different runs @@ -1510,17 +1511,6 @@ impl AppState { /// Current cached 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(&self, run_id: &RunId) -> Result { - self.stores - .runs - .get_cached_run(run_id) - .await - .map_err(|err| ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string()))? - .ok_or_else(|| ApiError::not_found("Run not found.")) - } - - /// Like [`Self::cached_run`], but returns only the shared projection — - /// no run summary clone or children count under the cache mutex. pub(crate) async fn cached_run_projection( &self, run_id: &RunId, @@ -2585,6 +2575,9 @@ pub(crate) fn build_app_state(config: AppStateConfig) -> anyhow::Result Option { - let cached = state.cached_run(id).await.ok()?; - Some(cached.projection.status) + state + .stores + .run_summaries + .get(id, Utc::now()) + .await + .ok()? + .map(|summary| summary.lifecycle.status) } async fn delete_run_sandbox_resource( @@ -2833,7 +2831,7 @@ async fn reject_active_delete_without_force( return Ok(()); } - match state.stores.runs.runs().find(run_id).await { + match state.stores.run_summaries.get(run_id, Utc::now()).await { Ok(Some(summary)) if summary.lifecycle.status.requires_force_to_delete() => { Err(ApiError::new( StatusCode::CONFLICT, @@ -3086,32 +3084,24 @@ fn failure_for_incomplete_run( } } -fn should_reconcile_run_on_startup(status: RunStatus) -> bool { - matches!( - status, - RunStatus::Starting - | RunStatus::Running - | RunStatus::Blocked { .. } - | RunStatus::Paused { .. } - | RunStatus::Removing - ) -} - pub(crate) async fn reconcile_incomplete_runs_on_startup( state: &Arc, ) -> anyhow::Result { + const RECONCILABLE_STATUSES: &[RunStatusKind] = &[ + RunStatusKind::Starting, + RunStatusKind::Running, + RunStatusKind::Blocked, + RunStatusKind::Paused, + RunStatusKind::Removing, + ]; let summaries = state .stores - .runs - .list_runs(&fabro_store::ListRunsQuery::default(), chrono::Utc::now()) + .run_summaries + .list_by_statuses(RECONCILABLE_STATUSES, chrono::Utc::now()) .await?; let mut reconciled = 0usize; for summary in summaries { - if !should_reconcile_run_on_startup(summary.lifecycle.status) { - continue; - } - let run_store = state.stores.runs.open_run(&summary.id).await?; let (error, reason) = failure_for_incomplete_run( summary.lifecycle.pending_control, @@ -3399,9 +3389,8 @@ async fn load_pending_control( ) -> anyhow::Result> { Ok(state .stores - .runs - .runs() - .find(&run_id) + .run_summaries + .get(&run_id, Utc::now()) .await? .and_then(|summary| summary.lifecycle.pending_control)) } @@ -3409,9 +3398,8 @@ async fn load_pending_control( async fn durable_run_status(state: &AppState, run_id: RunId) -> anyhow::Result> { Ok(state .stores - .runs - .runs() - .find(&run_id) + .run_summaries + .get(&run_id, Utc::now()) .await? .map(|summary| summary.lifecycle.status)) } @@ -3765,11 +3753,11 @@ async fn load_pending_interview( run_id: RunId, qid: &str, ) -> Result { - let cached = state - .cached_run(&run_id) + let projection = state + .cached_run_projection(&run_id) .await .map_err(IntoResponse::into_response)?; - let Some(record) = cached.projection.pending_interviews.get(qid) else { + let Some(record) = projection.pending_interviews.get(qid) else { return Err(ApiError::new( StatusCode::CONFLICT, "Question no longer exists or was already answered.", @@ -4556,8 +4544,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 cached = state.cached_run(run_id).await.ok()?; - cached.projection.archived_at.is_some().then(|| { + let projection = state.cached_run_projection(run_id).await.ok()?; + 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/automation_scheduler.rs b/lib/apps/fabro-server/src/server/automation_scheduler.rs index 3bceb22d3..16938446e 100644 --- a/lib/apps/fabro-server/src/server/automation_scheduler.rs +++ b/lib/apps/fabro-server/src/server/automation_scheduler.rs @@ -394,7 +394,6 @@ mod tests { AutomationDraft, AutomationGitWorkflowSource, AutomationTrigger, ScheduleTrigger, }; use fabro_static::EnvVars; - use fabro_store::ListRunsQuery; use fabro_types::{GitRunTarget, ResolvedAutomationGitWorkflowSource, RunStatus, RunTarget}; use super::*; @@ -488,16 +487,13 @@ mod tests { .build() } - async fn cached_runs(state: &AppState) -> Vec { + async fn stored_runs(state: &AppState) -> Vec { state .stores - .runs - .list_cached_runs(&ListRunsQuery::default(), Utc::now()) + .run_summaries + .list_all(Utc::now()) .await - .expect("cached runs should list") - .into_iter() - .map(|entry| entry.summary) - .collect() + .expect("stored runs should list") } fn prime_time() -> DateTime { @@ -651,7 +647,7 @@ mod tests { run_due_schedules_once(Arc::clone(&state), &mut planner, prime_time()).await; run_due_schedules_once(Arc::clone(&state), &mut planner, first_due_time()).await; - let runs = cached_runs(state.as_ref()).await; + let runs = stored_runs(state.as_ref()).await; assert_eq!(runs.len(), 1); assert_eq!( state @@ -714,7 +710,7 @@ mod tests { run_due_schedules_once(Arc::clone(&state), &mut planner, prime_time()).await; run_due_schedules_once(Arc::clone(&state), &mut planner, first_due_time()).await; - assert_eq!(cached_runs(state.as_ref()).await.len(), 1); + assert_eq!(stored_runs(state.as_ref()).await.len(), 1); } #[tokio::test] @@ -743,7 +739,7 @@ mod tests { let captured = materializer.captured_inputs(); assert_eq!(captured.len(), 1); assert_eq!(captured[0].workflow_source, Some(workflow_source.clone())); - let runs = cached_runs(state.as_ref()).await; + let runs = stored_runs(state.as_ref()).await; assert_eq!(runs.len(), 1); assert_eq!( runs[0] @@ -775,7 +771,7 @@ mod tests { run_due_schedules_once(Arc::clone(&state), &mut planner, prime_time()).await; run_due_schedules_once(Arc::clone(&state), &mut planner, first_due_time()).await; - assert!(cached_runs(state.as_ref()).await.is_empty()); + assert!(stored_runs(state.as_ref()).await.is_empty()); } #[tokio::test] @@ -792,7 +788,7 @@ mod tests { run_due_schedules_once(Arc::clone(&state), &mut planner, prime_time()).await; run_due_schedules_once(Arc::clone(&state), &mut planner, first_due_time()).await; - let mut trigger_ids = cached_runs(state.as_ref()) + let mut trigger_ids = stored_runs(state.as_ref()) .await .into_iter() .map(|run| run.automation.unwrap().trigger_id.unwrap()) @@ -813,7 +809,7 @@ mod tests { run_due_schedules_once(Arc::clone(&state), &mut planner, prime_time()).await; run_due_schedules_once(Arc::clone(&state), &mut planner, first_due_time()).await; - assert_eq!(cached_runs(state.as_ref()).await.len(), 1); + assert_eq!(stored_runs(state.as_ref()).await.len(), 1); assert!( state .runs @@ -825,7 +821,7 @@ mod tests { run_due_schedules_once(Arc::clone(&state), &mut planner, second_due_time()).await; - assert_eq!(cached_runs(state.as_ref()).await.len(), 2); + assert_eq!(stored_runs(state.as_ref()).await.len(), 2); } #[tokio::test] @@ -842,7 +838,7 @@ mod tests { run_due_schedules_once(Arc::clone(&state), &mut planner, first_due_time()).await; run_due_schedules_once(Arc::clone(&state), &mut planner, first_due_time()).await; - assert!(cached_runs(state.as_ref()).await.is_empty()); + assert!(stored_runs(state.as_ref()).await.is_empty()); assert_eq!(materializer.captured_inputs().len(), 1); assert!( state @@ -857,7 +853,7 @@ mod tests { run_due_schedules_once(Arc::clone(&state), &mut planner, second_due_time()).await; - assert!(cached_runs(state.as_ref()).await.is_empty()); + assert!(stored_runs(state.as_ref()).await.is_empty()); assert_eq!(materializer.captured_inputs().len(), 2); } @@ -883,7 +879,7 @@ mod tests { run_due_schedules_once(Arc::clone(&state), &mut planner, prime_time()).await; run_due_schedules_once(Arc::clone(&state), &mut planner, first_due_time()).await; - assert!(cached_runs(state.as_ref()).await.is_empty()); + assert!(stored_runs(state.as_ref()).await.is_empty()); assert_eq!(materializer.captured_inputs().len(), 1); } } diff --git a/lib/apps/fabro-server/src/server/handler/artifacts.rs b/lib/apps/fabro-server/src/server/handler/artifacts.rs index f0f2b6064..7773b7a3c 100644 --- a/lib/apps/fabro-server/src/server/handler/artifacts.rs +++ b/lib/apps/fabro-server/src/server/handler/artifacts.rs @@ -86,8 +86,8 @@ async fn get_checkpoint( Ok(id) => id, Err(response) => return response, }; - match state.cached_run(&id).await { - Ok(cached) => match cached.projection.current_checkpoint() { + match state.cached_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(run_id) + .cached_run_projection(run_id) .await .map(|_| ()) .map_err(IntoResponse::into_response) @@ -330,8 +330,8 @@ async fn download_run_artifacts( Ok(id) => id, Err(response) => return response, }; - let cached = match state.cached_run(&id).await { - Ok(cached) => cached, + let projection = match state.cached_run_projection(&id).await { + Ok(projection) => projection, Err(error) => return error.into_response(), }; let entries = match state.artifact_store.list_for_run(&id).await { @@ -345,7 +345,7 @@ async fn download_run_artifacts( .into_response(); } }; - let artifacts = latest_run_artifacts(entries, &cached.projection); + let artifacts = latest_run_artifacts(entries, &projection); let content_disposition = format!("attachment; filename=\"fabro-artifacts-{id}.zip\""); let body = artifact_archive_body(state.artifact_store.clone(), id, artifacts); diff --git a/lib/apps/fabro-server/src/server/handler/billing.rs b/lib/apps/fabro-server/src/server/handler/billing.rs index 38a22395c..c0839170a 100644 --- a/lib/apps/fabro-server/src/server/handler/billing.rs +++ b/lib/apps/fabro-server/src/server/handler/billing.rs @@ -69,11 +69,10 @@ async fn list_run_stages( Err(response) => return response, }; - let cached = match state.cached_run(&id).await { - Ok(cached) => cached, + let projection = match state.cached_run_projection(&id).await { + Ok(projection) => projection, Err(err) => return err.into_response(), }; - let projection = cached.projection; let now = Utc::now(); let graph = projection.spec().graph(); @@ -91,11 +90,10 @@ async fn get_run_billing( State(state): State>, Path(id): Path, ) -> Response { - let cached = match state.cached_run(&id).await { - Ok(cached) => cached, + let projection = match state.cached_run_projection(&id).await { + Ok(projection) => projection, Err(err) => return err.into_response(), }; - let projection = cached.projection; let catalog = state.catalog(); let rollup = fabro_workflow::billing_rollup_from_projection(&projection, Some(&catalog)); diff --git a/lib/apps/fabro-server/src/server/handler/graph.rs b/lib/apps/fabro-server/src/server/handler/graph.rs index 364b19cad..ec4f649a4 100644 --- a/lib/apps/fabro-server/src/server/handler/graph.rs +++ b/lib/apps/fabro-server/src/server/handler/graph.rs @@ -243,10 +243,9 @@ async fn load_run_dot_source(state: &AppState, id: &RunId) -> Result Router> { } async fn run_response(state: &AppState, id: RunId, status: StatusCode) -> Response { - match state.stores.runs.get_cached_summary(&id, Utc::now()).await { + match state.stores.run_summaries.get(&id, Utc::now()).await { Ok(Some(summary)) => { (status, Json(state.decorate_run_summary(summary).await)).into_response() } @@ -445,7 +445,7 @@ async fn cancel_run( if let Some(response) = reject_if_archived(state.as_ref(), &id).await { return response; } - let durable_summary = match state.stores.runs.runs().find(&id).await { + let durable_summary = match state.stores.run_summaries.get(&id, Utc::now()).await { Ok(summary) => summary, Err(err) => { return ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string()) @@ -1187,7 +1187,7 @@ async fn batch_run_archive_item( } }; - match state.stores.runs.get_cached_summary(&id, Utc::now()).await { + match state.stores.run_summaries.get(&id, Utc::now()).await { Ok(Some(summary)) => BatchRunLifecycleResult { run_id: id.to_string(), ok: true, 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 2582362e9..b17028fc0 100644 --- a/lib/apps/fabro-server/src/server/handler/pull_requests.rs +++ b/lib/apps/fabro-server/src/server/handler/pull_requests.rs @@ -151,8 +151,8 @@ async fn load_pull_request_record( state: &Arc, id: &RunId, ) -> Result { - let cached = state.cached_run(id).await?; - cached.projection.pull_request.clone().ok_or_else(|| { + let projection = state.cached_run_projection(id).await?; + 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}"), @@ -401,6 +401,7 @@ async fn create_run_pull_request( ) .into_response(); }; + state.enqueue_pull_request_creation(id, creation.requested_at); state.notify_pull_request_scheduler(); accepted_pull_request_creation_response(&id, creation) } diff --git a/lib/apps/fabro-server/src/server/handler/runs.rs b/lib/apps/fabro-server/src/server/handler/runs.rs index b2b6269cb..f10eeef7d 100644 --- a/lib/apps/fabro-server/src/server/handler/runs.rs +++ b/lib/apps/fabro-server/src/server/handler/runs.rs @@ -191,12 +191,7 @@ async fn link_run_parent( } }; let _parent_link_guard = state.parent_link_lock.lock().await; - let child = match state - .stores - .runs - .get_cached_summary(&child_id, Utc::now()) - .await - { + let child = match state.stores.run_summaries.get(&child_id, Utc::now()).await { Ok(Some(summary)) => summary, Ok(None) => return ApiError::not_found("Run not found.").into_response(), Err(err) => { @@ -242,12 +237,7 @@ async fn unlink_run_parent( State(state): State>, ) -> Response { let _parent_link_guard = state.parent_link_lock.lock().await; - let child = match state - .stores - .runs - .get_cached_summary(&child_id, Utc::now()) - .await - { + let child = match state.stores.run_summaries.get(&child_id, Utc::now()).await { Ok(Some(summary)) => summary, Ok(None) => return ApiError::not_found("Run not found.").into_response(), Err(err) => { @@ -297,8 +287,8 @@ async fn validate_parent_link( } let summary = state .stores - .runs - .get_cached_summary(¤t_id, Utc::now()) + .run_summaries + .get(¤t_id, Utc::now()) .await .map_err(|err| ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string()))?; let Some(summary) = summary else { @@ -485,7 +475,7 @@ async fn update_run( Ok(title) => title, Err(err) => return ApiError::bad_request(err.to_string()).into_response(), }; - let current = match state.stores.runs.get_cached_summary(&id, Utc::now()).await { + let current = match state.stores.run_summaries.get(&id, Utc::now()).await { Ok(Some(summary)) => summary, Ok(None) => return ApiError::not_found("Run not found.").into_response(), Err(err) => { @@ -518,7 +508,7 @@ async fn update_run( return ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string()).into_response(); } - match state.stores.runs.get_cached_summary(&id, Utc::now()).await { + match state.stores.run_summaries.get(&id, Utc::now()).await { Ok(Some(summary)) => ( StatusCode::OK, Json(state.decorate_run_summary(summary).await), @@ -934,8 +924,8 @@ async fn finalize_created_run( let created_at = created.run_id.created_at(); let summary = match state .stores - .runs - .get_cached_summary(&created.run_id, Utc::now()) + .run_summaries + .get(&created.run_id, Utc::now()) .await { Ok(Some(summary)) => summary, @@ -1618,25 +1608,20 @@ async fn get_run_settings( Ok(id) => id, Err(response) => return response, }; - let cached = match state.cached_run(&id).await { - Ok(cached) => cached, + let projection = match state.cached_run_projection(&id).await { + Ok(projection) => projection, Err(err) => return err.into_response(), }; - ( - StatusCode::OK, - Json(cached.projection.spec.settings.clone()), - ) - .into_response() + (StatusCode::OK, Json(projection.spec.settings.clone())).into_response() } async fn get_questions( RequireRunManagementTarget(id, _actor): RequireRunManagementTarget, State(state): State>, ) -> Response { - match state.cached_run(&id).await { - Ok(cached) => { - let questions = cached - .projection + match state.cached_run_projection(&id).await { + Ok(projection) => { + let questions = projection .pending_interviews .values() .map(api_question_from_pending_interview) @@ -1675,8 +1660,8 @@ async fn get_run_state( RequireRunManagementTarget(id, _actor): RequireRunManagementTarget, State(state): State>, ) -> Response { - match state.cached_run(&id).await { - Ok(cached) => Json(&*cached.projection).into_response(), + match state.cached_run_projection(&id).await { + Ok(projection) => Json(&*projection).into_response(), Err(err) => err.into_response(), } } @@ -1712,11 +1697,11 @@ async fn get_run_stage_context_window( Ok(stage_id) => stage_id, Err(response) => return response, }; - let cached = match state.cached_run(&id).await { - Ok(cached) => cached, + let projection = match state.cached_run_projection(&id).await { + Ok(projection) => projection, Err(err) => return err.into_response(), }; - let Some(stage) = cached.projection.stage(&stage_id) else { + let Some(stage) = projection.stage(&stage_id) else { return ApiError::not_found("Stage not found.").into_response(); }; @@ -1768,11 +1753,11 @@ 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 cached = match state.cached_run(&id).await { - Ok(cached) => cached, + let projection = match state.cached_run_projection(&id).await { + Ok(projection) => projection, Err(err) => return err.into_response(), }; - let Some(node) = cached.projection.stage(&stage_id) else { + let Some(node) = 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 b42d7920d..091730258 100644 --- a/lib/apps/fabro-server/src/server/handler/sandbox.rs +++ b/lib/apps/fabro-server/src/server/handler/sandbox.rs @@ -951,12 +951,11 @@ async fn load_run_sandbox_instance( state: &Arc, run_id: &RunId, ) -> Result { - let cached = state - .cached_run(run_id) + let projection = state + .cached_run_projection(run_id) .await .map_err(IntoResponse::into_response)?; - cached - .projection + projection .sandbox .clone() .and_then(fabro_types::RunSandbox::into_instance) diff --git a/lib/apps/fabro-server/src/server/handler/system.rs b/lib/apps/fabro-server/src/server/handler/system.rs index 9b53791d8..8ab5e668b 100644 --- a/lib/apps/fabro-server/src/server/handler/system.rs +++ b/lib/apps/fabro-server/src/server/handler/system.rs @@ -300,12 +300,7 @@ async fn get_system_df( Query(params): Query, ) -> Response { let storage_dir = state.server_storage_dir(); - let summaries = match state - .stores - .runs - .list_runs(&fabro_store::ListRunsQuery::default(), Utc::now()) - .await - { + let summaries = match state.stores.run_summaries.list_all(Utc::now()).await { Ok(summaries) => summaries, Err(err) => { return ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string()) @@ -366,12 +361,7 @@ async fn prune_runs( Json(body): Json, ) -> Response { let storage_dir = state.server_storage_dir(); - let summaries = match state - .stores - .runs - .list_runs(&fabro_store::ListRunsQuery::default(), Utc::now()) - .await - { + let summaries = match state.stores.run_summaries.list_all(Utc::now()).await { Ok(summaries) => summaries, Err(err) => { return ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string()) 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 7f70e9297..7d68e179c 100644 --- a/lib/apps/fabro-server/src/server/handler/worker_control.rs +++ b/lib/apps/fabro-server/src/server/handler/worker_control.rs @@ -35,14 +35,14 @@ async fn worker_control_stream( Query(query): Query, ws: WebSocketUpgrade, ) -> Response { - let cached = match state.cached_run(&id).await { - Ok(cached) => cached, + let projection = match state.cached_run_projection(&id).await { + Ok(projection) => projection, Err(err) => return err.into_response(), }; - if cached.projection.archived_at.is_some() { + if projection.archived_at.is_some() { return ApiError::new(StatusCode::CONFLICT, "Run is archived.").into_response(); } - if cached.projection.status.is_terminal() { + if projection.status.is_terminal() { return ApiError::new( StatusCode::CONFLICT, "Run is terminal and cannot accept worker control streams.", 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 2bebd4c97..d7ac3ec31 100644 --- a/lib/apps/fabro-server/src/server/pull_request_supervisor.rs +++ b/lib/apps/fabro-server/src/server/pull_request_supervisor.rs @@ -5,7 +5,7 @@ //! after a server restart), runs them under a bounded worker pool, and //! records a durable success or failure result. -use std::collections::HashMap; +use std::collections::{BTreeSet, HashMap, HashSet}; use std::sync::Arc; use std::time::Duration; @@ -22,11 +22,99 @@ use super::{AppState, pull_request, workflow_event}; const PULL_REQUEST_CREATION_TIMEOUT: Duration = Duration::from_mins(10); const PULL_REQUEST_CREATION_SCAN_INTERVAL: Duration = Duration::from_secs(30); const MAX_CONCURRENT_PULL_REQUEST_CREATIONS: usize = 4; +const PULL_REQUEST_CREATION_QUEUE_CAPACITY: usize = MAX_CONCURRENT_PULL_REQUEST_CREATIONS * 4; /// Stop retrying a run after this many worker attempts that could not even /// record a durable failure (store errors). Without a cap, such a run would /// re-run the whole attempt — including the LLM call — on every scan. const MAX_WORKER_FAILURES_PER_RUN: u32 = 3; +pub(super) struct PendingPullRequestCreationQueue { + capacity: usize, + ordered: BTreeSet<(chrono::DateTime, RunId)>, + run_ids: HashSet, +} + +impl PendingPullRequestCreationQueue { + fn new(capacity: usize) -> Self { + Self { + capacity, + ordered: BTreeSet::new(), + run_ids: HashSet::new(), + } + } + + fn push(&mut self, run_id: RunId, requested_at: chrono::DateTime) -> bool { + if self.ordered.len() >= self.capacity || !self.run_ids.insert(run_id) { + return false; + } + self.ordered.insert((requested_at, run_id)); + true + } + + fn pop(&mut self) -> Option { + let (_, run_id) = self.ordered.pop_first()?; + self.run_ids.remove(&run_id); + Some(run_id) + } + + #[cfg(test)] + fn len(&self) -> usize { + self.ordered.len() + } +} + +impl Default for PendingPullRequestCreationQueue { + fn default() -> Self { + Self::new(PULL_REQUEST_CREATION_QUEUE_CAPACITY) + } +} + +impl AppState { + pub(super) fn enqueue_pull_request_creation( + &self, + run_id: RunId, + requested_at: chrono::DateTime, + ) -> bool { + self.pull_request_creation_queue + .lock() + .expect("pull request creation queue lock poisoned") + .push(run_id, requested_at) + } + + fn pop_pull_request_creation(&self) -> Option { + self.pull_request_creation_queue + .lock() + .expect("pull request creation queue lock poisoned") + .pop() + } + + #[cfg(test)] + pub(super) fn clear_pull_request_creation_queue(&self) { + *self + .pull_request_creation_queue + .lock() + .expect("pull request creation queue lock poisoned") = + PendingPullRequestCreationQueue::default(); + } + + #[cfg(test)] + pub(super) fn pull_request_creation_queue_len(&self) -> usize { + self.pull_request_creation_queue + .lock() + .expect("pull request creation queue lock poisoned") + .len() + } + + #[cfg(test)] + pub(super) fn drain_pull_request_creation_queue(&self) -> Vec { + let mut queue = self + .pull_request_creation_queue + .lock() + .expect("pull request creation queue lock poisoned"); + std::iter::from_fn(|| queue.pop()).collect() + } +} + async fn append_pull_request_creation_failure( run_store: &fabro_store::RunDatabase, run_id: &RunId, @@ -160,6 +248,69 @@ pub(crate) fn spawn_pull_request_creation_supervisor(state: Arc) -> Jo ) } +pub(super) async fn recover_pending_pull_request_creations( + state: &AppState, + active: &HashMap, + failures: &HashMap, +) -> anyhow::Result<()> { + let candidates = state + .stores + .run_summaries + .list_pull_request_creation_candidate_run_ids() + .await?; + let candidate_count = candidates.len(); + let mut pending = Vec::new(); + let mut replay_errors = 0_usize; + + for run_id in candidates { + let projection = match state.stores.runs.open_run_reader(&run_id).await { + Ok(reader) => match reader.state().await { + Ok(projection) => projection, + Err(error) => { + replay_errors += 1; + warn!(%run_id, %error, "Failed to replay pull request creation candidate"); + continue; + } + }, + Err(error) => { + replay_errors += 1; + warn!(%run_id, %error, "Failed to open pull request creation candidate"); + continue; + } + }; + let Some(creation) = projection + .pull_request_creation + .as_ref() + .filter(|creation| creation.is_pending()) + else { + continue; + }; + if active.values().any(|active_id| active_id == &run_id) + || failures.get(&run_id).copied().unwrap_or(0) >= MAX_WORKER_FAILURES_PER_RUN + { + continue; + } + pending.push((creation.requested_at, run_id)); + } + + pending.sort_unstable(); + let pending_count = pending.len(); + let enqueued = pending + .into_iter() + .filter(|(requested_at, run_id)| { + state.enqueue_pull_request_creation(*run_id, *requested_at) + }) + .count(); + tracing::debug!( + candidate_count, + pending_count, + enqueued, + replay_errors, + "Recovered pending pull request creations" + ); + Ok(()) +} + async fn run_pull_request_creation_supervisor(state: Arc) { let shutdown = state.shutdown_token(); let mut workers = JoinSet::new(); @@ -171,39 +322,30 @@ async fn run_pull_request_creation_supervisor(state: Arc) { loop { if scan_requested { - match state - .stores - .runs - .pending_pull_request_creation_run_ids() - .await + if let Err(error) = + recover_pending_pull_request_creations(&state, &active, &failures).await { - Ok(pending) => { - let available = - MAX_CONCURRENT_PULL_REQUEST_CREATIONS.saturating_sub(active.len()); - let ready = pending - .into_iter() - .filter(|run_id| { - !active.values().any(|active_id| active_id == run_id) - && failures.get(run_id).copied().unwrap_or(0) - < MAX_WORKER_FAILURES_PER_RUN - }) - .take(available) - .collect::>(); - for run_id in ready { - let handle = workers.spawn( - process_pull_request_creation(Arc::clone(&state), run_id) - .instrument(info_span!("pull_request_creation", run_id = %run_id)), - ); - active.insert(handle.id(), run_id); - } - } - Err(err) => { - warn!(error = %err, "Failed to scan queued pull request creations"); - } + warn!(%error, "Failed to scan queued pull request creations"); } scan_requested = false; } + while active.len() < MAX_CONCURRENT_PULL_REQUEST_CREATIONS { + let Some(run_id) = state.pop_pull_request_creation() else { + break; + }; + if active.values().any(|active_id| active_id == &run_id) + || failures.get(&run_id).copied().unwrap_or(0) >= MAX_WORKER_FAILURES_PER_RUN + { + continue; + } + let handle = workers.spawn( + process_pull_request_creation(Arc::clone(&state), run_id) + .instrument(info_span!("pull_request_creation", run_id = %run_id)), + ); + active.insert(handle.id(), run_id); + } + if shutdown.is_cancelled() { break; } @@ -211,7 +353,7 @@ async fn run_pull_request_creation_supervisor(state: Arc) { if workers.is_empty() { tokio::select! { () = shutdown.cancelled() => break, - () = state.pull_request_scheduler_notified() => scan_requested = true, + () = state.pull_request_scheduler_notified() => {}, _ = scan_interval.tick() => scan_requested = true, } continue; @@ -219,7 +361,7 @@ async fn run_pull_request_creation_supervisor(state: Arc) { tokio::select! { () = shutdown.cancelled() => break, - () = state.pull_request_scheduler_notified() => scan_requested = true, + () = state.pull_request_scheduler_notified() => {}, _ = scan_interval.tick() => scan_requested = true, joined = workers.join_next_with_id() => { match joined { @@ -228,7 +370,6 @@ async fn run_pull_request_creation_supervisor(state: Arc) { match (run_id, result) { (Some(run_id), Ok(())) => { failures.remove(&run_id); - scan_requested = true; } (Some(run_id), Err(err)) => { // Deliberately no immediate rescan: the run's @@ -263,3 +404,44 @@ async fn run_pull_request_creation_supervisor(state: Arc) { } } } + +#[cfg(test)] +mod tests { + use chrono::{DateTime, Utc}; + + use super::PendingPullRequestCreationQueue; + + fn dt(value: &str) -> DateTime { + value.parse().expect("test timestamp should parse") + } + + #[test] + fn pull_request_creation_queue_deduplicates_orders_and_bounds_work() { + let requested_at = dt("2026-08-31T12:00:00Z"); + let earlier = requested_at - chrono::Duration::seconds(1); + let tied_low = fabro_types::RunId::new(); + let tied_high = fabro_types::RunId::new(); + let (tied_low, tied_high) = if tied_low < tied_high { + (tied_low, tied_high) + } else { + (tied_high, tied_low) + }; + let oldest = fabro_types::RunId::new(); + let overflow = fabro_types::RunId::new(); + let mut queue = PendingPullRequestCreationQueue::new(3); + + assert!(queue.push(tied_high, requested_at)); + assert!(queue.push(oldest, earlier)); + assert!(queue.push(tied_low, requested_at)); + assert!(!queue.push(tied_high, requested_at)); + assert!(!queue.push(overflow, requested_at)); + assert_eq!(queue.len(), 3); + assert_eq!(queue.pop(), Some(oldest)); + assert_eq!(queue.pop(), Some(tied_low)); + assert_eq!(queue.pop(), Some(tied_high)); + assert_eq!(queue.pop(), None); + + assert!(queue.push(overflow, requested_at)); + assert_eq!(queue.pop(), Some(overflow)); + } +} diff --git a/lib/apps/fabro-server/src/server/resource_sampler.rs b/lib/apps/fabro-server/src/server/resource_sampler.rs index e9e846ea6..741a4472e 100644 --- a/lib/apps/fabro-server/src/server/resource_sampler.rs +++ b/lib/apps/fabro-server/src/server/resource_sampler.rs @@ -138,8 +138,8 @@ impl ResourceSampler { let summaries = state .stores - .runs - .list_runs(&fabro_store::ListRunsQuery::default(), chrono::Utc::now()) + .run_summaries + .list_all(chrono::Utc::now()) .await .context("failed to list runs for resource sampling")?; let storage_path = storage_path.to_path_buf(); diff --git a/lib/apps/fabro-server/src/server/tests.rs b/lib/apps/fabro-server/src/server/tests.rs index 409800f22..4faace3fd 100644 --- a/lib/apps/fabro-server/src/server/tests.rs +++ b/lib/apps/fabro-server/src/server/tests.rs @@ -3397,8 +3397,8 @@ async fn create_run_with_explicit_title_skips_generated_title_work() { assert_eq!( state .stores - .runs - .get_cached_summary(&run_id, Utc::now()) + .run_summaries + .get(&run_id, Utc::now()) .await .unwrap() .unwrap() @@ -3467,8 +3467,8 @@ async fn generated_title_failure_leaves_deterministic_title_unchanged() { assert_eq!( state .stores - .runs - .get_cached_summary(&run_id, Utc::now()) + .run_summaries + .get(&run_id, Utc::now()) .await .unwrap() .unwrap() @@ -3516,8 +3516,8 @@ async fn generated_title_does_not_overwrite_user_title_edit() { assert_eq!( state .stores - .runs - .get_cached_summary(&run_id, Utc::now()) + .run_summaries + .get(&run_id, Utc::now()) .await .unwrap() .unwrap() @@ -4526,8 +4526,8 @@ async fn post_runs_create_regression_keeps_api_behavior_without_automation_metad assert_eq!(body["lifecycle"]["status"]["kind"], "submitted"); let summary = state .stores - .runs - .get_cached_summary(&run_id, Utc::now()) + .run_summaries + .get(&run_id, Utc::now()) .await .unwrap() .unwrap(); @@ -4566,8 +4566,8 @@ async fn create_run_from_manifest_helper_persists_without_automation_metadata() assert!(body["automation"].is_null()); let summary = state .stores - .runs - .get_cached_summary(&run_id, Utc::now()) + .run_summaries + .get(&run_id, Utc::now()) .await .unwrap() .unwrap(); @@ -4627,8 +4627,8 @@ async fn create_run_from_manifest_helper_persists_automation_metadata_and_exact_ ); let summary = state .stores - .runs - .get_cached_summary(&run_id, Utc::now()) + .run_summaries + .get(&run_id, Utc::now()) .await .unwrap() .unwrap(); @@ -4684,8 +4684,8 @@ async fn create_run_from_intent_helper_persists_automation_version_and_exact_tar assert_eq!(body["automation"]["id"], automation.id); let summary = state .stores - .runs - .get_cached_summary(&run_id, Utc::now()) + .run_summaries + .get(&run_id, Utc::now()) .await .unwrap() .unwrap(); @@ -5133,8 +5133,8 @@ async fn wait_for_run_title(state: &AppState, run_id: RunId, expected: &str) { for _ in 0..50 { let title = state .stores - .runs - .get_cached_summary(&run_id, Utc::now()) + .run_summaries + .get(&run_id, Utc::now()) .await .unwrap() .unwrap() @@ -10839,7 +10839,7 @@ async fn get_run_pull_request_returns_stored_github_association_when_github_pr_i } #[tokio::test] -async fn create_run_pull_request_creates_and_persists_record() { +async fn pull_request_creation_recovers_durable_request_after_crash_gap() { let github = MockServer::start(); let branch_mock = github.mock(|when, then| { when.method("GET") @@ -10942,9 +10942,11 @@ async fn create_run_pull_request_creates_and_persists_record() { assert_eq!(body["status"], "pending"); assert_eq!(body["model"], "gpt-5.4"); + assert_eq!(state.pull_request_creation_queue_len(), 1); // Starting the supervisor after the request simulates server recovery: // the durable pending event is enough to resume the operation. + state.clear_pull_request_creation_queue(); let supervisor = spawn_pull_request_creation_supervisor(Arc::clone(&state)); let creation_body = wait_for_pull_request_creation(&app, run_id).await; @@ -10982,7 +10984,7 @@ async fn create_run_pull_request_creates_and_persists_record() { } #[tokio::test] -async fn create_run_pull_request_returns_the_active_durable_request() { +async fn pull_request_creation_returns_the_active_durable_request() { let github = MockServer::start(); let (state, app, run_id) = Box::pin(pr_test_app_with_completed_run( Some("ghu_test"), @@ -11031,6 +11033,7 @@ async fn create_run_pull_request_returns_the_active_durable_request() { assert_eq!(first_body["id"], second_body["id"]); assert_eq!(first_body["model"], expected_default_model); + assert_eq!(state.pull_request_creation_queue_len(), 1); let run_store = state.stores.runs.open_run_reader(&run_id).await.unwrap(); let events = run_store.list_events().await.unwrap(); assert_eq!( @@ -11043,7 +11046,76 @@ async fn create_run_pull_request_returns_the_active_durable_request() { } #[tokio::test] -async fn create_run_pull_request_persists_generation_failure() { +async fn pull_request_creation_queue_overflow_recovers_from_indexed_scan() { + let state = test_app_state(); + let mut creation_ids = HashMap::new(); + for _ in 0..17 { + let run_id = RunId::new(); + let creation_id = fabro_types::PullRequestCreationId::new(); + creation_ids.insert(run_id, creation_id); + create_durable_run_with_events(&state, run_id, &[ + workflow_event::Event::PullRequestCreationRequested { + creation_id, + model: "test-model".to_string(), + force: false, + }, + ]) + .await; + } + + pull_request_supervisor::recover_pending_pull_request_creations( + state.as_ref(), + &HashMap::new(), + &HashMap::new(), + ) + .await + .unwrap(); + let first_batch = state.drain_pull_request_creation_queue(); + assert_eq!(first_batch.len(), 16); + + for run_id in first_batch { + let run_store = state.stores.runs.open_run(&run_id).await.unwrap(); + workflow_event::append_event( + &run_store, + &run_id, + &workflow_event::Event::PullRequestFailed { + creation_id: creation_ids.get(&run_id).copied(), + error: "test failure".to_string(), + }, + ) + .await + .unwrap(); + } + + pull_request_supervisor::recover_pending_pull_request_creations( + state.as_ref(), + &HashMap::new(), + &HashMap::new(), + ) + .await + .unwrap(); + let recovered = state.drain_pull_request_creation_queue(); + assert_eq!(recovered.len(), 1); + let recovered_id = recovered[0]; + let projection = state + .stores + .runs + .open_run_reader(&recovered_id) + .await + .unwrap() + .state() + .await + .unwrap(); + assert!( + projection + .pull_request_creation + .as_ref() + .is_some_and(fabro_types::PullRequestCreation::is_pending) + ); +} + +#[tokio::test] +async fn pull_request_creation_persists_generation_failure() { let github = MockServer::start(); let branch_mock = github.mock(|when, then| { when.method("GET") @@ -11145,8 +11217,8 @@ async fn create_run_pull_request_returns_conflict_when_record_exists() { } #[tokio::test] -async fn create_run_pull_request_rejects_missing_repo_origin() { - let (_state, app, run_id) = Box::pin(pr_test_app_with_completed_run(None, None, None)).await; +async fn pull_request_creation_rejects_missing_repo_origin_without_enqueue() { + let (state, app, run_id) = Box::pin(pr_test_app_with_completed_run(None, None, None)).await; let response = app .oneshot( @@ -11168,11 +11240,12 @@ async fn create_run_pull_request_rejects_missing_repo_origin() { let body = response_json!(response, StatusCode::BAD_REQUEST).await; assert_eq!(body["errors"][0]["code"], "missing_repo_origin"); + assert_eq!(state.pull_request_creation_queue_len(), 0); } #[tokio::test] -async fn create_run_pull_request_returns_service_unavailable_without_github_credentials() { - let (_state, app, run_id) = Box::pin(pr_test_app_with_completed_run( +async fn pull_request_creation_rejects_missing_credentials_without_enqueue() { + let (state, app, run_id) = Box::pin(pr_test_app_with_completed_run( None, None, Some("https://github.com/acme/widgets.git"), @@ -11198,6 +11271,8 @@ async fn create_run_pull_request_returns_service_unavailable_without_github_cred .unwrap(); let body = response_json!(response, StatusCode::SERVICE_UNAVAILABLE).await; + assert_eq!(state.pull_request_creation_queue_len(), 0); + assert_eq!(body["errors"][0]["code"], "integration_unavailable"); } @@ -12904,16 +12979,13 @@ async fn run_tool_worker_token_can_use_client_backend_routes_across_runs() { let cached = state .stores .runs - .get_cached_run(&created_child) + .get_cached_projection(&created_child) .await .unwrap() .expect("created run should be cached"); - assert_eq!( - cached.projection.spec.provenance.subject, - Principal::Worker { - run_id: parent_run_id, - }, - ); + assert_eq!(cached.spec.provenance.subject, Principal::Worker { + run_id: parent_run_id, + },); let response = app .clone() @@ -16653,9 +16725,8 @@ async fn cancel_run_overwrites_pending_pause_request() { let summary = state .stores - .runs - .runs() - .find(&run_id) + .run_summaries + .get(&run_id, Utc::now()) .await .unwrap() .unwrap(); @@ -16995,9 +17066,8 @@ async fn pause_run_rejects_when_control_is_already_pending() { let summary = state .stores - .runs - .runs() - .find(&run_id) + .run_summaries + .get(&run_id, Utc::now()) .await .unwrap() .unwrap(); @@ -17124,9 +17194,8 @@ async fn pause_run_immediately_pauses_blocked_run() { let summary = state .stores - .runs - .runs() - .find(&run_id) + .run_summaries + .get(&run_id, Utc::now()) .await .unwrap() .unwrap(); @@ -17164,9 +17233,8 @@ async fn unpause_run_sets_pending_control() { let summary = state .stores - .runs - .runs() - .find(&run_id) + .run_summaries + .get(&run_id, Utc::now()) .await .unwrap() .unwrap(); @@ -17249,9 +17317,8 @@ async fn unpause_run_returns_blocked_when_human_gate_is_still_unresolved() { let summary = state .stores - .runs - .runs() - .find(&run_id) + .run_summaries + .get(&run_id, Utc::now()) .await .unwrap() .unwrap(); @@ -17262,7 +17329,7 @@ async fn unpause_run_returns_blocked_when_human_gate_is_still_unresolved() { } #[tokio::test] -async fn startup_reconciliation_marks_inflight_runs_terminal() { +async fn reconcile_incomplete_runs_marks_inflight_runs_terminal() { let state = test_app_state(); create_durable_run_with_events(&state, fixtures::RUN_1, &[ diff --git a/lib/components/fabro-store/src/lib.rs b/lib/components/fabro-store/src/lib.rs index 4acbff476..1443ab83b 100644 --- a/lib/components/fabro-store/src/lib.rs +++ b/lib/components/fabro-store/src/lib.rs @@ -1,5 +1,3 @@ -use chrono::{DateTime, Utc}; - mod artifact_store; mod auth_code_store; pub mod auth_session_store; @@ -54,12 +52,5 @@ pub use run_summary_store::{ RunSummarySortDirection, RunSummaryStore, RunSummaryVisibility, }; pub use serializable_projection::SerializableProjection; -pub use slate::{CachedRunProjection, Database, RunDatabase, Runs, UnreadableRun}; +pub use slate::{Database, RunDatabase, UnreadableRun}; pub use types::EventPayload; - -#[derive(Debug, Default, Clone, PartialEq, Eq)] -pub struct ListRunsQuery { - pub start: Option>, - pub end: Option>, - pub parent_id: Option, -} diff --git a/lib/components/fabro-store/src/run_summary_store.rs b/lib/components/fabro-store/src/run_summary_store.rs index b9008a384..30f4890be 100644 --- a/lib/components/fabro-store/src/run_summary_store.rs +++ b/lib/components/fabro-store/src/run_summary_store.rs @@ -12,7 +12,7 @@ use sqlx::sqlite::{SqliteArguments, SqliteConnection, SqliteRow}; use sqlx::{Connection as _, QueryBuilder, Row as _, Sqlite, SqlitePool, Transaction}; use strum::VariantArray as _; -use crate::run_state::projected_billing; +use crate::run_state::{build_summary, projected_billing}; use crate::slate::CachedRunProjection; use crate::{Error, EventPayload, Result, keys}; @@ -503,6 +503,57 @@ ON CONFLICT(run_id) DO UPDATE SET deleted_at_ms = excluded.deleted_at_ms row.map(|row| decode_run_row(&row, now)).transpose() } + /// Every current run row, without the bounded HTTP-list visibility or + /// pagination semantics. + pub async fn list_all(&self, now: DateTime) -> Result> { + let mut query = QueryBuilder::::new(SELECT_RUN_SUMMARIES_SQL); + query.push(" ORDER BY created_at_ms DESC, id DESC"); + let rows = query.build().fetch_all(&self.pool).await?; + decode_run_rows(&rows, now) + } + + /// Every current run row whose durable status matches one of `statuses`. + pub async fn list_by_statuses( + &self, + statuses: &[RunStatusKind], + now: DateTime, + ) -> Result> { + if statuses.is_empty() { + return Ok(Vec::new()); + } + let mut query = QueryBuilder::::new(SELECT_RUN_SUMMARIES_SQL); + query.push(" WHERE status IN ("); + let mut separated = query.separated(", "); + for status in statuses { + separated.push_bind(status.to_string()); + } + separated.push_unseparated(") ORDER BY created_at_ms DESC, id DESC"); + let rows = query.build().fetch_all(&self.pool).await?; + decode_run_rows(&rows, now) + } + + /// Run ids that have ever recorded an explicit pull request creation + /// request. Callers replay these candidate histories to determine whether + /// their latest request is still pending. + pub async fn list_pull_request_creation_candidate_run_ids(&self) -> Result> { + sqlx::query_scalar::<_, String>( + "SELECT DISTINCT run_id FROM run_events \ + WHERE event_name = 'pull_request.creation_requested'", + ) + .fetch_all(&self.pool) + .await? + .into_iter() + .map(|stored_id| { + stored_id + .parse::() + .map_err(|_| Error::RunSummaryMismatch { + run_id: stored_id, + field: "id", + }) + }) + .collect() + } + /// Identity fields for every stored run, for selector resolution without /// decoding full summaries. pub async fn list_identities(&self) -> Result> { @@ -558,10 +609,7 @@ FROM runs", let rows = rows_query.build().fetch_all(&mut *transaction).await?; transaction.commit().await?; - let data = rows - .iter() - .map(|row| decode_run_row(row, now)) - .collect::>>()?; + let data = decode_run_rows(&rows, now)?; let total = u64::try_from(total).expect("COUNT(*) is non-negative"); let consumed = u64::from(query.offset).saturating_add(data.len() as u64); Ok(RunSummaryPage { @@ -942,7 +990,7 @@ struct PreparedRunSummary { impl PreparedRunSummary { fn from_entry(entry: &CachedRunProjection) -> Self { - let mut run = entry.summary.clone(); + let mut run = build_summary(&entry.projection, &entry.run_id); if run.timing.is_none() { let at = run .timestamps @@ -1433,6 +1481,10 @@ fn decode_run_row(row: &SqliteRow, now: DateTime) -> Result { Ok(run) } +fn decode_run_rows(rows: &[SqliteRow], now: DateTime) -> Result> { + rows.iter().map(|row| decode_run_row(row, now)).collect() +} + fn overlay_live_wall_time(run: &mut Run, now: DateTime) { if run.timestamps.completed_at.is_some() { return; @@ -1456,9 +1508,9 @@ mod tests { use chrono::{DateTime, Utc}; use fabro_types::{ AutomationRef, BilledTokenCounts, BlockedReason, Conclusion, DiffSummary, EventEnvelope, - FailureReason, Graph, PendingReason, RunDiff, RunId, RunProjection, RunSize, RunSpec, - RunStatus, RunStatusKind, RunTiming, SessionId, StageId, StageOutcome, SuccessReason, - WorkflowSettings, test_support, + FailureReason, Graph, PendingReason, PullRequestCreationId, RunDiff, RunId, RunProjection, + RunSize, RunSpec, RunStatus, RunStatusKind, RunTiming, SessionId, StageId, StageOutcome, + SuccessReason, WorkflowSettings, test_support, }; use strum::VariantArray as _; use tokio::time; @@ -2367,6 +2419,175 @@ mod tests { assert_eq!(archived.data[0].id, archived_id); } + #[tokio::test] + async fn list_all_is_unbounded_complete_and_newest_first() { + let (_directory, store) = store().await; + let created_at = dt("2026-07-11T12:00:00Z"); + let parent_id = run_id(created_at.timestamp_millis().cast_unsigned(), 1); + let child_id = run_id(created_at.timestamp_millis().cast_unsigned() + 1, 2); + let archived_id = run_id(created_at.timestamp_millis().cast_unsigned() + 2, 3); + let removing_id = run_id(created_at.timestamp_millis().cast_unsigned() + 3, 4); + let tied_low_id = run_id(created_at.timestamp_millis().cast_unsigned() + 4, 5); + let tied_high_id = run_id(created_at.timestamp_millis().cast_unsigned() + 4, 6); + + let mut projections = vec![projection(parent_id, "parent", created_at)]; + let mut child = projection( + child_id, + "child", + created_at + chrono::Duration::milliseconds(1), + ); + child.parent_id = Some(parent_id); + projections.push(child); + let mut archived = projection( + archived_id, + "archived", + created_at + chrono::Duration::milliseconds(2), + ); + archived.archived_at = Some(created_at + chrono::Duration::milliseconds(2)); + projections.push(archived); + let mut removing = projection( + removing_id, + "removing", + created_at + chrono::Duration::milliseconds(3), + ); + removing.status = RunStatus::Removing; + projections.push(removing); + projections.push(projection( + tied_low_id, + "tied-low", + created_at + chrono::Duration::milliseconds(4), + )); + projections.push(projection( + tied_high_id, + "tied-high", + created_at + chrono::Duration::milliseconds(4), + )); + for index in 0_u64..101 { + let timestamp_ms = created_at.timestamp_millis().cast_unsigned() + 10 + index; + projections.push(projection( + run_id(timestamp_ms, u128::from(index) + 10), + "bulk", + DateTime::from_timestamp_millis(timestamp_ms.cast_signed()).unwrap(), + )); + } + + let mut expected_ids = Vec::new(); + for projected in projections { + expected_ids.push(projected.spec.run_id); + store.upsert_projection(&entry(projected, 1)).await.unwrap(); + } + expected_ids.sort_by(|left, right| { + right + .created_at() + .cmp(&left.created_at()) + .then_with(|| right.cmp(left)) + }); + + let listed = store.list_all(created_at).await.unwrap(); + assert_eq!(listed.len(), 107); + assert_eq!( + listed.iter().map(|run| run.id).collect::>(), + expected_ids + ); + assert!(listed.iter().any(|run| run.id == archived_id)); + assert!(listed.iter().any(|run| run.id == removing_id)); + assert_eq!( + listed + .iter() + .find(|run| run.id == parent_id) + .unwrap() + .children_count, + 1 + ); + let tied = listed + .iter() + .filter(|run| run.id == tied_low_id || run.id == tied_high_id) + .map(|run| run.id) + .collect::>(); + assert_eq!(tied, vec![tied_high_id, tied_low_id]); + } + + #[tokio::test] + async fn list_by_statuses_is_exact_and_empty_is_empty() { + let (_directory, store) = store().await; + let created_at = dt("2026-07-11T12:00:00Z"); + for (index, kind) in RunStatusKind::VARIANTS.iter().enumerate() { + let id = run_id( + created_at.timestamp_millis().cast_unsigned() + u64::try_from(index).unwrap(), + u128::try_from(index).unwrap() + 1, + ); + let mut projected = projection(id, &kind.to_string(), id.created_at()); + projected.status = sample_status(*kind); + store.upsert_projection(&entry(projected, 1)).await.unwrap(); + } + + let startup_statuses = [ + RunStatusKind::Starting, + RunStatusKind::Running, + RunStatusKind::Blocked, + RunStatusKind::Paused, + RunStatusKind::Removing, + ]; + let listed = store + .list_by_statuses(&startup_statuses, created_at) + .await + .unwrap(); + assert_eq!( + listed + .iter() + .map(|run| run.lifecycle.status.kind()) + .collect::>(), + startup_statuses.into_iter().collect() + ); + assert_eq!(listed.len(), startup_statuses.len()); + assert!( + store + .list_by_statuses(&[], created_at) + .await + .unwrap() + .is_empty() + ); + } + + #[tokio::test] + async fn pull_request_creation_candidates_are_distinct_and_indexed_by_event_name() { + let (_directory, store) = store().await; + let created_at = dt("2026-08-27T12:00:00Z"); + let first_id = run_id(created_at.timestamp_millis().cast_unsigned(), 1); + let second_id = run_id(created_at.timestamp_millis().cast_unsigned() + 1, 2); + let unrelated_id = run_id(created_at.timestamp_millis().cast_unsigned() + 2, 3); + for id in [first_id, second_id, unrelated_id] { + store + .upsert_projection(&entry(projection(id, "candidate", id.created_at()), 3)) + .await + .unwrap(); + } + for (run_id, seq) in [(first_id, 2), (first_id, 3), (second_id, 2)] { + let request = sql_event_payload( + &run_id, + "pull_request.creation_requested", + None, + None, + None, + serde_json::json!({ + "creation_id": PullRequestCreationId::new(), + "model": "test-model", + "force": false, + }), + ); + seed_sql_event(&store, &run_id, seq, &request).await; + } + + let mut candidates = store + .list_pull_request_creation_candidate_run_ids() + .await + .unwrap(); + candidates.sort_unstable(); + let mut expected = vec![first_id, second_id]; + expected.sort_unstable(); + assert_eq!(candidates, expected); + } + #[tokio::test] async fn projection_persists_billing_diff_and_derived_size() { 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 9e92eb894..b5906013d 100644 --- a/lib/components/fabro-store/src/slate/mod.rs +++ b/lib/components/fabro-store/src/slate/mod.rs @@ -7,9 +7,9 @@ use std::sync::Arc; use std::time::Duration; use chrono::{DateTime, Utc}; -use fabro_types::{Run, RunId, SessionId}; +use fabro_types::{RunId, SessionId}; use object_store::ObjectStore; -pub use projection_cache::CachedRunProjection; +pub(crate) use projection_cache::CachedRunProjection; use projection_cache::RunProjectionCache; pub use run_store::RunDatabase; use run_store::RunDatabaseInner; @@ -17,9 +17,7 @@ use slatedb::config::{CompressionCodec, Settings}; use tokio::sync::{Mutex, MutexGuard, OnceCell}; use tracing::warn; -use crate::{ - BlobStore, Error, EventPayload, ListRunsQuery, Result, RunProjection, RunSummaryStore, keys, -}; +use crate::{BlobStore, Error, EventPayload, Result, RunProjection, RunSummaryStore, keys}; #[derive(Debug, Clone, PartialEq, Eq)] pub struct UnreadableRun { @@ -221,28 +219,6 @@ impl Database { self.open_run_database(run_id, true).await } - pub async fn list_runs(&self, query: &ListRunsQuery, now: DateTime) -> Result> { - Ok(self - .list_cached_runs(query, now) - .await? - .into_iter() - .map(|entry| entry.summary) - .collect()) - } - - pub async fn list_runs_with_projection( - &self, - query: &ListRunsQuery, - now: DateTime, - ) -> Result> { - Ok(self - .list_cached_runs(query, now) - .await? - .into_iter() - .map(|entry| (entry.summary, (*entry.projection).clone())) - .collect()) - } - pub async fn warm_projection_cache(&self) -> Result<()> { self.projection_cache_warmed .get_or_try_init(|| async { @@ -270,15 +246,6 @@ impl Database { Ok(()) } - pub async fn list_cached_runs( - &self, - query: &ListRunsQuery, - now: DateTime, - ) -> Result> { - self.warm_projection_cache().await?; - Ok(self.projection_cache.list(query, now)) - } - pub async fn list_unreadable_runs(&self) -> Result> { let run_ids = self.run_summary_store.list_run_ids().await?; let mut unreadable = Vec::new(); @@ -337,11 +304,6 @@ impl Database { Ok(()) } - pub async fn get_cached_run(&self, run_id: &RunId) -> Result> { - self.warm_projection_cache().await?; - Ok(self.projection_cache.get(run_id)) - } - pub async fn get_cached_projection( &self, run_id: &RunId, @@ -353,22 +315,6 @@ impl Database { .map(|(projection, _)| projection)) } - pub async fn get_cached_summary( - &self, - run_id: &RunId, - now: DateTime, - ) -> Result> { - self.warm_projection_cache().await?; - Ok(self.projection_cache.get_summary(run_id, now)) - } - - /// Run ids whose latest explicit pull request creation is still pending, - /// oldest request first. - pub async fn pending_pull_request_creation_run_ids(&self) -> Result> { - self.warm_projection_cache().await?; - Ok(self.projection_cache.pending_pull_request_creations()) - } - /// Resolves the run that owns `session_id` from the canonical typed /// creation event stored in SQLite. pub async fn find_session_owner(&self, session_id: &SessionId) -> Result> { @@ -421,30 +367,6 @@ impl Database { } Ok(deletes) } - - #[must_use] - pub fn runs(&self) -> Runs { - Runs { db: self.clone() } - } -} - -#[derive(Clone, Debug)] -pub struct Runs { - db: Database, -} - -impl Runs { - pub async fn get(&self, run_id: &RunId) -> Result { - self.db.open_run(run_id).await - } - - pub async fn find(&self, run_id: &RunId) -> Result> { - self.db.run_summary_store.get(run_id, Utc::now()).await - } - - pub async fn list(&self, query: &ListRunsQuery) -> Result> { - self.db.list_runs(query, Utc::now()).await - } } pub(crate) fn normalize_base_prefix(prefix: String) -> String { @@ -826,7 +748,8 @@ mod tests { append_created(&run_2, "run-2", dt("2026-03-27T12:00:10Z")).await; let summary = store - .list_runs(&ListRunsQuery::default(), Utc::now()) + .run_summary_store() + .list_all(Utc::now()) .await .unwrap(); assert_eq!(summary.len(), 2); @@ -846,7 +769,8 @@ mod tests { store.delete_run(&test_run_id("run-1")).await.unwrap(); assert!(store.open_run(&test_run_id("run-1")).await.is_err()); let remaining = store - .list_runs(&ListRunsQuery::default(), Utc::now()) + .run_summary_store() + .list_all(Utc::now()) .await .unwrap(); assert_eq!(remaining.len(), 1); @@ -1119,9 +1043,9 @@ mod tests { )); assert_eq!(run.list_events().await.unwrap(), events_before); assert_eq!(run.state().await.unwrap().status, RunStatus::Runnable); - let cached = store.get_cached_run(&run_id).await.unwrap().unwrap(); - assert_eq!(cached.last_seq, 4); - assert_eq!(cached.projection.status, RunStatus::Runnable); + let (projection, last_seq) = store.projection_cache.projection_snapshot(&run_id).unwrap(); + assert_eq!(last_seq, 4); + assert_eq!(projection.status, RunStatus::Runnable); } #[tokio::test] @@ -1138,10 +1062,12 @@ mod tests { .unwrap_err(); assert!(matches!(err, Error::EventRejected { .. })); - let entries = store - .list_cached_runs(&ListRunsQuery::default(), Utc::now()) - .await - .unwrap(); + let (projection, last_seq) = store.projection_cache.projection_snapshot(&run_id).unwrap(); + let entries = [CachedRunProjection::from_projection( + run_id, + Arc::unwrap_or_clone(projection), + last_seq, + )]; summaries.reconcile(&entries).await.unwrap(); let summary = summaries.get(&run_id, Utc::now()).await.unwrap().unwrap(); assert_eq!(summary.lifecycle.status, RunStatus::Runnable); @@ -1154,7 +1080,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.get_cached_run(&run_id).await.unwrap().unwrap(); + let cached_before = store.projection_cache.projection_snapshot(&run_id).unwrap(); summaries.close_pool().await; let result = run @@ -1170,9 +1096,9 @@ mod tests { result, Err(Error::Sqlite(sqlx::Error::PoolClosed)) )); - let cached = store.get_cached_run(&run_id).await.unwrap().unwrap(); - assert_eq!(cached.last_seq, cached_before.last_seq); - assert_eq!(cached.summary.title, cached_before.summary.title); + 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); } #[tokio::test] @@ -1241,7 +1167,8 @@ mod tests { .unwrap(); let summary = store - .list_runs(&ListRunsQuery::default(), Utc::now()) + .run_summary_store() + .list_all(Utc::now()) .await .unwrap(); assert_eq!(summary.len(), 1); @@ -1275,7 +1202,8 @@ mod tests { ); assert_eq!( store - .get_cached_summary(&test_run_id("run-3"), Utc::now()) + .run_summary_store() + .get(&test_run_id("run-3"), Utc::now()) .await .unwrap() .unwrap() @@ -1297,43 +1225,14 @@ mod tests { .unwrap(); assert_eq!( store - .get_cached_summary(&test_run_id("run-3"), Utc::now()) + .run_summary_store() + .get(&test_run_id("run-3"), Utc::now()) .await .unwrap() .unwrap() .parent_id, Some(test_run_id("run-2")) ); - assert!( - store - .list_runs( - &ListRunsQuery { - parent_id: Some(test_run_id("run-1")), - ..ListRunsQuery::default() - }, - Utc::now() - ) - .await - .unwrap() - .is_empty() - ); - assert_eq!( - store - .list_runs( - &ListRunsQuery { - parent_id: Some(test_run_id("run-2")), - ..ListRunsQuery::default() - }, - Utc::now() - ) - .await - .unwrap() - .into_iter() - .map(|summary| summary.id) - .collect::>(), - vec![test_run_id("run-3")] - ); - child .append_event(&event_payload( "run-3", @@ -1347,58 +1246,14 @@ mod tests { .unwrap(); assert_eq!( store - .get_cached_summary(&test_run_id("run-3"), Utc::now()) + .run_summary_store() + .get(&test_run_id("run-3"), Utc::now()) .await .unwrap() .unwrap() .parent_id, None ); - assert!( - store - .list_runs( - &ListRunsQuery { - parent_id: Some(test_run_id("run-2")), - ..ListRunsQuery::default() - }, - Utc::now() - ) - .await - .unwrap() - .is_empty() - ); - } - - #[tokio::test] - async fn list_runs_filters_by_parent_id() { - let (_object_store, store) = make_store(); - let parent = store.create_run(&test_run_id("run-1")).await.unwrap(); - let child = store.create_run(&test_run_id("run-2")).await.unwrap(); - let unrelated = store.create_run(&test_run_id("run-3")).await.unwrap(); - append_created(&parent, "run-1", dt("2026-03-27T12:00:00Z")).await; - append_created_with_parent( - &child, - "run-2", - dt("2026-03-27T12:00:10Z"), - test_run_id("run-1"), - ) - .await; - append_created(&unrelated, "run-3", dt("2026-03-27T12:00:20Z")).await; - - let summaries = store - .list_runs( - &ListRunsQuery { - parent_id: Some(test_run_id("run-1")), - ..ListRunsQuery::default() - }, - Utc::now(), - ) - .await - .unwrap(); - - assert_eq!(summaries.len(), 1); - assert_eq!(summaries[0].id, test_run_id("run-2")); - assert_eq!(summaries[0].parent_id, Some(test_run_id("run-1"))); } #[tokio::test] @@ -1426,7 +1281,8 @@ mod tests { append_created(&unrelated, "run-4", dt("2026-03-27T12:00:30Z")).await; let summaries = store - .list_runs(&ListRunsQuery::default(), Utc::now()) + .run_summary_store() + .list_all(Utc::now()) .await .unwrap(); @@ -1449,44 +1305,6 @@ mod tests { assert_eq!(unrelated_summary.children_count, 0); } - #[tokio::test] - async fn cached_summary_overlays_live_timing_without_mutating_cached_snapshot() { - let (_object_store, store) = make_store(); - let run = store.create_run(&test_run_id("run-1")).await.unwrap(); - append_created(&run, "run-1", dt("2026-03-27T12:00:00Z")).await; - run.append_event(&event_payload( - "run-1", - "2026-03-27T12:00:01Z", - "run.started", - &serde_json::json!({ "name": "Test run" }), - )) - .await - .unwrap(); - - let now = dt("2026-03-27T12:00:06Z"); - let expected = Some(fabro_types::RunTiming::wall_only(5_000)); - - let summary = store - .get_cached_summary(&test_run_id("run-1"), now) - .await - .unwrap() - .unwrap(); - assert_eq!(summary.timing, expected); - - let listed = store - .list_cached_runs(&ListRunsQuery::default(), now) - .await - .unwrap(); - assert_eq!(listed[0].summary.timing, expected); - - let cached = store - .get_cached_run(&test_run_id("run-1")) - .await - .unwrap() - .unwrap(); - assert_eq!(cached.summary.timing, None); - } - #[tokio::test] async fn control_effect_events_clear_pending_control_and_update_status() { let (_object_store, store) = make_store(); @@ -1552,7 +1370,8 @@ mod tests { .unwrap(); let summary = store - .list_runs(&ListRunsQuery::default(), Utc::now()) + .run_summary_store() + .list_all(Utc::now()) .await .unwrap(); assert_eq!(summary.len(), 1); @@ -1632,7 +1451,8 @@ mod tests { summaries, ); let summary = reopened - .list_runs(&ListRunsQuery::default(), Utc::now()) + .run_summary_store() + .list_all(Utc::now()) .await .unwrap(); assert_eq!(summary.len(), 1); @@ -1643,7 +1463,7 @@ mod tests { } #[tokio::test] - async fn projection_cache_warmup_lists_newest_first_and_applies_date_filters() { + async fn projection_cache_warmup_rebuilds_full_projections() { 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(); @@ -1661,42 +1481,20 @@ mod tests { ); reopened.warm_projection_cache().await.unwrap(); - let entries = reopened - .list_cached_runs(&ListRunsQuery::default(), Utc::now()) - .await + let (run_1_projection, run_1_last_seq) = reopened + .projection_cache + .projection_snapshot(&test_run_id("run-1")) .unwrap(); - assert_eq!( - entries.iter().map(|entry| entry.run_id).collect::>(), - vec![test_run_id("run-2"), test_run_id("run-1")] - ); - assert_eq!(entries[0].summary.lifecycle.status, RunStatus::Running); - assert_eq!(entries[0].projection.spec().run_id, test_run_id("run-2")); - assert_eq!(entries[0].last_seq, 4); - - let filtered = reopened - .list_cached_runs( - &ListRunsQuery { - start: Some(test_run_id("run-2").created_at()), - end: Some( - test_run_id("run-2").created_at() + chrono::Duration::seconds(1), - ), - parent_id: None, - }, - Utc::now(), - ) - .await - .unwrap(); - assert_eq!(filtered.len(), 1); - assert_eq!(filtered[0].run_id, test_run_id("run-2")); - - let cached = reopened - .get_cached_run(&test_run_id("run-1")) - .await - .unwrap() - .unwrap(); - assert_eq!(cached.summary.lifecycle.status, RunStatus::Succeeded { + assert_eq!(run_1_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); } #[tokio::test] @@ -1785,28 +1583,25 @@ mod tests { .await .unwrap(); - let cached = store - .get_cached_run(&test_run_id("run-1")) - .await - .unwrap() + let (projection, last_seq) = store + .projection_cache + .projection_snapshot(&test_run_id("run-1")) .unwrap(); - assert_eq!(cached.summary.lifecycle.status, RunStatus::Running); - assert_eq!(cached.last_seq, 7); + assert_eq!(projection.status, RunStatus::Running); + assert_eq!(last_seq, 7); assert_eq!( - cached - .projection + projection .stage(&StageId::new("review", 1)) .unwrap() .effective_state(), fabro_types::StageState::Running ); assert_eq!( - cached.projection.pending_interviews["q-1"].question.text, + projection.pending_interviews["q-1"].question.text, "Approve deploy?" ); assert_eq!( - cached - .projection + projection .current_checkpoint() .unwrap() .git_commit_sha @@ -1814,58 +1609,21 @@ mod tests { Some("abc123") ); - let cached_summaries = store - .list_runs(&ListRunsQuery::default(), Utc::now()) - .await - .unwrap(); - let projected = store - .list_runs_with_projection(&ListRunsQuery::default(), Utc::now()) - .await - .unwrap(); - assert_eq!(cached_summaries, vec![cached.summary.clone()]); - assert_eq!(projected[0].0, cached.summary); - assert_eq!( - projected[0] - .1 - .current_checkpoint() - .unwrap() - .git_commit_sha - .as_deref(), - cached - .projection - .current_checkpoint() - .unwrap() - .git_commit_sha - .as_deref() - ); let comparison_time = dt("2026-03-27T12:00:10Z"); - let cache_summary = store - .get_cached_summary(&test_run_id("run-1"), comparison_time) - .await - .unwrap() - .unwrap(); let sql_summary = summaries .get(&test_run_id("run-1"), comparison_time) .await .unwrap() .unwrap(); - assert_eq!(sql_summary, cache_summary); + assert_eq!(sql_summary.lifecycle.status, RunStatus::Running); store.delete_run(&test_run_id("run-1")).await.unwrap(); assert!( store - .get_cached_run(&test_run_id("run-1")) - .await - .unwrap() + .projection_cache + .projection_snapshot(&test_run_id("run-1")) .is_none() ); - assert!( - store - .list_cached_runs(&ListRunsQuery::default(), Utc::now()) - .await - .unwrap() - .is_empty() - ); assert!( summaries .get(&test_run_id("run-1"), Utc::now()) @@ -1985,9 +1743,13 @@ mod tests { reason: FailureReason::WorkflowError, }); - let cached = reopened.get_cached_run(&run_id).await.unwrap().unwrap(); - assert_eq!(cached.summary.title, "Renamed failed run"); - assert_eq!(cached.summary.lifecycle.status, RunStatus::Failed { + let cached = reopened + .get_cached_projection(&run_id) + .await + .unwrap() + .unwrap(); + assert_eq!(cached.title, "Renamed failed run"); + assert_eq!(cached.status, RunStatus::Failed { reason: FailureReason::WorkflowError, }); } diff --git a/lib/components/fabro-store/src/slate/projection_cache.rs b/lib/components/fabro-store/src/slate/projection_cache.rs index bd5171135..33a0842a8 100644 --- a/lib/components/fabro-store/src/slate/projection_cache.rs +++ b/lib/components/fabro-store/src/slate/projection_cache.rs @@ -1,26 +1,19 @@ -use std::collections::{BTreeSet, HashMap}; +use std::collections::HashMap; use std::sync::{Arc, Mutex, MutexGuard}; -use chrono::{DateTime, Utc}; -use fabro_types::{Run, RunId, RunProjection}; - -use crate::ListRunsQuery; -use crate::run_state::build_summary; +use fabro_types::{RunId, RunProjection}; #[derive(Debug, Clone)] -pub struct CachedRunProjection { - pub run_id: RunId, - pub summary: Run, - pub projection: Arc, - pub last_seq: u32, +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 { - let summary = build_summary(&projection, &run_id); Self { run_id, - summary, projection: Arc::new(projection), last_seq, } @@ -37,76 +30,23 @@ pub(crate) struct RunProjectionCache { #[derive(Debug, Default)] struct RunProjectionCacheState { - entries: HashMap, - children_by_parent: HashMap>, + entries: HashMap, } impl RunProjectionCacheState { fn replace_all(&mut self, entries: Vec) { self.entries.clear(); - self.children_by_parent.clear(); for entry in entries { self.insert(entry); } } fn insert(&mut self, entry: CachedRunProjection) { - let run_id = entry.run_id; - let parent_id = entry.summary.parent_id; - if let Some(previous) = self.entries.insert(run_id, entry) { - self.remove_parent_index(&previous); - } - if let Some(parent_id) = parent_id { - self.children_by_parent - .entry(parent_id) - .or_default() - .insert(run_id); - } + self.entries.insert(entry.run_id, entry); } fn remove(&mut self, run_id: &RunId) { - if let Some(entry) = self.entries.remove(run_id) { - self.remove_parent_index(&entry); - } - } - - fn remove_parent_index(&mut self, entry: &CachedRunProjection) { - let Some(parent_id) = entry.summary.parent_id else { - return; - }; - self.remove_parent_link(&parent_id, &entry.run_id); - } - - fn remove_parent_link(&mut self, parent_id: &RunId, run_id: &RunId) { - let Some(children) = self.children_by_parent.get_mut(parent_id) else { - return; - }; - children.remove(run_id); - if children.is_empty() { - self.children_by_parent.remove(parent_id); - } - } - - fn count_children(&self, run_id: &RunId) -> u64 { - self.children_by_parent - .get(run_id) - .map_or(0, |children| children.len() as u64) - } - - fn with_children_count(&self, mut entry: CachedRunProjection) -> CachedRunProjection { - entry.summary.children_count = self.count_children(&entry.run_id); - entry - } -} - -/// Apply read-time overlays to a cached entry. Pure: does not touch the cache -/// state, so it can run outside the cache mutex. -fn apply_read_overlays(entry: &mut CachedRunProjection, now: DateTime) { - // `Conclusion::timing` is the authoritative terminal snapshot and is - // already present in cached terminal summaries. Only fill missing timing - // with the best-effort live projection. - if entry.summary.timing.is_none() { - entry.summary.timing = entry.projection.live_run_timing(now); + self.entries.remove(run_id); } } @@ -125,66 +65,6 @@ impl RunProjectionCache { self.lock().insert(entry); } - pub(crate) fn list( - &self, - query: &ListRunsQuery, - now: DateTime, - ) -> Vec { - let entries = { - let state = self.lock(); - let raw = match query.parent_id { - Some(parent_id) => state - .children_by_parent - .get(&parent_id) - .into_iter() - .flat_map(|children| children.iter()) - .filter_map(|run_id| state.entries.get(run_id).cloned()) - .collect::>(), - None => state.entries.values().cloned().collect::>(), - }; - raw.into_iter() - .map(|entry| state.with_children_count(entry)) - .collect::>() - }; - let mut entries = entries - .into_iter() - .filter(|entry| { - let created_at = entry.run_id.created_at(); - if query.start.is_some_and(|start| created_at < start) { - return false; - } - if query.end.is_some_and(|end| created_at > end) { - return false; - } - true - }) - .collect::>(); - // Apply per-entry live overlays outside the cache mutex, after any - // date filtering so skipped entries do not sum stage timings. - for entry in &mut entries { - apply_read_overlays(entry, now); - } - entries.sort_by(|left, right| { - right - .run_id - .created_at() - .cmp(&left.run_id.created_at()) - .then_with(|| right.run_id.cmp(&left.run_id)) - }); - entries - } - - pub(crate) fn get(&self, run_id: &RunId) -> Option { - let state = self.lock(); - state - .entries - .get(run_id) - .cloned() - .map(|entry| state.with_children_count(entry)) - } - - /// Projection and last sequence for `run_id`, without the summary clone - /// and children count that `get` computes under the cache mutex. pub(crate) fn projection_snapshot(&self, run_id: &RunId) -> Option<(Arc, u32)> { self.lock() .entries @@ -192,38 +72,6 @@ impl RunProjectionCache { .map(|entry| (Arc::clone(&entry.projection), entry.last_seq)) } - /// Run ids whose latest explicit pull request creation is still pending, - /// oldest request first. Clones only ids and timestamps, so callers can - /// poll on an interval without materializing run summaries. - pub(crate) fn pending_pull_request_creations(&self) -> Vec { - let mut pending = self - .lock() - .entries - .values() - .filter_map(|entry| { - let creation = entry.projection.pull_request_creation.as_ref()?; - creation - .is_pending() - .then_some((creation.requested_at, entry.run_id)) - }) - .collect::>(); - pending.sort_unstable(); - pending.into_iter().map(|(_, run_id)| run_id).collect() - } - - pub(crate) fn get_summary(&self, run_id: &RunId, now: DateTime) -> Option { - let mut entry = { - let state = self.lock(); - state - .entries - .get(run_id) - .cloned() - .map(|entry| state.with_children_count(entry))? - }; - apply_read_overlays(&mut entry, now); - Some(entry.summary) - } - pub(crate) fn remove(&self, run_id: &RunId) { self.lock().remove(run_id); } diff --git a/lib/components/fabro-workflow/src/run_lookup.rs b/lib/components/fabro-workflow/src/run_lookup.rs index d7ba5993f..47d6d1a4e 100644 --- a/lib/components/fabro-workflow/src/run_lookup.rs +++ b/lib/components/fabro-workflow/src/run_lookup.rs @@ -234,7 +234,8 @@ fn scan_orphan_runs(base: &Path) -> Result> { pub async fn scan_runs_combined(store: &Database, base: &Path) -> Result> { let store_runs = store - .list_runs(&fabro_store::ListRunsQuery::default(), Utc::now()) + .run_summary_store() + .list_all(Utc::now()) .await .unwrap_or_default(); scan_runs_with_summaries(&store_runs, base) diff --git a/lib/components/fabro-workflow/tests/it/daytona_integration.rs b/lib/components/fabro-workflow/tests/it/daytona_integration.rs index f5831371a..5ed53dab3 100644 --- a/lib/components/fabro-workflow/tests/it/daytona_integration.rs +++ b/lib/components/fabro-workflow/tests/it/daytona_integration.rs @@ -82,27 +82,23 @@ fn load_run_checkpoint(run_dir: &Path) -> Result Result Result Result Date: Mon, 31 Aug 2026 14:58:23 -0400 Subject: [PATCH 2/2] Simplify pull request recovery and summary store queries - Load recovery candidates from the warm projection cache instead of replaying each run's full event history, and check dispatch eligibility before any I/O - Extract the shared can_dispatch predicate used by both the recovery scan and the worker dispatch loop - Drop load_durable_run_status, now identical to durable_run_status - Share parse_stored_run_id across the three stored-id decode sites - Reuse push_order for the canonical run ordering in list_all and list_by_statuses - Replace the test-only queue clear accessor with the existing drain Co-Authored-By: Claude Fable 5 --- lib/apps/fabro-server/src/server.rs | 12 +---- .../src/server/pull_request_supervisor.rs | 52 ++++++++----------- lib/apps/fabro-server/src/server/tests.rs | 2 +- .../fabro-store/src/run_summary_store.rs | 49 ++++++++--------- 4 files changed, 49 insertions(+), 66 deletions(-) diff --git a/lib/apps/fabro-server/src/server.rs b/lib/apps/fabro-server/src/server.rs index eee8d3ec0..cfc671909 100644 --- a/lib/apps/fabro-server/src/server.rs +++ b/lib/apps/fabro-server/src/server.rs @@ -2638,7 +2638,7 @@ async fn delete_run_internal( }; let had_managed_run = managed_run.is_some(); let durable_status = if managed_run.is_some() { - load_durable_run_status(state, &id).await + durable_run_status(state, id).await.ok().flatten() } else { None }; @@ -2705,16 +2705,6 @@ async fn delete_run_internal( } } -async fn load_durable_run_status(state: &AppState, id: &RunId) -> Option { - state - .stores - .run_summaries - .get(id, Utc::now()) - .await - .ok()? - .map(|summary| summary.lifecycle.status) -} - async fn delete_run_sandbox_resource( state: &AppState, id: RunId, 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 d7ac3ec31..60fec9a6e 100644 --- a/lib/apps/fabro-server/src/server/pull_request_supervisor.rs +++ b/lib/apps/fabro-server/src/server/pull_request_supervisor.rs @@ -88,15 +88,6 @@ impl AppState { .pop() } - #[cfg(test)] - pub(super) fn clear_pull_request_creation_queue(&self) { - *self - .pull_request_creation_queue - .lock() - .expect("pull request creation queue lock poisoned") = - PendingPullRequestCreationQueue::default(); - } - #[cfg(test)] pub(super) fn pull_request_creation_queue_len(&self) -> usize { self.pull_request_creation_queue @@ -260,21 +251,18 @@ pub(super) async fn recover_pending_pull_request_creations( .await?; let candidate_count = candidates.len(); let mut pending = Vec::new(); - let mut replay_errors = 0_usize; + let mut load_errors = 0_usize; for run_id in candidates { - let projection = match state.stores.runs.open_run_reader(&run_id).await { - Ok(reader) => match reader.state().await { - Ok(projection) => projection, - Err(error) => { - replay_errors += 1; - warn!(%run_id, %error, "Failed to replay pull request creation candidate"); - continue; - } - }, + if !can_dispatch(&run_id, active, failures) { + continue; + } + let projection = match state.stores.runs.get_cached_projection(&run_id).await { + Ok(Some(projection)) => projection, + Ok(None) => continue, Err(error) => { - replay_errors += 1; - warn!(%run_id, %error, "Failed to open pull request creation candidate"); + load_errors += 1; + warn!(%run_id, %error, "Failed to load pull request creation candidate"); continue; } }; @@ -285,11 +273,6 @@ pub(super) async fn recover_pending_pull_request_creations( else { continue; }; - if active.values().any(|active_id| active_id == &run_id) - || failures.get(&run_id).copied().unwrap_or(0) >= MAX_WORKER_FAILURES_PER_RUN - { - continue; - } pending.push((creation.requested_at, run_id)); } @@ -305,12 +288,23 @@ pub(super) async fn recover_pending_pull_request_creations( candidate_count, pending_count, enqueued, - replay_errors, + load_errors, "Recovered pending pull request creations" ); Ok(()) } +/// Whether the supervisor may hand `run_id` to a worker right now: not +/// already being processed, and not past the store-failure retry cap. +fn can_dispatch( + run_id: &RunId, + active: &HashMap, + failures: &HashMap, +) -> bool { + !active.values().any(|active_id| active_id == run_id) + && failures.get(run_id).copied().unwrap_or(0) < MAX_WORKER_FAILURES_PER_RUN +} + async fn run_pull_request_creation_supervisor(state: Arc) { let shutdown = state.shutdown_token(); let mut workers = JoinSet::new(); @@ -334,9 +328,7 @@ async fn run_pull_request_creation_supervisor(state: Arc) { let Some(run_id) = state.pop_pull_request_creation() else { break; }; - if active.values().any(|active_id| active_id == &run_id) - || failures.get(&run_id).copied().unwrap_or(0) >= MAX_WORKER_FAILURES_PER_RUN - { + if !can_dispatch(&run_id, &active, &failures) { continue; } let handle = workers.spawn( diff --git a/lib/apps/fabro-server/src/server/tests.rs b/lib/apps/fabro-server/src/server/tests.rs index 4faace3fd..25bb99803 100644 --- a/lib/apps/fabro-server/src/server/tests.rs +++ b/lib/apps/fabro-server/src/server/tests.rs @@ -10946,7 +10946,7 @@ async fn pull_request_creation_recovers_durable_request_after_crash_gap() { // Starting the supervisor after the request simulates server recovery: // the durable pending event is enough to resume the operation. - state.clear_pull_request_creation_queue(); + let _ = state.drain_pull_request_creation_queue(); let supervisor = spawn_pull_request_creation_supervisor(Arc::clone(&state)); let creation_body = wait_for_pull_request_creation(&app, run_id).await; diff --git a/lib/components/fabro-store/src/run_summary_store.rs b/lib/components/fabro-store/src/run_summary_store.rs index 30f4890be..18559d5c8 100644 --- a/lib/components/fabro-store/src/run_summary_store.rs +++ b/lib/components/fabro-store/src/run_summary_store.rs @@ -342,14 +342,7 @@ ON CONFLICT(singleton) DO NOTHING .fetch_all(&self.pool) .await? .into_iter() - .map(|stored_id| { - stored_id - .parse::() - .map_err(|_| Error::RunSummaryMismatch { - run_id: stored_id, - field: "id", - }) - }) + .map(parse_stored_run_id) .collect() } @@ -507,7 +500,12 @@ ON CONFLICT(run_id) DO UPDATE SET deleted_at_ms = excluded.deleted_at_ms /// pagination semantics. pub async fn list_all(&self, now: DateTime) -> Result> { let mut query = QueryBuilder::::new(SELECT_RUN_SUMMARIES_SQL); - query.push(" ORDER BY created_at_ms DESC, id DESC"); + push_order( + &mut query, + RunSummarySort::CreatedAt, + RunSummarySortDirection::Desc, + now, + ); let rows = query.build().fetch_all(&self.pool).await?; decode_run_rows(&rows, now) } @@ -527,7 +525,13 @@ ON CONFLICT(run_id) DO UPDATE SET deleted_at_ms = excluded.deleted_at_ms for status in statuses { separated.push_bind(status.to_string()); } - separated.push_unseparated(") ORDER BY created_at_ms DESC, id DESC"); + separated.push_unseparated(")"); + push_order( + &mut query, + RunSummarySort::CreatedAt, + RunSummarySortDirection::Desc, + now, + ); let rows = query.build().fetch_all(&self.pool).await?; decode_run_rows(&rows, now) } @@ -543,14 +547,7 @@ ON CONFLICT(run_id) DO UPDATE SET deleted_at_ms = excluded.deleted_at_ms .fetch_all(&self.pool) .await? .into_iter() - .map(|stored_id| { - stored_id - .parse::() - .map_err(|_| Error::RunSummaryMismatch { - run_id: stored_id, - field: "id", - }) - }) + .map(parse_stored_run_id) .collect() } @@ -569,12 +566,7 @@ FROM runs", rows.iter() .map(|row| { let stored_id: String = row.try_get("id")?; - let id = stored_id - .parse::() - .map_err(|_| Error::RunSummaryMismatch { - run_id: stored_id, - field: "id", - })?; + let id = parse_stored_run_id(stored_id)?; Ok(RunSummaryIdentity { id, workflow_slug: row.try_get("workflow_slug")?, @@ -1462,6 +1454,15 @@ fn push_order( builder.push(", id DESC"); } +fn parse_stored_run_id(stored_id: String) -> Result { + stored_id + .parse::() + .map_err(|_| Error::RunSummaryMismatch { + run_id: stored_id, + field: "id", + }) +} + fn decode_run_row(row: &SqliteRow, now: DateTime) -> Result { let stored_id: String = row.try_get("id")?; let summary_json: String = row.try_get("summary_json")?;