From 0787b4fac8f01e0556fca757198fcb995f0f3b58 Mon Sep 17 00:00:00 2001 From: Bryan Helmkamp Date: Sun, 5 Apr 2026 10:30:17 -0400 Subject: [PATCH] fix(server): persist cancelled run status in store Keep durable run summaries aligned with in-memory cancellation state, including runs cancelled before startup completes, and update server coverage to assert the durable cancelled reason. --- lib/crates/fabro-server/src/server.rs | 153 +++++++++++++++--------- lib/crates/fabro-server/tests/it/api.rs | 14 +-- 2 files changed, 104 insertions(+), 63 deletions(-) diff --git a/lib/crates/fabro-server/src/server.rs b/lib/crates/fabro-server/src/server.rs index 9bad67730..96e8d4ccf 100644 --- a/lib/crates/fabro-server/src/server.rs +++ b/lib/crates/fabro-server/src/server.rs @@ -47,12 +47,11 @@ use crate::sessions::{SessionStore, new_session_store}; use crate::static_files; use crate::web_auth; use fabro_interview::{Answer, Interviewer, QuestionType, WebInterviewer}; -use fabro_workflow::event::Emitter; +use fabro_workflow::event::{self as workflow_event, Emitter}; use fabro_workflow::operations::{self, CreateRunInput, WorkflowInput}; use fabro_workflow::pipeline::Persisted; use fabro_workflow::records::Checkpoint; use fabro_workflow::run_status::RunStatus as WorkflowRunStatus; -#[cfg(test)] use fabro_workflow::run_status::StatusReason as WorkflowStatusReason; use fabro_api::types::AggregateUsageTotals; @@ -686,6 +685,21 @@ fn clear_live_run_state(run: &mut ManagedRun) { run.cancel_token = None; } +async fn persist_cancelled_run_status(state: &AppState, run_id: RunId) -> anyhow::Result<()> { + let run_store = state.store.open_run(&run_id).await?; + workflow_event::append_event( + &run_store, + &run_id, + &workflow_event::Event::WorkflowRunFailed { + error: FabroError::Cancelled, + duration_ms: 0, + reason: Some(WorkflowStatusReason::Cancelled), + git_commit_sha: None, + }, + ) + .await +} + fn managed_run( dot_source: String, status: RunStatus, @@ -979,18 +993,28 @@ async fn execute_run(state: Arc, run_id: RunId) { let emitter = Arc::new(emitter); // Transition to Running, populate interviewer - { + let cancelled_during_setup = { let mut runs = state.runs.lock().expect("runs lock poisoned"); if let Some(managed_run) = runs.get_mut(&run_id) { if managed_run.status != RunStatus::Starting { // Was cancelled during setup clear_live_run_state(managed_run); state.scheduler_notify.notify_one(); - return; + true + } else { + managed_run.status = RunStatus::Running; + managed_run.interviewer = Some(Arc::clone(&interviewer)); + false } - managed_run.status = RunStatus::Running; - managed_run.interviewer = Some(Arc::clone(&interviewer)); + } else { + false } + }; + if cancelled_during_setup { + if let Err(err) = persist_cancelled_run_status(state.as_ref(), run_id).await { + error!(run_id = %run_id, error = %err, "Failed to persist cancelled run status"); + } + return; } let run_store = match state.store.open_run(&run_id).await { @@ -1660,34 +1684,52 @@ async fn cancel_run( Ok(id) => id, Err(response) => return response, }; - let mut runs = state.runs.lock().expect("runs lock poisoned"); - match runs.get_mut(&id) { - Some(managed_run) => match managed_run.status { - RunStatus::Submitted | RunStatus::Queued | RunStatus::Starting | RunStatus::Running => { - if let Some(token) = &managed_run.cancel_token { - token.store(true, Ordering::Relaxed); + let (created_at, persist_cancelled_status) = { + let mut runs = state.runs.lock().expect("runs lock poisoned"); + match runs.get_mut(&id) { + Some(managed_run) => match managed_run.status { + RunStatus::Submitted + | RunStatus::Queued + | RunStatus::Starting + | RunStatus::Running => { + if let Some(token) = &managed_run.cancel_token { + token.store(true, Ordering::Relaxed); + } + if let Some(cancel_tx) = managed_run.cancel_tx.take() { + let _ = cancel_tx.send(()); + } + let persist_cancelled_status = + matches!(managed_run.status, RunStatus::Submitted | RunStatus::Queued); + managed_run.status = RunStatus::Cancelled; + (managed_run.created_at, persist_cancelled_status) } - if let Some(cancel_tx) = managed_run.cancel_tx.take() { - let _ = cancel_tx.send(()); + _ => { + return ApiError::new(StatusCode::CONFLICT, "Run is not cancellable.") + .into_response(); } - managed_run.status = RunStatus::Cancelled; - let created_at = managed_run.created_at; - ( - StatusCode::OK, - Json(RunStatusResponse { - id: id.to_string(), - status: RunStatus::Cancelled, - error: None, - queue_position: None, - created_at, - }), - ) - .into_response() - } - _ => ApiError::new(StatusCode::CONFLICT, "Run is not cancellable.").into_response(), - }, - None => ApiError::not_found("Run not found.").into_response(), + }, + None => return ApiError::not_found("Run not found.").into_response(), + } + }; + + if persist_cancelled_status { + if let Err(err) = persist_cancelled_run_status(state.as_ref(), id).await { + return ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string()) + .into_response(); + } } + + ( + StatusCode::OK, + Json(RunStatusResponse { + id: id.to_string(), + status: RunStatus::Cancelled, + error: None, + queue_position: None, + created_at, + }), + ) + .into_response() } async fn pause_run( @@ -3348,7 +3390,7 @@ mod tests { #[tokio::test] async fn cancel_queued_run_succeeds() { let state = create_app_state(); - let app = build_router(state, AuthMode::Disabled); + let app = build_router(Arc::clone(&state), AuthMode::Disabled); // Submit a run (no start, stays submitted) let req = Request::builder() @@ -3383,7 +3425,13 @@ mod tests { let response = app.oneshot(req).await.unwrap(); let body = body_json(response.into_body()).await; - assert_eq!(body["status"].as_str().unwrap(), "cancelled"); + assert_eq!(body["status"].as_str().unwrap(), "failed"); + assert_eq!(body["status_reason"].as_str().unwrap(), "cancelled"); + + let run_store = state.store.open_run_reader(&run_id).await.unwrap(); + let status = run_store.state().await.unwrap().status.unwrap(); + assert_eq!(status.status, WorkflowRunStatus::Failed); + assert_eq!(status.reason, Some(WorkflowStatusReason::Cancelled)); } #[tokio::test(flavor = "multi_thread", worker_threads = 2)] @@ -3481,30 +3529,30 @@ mod tests { let app = build_router(state, AuthMode::Disabled); // Create and start two runs (no scheduler, both stay queued) - let mut run_ids = Vec::new(); - let id = create_and_start_run(&app, MINIMAL_DOT).await; - run_ids.push(id); - let id = create_and_start_run(&app, MINIMAL_DOT).await; - run_ids.push(id); + let first_run_id = create_and_start_run(&app, MINIMAL_DOT).await; + let second_run_id = create_and_start_run(&app, MINIMAL_DOT).await; - // Check queue positions via individual status + // Check queue positions via the live board endpoint. let req = Request::builder() .method("GET") - .uri(api(&format!("/runs/{}", run_ids[0]))) + .uri(api("/boards/runs")) .body(Body::empty()) .unwrap(); let response = app.clone().oneshot(req).await.unwrap(); let body = body_json(response.into_body()).await; - assert_eq!(body["queue_position"].as_i64().unwrap(), 1); + let items = body["data"].as_array().unwrap(); - let req = Request::builder() - .method("GET") - .uri(api(&format!("/runs/{}", run_ids[1]))) - .body(Body::empty()) + let first = items + .iter() + .find(|item| item["id"].as_str() == Some(first_run_id.as_str())) .unwrap(); - let response = app.clone().oneshot(req).await.unwrap(); - let body = body_json(response.into_body()).await; - assert_eq!(body["queue_position"].as_i64().unwrap(), 2); + assert_eq!(first["queue_position"].as_i64().unwrap(), 1); + + let second = items + .iter() + .find(|item| item["id"].as_str() == Some(second_run_id.as_str())) + .unwrap(); + assert_eq!(second["queue_position"].as_i64().unwrap(), 2); } #[tokio::test(flavor = "multi_thread", worker_threads = 2)] @@ -3513,11 +3561,8 @@ mod tests { let app = test_app_with_scheduler(state); // Create and start two runs with max_concurrent_runs=1 - let mut run_ids = Vec::new(); - let id = create_and_start_run(&app, MINIMAL_DOT).await; - run_ids.push(id); - let id = create_and_start_run(&app, MINIMAL_DOT).await; - run_ids.push(id); + create_and_start_run(&app, MINIMAL_DOT).await; + create_and_start_run(&app, MINIMAL_DOT).await; // Give scheduler time to pick up the first run tokio::time::sleep(std::time::Duration::from_millis(50)).await; @@ -3525,7 +3570,7 @@ mod tests { // Check statuses: at most 1 should be starting/running, the other queued let req = Request::builder() .method("GET") - .uri(api("/runs")) + .uri(api("/boards/runs")) .body(Body::empty()) .unwrap(); let response = app.clone().oneshot(req).await.unwrap(); diff --git a/lib/crates/fabro-server/tests/it/api.rs b/lib/crates/fabro-server/tests/it/api.rs index eb1aca048..368f7452f 100644 --- a/lib/crates/fabro-server/tests/it/api.rs +++ b/lib/crates/fabro-server/tests/it/api.rs @@ -635,15 +635,11 @@ mod server_lifecycle { let body = body_json(response.into_body()).await; assert_eq!(body["status"], "cancelled"); - // Verify status is cancelled - let req = Request::builder() - .method("GET") - .uri(api(&format!("/runs/{run_id}"))) - .body(Body::empty()) - .unwrap(); - let response = app.clone().oneshot(req).await.unwrap(); - let body = body_json(response.into_body()).await; - assert_eq!(body["status"], "cancelled"); + // Verify the durable store view converges to cancelled failure. + let status = wait_for_run_status(&app, &run_id, &["failed"]).await; + assert_eq!(status, "failed"); + let body = run_json(&app, &run_id).await; + assert_eq!(body["status_reason"], "cancelled"); } }