diff --git a/docs/public/api-reference/fabro-api.yaml b/docs/public/api-reference/fabro-api.yaml index ac4a739ac..bd41cc8ae 100644 --- a/docs/public/api-reference/fabro-api.yaml +++ b/docs/public/api-reference/fabro-api.yaml @@ -2732,6 +2732,20 @@ paths: schema: $ref: "#/components/schemas/DiskUsageResponse" + /api/v1/system/repair/runs: + get: + operationId: getSystemRepairRuns + tags: [System] + summary: List Run Repair Issues + description: Lists cataloged runs that cannot be loaded from durable storage. + responses: + "200": + description: Run repair issues + content: + application/json: + schema: + $ref: "#/components/schemas/SystemRepairRunsResponse" + /api/v1/system/prune/runs: post: operationId: pruneRuns @@ -8017,6 +8031,34 @@ components: format: int64 description: Runs currently queued or executing. + SystemRepairRunsResponse: + description: Runs that need manual repair or deletion because they cannot be loaded. + type: object + properties: + runs: + type: array + items: + $ref: "#/components/schemas/SystemRepairRunIssue" + total_count: + type: integer + format: int64 + description: Count of run repair issues. + + SystemRepairRunIssue: + description: One cataloged run that cannot be loaded from durable storage. + type: object + properties: + run_id: + type: string + description: Run identifier. + created_at: + type: string + format: date-time + description: Timestamp encoded in the run identifier. + error: + type: string + description: Error produced while loading the run projection. + DiskUsageResponse: description: Disk usage summary for server-managed data. type: object diff --git a/lib/crates/fabro-cli/src/args.rs b/lib/crates/fabro-cli/src/args.rs index 4446d51b6..8ef5a7cc6 100644 --- a/lib/crates/fabro-cli/src/args.rs +++ b/lib/crates/fabro-cli/src/args.rs @@ -772,6 +772,24 @@ pub(crate) struct SystemEventsArgs { pub(crate) run_ids: Vec, } +#[derive(Args)] +pub(crate) struct SystemRepairArgs { + #[command(subcommand)] + pub(crate) command: SystemRepairCommand, +} + +#[derive(Subcommand)] +pub(crate) enum SystemRepairCommand { + /// List runs that cannot be loaded from durable storage + Runs(SystemRepairRunsArgs), +} + +#[derive(Args)] +pub(crate) struct SystemRepairRunsArgs { + #[command(flatten)] + pub(crate) connection: ServerConnectionArgs, +} + #[derive(Args)] pub(crate) struct SettingsArgs { #[command(flatten)] @@ -1212,6 +1230,9 @@ impl Commands { SystemCommand::Prune(_) => "system prune", SystemCommand::Df(_) => "system df", SystemCommand::Events(_) => "system events", + SystemCommand::Repair(args) => match &args.command { + SystemRepairCommand::Runs(_) => "system repair runs", + }, }, Self::SendAnalytics { .. } => "__send_analytics", Self::SendPanic { .. } => "__send_panic", @@ -1373,6 +1394,8 @@ pub(crate) enum SystemCommand { Df(DfArgs), /// Stream run events from the server Events(SystemEventsArgs), + /// Inspect and repair durable server data + Repair(SystemRepairArgs), } #[derive(Args)] diff --git a/lib/crates/fabro-cli/src/commands/runs/rm.rs b/lib/crates/fabro-cli/src/commands/runs/rm.rs index 31cacfd99..c959a4b4f 100644 --- a/lib/crates/fabro-cli/src/commands/runs/rm.rs +++ b/lib/crates/fabro-cli/src/commands/runs/rm.rs @@ -21,6 +21,29 @@ async fn remove_from(args: &RunsRemoveArgs, ctx: &CommandContext) -> Result<()> let mut errors = Vec::new(); for identifier in &args.runs { + if args.force { + if let Ok(run_id) = identifier.parse::() { + let run_id_string = run_id.to_string(); + if let Err(err) = delete_server_run(client, &run_id, true).await { + let error = err.to_string(); + if !json { + fabro_util::printerr!(printer, "error: {identifier}: {error}"); + } + errors.push(serde_json::json!({ + "identifier": identifier, + "error": error, + })); + had_errors = true; + continue; + } + removed.push(run_id_string.clone()); + if !json { + fabro_util::printerr!(printer, "{}", short_run_id(&run_id_string)); + } + continue; + } + } + let run = match client.resolve_run(identifier).await { Ok(run) => run, Err(err) => { diff --git a/lib/crates/fabro-cli/src/commands/system/mod.rs b/lib/crates/fabro-cli/src/commands/system/mod.rs index 269c2a2f6..d132fcbda 100644 --- a/lib/crates/fabro-cli/src/commands/system/mod.rs +++ b/lib/crates/fabro-cli/src/commands/system/mod.rs @@ -2,6 +2,7 @@ mod df; mod events; mod info; mod prune; +mod repair; use anyhow::Result; @@ -14,5 +15,6 @@ pub(crate) async fn dispatch(ns: SystemNamespace, base_ctx: &CommandContext) -> SystemCommand::Prune(args) => prune::prune_command(&args, base_ctx).await, SystemCommand::Df(args) => df::df_command(&args, base_ctx).await, SystemCommand::Events(args) => events::events_command(&args, base_ctx).await, + SystemCommand::Repair(args) => repair::repair_command(&args, base_ctx).await, } } diff --git a/lib/crates/fabro-cli/src/commands/system/repair.rs b/lib/crates/fabro-cli/src/commands/system/repair.rs new file mode 100644 index 000000000..80edca4f4 --- /dev/null +++ b/lib/crates/fabro-cli/src/commands/system/repair.rs @@ -0,0 +1,65 @@ +use anyhow::Result; +use chrono::{DateTime, Utc}; +use fabro_api::types; +use fabro_util::printer::Printer; + +use crate::args::{SystemRepairArgs, SystemRepairCommand, SystemRepairRunsArgs}; +use crate::command_context::CommandContext; +use crate::shared::print_json_pretty; + +pub(super) async fn repair_command( + args: &SystemRepairArgs, + base_ctx: &CommandContext, +) -> Result<()> { + match &args.command { + SystemRepairCommand::Runs(args) => repair_runs_command(args, base_ctx).await, + } +} + +async fn repair_runs_command(args: &SystemRepairRunsArgs, base_ctx: &CommandContext) -> Result<()> { + let ctx = base_ctx.with_connection(&args.connection)?; + let server = ctx.server().await?; + let response = server.get_system_repair_runs().await?; + repair_runs_from(&response, ctx.json_output(), ctx.printer()) +} + +fn repair_runs_from( + response: &types::SystemRepairRunsResponse, + json_output: bool, + printer: Printer, +) -> Result<()> { + if json_output { + print_json_pretty(response)?; + return Ok(()); + } + + let runs = response.runs.as_slice(); + if runs.is_empty() { + fabro_util::printout!(printer, "No run repair issues found."); + return Ok(()); + } + + fabro_util::printout!(printer, "Unreadable runs:"); + for run in runs { + fabro_util::printout!( + printer, + " {} {} {}", + run.run_id.as_deref().unwrap_or("-"), + format_created_at(run.created_at.as_ref()), + run.error.as_deref().unwrap_or("-"), + ); + } + + fabro_util::printout!(printer, ""); + fabro_util::printout!(printer, "Delete with:"); + for run in runs { + if let Some(run_id) = run.run_id.as_deref() { + fabro_util::printout!(printer, " fabro rm --force {run_id}"); + } + } + Ok(()) +} + +fn format_created_at(created_at: Option<&DateTime>) -> String { + created_at.map_or_else(|| "-".to_string(), DateTime::to_rfc3339) +} diff --git a/lib/crates/fabro-cli/tests/it/cmd/mod.rs b/lib/crates/fabro-cli/tests/it/cmd/mod.rs index 0d509cb3d..906637a6b 100644 --- a/lib/crates/fabro-cli/tests/it/cmd/mod.rs +++ b/lib/crates/fabro-cli/tests/it/cmd/mod.rs @@ -59,6 +59,7 @@ mod system_df; mod system_events; mod system_info; mod system_prune; +mod system_repair; #[cfg(debug_assertions)] mod test_panic; mod top_level; diff --git a/lib/crates/fabro-cli/tests/it/cmd/rm.rs b/lib/crates/fabro-cli/tests/it/cmd/rm.rs index e5ebab4b7..6b16ef84f 100644 --- a/lib/crates/fabro-cli/tests/it/cmd/rm.rs +++ b/lib/crates/fabro-cli/tests/it/cmd/rm.rs @@ -152,35 +152,6 @@ fn rm_force_removes_active_run() { let context = test_context!(); let run_id = unique_run_id(); let server = MockServer::start(); - let resolve_mock = server.mock(|when, then| { - when.method("GET") - .path("/api/v1/runs/resolve") - .query_param("selector", &run_id); - then.status(200) - .header("Content-Type", "application/json") - .body( - serde_json::json!({ - "run_id": run_id, - "workflow_name": "Active Workflow", - "workflow_slug": "active-workflow", - "goal": "Active goal", - "title": "Active goal", - "labels": {}, - "source_directory": null, - "repository": { "name": "unknown" }, - "start_time": "2026-04-05T12:00:00Z", - "created_at": "2026-04-05T12:00:00Z", - "status": { - "kind": "running" - }, - "pending_control": null, - "duration_ms": 123, - "elapsed_secs": 0, - "total_usd_micros": null - }) - .to_string(), - ); - }); let delete_mock = server.mock(|when, then| { when.method("DELETE") .path(format!("/api/v1/runs/{run_id}")) @@ -203,7 +174,6 @@ fn rm_force_removes_active_run() { ----- stderr ----- [ULID] "); - resolve_mock.assert(); delete_mock.assert(); } diff --git a/lib/crates/fabro-cli/tests/it/cmd/system.rs b/lib/crates/fabro-cli/tests/it/cmd/system.rs index 7b5e3d09c..6ef4b7756 100644 --- a/lib/crates/fabro-cli/tests/it/cmd/system.rs +++ b/lib/crates/fabro-cli/tests/it/cmd/system.rs @@ -18,6 +18,7 @@ fn help() { prune Delete old workflow runs df Show disk usage events Stream run events from the server + repair Inspect and repair durable server data help Print this message or the help of the given subcommand(s) Options: diff --git a/lib/crates/fabro-cli/tests/it/cmd/system_repair.rs b/lib/crates/fabro-cli/tests/it/cmd/system_repair.rs new file mode 100644 index 000000000..c5d5a1d3e --- /dev/null +++ b/lib/crates/fabro-cli/tests/it/cmd/system_repair.rs @@ -0,0 +1,129 @@ +use fabro_test::{fabro_snapshot, test_context}; +use httpmock::MockServer; +use serde_json::Value; + +#[test] +fn help() { + let context = test_context!(); + let mut cmd = context.command(); + cmd.args(["system", "repair", "runs", "--help"]); + fabro_snapshot!(context.filters(), cmd, @" + success: true + exit_code: 0 + ----- stdout ----- + List runs that cannot be loaded from durable storage + + Usage: fabro system repair runs [OPTIONS] + + Options: + --json Output as JSON [env: FABRO_JSON=] + --storage-dir Local storage directory (default: ~/.fabro/storage) [env: FABRO_STORAGE_DIR=] + --debug Enable DEBUG-level logging (default is INFO) [env: FABRO_DEBUG=] + --server Fabro server target: http(s) URL or absolute Unix socket path [env: FABRO_SERVER=] + --no-upgrade-check Disable automatic upgrade check [env: FABRO_NO_UPGRADE_CHECK=true] + --quiet Suppress non-essential output [env: FABRO_QUIET=] + --verbose Enable verbose output [env: FABRO_VERBOSE=] + -h, --help Print help + ----- stderr ----- + "); +} + +#[test] +fn system_repair_runs_reports_unreadable_runs() { + let context = test_context!(); + let server = MockServer::start(); + let mock = server.mock(|when, then| { + when.method("GET").path("/api/v1/system/repair/runs"); + then.status(200) + .header("Content-Type", "application/json") + .body( + serde_json::json!({ + "runs": [{ + "run_id": "01KQT1TNZ0QXK0QHP10G0V5X84", + "created_at": "2026-05-05T20:46:33Z", + "error": "Serialization error: missing field `integrations`", + }], + "total_count": 1, + }) + .to_string(), + ); + }); + context.set_http_target(&server.base_url()); + + let output = context + .command() + .args(["system", "repair", "runs"]) + .output() + .expect("command should run"); + + assert!(output.status.success(), "system repair runs failed"); + let stdout = String::from_utf8(output.stdout).expect("stdout should be valid UTF-8"); + assert!(stdout.contains("Unreadable runs:"), "{stdout}"); + assert!(stdout.contains("01KQT1TNZ0QXK0QHP10G0V5X84"), "{stdout}"); + assert!(stdout.contains("missing field `integrations`"), "{stdout}"); + assert!( + stdout.contains("fabro rm --force 01KQT1TNZ0QXK0QHP10G0V5X84"), + "{stdout}" + ); + mock.assert(); +} + +#[test] +fn system_repair_runs_json_emits_api_response() { + let context = test_context!(); + let server = MockServer::start(); + let mock = server.mock(|when, then| { + when.method("GET").path("/api/v1/system/repair/runs"); + then.status(200) + .header("Content-Type", "application/json") + .body( + serde_json::json!({ + "runs": [{ + "run_id": "01KQT1TNZ0QXK0QHP10G0V5X84", + "created_at": "2026-05-05T20:46:33Z", + "error": "Serialization error: missing field `integrations`", + }], + "total_count": 1, + }) + .to_string(), + ); + }); + context.set_http_target(&server.base_url()); + + let output = context + .command() + .args(["--json", "system", "repair", "runs"]) + .output() + .expect("command should run"); + + assert!(output.status.success()); + let value: Value = + serde_json::from_slice(&output.stdout).expect("system repair JSON should parse"); + assert_eq!(value["total_count"], 1); + assert_eq!(value["runs"][0]["run_id"], "01KQT1TNZ0QXK0QHP10G0V5X84"); + mock.assert(); +} + +#[test] +fn system_repair_runs_reports_empty_state() { + let context = test_context!(); + let server = MockServer::start(); + let mock = server.mock(|when, then| { + when.method("GET").path("/api/v1/system/repair/runs"); + then.status(200) + .header("Content-Type", "application/json") + .body(serde_json::json!({ "runs": [], "total_count": 0 }).to_string()); + }); + context.set_http_target(&server.base_url()); + + let output = context + .command() + .args(["system", "repair", "runs"]) + .output() + .expect("command should run"); + + assert!(output.status.success()); + let stdout = String::from_utf8(output.stdout).expect("stdout should be valid UTF-8"); + assert!(stdout.contains("No run repair issues found."), "{stdout}"); + mock.assert(); +} diff --git a/lib/crates/fabro-client/src/client.rs b/lib/crates/fabro-client/src/client.rs index 9fa4cea5b..e7c111e7f 100644 --- a/lib/crates/fabro-client/src/client.rs +++ b/lib/crates/fabro-client/src/client.rs @@ -686,6 +686,13 @@ impl Client { Ok(response.into_inner()) } + pub async fn get_system_repair_runs(&self) -> Result { + let response = self + .send_api(|client| async move { client.get_system_repair_runs().send().await }) + .await?; + Ok(response.into_inner()) + } + pub async fn prune_runs( &self, body: types::PruneRunsRequest, diff --git a/lib/crates/fabro-server/src/demo/mod.rs b/lib/crates/fabro-server/src/demo/mod.rs index 539731c82..3a5452224 100644 --- a/lib/crates/fabro-server/src/demo/mod.rs +++ b/lib/crates/fabro-server/src/demo/mod.rs @@ -758,6 +758,20 @@ pub(crate) async fn get_system_disk_usage( .into_response() } +pub(crate) async fn get_system_repair_runs( + _auth: RequiredUser, + State(_state): State>, +) -> Response { + ( + StatusCode::OK, + Json(json!({ + "runs": [], + "total_count": 0 + })), + ) + .into_response() +} + pub(crate) async fn prune_runs( _auth: RequiredUser, State(_state): State>, diff --git a/lib/crates/fabro-server/src/server.rs b/lib/crates/fabro-server/src/server.rs index b8abf2716..0520c0424 100644 --- a/lib/crates/fabro-server/src/server.rs +++ b/lib/crates/fabro-server/src/server.rs @@ -34,8 +34,8 @@ pub use fabro_api::types::{ RewindRequest, RewindResponse, RunArtifactEntry, RunArtifactListResponse, RunBilling, RunBillingStage, RunBillingTotals, RunError, RunManifest, RunStage, RunStatusResponse, SandboxFileEntry, SandboxFileListResponse, SshAccessRequest, SshAccessResponse, StageState, - StartRunRequest, SubmitAnswerRequest, SystemFeatures, SystemInfoResponse, SystemRunCounts, - TimelineEntryResponse, WriteBlobResponse, + StartRunRequest, SubmitAnswerRequest, SystemFeatures, SystemInfoResponse, SystemRepairRunIssue, + SystemRepairRunsResponse, SystemRunCounts, TimelineEntryResponse, WriteBlobResponse, }; use fabro_auth::{ CredentialSource, VaultCredentialSource, auth_issue_message, parse_credential_secret, diff --git a/lib/crates/fabro-server/src/server/handler/mod.rs b/lib/crates/fabro-server/src/server/handler/mod.rs index 102dfc9ce..451b9e7d9 100644 --- a/lib/crates/fabro-server/src/server/handler/mod.rs +++ b/lib/crates/fabro-server/src/server/handler/mod.rs @@ -104,6 +104,7 @@ pub(super) fn demo_routes() -> Router> { .route("/settings", get(demo::get_server_settings)) .route("/system/info", get(demo::get_system_info)) .route("/system/df", get(demo::get_system_disk_usage)) + .route("/system/repair/runs", get(demo::get_system_repair_runs)) .route("/system/prune/runs", post(demo::prune_runs)) .route("/billing", get(demo::get_aggregate_billing)) .merge(runs::manifest_routes()) diff --git a/lib/crates/fabro-server/src/server/handler/system.rs b/lib/crates/fabro-server/src/server/handler/system.rs index e318475c1..07455b70e 100644 --- a/lib/crates/fabro-server/src/server/handler/system.rs +++ b/lib/crates/fabro-server/src/server/handler/system.rs @@ -4,9 +4,10 @@ use super::super::{ AggregateBilling, AggregateBillingTotals, ApiError, AppState, BilledTokenCounts, BillingByModel, DfParams, FABRO_VERSION, GithubIntegrationStrategy, IntoResponse, Json, ModelReference, Path, PruneRunsRequest, PruneRunsResponse, Query, RequiredUser, Response, - Router, RunStatus, State, StatusCode, SystemInfoResponse, SystemRunCounts, - build_disk_usage_response, build_prune_plan, delete_run_internal, diagnostics, get, post, - resolve_interp_string, spawn_blocking, system_features, system_sandbox_provider, to_i64, + Router, RunStatus, State, StatusCode, SystemInfoResponse, SystemRepairRunIssue, + SystemRepairRunsResponse, SystemRunCounts, build_disk_usage_response, build_prune_plan, + delete_run_internal, diagnostics, get, post, resolve_interp_string, spawn_blocking, + system_features, system_sandbox_provider, to_i64, }; pub(super) fn routes() -> Router> { @@ -16,6 +17,7 @@ pub(super) fn routes() -> Router> { .route("/settings", get(get_server_settings)) .route("/system/info", get(get_system_info)) .route("/system/df", get(get_system_df)) + .route("/system/repair/runs", get(get_system_repair_runs)) .route("/system/prune/runs", post(prune_runs)) .route("/billing", get(get_aggregate_billing)) } @@ -117,6 +119,37 @@ async fn get_system_df( (StatusCode::OK, Json(response)).into_response() } +async fn get_system_repair_runs( + _auth: RequiredUser, + State(state): State>, +) -> Response { + let issues = match state.store.list_unreadable_runs().await { + Ok(issues) => issues, + Err(err) => { + return ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string()) + .into_response(); + } + }; + let total_count = issues.len(); + let runs = issues + .into_iter() + .map(|issue| SystemRepairRunIssue { + run_id: Some(issue.run_id.to_string()), + created_at: Some(issue.created_at), + error: Some(issue.error), + }) + .collect(); + + ( + StatusCode::OK, + Json(SystemRepairRunsResponse { + runs, + total_count: Some(to_i64(total_count)), + }), + ) + .into_response() +} + async fn prune_runs( _auth: RequiredUser, State(state): State>, diff --git a/lib/crates/fabro-server/src/server/tests.rs b/lib/crates/fabro-server/src/server/tests.rs index 1129ce52b..0cedf8c0a 100644 --- a/lib/crates/fabro-server/src/server/tests.rs +++ b/lib/crates/fabro-server/src/server/tests.rs @@ -2037,6 +2037,43 @@ url = "http://127.0.0.1:32276" ); } +#[tokio::test] +async fn system_repair_runs_lists_catalog_entries_without_projection() { + let state = test_app_state(); + let app = crate::test_support::build_test_router(Arc::clone(&state)); + let run_id = RunId::new(); + state + .store + .catalog_index() + .await + .unwrap() + .add(&run_id) + .await + .unwrap(); + + let response = app + .oneshot( + Request::builder() + .method(Method::GET) + .uri(api("/system/repair/runs")) + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + + let body = response_json!(response, StatusCode::OK).await; + assert_eq!(body["total_count"], 1); + assert_eq!(body["runs"][0]["run_id"], run_id.to_string()); + let created_at = body["runs"][0]["created_at"] + .as_str() + .unwrap() + .parse::>() + .unwrap(); + assert_eq!(created_at, run_id.created_at()); + assert_eq!(body["runs"][0]["error"], "run has no events"); +} + #[tokio::test] async fn create_run_response_omits_web_url_when_web_disabled() { let state = test_app_state_with_options( diff --git a/lib/crates/fabro-store/src/lib.rs b/lib/crates/fabro-store/src/lib.rs index b22e7b2c3..604893c91 100644 --- a/lib/crates/fabro-store/src/lib.rs +++ b/lib/crates/fabro-store/src/lib.rs @@ -24,7 +24,7 @@ pub use run_state::RunProjectionReducer; pub use serializable_projection::SerializableProjection; pub use slate::{ AuthCode, AuthCodeStore, Blob, BlobStore, CachedRunProjection, ConsumeOutcome, Database, - RefreshToken, RefreshTokenStore, RunCatalogIndex, RunDatabase, Runs, + RefreshToken, RefreshTokenStore, RunCatalogIndex, RunDatabase, Runs, UnreadableRun, }; pub use types::EventPayload; diff --git a/lib/crates/fabro-store/src/slate/mod.rs b/lib/crates/fabro-store/src/slate/mod.rs index f4ea78343..8cde4e350 100644 --- a/lib/crates/fabro-store/src/slate/mod.rs +++ b/lib/crates/fabro-store/src/slate/mod.rs @@ -13,6 +13,7 @@ use std::time::Duration; pub use auth_codes::{AuthCode, AuthCodeStore}; pub use auth_tokens::{ConsumeOutcome, RefreshToken, RefreshTokenStore}; pub use blob_store::{Blob, BlobStore}; +use chrono::{DateTime, Utc}; use fabro_types::{RunId, RunSummary}; use object_store::ObjectStore; pub use projection_cache::CachedRunProjection; @@ -26,6 +27,13 @@ use tracing::warn; use crate::{Error, ListRunsQuery, Result, RunProjection, keys}; +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct UnreadableRun { + pub run_id: RunId, + pub created_at: DateTime, + pub error: String, +} + #[derive(Clone)] pub struct Database { object_store: Arc, @@ -242,6 +250,38 @@ impl Database { Ok(self.projection_cache.list(query).await) } + pub async fn list_unreadable_runs(&self) -> Result> { + let db = self.open_db().await?; + let run_ids = self + .catalog_index() + .await? + .list(&ListRunsQuery::default()) + .await?; + let mut unreadable = Vec::new(); + for run_id in run_ids { + match RunDatabase::build_cached_projection(&db, &run_id).await { + Ok(Some(_)) => {} + Ok(None) => unreadable.push(UnreadableRun { + run_id, + created_at: run_id.created_at(), + error: "run has no events".to_string(), + }), + Err(err) => unreadable.push(UnreadableRun { + run_id, + created_at: run_id.created_at(), + error: err.to_string(), + }), + } + } + unreadable.sort_by(|left, right| { + right + .created_at + .cmp(&left.created_at) + .then_with(|| right.run_id.cmp(&left.run_id)) + }); + Ok(unreadable) + } + pub async fn get_cached_run(&self, run_id: &RunId) -> Result> { self.warm_projection_cache().await?; Ok(self.projection_cache.get(run_id).await) @@ -821,6 +861,63 @@ mod tests { assert!(reopened.runs().find(&bad_run_id).await.unwrap().is_none()); } + #[tokio::test] + async fn list_unreadable_runs_reports_catalog_entries_that_fail_projection() { + let (object_store, store) = make_store(); + let good_run = store.create_run(&test_run_id("run-1")).await.unwrap(); + append_completed(&good_run, "run-1", dt("2026-03-27T12:00:00Z")).await; + + let bad_run_id = test_run_id("run-2"); + store + .catalog_index() + .await + .unwrap() + .add(&bad_run_id) + .await + .unwrap(); + let mut run_spec = serde_json::to_value(sample_run_spec("run-2")).unwrap(); + let run_settings = run_spec + .get_mut("settings") + .and_then(|settings| settings.get_mut("run")) + .and_then(serde_json::Value::as_object_mut) + .unwrap(); + run_settings.remove("integrations"); + let db = store.open_db().await.unwrap(); + db.put( + keys::run_event_key(&bad_run_id, 1, 0), + serde_json::to_vec(&serde_json::json!({ + "id": "evt-run-2-run.created", + "ts": "2026-03-27T12:00:10Z", + "run_id": bad_run_id, + "event": "run.created", + "properties": { + "settings": run_spec["settings"], + "graph": run_spec["graph"], + "workflow_slug": run_spec["workflow_slug"], + "source_directory": run_spec["source_directory"], + "run_dir": "/tmp/run-2", + "git": run_spec["git"], + "labels": run_spec["labels"], + }, + })) + .unwrap(), + ) + .await + .unwrap(); + + let reopened = Database::new(object_store, "runs", Duration::from_millis(1), None); + let unreadable = reopened.list_unreadable_runs().await.unwrap(); + + assert_eq!(unreadable.len(), 1); + assert_eq!(unreadable[0].run_id, bad_run_id); + assert_eq!(unreadable[0].created_at, bad_run_id.created_at()); + assert!( + unreadable[0].error.contains("missing field `integrations`"), + "expected missing integrations error, got: {}", + unreadable[0].error + ); + } + #[tokio::test] async fn append_event_refreshes_projection_cache_and_delete_removes_it() { let (_object_store, store) = make_store(); diff --git a/lib/packages/fabro-api-client/src/.openapi-generator/FILES b/lib/packages/fabro-api-client/src/.openapi-generator/FILES index eac69187c..31c539fbb 100644 --- a/lib/packages/fabro-api-client/src/.openapi-generator/FILES +++ b/lib/packages/fabro-api-client/src/.openapi-generator/FILES @@ -307,6 +307,8 @@ models/success-reason.ts models/system-actor-kind.ts models/system-features.ts models/system-info-response.ts +models/system-repair-run-issue.ts +models/system-repair-runs-response.ts models/system-run-counts.ts models/teams-integration-settings.ts models/terminal-status.ts diff --git a/lib/packages/fabro-api-client/src/api/system-api.ts b/lib/packages/fabro-api-client/src/api/system-api.ts index 33a69928d..a18e7b642 100644 --- a/lib/packages/fabro-api-client/src/api/system-api.ts +++ b/lib/packages/fabro-api-client/src/api/system-api.ts @@ -31,6 +31,8 @@ import type { PruneRunsRequest } from '../models'; import type { PruneRunsResponse } from '../models'; // @ts-ignore import type { SystemInfoResponse } from '../models'; +// @ts-ignore +import type { SystemRepairRunsResponse } from '../models'; /** * SystemApi - axios parameter creator */ @@ -154,6 +156,42 @@ export const SystemApiAxiosParamCreator = function (configuration?: Configuratio options: localVarRequestOptions, }; }, + /** + * Lists cataloged runs that cannot be loaded from durable storage. + * @summary List Run Repair Issues + * @param {*} [options] Override http request option. + * @throws {RequiredError} + */ + getSystemRepairRuns: async (options: RawAxiosRequestConfig = {}): Promise => { + const localVarPath = `/api/v1/system/repair/runs`; + // use dummy base URL string because the URL constructor only accepts absolute URLs. + const localVarUrlObj = new URL(localVarPath, DUMMY_BASE_URL); + let baseOptions; + if (configuration) { + baseOptions = configuration.baseOptions; + } + + const localVarRequestOptions = { method: 'GET', ...baseOptions, ...options}; + const localVarHeaderParameter = {} as any; + const localVarQueryParameter = {} as any; + + // authentication SessionCookie required + + // authentication BearerAuth required + // http bearer authentication required + await setBearerAuthToObject(localVarHeaderParameter, configuration) + + localVarHeaderParameter['Accept'] = 'application/json'; + + setSearchParams(localVarUrlObj, localVarQueryParameter); + let headersFromBaseOptions = baseOptions && baseOptions.headers ? baseOptions.headers : {}; + localVarRequestOptions.headers = {...localVarHeaderParameter, ...headersFromBaseOptions, ...options.headers}; + + return { + url: toPathString(localVarUrlObj), + options: localVarRequestOptions, + }; + }, /** * Deletes completed runs matching the provided filters, or previews the deletion set when dry-run is enabled. * @summary Prune Runs @@ -242,6 +280,18 @@ export const SystemApiFp = function(configuration?: Configuration) { const localVarOperationServerBasePath = operationServerMap['SystemApi.getSystemInfo']?.[localVarOperationServerIndex]?.url; return (axios, basePath) => createRequestFunction(localVarAxiosArgs, globalAxios, BASE_PATH, configuration)(axios, localVarOperationServerBasePath || basePath); }, + /** + * Lists cataloged runs that cannot be loaded from durable storage. + * @summary List Run Repair Issues + * @param {*} [options] Override http request option. + * @throws {RequiredError} + */ + async getSystemRepairRuns(options?: RawAxiosRequestConfig): Promise<(axios?: AxiosInstance, basePath?: string) => AxiosPromise> { + const localVarAxiosArgs = await localVarAxiosParamCreator.getSystemRepairRuns(options); + const localVarOperationServerIndex = configuration?.serverIndex ?? 0; + const localVarOperationServerBasePath = operationServerMap['SystemApi.getSystemRepairRuns']?.[localVarOperationServerIndex]?.url; + return (axios, basePath) => createRequestFunction(localVarAxiosArgs, globalAxios, BASE_PATH, configuration)(axios, localVarOperationServerBasePath || basePath); + }, /** * Deletes completed runs matching the provided filters, or previews the deletion set when dry-run is enabled. * @summary Prune Runs @@ -293,6 +343,15 @@ export const SystemApiFactory = function (configuration?: Configuration, basePat getSystemInfo(options?: RawAxiosRequestConfig): AxiosPromise { return localVarFp.getSystemInfo(options).then((request) => request(axios, basePath)); }, + /** + * Lists cataloged runs that cannot be loaded from durable storage. + * @summary List Run Repair Issues + * @param {*} [options] Override http request option. + * @throws {RequiredError} + */ + getSystemRepairRuns(options?: RawAxiosRequestConfig): AxiosPromise { + return localVarFp.getSystemRepairRuns(options).then((request) => request(axios, basePath)); + }, /** * Deletes completed runs matching the provided filters, or previews the deletion set when dry-run is enabled. * @summary Prune Runs @@ -342,6 +401,16 @@ export class SystemApi extends BaseAPI { return SystemApiFp(this.configuration).getSystemInfo(options).then((request) => request(this.axios, this.basePath)); } + /** + * Lists cataloged runs that cannot be loaded from durable storage. + * @summary List Run Repair Issues + * @param {*} [options] Override http request option. + * @throws {RequiredError} + */ + public getSystemRepairRuns(options?: RawAxiosRequestConfig) { + return SystemApiFp(this.configuration).getSystemRepairRuns(options).then((request) => request(this.axios, this.basePath)); + } + /** * Deletes completed runs matching the provided filters, or previews the deletion set when dry-run is enabled. * @summary Prune Runs diff --git a/lib/packages/fabro-api-client/src/models/index.ts b/lib/packages/fabro-api-client/src/models/index.ts index 3897be725..b50c5e824 100644 --- a/lib/packages/fabro-api-client/src/models/index.ts +++ b/lib/packages/fabro-api-client/src/models/index.ts @@ -286,6 +286,8 @@ export * from './success-reason'; export * from './system-actor-kind'; export * from './system-features'; export * from './system-info-response'; +export * from './system-repair-run-issue'; +export * from './system-repair-runs-response'; export * from './system-run-counts'; export * from './teams-integration-settings'; export * from './terminal-status'; diff --git a/lib/packages/fabro-api-client/src/models/run-integrations-github-settings.ts b/lib/packages/fabro-api-client/src/models/run-integrations-github-settings.ts index 740e2967b..8502a9339 100644 --- a/lib/packages/fabro-api-client/src/models/run-integrations-github-settings.ts +++ b/lib/packages/fabro-api-client/src/models/run-integrations-github-settings.ts @@ -19,4 +19,3 @@ export interface RunIntegrationsGithubSettings { } - diff --git a/lib/packages/fabro-api-client/src/models/run-integrations-settings.ts b/lib/packages/fabro-api-client/src/models/run-integrations-settings.ts index de44e953f..4b6dcd6ae 100644 --- a/lib/packages/fabro-api-client/src/models/run-integrations-settings.ts +++ b/lib/packages/fabro-api-client/src/models/run-integrations-settings.ts @@ -22,4 +22,3 @@ export interface RunIntegrationsSettings { } - diff --git a/lib/packages/fabro-api-client/src/models/system-repair-run-issue.ts b/lib/packages/fabro-api-client/src/models/system-repair-run-issue.ts new file mode 100644 index 000000000..cd5fd76ad --- /dev/null +++ b/lib/packages/fabro-api-client/src/models/system-repair-run-issue.ts @@ -0,0 +1,34 @@ +/* tslint:disable */ +/* eslint-disable */ +/** + * Fabro Run API + * HTTP API for managing Fabro workflow run executions. + * + * The version of the OpenAPI document: 0.1.0 + * + * + * NOTE: This class is auto generated by OpenAPI Generator (https://openapi-generator.tech). + * https://openapi-generator.tech + * Do not edit the class manually. + */ + + + +/** + * One cataloged run that cannot be loaded from durable storage. + */ +export interface SystemRepairRunIssue { + /** + * Run identifier. + */ + 'run_id'?: string; + /** + * Timestamp encoded in the run identifier. + */ + 'created_at'?: string; + /** + * Error produced while loading the run projection. + */ + 'error'?: string; +} + diff --git a/lib/packages/fabro-api-client/src/models/system-repair-runs-response.ts b/lib/packages/fabro-api-client/src/models/system-repair-runs-response.ts new file mode 100644 index 000000000..7fbe34534 --- /dev/null +++ b/lib/packages/fabro-api-client/src/models/system-repair-runs-response.ts @@ -0,0 +1,30 @@ +/* tslint:disable */ +/* eslint-disable */ +/** + * Fabro Run API + * HTTP API for managing Fabro workflow run executions. + * + * The version of the OpenAPI document: 0.1.0 + * + * + * NOTE: This class is auto generated by OpenAPI Generator (https://openapi-generator.tech). + * https://openapi-generator.tech + * Do not edit the class manually. + */ + + +// May contain unused imports in some cases +// @ts-ignore +import type { SystemRepairRunIssue } from './system-repair-run-issue'; + +/** + * Runs that need manual repair or deletion because they cannot be loaded. + */ +export interface SystemRepairRunsResponse { + 'runs'?: Array; + /** + * Count of run repair issues. + */ + 'total_count'?: number; +} +