diff --git a/docs/public/api-reference/fabro-api.yaml b/docs/public/api-reference/fabro-api.yaml index ec066dc94..2b82081f6 100644 --- a/docs/public/api-reference/fabro-api.yaml +++ b/docs/public/api-reference/fabro-api.yaml @@ -2386,12 +2386,11 @@ paths: tags: [Runs] summary: Retry Run description: > - Creates a new run from the terminal source run's last checkpoint and - starts it. When the source failed on a stage, that stage runs again on - the files of the stage before it; otherwise the new run continues from - the last checkpoint as it stands. The new run records `retried_from` - and `fork_source_ref`; the source run is left unchanged. Active and - archived runs are not retryable. + Creates a new run from the terminal source run's saved specification + and starts the workflow from the beginning in a fresh workspace. + No Git checkpoint is required. The new run records `retried_from` + without a `fork_source_ref`; the source run is left unchanged. + Active and archived runs are not retryable. parameters: - $ref: "#/components/parameters/RunId" responses: @@ -2402,7 +2401,7 @@ paths: schema: $ref: "#/components/schemas/Run" "400": - description: The source has no checkpoint to retry from + description: Invalid retry request headers: x-request-id: $ref: "#/components/headers/XRequestId" diff --git a/docs/public/reference/cli.mdx b/docs/public/reference/cli.mdx index 49d17f763..6cf2ebea1 100644 --- a/docs/public/reference/cli.mdx +++ b/docs/public/reference/cli.mdx @@ -90,7 +90,7 @@ fabro [OPTIONS] [COMMAND] | `fabro provider` | Provider operations | | `fabro repo` | Repository commands | | `fabro resume` | Resume an interrupted workflow run | -| `fabro retry` | Retry a finished workflow run from its last checkpoint in a new run | +| `fabro retry` | Retry a finished workflow from the start in a new run | | `fabro rewind` | Rewind a workflow run to an earlier checkpoint, replacing it | | `fabro rm` | Remove one or more workflow runs | | `fabro run` | Register a workflow version, create a run, and start it | @@ -1026,7 +1026,7 @@ fabro resume [OPTIONS] ### `fabro retry` -Retry a finished workflow run from its last checkpoint in a new run +Retry a finished workflow from the start in a new run ```bash fabro retry [OPTIONS] diff --git a/lib/apps/fabro-cli/src/args.rs b/lib/apps/fabro-cli/src/args.rs index 9db43de8f..9ffe353c1 100644 --- a/lib/apps/fabro-cli/src/args.rs +++ b/lib/apps/fabro-cli/src/args.rs @@ -1321,7 +1321,7 @@ pub(crate) enum RunCommands { Logs(LogsArgs), /// Resume an interrupted workflow run Resume(ResumeArgs), - /// Retry a finished workflow run from its last checkpoint in a new run + /// Retry a finished workflow from the start in a new run Retry(RetryArgs), /// Fork a workflow run from an earlier checkpoint into a new run Fork(ForkArgs), diff --git a/lib/apps/fabro-cli/src/commands/run/retry.rs b/lib/apps/fabro-cli/src/commands/run/retry.rs index 2fae3e865..566ffdfac 100644 --- a/lib/apps/fabro-cli/src/commands/run/retry.rs +++ b/lib/apps/fabro-cli/src/commands/run/retry.rs @@ -5,8 +5,7 @@ use crate::args::RetryArgs; use crate::command_context::CommandContext; use crate::shared::print_json_pretty; -/// Retry a finished run from its last checkpoint in a new run, which starts -/// at once. +/// Retry a finished workflow from the start in a new run. pub(crate) async fn run(args: &RetryArgs, base_ctx: &CommandContext) -> Result<()> { let printer = base_ctx.printer(); let ctx = base_ctx.with_target(&args.server)?; diff --git a/lib/apps/fabro-cli/tests/it/cmd/fabro.rs b/lib/apps/fabro-cli/tests/it/cmd/fabro.rs index 52842f5db..7fb2d7488 100644 --- a/lib/apps/fabro-cli/tests/it/cmd/fabro.rs +++ b/lib/apps/fabro-cli/tests/it/cmd/fabro.rs @@ -19,7 +19,7 @@ fn help() { events View the event log of a workflow run logs View the raw worker tracing log of a workflow run resume Resume an interrupted workflow run - retry Retry a finished workflow run from its last checkpoint in a new run + retry Retry a finished workflow from the start in a new run fork Fork a workflow run from an earlier checkpoint into a new run rewind Rewind a workflow run to an earlier checkpoint, replacing it timeline Show the checkpoint timeline of a workflow run diff --git a/lib/apps/fabro-cli/tests/it/scenario/petri_fork.rs b/lib/apps/fabro-cli/tests/it/scenario/petri_fork.rs index b1b8ab053..04a022d51 100644 --- a/lib/apps/fabro-cli/tests/it/scenario/petri_fork.rs +++ b/lib/apps/fabro-cli/tests/it/scenario/petri_fork.rs @@ -251,11 +251,10 @@ async fn a_fork_at_the_first_stage_continues_with_the_rest_on_its_files() { server.shutdown(); } -/// A retry of a run whose last stage failed reruns that stage: the failure -/// was transient, so the retry passes it on the files of the stage before -/// and finishes the run. +/// A retry starts the workflow over in a new run: every stage runs again +/// in a fresh workspace, so a transient failure passes the second time. #[tokio::test(flavor = "multi_thread")] -async fn a_retry_reruns_the_failed_stage_and_succeeds_when_the_failure_was_transient() { +async fn a_retry_starts_over_and_succeeds_when_the_failure_was_transient() { let context = test_context!(); let server = RunningServer::start().await; let marker = context.temp_dir.join("flaky.marker"); @@ -280,22 +279,21 @@ async fn a_retry_reruns_the_failed_stage_and_succeeds_when_the_failure_was_trans assert_eq!(nodes(&retry_timeline), [ "start", "one", "flaky", "three", "exit" ]); - assert_eq!(retry_timeline["forked_from"]["source_run_id"], source); - assert_eq!(retry_timeline["forked_from"]["rerun_last"], true); + assert!(retry_timeline["forked_from"].is_null()); let retry_workspace = workspace(&server, &retry); assert_eq!(read(&retry_workspace, "one.txt"), "one\n"); assert_eq!(read(&retry_workspace, "flaky.txt"), "flaky\n"); assert_eq!(read(&retry_workspace, "three.txt"), "three\n"); assert_eq!(commit_subjects(&retry_workspace), [ - format!("fabro({source}): start (success)"), - format!("fabro({source}): one (success)"), + format!("fabro({retry}): start (success)"), + format!("fabro({retry}): one (success)"), format!("fabro({retry}): flaky (success)"), format!("fabro({retry}): three (success)"), format!("fabro({retry}): exit (success)"), ]); let state = run_json(&server, &format!("runs/{retry}/state")).await; assert_eq!(state["retried_from"], source); - assert_eq!(state["spec"]["fork_source_ref"]["source_run_id"], source); + assert!(state["spec"]["fork_source_ref"].is_null()); // The source is untouched, and a second retry is refused for the // running or archived cases alone: it is terminal, so it may retry again. diff --git a/lib/apps/fabro-server/src/server/handler/lineage.rs b/lib/apps/fabro-server/src/server/handler/lineage.rs index 05fde2318..b2d43266f 100644 --- a/lib/apps/fabro-server/src/server/handler/lineage.rs +++ b/lib/apps/fabro-server/src/server/handler/lineage.rs @@ -9,9 +9,9 @@ //! (`fabro_petri::fork`), and queues it in resume mode, so its worker //! acquires a fresh workspace, restores the checkpoint's commit into it and //! continues from the position. A rewind is a fork of a terminal run that -//! archives the source and records `run.superseded` on it; a retry is a -//! fork of a terminal run at its last checkpoint, with the failed stage run -//! again when the run failed on one. +//! archives the source and records `run.superseded` on it. A retry is not a +//! fork: it is a new run from the source's saved spec, started from the +//! beginning, so it needs no checkpoint or workspace from the source. use std::sync::Arc; @@ -53,7 +53,6 @@ pub(super) fn routes() -> Router> { enum ForkKind { Fork, Rewind, - Retry, } /// A fork made and queued. @@ -86,14 +85,13 @@ async fn run_timeline( async fn fork_run( RequireRunManagementTarget(id, actor): RequireRunManagementTarget, State(state): State>, - headers: HeaderMap, body: Option>, ) -> Response { let target = match parse_fork_target(body.and_then(|Json(body)| body.target)) { Ok(target) => target, Err(err) => return err.into_response(), }; - match fork_at(state.as_ref(), id, actor, &headers, ForkKind::Fork, target).await { + match fork_at(state.as_ref(), id, actor, ForkKind::Fork, target).await { Ok(outcome) => ( StatusCode::OK, Json(api::ForkResponse { @@ -114,23 +112,13 @@ async fn fork_run( async fn rewind_run( RequireRunManagementTarget(id, actor): RequireRunManagementTarget, State(state): State>, - headers: HeaderMap, body: Option>, ) -> Response { let target = match parse_fork_target(body.and_then(|Json(body)| body.target)) { Ok(target) => target, Err(err) => return err.into_response(), }; - let outcome = match fork_at( - state.as_ref(), - id, - actor.clone(), - &headers, - ForkKind::Rewind, - target, - ) - .await - { + let outcome = match fork_at(state.as_ref(), id, actor.clone(), ForkKind::Rewind, target).await { Ok(outcome) => outcome, Err(err) => return err.into_response(), }; @@ -175,8 +163,29 @@ async fn retry_run( State(state): State>, headers: HeaderMap, ) -> Response { - match fork_at(state.as_ref(), id, actor, &headers, ForkKind::Retry, None).await { - Ok(outcome) => run_response(state.as_ref(), outcome.new_run_id, StatusCode::CREATED).await, + let result = async { + let source = run_records::require_projection(state.as_ref(), id).await?; + operations::ensure_retryable(&source, &id).map_err(workflow_operation_error)?; + let new_run_id = RunId::new(); + let storage = Storage::new(state.server_storage_dir()); + let run_dir = storage.run_scratch(&new_run_id).root().to_path_buf(); + Box::pin(operations::persist_retried_run( + state.store_ref().as_ref(), + &source, + new_run_id, + &run_dir, + run_provenance(&headers, &actor), + state.run_web_url(&new_run_id), + )) + .await + .map_err(workflow_operation_error)?; + let projection = run_records::require_projection(state.as_ref(), new_run_id).await?; + queue_run(state.as_ref(), new_run_id, &projection, false, actor).await?; + Ok::<_, ApiError>(new_run_id) + } + .await; + match result { + Ok(id) => run_response(state.as_ref(), id, StatusCode::CREATED).await, Err(err) => err.into_response(), } } @@ -252,7 +261,6 @@ async fn fork_at( state: &AppState, id: RunId, actor: Principal, - headers: &HeaderMap, kind: ForkKind, target: Option, ) -> Result { @@ -260,17 +268,13 @@ async fn fork_at( match kind { ForkKind::Fork => operations::ensure_forkable(&source, &id), ForkKind::Rewind => operations::ensure_rewindable(&source, &id), - ForkKind::Retry => operations::ensure_retryable(&source, &id), } .map_err(workflow_operation_error)?; let timeline = timeline(state, id).await?; - let entry = match kind { - ForkKind::Retry => timeline.latest(), - ForkKind::Fork | ForkKind::Rewind => timeline.resolve_or_latest(target.as_ref()), - } - .map_err(workflow_operation_error)?; + let entry = timeline + .resolve_or_latest(target.as_ref()) + .map_err(workflow_operation_error)?; let resolved = ResolvedForkTarget::of(entry).map_err(workflow_operation_error)?; - let rerun_last = kind == ForkKind::Retry && operations::reruns_last(source.status); // A position Petri would refuse is refused before the new run exists. let position = petri_fork::position(resolved.position.execution, resolved.position.firing); @@ -285,15 +289,14 @@ async fn fork_at( let storage = Storage::new(state.server_storage_dir()); let source_run_dir = storage.run_scratch(&id).root().to_path_buf(); let run_dir = storage.run_scratch(&new_run_id).root().to_path_buf(); - let provenance = (kind == ForkKind::Retry).then(|| run_provenance(headers, &actor)); operations::persist_forked_run(state.store_ref().as_ref(), &operations::ForkedRunInput { source: &source, new_run_id, run_dir: run_dir.clone(), checkpoint_sha: resolved.checkpoint_sha.clone(), - provenance, + provenance: None, web_url: state.run_web_url(&new_run_id), - retried_from: (kind == ForkKind::Retry).then_some(id), + retried_from: None, }) .await .map_err(workflow_operation_error)?; @@ -308,7 +311,7 @@ async fn fork_at( &state.stores.run_summaries, ))), position, - rerun_last, + rerun_last: false, settings: source.spec.settings.run.clone(), }) .await; @@ -337,7 +340,7 @@ async fn fork_at( source_run_id: id, new_run_id, target: resolved, - rerun_last, + rerun_last: false, }) } diff --git a/lib/components/fabro-workflow/src/operations/fork.rs b/lib/components/fabro-workflow/src/operations/fork.rs index 241892f8b..10a7c838e 100644 --- a/lib/components/fabro-workflow/src/operations/fork.rs +++ b/lib/components/fabro-workflow/src/operations/fork.rs @@ -117,13 +117,28 @@ pub fn forked_run_record(input: &ForkedRunInput<'_>) -> RunCreatedRecord { /// (`run.created` and the `submitted` transition), which wake its /// projector. pub async fn persist_forked_run(store: &Database, input: &ForkedRunInput<'_>) -> Result<(), Error> { - fs::create_dir_all(&input.run_dir).await.map_err(|err| { + persist_new_run( + store, + input.new_run_id, + &input.run_dir, + forked_run_record(input), + ) + .await +} + +pub(super) async fn persist_new_run( + store: &Database, + run_id: RunId, + run_dir: &std::path::Path, + created: RunCreatedRecord, +) -> Result<(), Error> { + fs::create_dir_all(run_dir).await.map_err(|err| { Error::Io(format!( "creating run directory {}: {err}", - input.run_dir.display() + run_dir.display() )) })?; - let created = PlatformRecord::RunCreated(forked_run_record(input)); + let created = PlatformRecord::RunCreated(created); let submitted = PlatformRecord::RunLifecycle( RunLifecycleRecord::new(RunLifecycleKind::Submitted).with_status(RunStatus::Submitted), ); @@ -131,11 +146,11 @@ pub async fn persist_forked_run(store: &Database, input: &ForkedRunInput<'_>) -> let platform_records = summaries.platform_records(); for record in [created, submitted] { platform_records - .append(&input.new_run_id, &record, None) + .append(&run_id, &record, None) .await .map_err(|err| Error::engine_with_source("run store operation failed", err))?; } - summaries.notify_platform_record(input.new_run_id); + summaries.notify_platform_record(run_id); Ok(()) } diff --git a/lib/components/fabro-workflow/src/operations/mod.rs b/lib/components/fabro-workflow/src/operations/mod.rs index 91aa55155..247cff461 100644 --- a/lib/components/fabro-workflow/src/operations/mod.rs +++ b/lib/components/fabro-workflow/src/operations/mod.rs @@ -14,7 +14,7 @@ pub use fork::{ ForkedRunInput, ResolvedForkTarget, ensure_forkable, ensure_terminal, forked_run_record, persist_forked_run, }; -pub use retry::{ensure_retryable, reruns_last}; +pub use retry::{ensure_retryable, persist_retried_run}; pub use rewind::{ensure_rewindable, superseded_record}; pub use timeline::{ ForkTarget, RunTimeline, StageLabel, StageLabels, TimelineEntry, TimelinePosition, diff --git a/lib/components/fabro-workflow/src/operations/retry.rs b/lib/components/fabro-workflow/src/operations/retry.rs index 0b493e591..e72d04b33 100644 --- a/lib/components/fabro-workflow/src/operations/retry.rs +++ b/lib/components/fabro-workflow/src/operations/retry.rs @@ -1,47 +1,38 @@ -//! Retrying a run: a fork from its last checkpoint. -//! -//! A retry forks a terminal run at its last checkpoint. When the run failed -//! on a stage (its last durable finish is the failed stage's), the position's -//! firing runs again, so the retry reruns the failed stage on the files of -//! the stage before it; a run that succeeded, was cancelled or died forks at -//! the last position as it stands, and continues from there. +//! Retrying starts a new execution from the original spec. It does not +//! require Git checkpoints or reconstruct the previous workspace. -use fabro_types::{FailureReason, RunId, RunProjection, RunStatus}; +use std::path::Path; -use super::fork::{ensure_forkable, ensure_terminal}; +use fabro_store::platform_records::RunCreatedRecord; +use fabro_store::{Database, RunProjection}; +use fabro_types::{RunId, RunProvenance}; + +use super::fork; use crate::error::Error; -/// A run can be retried when it is terminal and not archived. pub fn ensure_retryable(source: &RunProjection, run_id: &RunId) -> Result<(), Error> { - ensure_forkable(source, run_id)?; - ensure_terminal(source, run_id, "retry") + fork::ensure_forkable(source, run_id)?; + fork::ensure_terminal(source, run_id, "retry") } -/// Whether the retry reruns the last checkpointed stage: it does when the -/// run failed on its own terms, since that stage's finish is the failure. -#[must_use] -pub fn reruns_last(status: RunStatus) -> bool { - matches!( - status, - RunStatus::Failed { reason } if reason != FailureReason::Cancelled - ) -} - -#[cfg(test)] -mod tests { - use super::*; - - #[test] - fn a_failed_run_reruns_its_last_stage_and_the_others_continue() { - assert!(reruns_last(RunStatus::Failed { - reason: FailureReason::WorkflowError, - })); - assert!(!reruns_last(RunStatus::Failed { - reason: FailureReason::Cancelled, - })); - assert!(!reruns_last(RunStatus::Dead)); - assert!(!reruns_last(RunStatus::Succeeded { - reason: fabro_types::SuccessReason::Completed, - })); - } +pub async fn persist_retried_run( + store: &Database, + source: &RunProjection, + run_id: RunId, + run_dir: &Path, + provenance: RunProvenance, + web_url: Option, +) -> Result<(), Error> { + let mut spec = source.spec.clone(); + spec.run_id = run_id; + spec.fork_source_ref = None; + spec.provenance = provenance; + let created = RunCreatedRecord { + spec, + title: Some(source.title().into_owned()), + parent_id: source.parent_id, + retried_from: Some(source.spec.run_id), + web_url, + }; + fork::persist_new_run(store, run_id, run_dir, created).await } diff --git a/lib/packages/fabro-api-client/src/api/runs-api.ts b/lib/packages/fabro-api-client/src/api/runs-api.ts index cdd8f5c80..0f7e672c8 100644 --- a/lib/packages/fabro-api-client/src/api/runs-api.ts +++ b/lib/packages/fabro-api-client/src/api/runs-api.ts @@ -1164,7 +1164,7 @@ export const RunsApiAxiosParamCreator = function (configuration?: Configuration) }; }, /** - * Creates a new run from the terminal source run\'s last checkpoint and starts it. When the source failed on a stage, that stage runs again on the files of the stage before it; otherwise the new run continues from the last checkpoint as it stands. The new run records `retried_from` and `fork_source_ref`; the source run is left unchanged. Active and archived runs are not retryable. + * Creates a new run from the terminal source run\'s saved specification and starts the workflow from the beginning in a fresh workspace. No Git checkpoint is required. The new run records `retried_from` without a `fork_source_ref`; the source run is left unchanged. Active and archived runs are not retryable. * @summary Retry Run * @param {string} id Unique run identifier (ULID). * @param {*} [options] Override http request option. @@ -1925,7 +1925,7 @@ export const RunsApiFp = function(configuration?: Configuration) { return (axios, basePath) => createRequestFunction(localVarAxiosArgs, globalAxios, BASE_PATH, configuration)(axios, localVarOperationServerBasePath || basePath); }, /** - * Creates a new run from the terminal source run\'s last checkpoint and starts it. When the source failed on a stage, that stage runs again on the files of the stage before it; otherwise the new run continues from the last checkpoint as it stands. The new run records `retried_from` and `fork_source_ref`; the source run is left unchanged. Active and archived runs are not retryable. + * Creates a new run from the terminal source run\'s saved specification and starts the workflow from the beginning in a fresh workspace. No Git checkpoint is required. The new run records `retried_from` without a `fork_source_ref`; the source run is left unchanged. Active and archived runs are not retryable. * @summary Retry Run * @param {string} id Unique run identifier (ULID). * @param {*} [options] Override http request option. @@ -2331,7 +2331,7 @@ export const RunsApiFactory = function (configuration?: Configuration, basePath? return localVarFp.retrieveRunGraphSource(id, options).then((request) => request(axios, basePath)); }, /** - * Creates a new run from the terminal source run\'s last checkpoint and starts it. When the source failed on a stage, that stage runs again on the files of the stage before it; otherwise the new run continues from the last checkpoint as it stands. The new run records `retried_from` and `fork_source_ref`; the source run is left unchanged. Active and archived runs are not retryable. + * Creates a new run from the terminal source run\'s saved specification and starts the workflow from the beginning in a fresh workspace. No Git checkpoint is required. The new run records `retried_from` without a `fork_source_ref`; the source run is left unchanged. Active and archived runs are not retryable. * @summary Retry Run * @param {string} id Unique run identifier (ULID). * @param {*} [options] Override http request option. @@ -2730,7 +2730,7 @@ export class RunsApi extends BaseAPI { } /** - * Creates a new run from the terminal source run\'s last checkpoint and starts it. When the source failed on a stage, that stage runs again on the files of the stage before it; otherwise the new run continues from the last checkpoint as it stands. The new run records `retried_from` and `fork_source_ref`; the source run is left unchanged. Active and archived runs are not retryable. + * Creates a new run from the terminal source run\'s saved specification and starts the workflow from the beginning in a fresh workspace. No Git checkpoint is required. The new run records `retried_from` without a `fork_source_ref`; the source run is left unchanged. Active and archived runs are not retryable. * @summary Retry Run * @param {string} id Unique run identifier (ULID). * @param {*} [options] Override http request option.