scriptoriumadmin/fabro
1
0
Fork 0
mirror of https://github.com/fabro-sh/fabro.git synced 2026-09-07 08:27:12 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 3

Merge pull request #830 from fabro-sh/codex/sql-run-summary-queries

Browse source 
Move cross-run summaries and pull request recovery to SQLite
This commit is contained in:
Scott Werner 2026-09-01 15:19:55 -04:00 • committed by GitHub
commit d260ae89b5
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
23 changed files with 772 additions and 775 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
lib/apps/fabro-server/src/run_files.rs
View file

@ -1195,7 +1195,7 @@ async fn load_projection(
state: &Arc<AppState>,
run_id: &RunId,
) -> std::result::Result<Arc<fabro_store::RunProjection>, ApiError> {
Ok(state.cached_run(run_id).await?.projection)
state.cached_run_projection(run_id).await
}
async fn reconnect_run_sandbox(

80
lib/apps/fabro-server/src/server.rs
View file

@ -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<pull_request_supervisor::PendingPullRequestCreationQueue>,
global_event_tx: broadcast::Sender<EventEnvelope>,
/// 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<CachedRunProjection, ApiError> {
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<Arc<AppS
scheduler_notify: Notify::new(),
automation_scheduler_notify: Notify::new(),
pull_request_scheduler_notify: Notify::new(),
pull_request_creation_queue: Mutex::new(
pull_request_supervisor::PendingPullRequestCreationQueue::default(),
),
global_event_tx,
files_in_flight: new_files_in_flight(),
pull_request_create_locks: KeyedMutex::new(),
@ -2645,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
};
@ -2712,11 +2705,6 @@ async fn delete_run_internal(
}
}
async fn load_durable_run_status(state: &AppState, id: &RunId) -> Option<RunStatus> {
let cached = state.cached_run(id).await.ok()?;
Some(cached.projection.status)
}
async fn delete_run_sandbox_resource(
state: &AppState,
id: RunId,
@ -2833,7 +2821,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 +3074,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<AppState>,
) -> anyhow::Result<usize> {
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 +3379,8 @@ async fn load_pending_control(
) -> anyhow::Result<Option<RunControlAction>> {
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 +3388,8 @@ async fn load_pending_control(
async fn durable_run_status(state: &AppState, run_id: RunId) -> anyhow::Result<Option<RunStatus>> {
Ok(state
.stores
.runs
.runs()
.find(&run_id)
.run_summaries
.get(&run_id, Utc::now())
.await?
.map(|summary| summary.lifecycle.status))
}
@ -3765,11 +3743,11 @@ async fn load_pending_interview(
run_id: RunId,
qid: &str,
) -> Result<LoadedPendingInterview, Response> {
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 +4534,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<Response> {
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),

32
lib/apps/fabro-server/src/server/automation_scheduler.rs
View file

@ -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<fabro_types::Run> {
async fn stored_runs(state: &AppState) -> Vec<fabro_types::Run> {
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<Utc> {
@ -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);
}
}

12
lib/apps/fabro-server/src/server/handler/artifacts.rs
View file

@ -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);

10
lib/apps/fabro-server/src/server/handler/billing.rs
View file

@ -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<Arc<AppState>>,
Path(id): Path<RunId>,
) -> 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));

3
lib/apps/fabro-server/src/server/handler/graph.rs
View file

@ -243,10 +243,9 @@ async fn load_run_dot_source(state: &AppState, id: &RunId) -> Result<String, Res
Some(dot)
} else {
state
.cached_run(id)
.cached_run_projection(id)
.await
.map_err(IntoResponse::into_response)?
.projection
.spec
.graph_source
.clone()

6
lib/apps/fabro-server/src/server/handler/lifecycle.rs
View file

@ -41,7 +41,7 @@ pub(super) fn routes() -> Router<Arc<AppState>> {
}
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,

5
lib/apps/fabro-server/src/server/handler/pull_requests.rs
View file

@ -151,8 +151,8 @@ async fn load_pull_request_record(
state: &Arc<AppState>,
id: &RunId,
) -> Result<PullRequestLink, ApiError> {
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)
}

59
lib/apps/fabro-server/src/server/handler/runs.rs
View file

@ -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<Arc<AppState>>,
) -> 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(&current_id, Utc::now())
.run_summaries
.get(&current_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<Arc<AppState>>,
) -> 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<Arc<AppState>>,
) -> 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();
};

7
lib/apps/fabro-server/src/server/handler/sandbox.rs
View file

@ -951,12 +951,11 @@ async fn load_run_sandbox_instance(
state: &Arc<AppState>,
run_id: &RunId,
) -> Result<fabro_types::RunSandboxInstance, Response> {
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)

14
lib/apps/fabro-server/src/server/handler/system.rs
View file

@ -300,12 +300,7 @@ async fn get_system_df(
Query(params): Query<DfParams>,
) -> 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<PruneRunsRequest>,
) -> 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())

8
lib/apps/fabro-server/src/server/handler/worker_control.rs
View file

@ -35,14 +35,14 @@ async fn worker_control_stream(
Query(query): Query<WorkerControlStreamQuery>,
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.",

238
lib/apps/fabro-server/src/server/pull_request_supervisor.rs
View file

@ -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,90 @@ 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<chrono::Utc>, RunId)>,
run_ids: HashSet<RunId>,
}
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<chrono::Utc>) -> 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<RunId> {
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<chrono::Utc>,
) -> 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<RunId> {
self.pull_request_creation_queue
.lock()
.expect("pull request creation queue lock poisoned")
.pop()
}
#[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<RunId> {
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 +239,72 @@ pub(crate) fn spawn_pull_request_creation_supervisor(state: Arc<AppState>) -> Jo
)
}
pub(super) async fn recover_pending_pull_request_creations(
state: &AppState,
active: &HashMap<task::Id, RunId>,
failures: &HashMap<RunId, u32>,
) -> 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 load_errors = 0_usize;
for run_id in candidates {
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) => {
load_errors += 1;
warn!(%run_id, %error, "Failed to load pull request creation candidate");
continue;
}
};
let Some(creation) = projection
.pull_request_creation
.as_ref()
.filter(|creation| creation.is_pending())
else {
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,
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<task::Id, RunId>,
failures: &HashMap<RunId, u32>,
) -> 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<AppState>) {
let shutdown = state.shutdown_token();
let mut workers = JoinSet::new();
@ -171,39 +316,28 @@ async fn run_pull_request_creation_supervisor(state: Arc<AppState>) {
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::<Vec<_>>();
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 !can_dispatch(&run_id, &active, &failures) {
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 +345,7 @@ async fn run_pull_request_creation_supervisor(state: Arc<AppState>) {
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 +353,7 @@ async fn run_pull_request_creation_supervisor(state: Arc<AppState>) {
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 +362,6 @@ async fn run_pull_request_creation_supervisor(state: Arc<AppState>) {
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 +396,44 @@ async fn run_pull_request_creation_supervisor(state: Arc<AppState>) {
}
}
}
#[cfg(test)]
mod tests {
use chrono::{DateTime, Utc};
use super::PendingPullRequestCreationQueue;
fn dt(value: &str) -> DateTime<Utc> {
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));
}
}

4
lib/apps/fabro-server/src/server/resource_sampler.rs
View file

@ -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();

159
lib/apps/fabro-server/src/server/tests.rs
View file

@ -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.
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;
@ -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, &[

11
lib/components/fabro-store/src/lib.rs
View file

@ -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<DateTime<Utc>>,
pub end: Option<DateTime<Utc>>,
pub parent_id: Option<fabro_types::RunId>,
}

268
lib/components/fabro-store/src/run_summary_store.rs
View file

@ -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};
@ -342,14 +342,7 @@ ON CONFLICT(singleton) DO NOTHING
.fetch_all(&self.pool)
.await?
.into_iter()
.map(|stored_id| {
stored_id
.parse::<RunId>()
.map_err(|_| Error::RunSummaryMismatch {
run_id: stored_id,
field: "id",
})
})
.map(parse_stored_run_id)
.collect()
}
@ -503,6 +496,61 @@ 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<Utc>) -> Result<Vec<Run>> {
let mut query = QueryBuilder::<Sqlite>::new(SELECT_RUN_SUMMARIES_SQL);
push_order(
&mut query,
RunSummarySort::CreatedAt,
RunSummarySortDirection::Desc,
now,
);
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<Utc>,
) -> Result<Vec<Run>> {
if statuses.is_empty() {
return Ok(Vec::new());
}
let mut query = QueryBuilder::<Sqlite>::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(")");
push_order(
&mut query,
RunSummarySort::CreatedAt,
RunSummarySortDirection::Desc,
now,
);
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<Vec<RunId>> {
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(parse_stored_run_id)
.collect()
}
/// Identity fields for every stored run, for selector resolution without
/// decoding full summaries.
pub async fn list_identities(&self) -> Result<Vec<RunSummaryIdentity>> {
@ -518,12 +566,7 @@ FROM runs",
rows.iter()
.map(|row| {
let stored_id: String = row.try_get("id")?;
let id = stored_id
.parse::<RunId>()
.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")?,
@ -558,10 +601,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::<Result<Vec<_>>>()?;
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 +982,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
@ -1414,6 +1454,15 @@ fn push_order(
builder.push(", id DESC");
}
fn parse_stored_run_id(stored_id: String) -> Result<RunId> {
stored_id
.parse::<RunId>()
.map_err(|_| Error::RunSummaryMismatch {
run_id: stored_id,
field: "id",
})
}
fn decode_run_row(row: &SqliteRow, now: DateTime<Utc>) -> Result<Run> {
let stored_id: String = row.try_get("id")?;
let summary_json: String = row.try_get("summary_json")?;
@ -1433,6 +1482,10 @@ fn decode_run_row(row: &SqliteRow, now: DateTime<Utc>) -> Result<Run> {
Ok(run)
}
fn decode_run_rows(rows: &[SqliteRow], now: DateTime<Utc>) -> Result<Vec<Run>> {
rows.iter().map(|row| decode_run_row(row, now)).collect()
}
fn overlay_live_wall_time(run: &mut Run, now: DateTime<Utc>) {
if run.timestamps.completed_at.is_some() {
return;
@ -1456,9 +1509,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 +2420,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::<Vec<_>>(),
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::<Vec<_>>();
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::<std::collections::HashSet<_>>(),
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;

366
lib/components/fabro-store/src/slate/mod.rs
View file

@ -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<Utc>) -> Result<Vec<Run>> {
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<Utc>,
) -> Result<Vec<(Run, RunProjection)>> {
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<Utc>,
) -> Result<Vec<CachedRunProjection>> {
self.warm_projection_cache().await?;
Ok(self.projection_cache.list(query, now))
}
pub async fn list_unreadable_runs(&self) -> Result<Vec<UnreadableRun>> {
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<Option<CachedRunProjection>> {
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<Utc>,
) -> Result<Option<Run>> {
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<Vec<RunId>> {
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<Option<RunId>> {
@ -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<RunDatabase> {
self.db.open_run(run_id).await
}
pub async fn find(&self, run_id: &RunId) -> Result<Option<Run>> {
self.db.run_summary_store.get(run_id, Utc::now()).await
}
pub async fn list(&self, query: &ListRunsQuery) -> Result<Vec<Run>> {
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<_>>(),
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<_>>(),
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,
});
}

170
lib/components/fabro-store/src/slate/projection_cache.rs
View file

@ -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<RunProjection>,
pub last_seq: u32,
pub(crate) struct CachedRunProjection {
pub(crate) run_id: RunId,
pub(crate) projection: Arc<RunProjection>,
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<RunId, CachedRunProjection>,
children_by_parent: HashMap<RunId, BTreeSet<RunId>>,
entries: HashMap<RunId, CachedRunProjection>,
}
impl RunProjectionCacheState {
fn replace_all(&mut self, entries: Vec<CachedRunProjection>) {
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<Utc>) {
// `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<Utc>,
) -> Vec<CachedRunProjection> {
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::<Vec<_>>(),
None => state.entries.values().cloned().collect::<Vec<_>>(),
};
raw.into_iter()
.map(|entry| state.with_children_count(entry))
.collect::<Vec<_>>()
};
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::<Vec<_>>();
// 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<CachedRunProjection> {
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<RunProjection>, 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<RunId> {
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::<Vec<_>>();
pending.sort_unstable();
pending.into_iter().map(|(_, run_id)| run_id).collect()
}
pub(crate) fn get_summary(&self, run_id: &RunId, now: DateTime<Utc>) -> Option<Run> {
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);
}

3
lib/components/fabro-workflow/src/run_lookup.rs
View file

@ -234,7 +234,8 @@ fn scan_orphan_runs(base: &Path) -> Result<Vec<RunInfo>> {
pub async fn scan_runs_combined(store: &Database, base: &Path) -> Result<Vec<RunInfo>> {
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)

42
lib/components/fabro-workflow/tests/it/daytona_integration.rs
View file

@ -82,27 +82,23 @@ fn load_run_checkpoint(run_dir: &Path) -> Result<Checkpoint, Box<dyn std::error:
let runtime = tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()?;
let run_id =
if uses_shared_store {
run_dir
.file_name()
.ok_or("run dir should have file name")?
.to_string_lossy()
.rsplit('-')
.next()
.ok_or("run dir should contain run id suffix")?
.parse()?
} else {
runtime
.block_on(store.list_runs(
&fabro_store::ListRunsQuery::default(),
chrono::Utc::now(),
))?
.into_iter()
.next()
.ok_or("test store should contain one run")?
.id
};
let run_id = if uses_shared_store {
run_dir
.file_name()
.ok_or("run dir should have file name")?
.to_string_lossy()
.rsplit('-')
.next()
.ok_or("run dir should contain run id suffix")?
.parse()?
} else {
runtime
.block_on(store.run_summary_store().list_all(chrono::Utc::now()))?
.into_iter()
.next()
.ok_or("test store should contain one run")?
.id
};
let run = runtime.block_on(store.open_run_reader(&run_id))?;
let state = runtime.block_on(async {
for attempt in 0..20 {
@ -135,9 +131,7 @@ fn load_run_checkpoint(run_dir: &Path) -> Result<Checkpoint, Box<dyn std::error:
.parse()?
} else {
runtime
.block_on(
store.list_runs(&fabro_store::ListRunsQuery::default(), chrono::Utc::now()),
)?
.block_on(store.run_summary_store().list_all(chrono::Utc::now()))?
.into_iter()
.next()
.ok_or("test store should contain one run")?

46
lib/components/fabro-workflow/tests/it/integration.rs
View file

@ -131,27 +131,23 @@ fn load_run_checkpoint(run_dir: &Path) -> Result<Checkpoint, Box<dyn std::error:
let runtime = tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()?;
let run_id =
if uses_shared_store {
run_dir
.file_name()
.ok_or("run dir should have file name")?
.to_string_lossy()
.rsplit('-')
.next()
.ok_or("run dir should contain run id suffix")?
.parse()?
} else {
runtime
.block_on(store.list_runs(
&fabro_store::ListRunsQuery::default(),
chrono::Utc::now(),
))?
.into_iter()
.next()
.ok_or("test store should contain one run")?
.id
};
let run_id = if uses_shared_store {
run_dir
.file_name()
.ok_or("run dir should have file name")?
.to_string_lossy()
.rsplit('-')
.next()
.ok_or("run dir should contain run id suffix")?
.parse()?
} else {
runtime
.block_on(store.run_summary_store().list_all(chrono::Utc::now()))?
.into_iter()
.next()
.ok_or("test store should contain one run")?
.id
};
let run = runtime.block_on(store.open_run_reader(&run_id))?;
let state = runtime.block_on(async {
for attempt in 0..20 {
@ -184,9 +180,7 @@ fn load_run_checkpoint(run_dir: &Path) -> Result<Checkpoint, Box<dyn std::error:
.parse()?
} else {
runtime
.block_on(
store.list_runs(&fabro_store::ListRunsQuery::default(), chrono::Utc::now()),
)?
.block_on(store.run_summary_store().list_all(chrono::Utc::now()))?
.into_iter()
.next()
.ok_or("test store should contain one run")?
@ -267,9 +261,7 @@ fn resolve_checkpoint_text(
.parse()?
} else {
runtime
.block_on(
store.list_runs(&fabro_store::ListRunsQuery::default(), chrono::Utc::now()),
)?
.block_on(store.run_summary_store().list_all(chrono::Utc::now()))?
.into_iter()
.next()
.ok_or("test store should contain one run")?

2
lib/foundation/fabro-db/tests/sqlite.rs
View file

@ -946,7 +946,7 @@ async fn run_events_schema_query_plans_use_candidate_indexes_including_session_o
"run_events_by_session_owner",
),
(
"EXPLAIN QUERY PLAN SELECT * FROM run_events WHERE event_name = 'pull_request.creation_requested' ORDER BY run_id, seq",
"EXPLAIN QUERY PLAN SELECT DISTINCT run_id FROM run_events WHERE event_name = 'pull_request.creation_requested'",
"run_events_by_pull_request_creation_request",
),
] {