mirror of
https://github.com/fabro-sh/fabro.git
synced 2026-10-09 03:20:56 +00:00
refactor(server): drop duplicate run status adapters
Use the unified status family directly in server.rs and remove the manual conversion helpers that only existed to bridge duplicate API/domain types.
This commit is contained in:
Bryan Helmkamp
parent
eb7e65b6b4
commit
e92a2ba9db
1 changed files with 66 additions and 132 deletions
|
|
@ -25,18 +25,17 @@ use bytes::Bytes;
|
|||
pub use fabro_api::types::{
|
||||
AggregateBilling, AggregateBillingTotals, ApiQuestion, ApiQuestionOption, AppendEventResponse,
|
||||
ArtifactEntry, ArtifactListResponse, BilledTokenCounts as ApiBilledTokenCounts, BillingByModel,
|
||||
BillingStageRef, BlockedReason as ApiBlockedReason, CompletionContentPart, CompletionMessage,
|
||||
CompletionMessageRole, CompletionResponse, CompletionToolChoiceMode, CompletionUsage,
|
||||
CreateCompletionRequest, CreateSecretRequest, DeleteSecretRequest, DiskUsageResponse,
|
||||
DiskUsageRunRow, DiskUsageSummaryRow, EventEnvelope as ApiEventEnvelope, ModelReference,
|
||||
PaginatedEventList, PaginatedRunList, PaginationMeta, PreflightResponse, PreviewUrlRequest,
|
||||
PreviewUrlResponse, PruneRunEntry, PruneRunsRequest, PruneRunsResponse,
|
||||
QuestionType as ApiQuestionType, RenderWorkflowGraphDirection, RenderWorkflowGraphRequest,
|
||||
RunArtifactEntry, RunArtifactListResponse, RunBilling, RunBillingStage, RunBillingTotals,
|
||||
RunControlAction as ApiRunControlAction, RunError, RunManifest, RunStage, RunStatus,
|
||||
RunStatusResponse, SandboxFileEntry, SandboxFileListResponse, SecretType as ApiSecretType,
|
||||
ServerSettings, SshAccessRequest, SshAccessResponse, StageStatus as ApiStageStatus,
|
||||
StartRunRequest, StatusReason as ApiStatusReason, SubmitAnswerRequest, SystemFeatures,
|
||||
BillingStageRef, CompletionContentPart, CompletionMessage, CompletionMessageRole,
|
||||
CompletionResponse, CompletionToolChoiceMode, CompletionUsage, CreateCompletionRequest,
|
||||
CreateSecretRequest, DeleteSecretRequest, DiskUsageResponse, DiskUsageRunRow,
|
||||
DiskUsageSummaryRow, EventEnvelope as ApiEventEnvelope, ModelReference, PaginatedEventList,
|
||||
PaginatedRunList, PaginationMeta, PreflightResponse, PreviewUrlRequest, PreviewUrlResponse,
|
||||
PruneRunEntry, PruneRunsRequest, PruneRunsResponse, QuestionType as ApiQuestionType,
|
||||
RenderWorkflowGraphDirection, RenderWorkflowGraphRequest, RunArtifactEntry,
|
||||
RunArtifactListResponse, RunBilling, RunBillingStage, RunBillingTotals, RunError, RunManifest,
|
||||
RunStage, RunStatusResponse, SandboxFileEntry, SandboxFileListResponse,
|
||||
SecretType as ApiSecretType, ServerSettings, SshAccessRequest, SshAccessResponse,
|
||||
StageStatus as ApiStageStatus, StartRunRequest, SubmitAnswerRequest, SystemFeatures,
|
||||
SystemInfoResponse, SystemRunCounts, WriteBlobResponse,
|
||||
};
|
||||
use fabro_auth::parse_credential_secret;
|
||||
|
|
@ -86,9 +85,7 @@ use fabro_workflow::records::Checkpoint;
|
|||
use fabro_workflow::run_lookup::{
|
||||
RunInfo, StatusFilter, filter_runs, scan_runs_with_summaries, scratch_base,
|
||||
};
|
||||
use fabro_workflow::run_status::{
|
||||
RunStatus as WorkflowRunStatus, StatusReason as WorkflowStatusReason,
|
||||
};
|
||||
use fabro_workflow::run_status::{RunStatus, StatusReason};
|
||||
use jsonwebtoken::{Algorithm, DecodingKey, EncodingKey, Header, Validation};
|
||||
use object_store::memory::InMemory as MemoryObjectStore;
|
||||
use rand::TryRngCore;
|
||||
|
|
@ -2724,16 +2721,14 @@ fn test_secret_store_path() -> PathBuf {
|
|||
dir.join("secrets.json")
|
||||
}
|
||||
|
||||
fn board_column(status: WorkflowRunStatus) -> Option<&'static str> {
|
||||
fn board_column(status: RunStatus) -> Option<&'static str> {
|
||||
match status {
|
||||
WorkflowRunStatus::Submitted | WorkflowRunStatus::Queued | WorkflowRunStatus::Starting => {
|
||||
Some("initializing")
|
||||
}
|
||||
WorkflowRunStatus::Running | WorkflowRunStatus::Paused => Some("running"),
|
||||
WorkflowRunStatus::Blocked => Some("blocked"),
|
||||
WorkflowRunStatus::Succeeded => Some("succeeded"),
|
||||
WorkflowRunStatus::Failed | WorkflowRunStatus::Dead => Some("failed"),
|
||||
WorkflowRunStatus::Removing | WorkflowRunStatus::Archived => None,
|
||||
RunStatus::Submitted | RunStatus::Queued | RunStatus::Starting => Some("initializing"),
|
||||
RunStatus::Running | RunStatus::Paused => Some("running"),
|
||||
RunStatus::Blocked => Some("blocked"),
|
||||
RunStatus::Succeeded => Some("succeeded"),
|
||||
RunStatus::Failed | RunStatus::Dead => Some("failed"),
|
||||
RunStatus::Removing | RunStatus::Archived => None,
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -2788,9 +2783,9 @@ fn summary_to_api_run_summary(summary: fabro_store::RunSummary) -> serde_json::V
|
|||
"repository": { "name": repository },
|
||||
"start_time": summary.start_time.map(|time| time.to_rfc3339()),
|
||||
"status": summary.status,
|
||||
"status_reason": summary.status_reason.map(api_status_reason),
|
||||
"blocked_reason": summary.blocked_reason.map(api_blocked_reason),
|
||||
"pending_control": summary.pending_control.map(api_pending_control),
|
||||
"status_reason": summary.status_reason,
|
||||
"blocked_reason": summary.blocked_reason,
|
||||
"pending_control": summary.pending_control,
|
||||
"duration_ms": summary.duration_ms,
|
||||
"elapsed_secs": elapsed_secs(summary.duration_ms),
|
||||
"total_usd_micros": summary.total_usd_micros,
|
||||
|
|
@ -2922,7 +2917,7 @@ async fn list_runs(
|
|||
let include_archived = params.include_archived;
|
||||
let items = runs
|
||||
.into_iter()
|
||||
.filter(|summary| include_archived || summary.status != WorkflowRunStatus::Archived)
|
||||
.filter(|summary| include_archived || summary.status != RunStatus::Archived)
|
||||
.map(summary_to_api_run_summary)
|
||||
.collect::<Vec<_>>();
|
||||
let (data, has_more) = paginate_items(items, ¶ms.pagination());
|
||||
|
|
@ -3380,29 +3375,26 @@ struct LiveWorkerProcess {
|
|||
fn failure_for_incomplete_run(
|
||||
pending_control: Option<RunControlAction>,
|
||||
terminated_message: String,
|
||||
) -> (WorkflowError, Option<WorkflowStatusReason>) {
|
||||
) -> (WorkflowError, Option<StatusReason>) {
|
||||
if pending_control == Some(RunControlAction::Cancel) {
|
||||
(
|
||||
WorkflowError::Cancelled,
|
||||
Some(WorkflowStatusReason::Cancelled),
|
||||
)
|
||||
(WorkflowError::Cancelled, Some(StatusReason::Cancelled))
|
||||
} else {
|
||||
(
|
||||
WorkflowError::engine(terminated_message),
|
||||
Some(WorkflowStatusReason::Terminated),
|
||||
Some(StatusReason::Terminated),
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
fn should_reconcile_run_on_startup(status: WorkflowRunStatus) -> bool {
|
||||
fn should_reconcile_run_on_startup(status: RunStatus) -> bool {
|
||||
matches!(
|
||||
status,
|
||||
WorkflowRunStatus::Queued
|
||||
| WorkflowRunStatus::Starting
|
||||
| WorkflowRunStatus::Running
|
||||
| WorkflowRunStatus::Blocked
|
||||
| WorkflowRunStatus::Paused
|
||||
| WorkflowRunStatus::Removing
|
||||
RunStatus::Queued
|
||||
| RunStatus::Starting
|
||||
| RunStatus::Running
|
||||
| RunStatus::Blocked
|
||||
| RunStatus::Paused
|
||||
| RunStatus::Removing
|
||||
)
|
||||
}
|
||||
|
||||
|
|
@ -3562,7 +3554,7 @@ async fn persist_cancelled_run_status(state: &AppState, run_id: RunId) -> anyhow
|
|||
&workflow_event::Event::WorkflowRunFailed {
|
||||
error: WorkflowError::Cancelled,
|
||||
duration_ms: 0,
|
||||
reason: Some(WorkflowStatusReason::Cancelled),
|
||||
reason: Some(StatusReason::Cancelled),
|
||||
git_commit_sha: None,
|
||||
final_patch: None,
|
||||
},
|
||||
|
|
@ -3618,28 +3610,6 @@ fn managed_run(
|
|||
}
|
||||
}
|
||||
|
||||
fn api_status_from_workflow(status: WorkflowRunStatus) -> RunStatus {
|
||||
match status {
|
||||
WorkflowRunStatus::Submitted => RunStatus::Submitted,
|
||||
WorkflowRunStatus::Queued => RunStatus::Queued,
|
||||
WorkflowRunStatus::Starting => RunStatus::Starting,
|
||||
WorkflowRunStatus::Running => RunStatus::Running,
|
||||
WorkflowRunStatus::Blocked => RunStatus::Blocked,
|
||||
WorkflowRunStatus::Paused => RunStatus::Paused,
|
||||
WorkflowRunStatus::Removing => RunStatus::Removing,
|
||||
WorkflowRunStatus::Succeeded => RunStatus::Succeeded,
|
||||
WorkflowRunStatus::Failed => RunStatus::Failed,
|
||||
WorkflowRunStatus::Dead => RunStatus::Dead,
|
||||
WorkflowRunStatus::Archived => RunStatus::Archived,
|
||||
}
|
||||
}
|
||||
|
||||
fn api_blocked_reason(reason: BlockedReason) -> ApiBlockedReason {
|
||||
match reason {
|
||||
BlockedReason::HumanInputRequired => ApiBlockedReason::HumanInputRequired,
|
||||
}
|
||||
}
|
||||
|
||||
fn worker_mode_arg(mode: RunExecutionMode) -> &'static str {
|
||||
match mode {
|
||||
RunExecutionMode::Start => "start",
|
||||
|
|
@ -3647,43 +3617,19 @@ fn worker_mode_arg(mode: RunExecutionMode) -> &'static str {
|
|||
}
|
||||
}
|
||||
|
||||
fn api_status_reason(reason: WorkflowStatusReason) -> ApiStatusReason {
|
||||
match reason {
|
||||
WorkflowStatusReason::Completed => ApiStatusReason::Completed,
|
||||
WorkflowStatusReason::PartialSuccess => ApiStatusReason::PartialSuccess,
|
||||
WorkflowStatusReason::WorkflowError => ApiStatusReason::WorkflowError,
|
||||
WorkflowStatusReason::Cancelled => ApiStatusReason::Cancelled,
|
||||
WorkflowStatusReason::Terminated => ApiStatusReason::Terminated,
|
||||
WorkflowStatusReason::TransientInfra => ApiStatusReason::TransientInfra,
|
||||
WorkflowStatusReason::BudgetExhausted => ApiStatusReason::BudgetExhausted,
|
||||
WorkflowStatusReason::LaunchFailed => ApiStatusReason::LaunchFailed,
|
||||
WorkflowStatusReason::BootstrapFailed => ApiStatusReason::BootstrapFailed,
|
||||
WorkflowStatusReason::SandboxInitFailed => ApiStatusReason::SandboxInitFailed,
|
||||
WorkflowStatusReason::SandboxInitializing => ApiStatusReason::SandboxInitializing,
|
||||
}
|
||||
}
|
||||
|
||||
fn api_pending_control(action: RunControlAction) -> ApiRunControlAction {
|
||||
match action {
|
||||
RunControlAction::Cancel => ApiRunControlAction::Cancel,
|
||||
RunControlAction::Pause => ApiRunControlAction::Pause,
|
||||
RunControlAction::Unpause => ApiRunControlAction::Unpause,
|
||||
}
|
||||
}
|
||||
|
||||
async fn load_run_status_metadata(
|
||||
state: &AppState,
|
||||
run_id: RunId,
|
||||
) -> (
|
||||
Option<ApiStatusReason>,
|
||||
Option<ApiBlockedReason>,
|
||||
Option<ApiRunControlAction>,
|
||||
Option<StatusReason>,
|
||||
Option<BlockedReason>,
|
||||
Option<RunControlAction>,
|
||||
) {
|
||||
match state.store.runs().find(&run_id).await {
|
||||
Ok(Some(summary)) => (
|
||||
summary.status_reason.map(api_status_reason),
|
||||
summary.blocked_reason.map(api_blocked_reason),
|
||||
summary.pending_control.map(api_pending_control),
|
||||
summary.status_reason,
|
||||
summary.blocked_reason,
|
||||
summary.pending_control,
|
||||
),
|
||||
_ => (None, None, None),
|
||||
}
|
||||
|
|
@ -4365,7 +4311,7 @@ async fn start_run(
|
|||
} else if let Some(record) = run_state.status.as_ref() {
|
||||
if !matches!(
|
||||
record.status,
|
||||
WorkflowRunStatus::Submitted | WorkflowRunStatus::Queued | WorkflowRunStatus::Starting
|
||||
RunStatus::Submitted | RunStatus::Queued | RunStatus::Starting
|
||||
) {
|
||||
return ApiError::new(
|
||||
StatusCode::CONFLICT,
|
||||
|
|
@ -4792,7 +4738,7 @@ async fn execute_run_subprocess(state: Arc<AppState>, run_id: RunId) {
|
|||
&workflow_event::Event::WorkflowRunFailed {
|
||||
error: WorkflowError::engine(err.to_string()),
|
||||
duration_ms: 0,
|
||||
reason: Some(WorkflowStatusReason::LaunchFailed),
|
||||
reason: Some(StatusReason::LaunchFailed),
|
||||
git_commit_sha: None,
|
||||
final_patch: None,
|
||||
},
|
||||
|
|
@ -4814,7 +4760,7 @@ async fn execute_run_subprocess(state: Arc<AppState>, run_id: RunId) {
|
|||
&workflow_event::Event::WorkflowRunFailed {
|
||||
error: WorkflowError::engine(message.clone()),
|
||||
duration_ms: 0,
|
||||
reason: Some(WorkflowStatusReason::LaunchFailed),
|
||||
reason: Some(StatusReason::LaunchFailed),
|
||||
git_commit_sha: None,
|
||||
final_patch: None,
|
||||
},
|
||||
|
|
@ -4844,7 +4790,7 @@ async fn execute_run_subprocess(state: Arc<AppState>, run_id: RunId) {
|
|||
&workflow_event::Event::WorkflowRunFailed {
|
||||
error: WorkflowError::engine(message.clone()),
|
||||
duration_ms: 0,
|
||||
reason: Some(WorkflowStatusReason::LaunchFailed),
|
||||
reason: Some(StatusReason::LaunchFailed),
|
||||
git_commit_sha: None,
|
||||
final_patch: None,
|
||||
},
|
||||
|
|
@ -4865,7 +4811,7 @@ async fn execute_run_subprocess(state: Arc<AppState>, run_id: RunId) {
|
|||
&workflow_event::Event::WorkflowRunFailed {
|
||||
error: WorkflowError::engine(message.clone()),
|
||||
duration_ms: 0,
|
||||
reason: Some(WorkflowStatusReason::LaunchFailed),
|
||||
reason: Some(StatusReason::LaunchFailed),
|
||||
git_commit_sha: None,
|
||||
final_patch: None,
|
||||
},
|
||||
|
|
@ -4898,7 +4844,7 @@ async fn execute_run_subprocess(state: Arc<AppState>, run_id: RunId) {
|
|||
&workflow_event::Event::WorkflowRunFailed {
|
||||
error: WorkflowError::engine(err.to_string()),
|
||||
duration_ms: 0,
|
||||
reason: Some(WorkflowStatusReason::Terminated),
|
||||
reason: Some(StatusReason::Terminated),
|
||||
git_commit_sha: None,
|
||||
final_patch: None,
|
||||
},
|
||||
|
|
@ -4984,7 +4930,7 @@ async fn execute_run_subprocess(state: Arc<AppState>, run_id: RunId) {
|
|||
let mut runs = state.runs.lock().expect("runs lock poisoned");
|
||||
if let Some(managed_run) = runs.get_mut(&run_id) {
|
||||
if let Some(status) = final_state.status.as_ref() {
|
||||
managed_run.status = api_status_from_workflow(status.status);
|
||||
managed_run.status = status.status;
|
||||
} else if !wait_status.success() {
|
||||
managed_run.status = RunStatus::Failed;
|
||||
}
|
||||
|
|
@ -6306,7 +6252,7 @@ async fn reject_if_archived(state: &AppState, run_id: &RunId) -> Option<Response
|
|||
let run_store = state.store.open_run_reader(run_id).await.ok()?;
|
||||
let projection = run_store.state().await.ok()?;
|
||||
let status = projection.status.as_ref()?.status;
|
||||
(status == WorkflowRunStatus::Archived).then(|| {
|
||||
(status == RunStatus::Archived).then(|| {
|
||||
ApiError::new(
|
||||
StatusCode::CONFLICT,
|
||||
operations::archived_rejection_message(run_id),
|
||||
|
|
@ -6754,9 +6700,9 @@ async fn archive_status_response(state: &AppState, id: RunId) -> Response {
|
|||
)
|
||||
.into_response();
|
||||
};
|
||||
let status = api_status_from_workflow(record.status);
|
||||
let status_reason = record.status_reason.map(api_status_reason);
|
||||
let blocked_reason = record.blocked_reason.map(api_blocked_reason);
|
||||
let status = record.status;
|
||||
let status_reason = record.status_reason;
|
||||
let blocked_reason = record.blocked_reason;
|
||||
(
|
||||
StatusCode::OK,
|
||||
Json(RunStatusResponse {
|
||||
|
|
@ -9988,8 +9934,8 @@ level = "debug"
|
|||
|
||||
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.status_reason, Some(WorkflowStatusReason::Cancelled));
|
||||
assert_eq!(status.status, RunStatus::Failed);
|
||||
assert_eq!(status.status_reason, Some(StatusReason::Cancelled));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
|
|
@ -10145,7 +10091,7 @@ level = "debug"
|
|||
assert_eq!(body["pending_control"], serde_json::Value::Null);
|
||||
|
||||
let summary = state.store.runs().find(&run_id).await.unwrap().unwrap();
|
||||
assert_eq!(summary.status, WorkflowRunStatus::Paused);
|
||||
assert_eq!(summary.status, RunStatus::Paused);
|
||||
assert_eq!(
|
||||
summary.blocked_reason,
|
||||
Some(BlockedReason::HumanInputRequired)
|
||||
|
|
@ -10231,7 +10177,7 @@ level = "debug"
|
|||
assert_eq!(body["pending_control"], serde_json::Value::Null);
|
||||
|
||||
let summary = state.store.runs().find(&run_id).await.unwrap().unwrap();
|
||||
assert_eq!(summary.status, WorkflowRunStatus::Blocked);
|
||||
assert_eq!(summary.status, RunStatus::Blocked);
|
||||
assert_eq!(
|
||||
summary.blocked_reason,
|
||||
Some(BlockedReason::HumanInputRequired)
|
||||
|
|
@ -10282,7 +10228,7 @@ level = "debug"
|
|||
.state()
|
||||
.await
|
||||
.unwrap();
|
||||
assert_eq!(run_1.status.unwrap().status, WorkflowRunStatus::Submitted);
|
||||
assert_eq!(run_1.status.unwrap().status, RunStatus::Submitted);
|
||||
|
||||
let run_2 = state
|
||||
.store
|
||||
|
|
@ -10293,11 +10239,8 @@ level = "debug"
|
|||
.await
|
||||
.unwrap();
|
||||
let run_2_status = run_2.status.unwrap();
|
||||
assert_eq!(run_2_status.status, WorkflowRunStatus::Failed);
|
||||
assert_eq!(
|
||||
run_2_status.status_reason,
|
||||
Some(WorkflowStatusReason::Terminated)
|
||||
);
|
||||
assert_eq!(run_2_status.status, RunStatus::Failed);
|
||||
assert_eq!(run_2_status.status_reason, Some(StatusReason::Terminated));
|
||||
|
||||
let run_3 = state
|
||||
.store
|
||||
|
|
@ -10308,11 +10251,8 @@ level = "debug"
|
|||
.await
|
||||
.unwrap();
|
||||
let run_3_status = run_3.status.unwrap();
|
||||
assert_eq!(run_3_status.status, WorkflowRunStatus::Failed);
|
||||
assert_eq!(
|
||||
run_3_status.status_reason,
|
||||
Some(WorkflowStatusReason::Cancelled)
|
||||
);
|
||||
assert_eq!(run_3_status.status, RunStatus::Failed);
|
||||
assert_eq!(run_3_status.status_reason, Some(StatusReason::Cancelled));
|
||||
assert_eq!(run_3.pending_control, None);
|
||||
}
|
||||
|
||||
|
|
@ -10383,11 +10323,8 @@ level = "debug"
|
|||
.await
|
||||
.unwrap();
|
||||
let run_status = run_state.status.unwrap();
|
||||
assert_eq!(run_status.status, WorkflowRunStatus::Failed);
|
||||
assert_eq!(
|
||||
run_status.status_reason,
|
||||
Some(WorkflowStatusReason::Terminated)
|
||||
);
|
||||
assert_eq!(run_status.status, RunStatus::Failed);
|
||||
assert_eq!(run_status.status_reason, Some(StatusReason::Terminated));
|
||||
}
|
||||
|
||||
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
||||
|
|
@ -10473,8 +10410,8 @@ timeout = "30s"
|
|||
let mut status_record = None;
|
||||
for _ in 0..50 {
|
||||
if let Some(record) = run_store.state().await.unwrap().status {
|
||||
if record.status == WorkflowRunStatus::Failed
|
||||
&& record.status_reason == Some(WorkflowStatusReason::Cancelled)
|
||||
if record.status == RunStatus::Failed
|
||||
&& record.status_reason == Some(StatusReason::Cancelled)
|
||||
{
|
||||
status_record = Some(record);
|
||||
break;
|
||||
|
|
@ -10484,11 +10421,8 @@ timeout = "30s"
|
|||
}
|
||||
|
||||
let status_record = status_record.expect("status record should be persisted");
|
||||
assert_eq!(status_record.status, WorkflowRunStatus::Failed);
|
||||
assert_eq!(
|
||||
status_record.status_reason,
|
||||
Some(WorkflowStatusReason::Cancelled)
|
||||
);
|
||||
assert_eq!(status_record.status, RunStatus::Failed);
|
||||
assert_eq!(status_record.status_reason, Some(StatusReason::Cancelled));
|
||||
}
|
||||
|
||||
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue