mirror of
https://github.com/fabro-sh/fabro.git
synced 2026-09-11 22:53:00 +00:00
refactor(archive): consolidate helpers after review
Three cleanups from `/simplify` review: - Promote `archived_rejection_message` and `ensure_not_archived` to `pub` via operations/mod.rs and reuse them from `resume`, the CLI rewind caller, and the server's `reject_if_archived` guard so the canonical error string lives in exactly one place. - Tighten `RewindInput.current_status` from `Option<RunStatus>` to `RunStatus`. The runtime check for None was enforcing a compile-time invariant. CLI callers already load the projection and now surface a clean error up-front if it's missing. Drop the None-branch test that existed only to cover the removed runtime check. - Collapse `archive_run` / `unarchive_run` HTTP handlers into a shared `run_archive_action` body with an `ArchiveAction` enum, mirroring the CLI pattern. Removes ~20 lines of copy-paste and unifies error-mapping. Also drop narrative comments that referenced plan unit numbers in the scenario tests, and clean up the convoluted `ps_runs` helper pattern that built an empty-slot arg vec before filling it in. No behavior change. Full workspace: 4185 tests pass, clippy clean.
This commit is contained in:
Bryan Helmkamp
parent
828a09636e
commit
48eb5efed6
8 changed files with 59 additions and 109 deletions
|
|
@ -44,7 +44,11 @@ pub(crate) async fn run(
|
|||
let run = lookup.resolve(&args.run_id)?;
|
||||
let run_id = run.run_id();
|
||||
let state = lookup.client().get_run_state(&run_id).await?;
|
||||
let current_status = state.status.as_ref().map(|record| record.status);
|
||||
let current_status = state
|
||||
.status
|
||||
.as_ref()
|
||||
.map(|record| record.status)
|
||||
.context("run has no recorded status — cannot rewind")?;
|
||||
let record = state.run.context("Failed to load run record from store")?;
|
||||
ensure_matching_repo_origin(record.repo_origin_url.as_deref(), "rewind")?;
|
||||
let store = Store::new(repo);
|
||||
|
|
|
|||
|
|
@ -1,21 +1,15 @@
|
|||
//! End-to-end CLI lifecycle for the archived run status: run → archive →
|
||||
//! hide/show → unarchive → still-deletable.
|
||||
|
||||
use fabro_test::test_context;
|
||||
use serde_json::Value;
|
||||
|
||||
use crate::cmd::support::setup_completed_fast_dry_run;
|
||||
|
||||
fn ps_runs(context: &fabro_test::TestContext, include_archived: bool) -> Vec<Value> {
|
||||
let mut cmd = context.ps();
|
||||
let mut args = vec!["--json", "--label", ""];
|
||||
// `--label` needs the actual label value; we replace the empty slot below.
|
||||
let label = context.test_case_label();
|
||||
args[2] = &label;
|
||||
let mut cmd = context.ps();
|
||||
if include_archived {
|
||||
cmd.args(["-a"]);
|
||||
cmd.arg("-a");
|
||||
}
|
||||
cmd.args(args);
|
||||
cmd.args(["--json", "--label", &label]);
|
||||
let output = cmd.output().expect("ps should execute");
|
||||
assert!(
|
||||
output.status.success(),
|
||||
|
|
@ -31,13 +25,11 @@ fn archive_lifecycle_end_to_end() {
|
|||
let context = test_context!();
|
||||
let run = setup_completed_fast_dry_run(&context);
|
||||
|
||||
// 1. Baseline: `ps -a` sees the succeeded run.
|
||||
let visible = ps_runs(&context, true);
|
||||
assert_eq!(visible.len(), 1);
|
||||
assert_eq!(visible[0]["run_id"], run.run_id);
|
||||
assert_eq!(visible[0]["status"], "succeeded");
|
||||
|
||||
// 2. Archive the run.
|
||||
let archive = context
|
||||
.command()
|
||||
.args(["archive", &run.run_id])
|
||||
|
|
@ -49,20 +41,17 @@ fn archive_lifecycle_end_to_end() {
|
|||
String::from_utf8_lossy(&archive.stderr)
|
||||
);
|
||||
|
||||
// 3. Default `ps` hides archived runs.
|
||||
let default_visible = ps_runs(&context, false);
|
||||
assert!(
|
||||
default_visible.is_empty(),
|
||||
"default ps should hide archived, got {default_visible:?}"
|
||||
);
|
||||
|
||||
// 4. `ps -a` surfaces the run with status `archived`.
|
||||
let with_archived = ps_runs(&context, true);
|
||||
assert_eq!(with_archived.len(), 1);
|
||||
assert_eq!(with_archived[0]["run_id"], run.run_id);
|
||||
assert_eq!(with_archived[0]["status"], "archived");
|
||||
|
||||
// 5. Unarchive restores the prior terminal status.
|
||||
let unarchive = context
|
||||
.command()
|
||||
.args(["unarchive", &run.run_id])
|
||||
|
|
@ -76,7 +65,8 @@ fn archive_lifecycle_end_to_end() {
|
|||
let restored = ps_runs(&context, true);
|
||||
assert_eq!(restored[0]["status"], "succeeded");
|
||||
|
||||
// 6. Archived runs remain delete-able (plan Scope Boundaries).
|
||||
// `rm` must remain available on archived runs — archive and delete are
|
||||
// orthogonal per the plan's Scope Boundaries.
|
||||
context
|
||||
.command()
|
||||
.args(["archive", &run.run_id])
|
||||
|
|
|
|||
|
|
@ -5996,19 +5996,13 @@ 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;
|
||||
if status == WorkflowRunStatus::Archived {
|
||||
Some(
|
||||
ApiError::new(
|
||||
StatusCode::CONFLICT,
|
||||
format!(
|
||||
"run {run_id} is archived; run `fabro unarchive {run_id}` to restore it and try again"
|
||||
),
|
||||
)
|
||||
.into_response(),
|
||||
(status == WorkflowRunStatus::Archived).then(|| {
|
||||
ApiError::new(
|
||||
StatusCode::CONFLICT,
|
||||
operations::archived_rejection_message(run_id),
|
||||
)
|
||||
} else {
|
||||
None
|
||||
}
|
||||
.into_response()
|
||||
})
|
||||
}
|
||||
|
||||
fn schedule_worker_kill(state: Arc<AppState>, run_id: RunId, worker_pid: u32) {
|
||||
|
|
@ -6381,35 +6375,44 @@ async fn archive_run(
|
|||
State(state): State<Arc<AppState>>,
|
||||
Path(id): Path<String>,
|
||||
) -> Response {
|
||||
let id = match parse_run_id_path(&id) {
|
||||
Ok(id) => id,
|
||||
Err(response) => return response,
|
||||
};
|
||||
let actor = actor_from_subject(&subject);
|
||||
match operations::archive(&state.store, &id, actor).await {
|
||||
Ok(_) => archive_status_response(state.as_ref(), id).await,
|
||||
Err(WorkflowError::Precondition(message)) => {
|
||||
ApiError::new(StatusCode::CONFLICT, message).into_response()
|
||||
}
|
||||
Err(WorkflowError::RunNotFound(_)) => ApiError::not_found("Run not found.").into_response(),
|
||||
Err(err) => {
|
||||
ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string()).into_response()
|
||||
}
|
||||
}
|
||||
run_archive_action(state, subject, id, ArchiveAction::Archive).await
|
||||
}
|
||||
|
||||
async fn unarchive_run(
|
||||
subject: AuthenticatedSubject,
|
||||
State(state): State<Arc<AppState>>,
|
||||
Path(id): Path<String>,
|
||||
) -> Response {
|
||||
run_archive_action(state, subject, id, ArchiveAction::Unarchive).await
|
||||
}
|
||||
|
||||
#[derive(Clone, Copy)]
|
||||
enum ArchiveAction {
|
||||
Archive,
|
||||
Unarchive,
|
||||
}
|
||||
|
||||
async fn run_archive_action(
|
||||
state: Arc<AppState>,
|
||||
subject: AuthenticatedSubject,
|
||||
id: String,
|
||||
action: ArchiveAction,
|
||||
) -> Response {
|
||||
let id = match parse_run_id_path(&id) {
|
||||
Ok(id) => id,
|
||||
Err(response) => return response,
|
||||
};
|
||||
let actor = actor_from_subject(&subject);
|
||||
match operations::unarchive(&state.store, &id, actor).await {
|
||||
Ok(_) => archive_status_response(state.as_ref(), id).await,
|
||||
let result = match action {
|
||||
ArchiveAction::Archive => operations::archive(&state.store, &id, actor)
|
||||
.await
|
||||
.map(|_| ()),
|
||||
ArchiveAction::Unarchive => operations::unarchive(&state.store, &id, actor)
|
||||
.await
|
||||
.map(|_| ()),
|
||||
};
|
||||
match result {
|
||||
Ok(()) => archive_status_response(state.as_ref(), id).await,
|
||||
Err(WorkflowError::Precondition(message)) => {
|
||||
ApiError::new(StatusCode::CONFLICT, message).into_response()
|
||||
}
|
||||
|
|
|
|||
|
|
@ -1,7 +1,3 @@
|
|||
//! End-to-end HTTP coverage for the archived run status: the R14 mutation
|
||||
//! rejection contract, archive/unarchive status codes, and the
|
||||
//! `include_archived` listing filter.
|
||||
|
||||
use axum::body::Body;
|
||||
use axum::http::{Request, StatusCode};
|
||||
use tower::ServiceExt;
|
||||
|
|
@ -23,7 +19,6 @@ async fn archived_runs_reject_mutations_with_actionable_body() {
|
|||
let status = wait_for_run_status(&app, &run_id, &["succeeded", "failed"]).await;
|
||||
assert_eq!(status, "succeeded");
|
||||
|
||||
// Archive the run.
|
||||
let req = Request::builder()
|
||||
.method("POST")
|
||||
.uri(api(&format!("/runs/{run_id}/archive")))
|
||||
|
|
@ -34,8 +29,6 @@ async fn archived_runs_reject_mutations_with_actionable_body() {
|
|||
let body = body_json(response.into_body()).await;
|
||||
assert_eq!(body["status"], "archived");
|
||||
|
||||
// Every mutation endpoint that guards against archived runs returns 409 with
|
||||
// an actionable "unarchive first" body.
|
||||
for path in &["/cancel", "/pause", "/unpause", "/start"] {
|
||||
let req = Request::builder()
|
||||
.method("POST")
|
||||
|
|
@ -56,7 +49,6 @@ async fn archived_runs_reject_mutations_with_actionable_body() {
|
|||
);
|
||||
}
|
||||
|
||||
// Client-supplied lifecycle events on the archived run are rejected.
|
||||
let req = Request::builder()
|
||||
.method("POST")
|
||||
.uri(api(&format!("/runs/{run_id}/events")))
|
||||
|
|
@ -79,10 +71,9 @@ async fn archived_runs_reject_mutations_with_actionable_body() {
|
|||
"expected 409 for POST /runs/{{id}}/events on archived run"
|
||||
);
|
||||
|
||||
// Additional write surfaces that the Unit 4 audit guards but the scenario
|
||||
// loop doesn't cover. The `reject_if_archived` check runs before each
|
||||
// endpoint's state-specific lookups, so synthetic stage/question/filename
|
||||
// values are fine — the archive guard fires first.
|
||||
// The archive guard runs before each endpoint's state-specific lookups, so
|
||||
// synthetic stage/question/filename values are enough to drive these
|
||||
// write surfaces into the guard.
|
||||
for (method, path, body, content_type) in [
|
||||
(
|
||||
"POST",
|
||||
|
|
|
|||
|
|
@ -15,14 +15,14 @@ fn map_open_run_error(run_id: &RunId, err: StoreError) -> Error {
|
|||
/// by the operations layer, the CLI rewind precheck, and the server HTTP
|
||||
/// guards so the user sees the same actionable guidance everywhere.
|
||||
#[must_use]
|
||||
pub(crate) fn archived_rejection_message(run_id: &RunId) -> String {
|
||||
pub fn archived_rejection_message(run_id: &RunId) -> String {
|
||||
format!("run {run_id} is archived; run `fabro unarchive {run_id}` to restore it and try again")
|
||||
}
|
||||
|
||||
/// Returns `Err(Error::Precondition)` when the given status represents an
|
||||
/// archived run. Use this at any mutation entry point that would otherwise
|
||||
/// transition or emit events against the run (rewind, resume, etc.).
|
||||
pub(crate) fn ensure_not_archived(status: Option<RunStatus>, run_id: &RunId) -> Result<(), Error> {
|
||||
pub fn ensure_not_archived(status: Option<RunStatus>, run_id: &RunId) -> Result<(), Error> {
|
||||
if status == Some(RunStatus::Archived) {
|
||||
Err(Error::Precondition(archived_rejection_message(run_id)))
|
||||
} else {
|
||||
|
|
|
|||
|
|
@ -10,7 +10,10 @@ mod start;
|
|||
mod test_support;
|
||||
mod validate;
|
||||
|
||||
pub use archive::{ArchiveOutcome, UnarchiveOutcome, archive, unarchive};
|
||||
pub use archive::{
|
||||
ArchiveOutcome, UnarchiveOutcome, archive, archived_rejection_message, ensure_not_archived,
|
||||
unarchive,
|
||||
};
|
||||
pub use create::{CreateRunInput, CreatedRun, create, make_run_dir};
|
||||
pub use fork::{ForkRunInput, fork};
|
||||
pub use rebuild_meta::{
|
||||
|
|
|
|||
|
|
@ -15,12 +15,7 @@ pub async fn resume(run_dir: &Path, services: StartServices) -> Result<Started,
|
|||
.map_err(|err| Error::engine(err.to_string()))?;
|
||||
|
||||
if let Some(record) = state.status {
|
||||
if record.status == RunStatus::Archived {
|
||||
return Err(Error::Precondition(format!(
|
||||
"run {run_id} is archived; run `fabro unarchive {run_id}` to restore it and try again",
|
||||
run_id = services.run_id,
|
||||
)));
|
||||
}
|
||||
super::archive::ensure_not_archived(Some(record.status), &services.run_id)?;
|
||||
if record.status == RunStatus::Succeeded {
|
||||
return Err(Error::Precondition(
|
||||
"run already finished successfully — nothing to resume".to_string(),
|
||||
|
|
|
|||
|
|
@ -118,10 +118,10 @@ pub struct RewindInput {
|
|||
pub run_id: RunId,
|
||||
pub target: RewindTarget,
|
||||
pub push: bool,
|
||||
/// Current durable run status. Callers must thread this through so rewind
|
||||
/// can enforce the archived-run precondition itself instead of relying on
|
||||
/// an upstream check.
|
||||
pub current_status: Option<RunStatus>,
|
||||
/// Current durable run status. Callers must load this from the projection
|
||||
/// store before calling rewind so the archived-run precondition can be
|
||||
/// enforced here rather than by an upstream check that can drift.
|
||||
pub current_status: RunStatus,
|
||||
}
|
||||
|
||||
pub fn build_timeline(store: &Store, run_id: &str) -> Result<RunTimeline> {
|
||||
|
|
@ -254,13 +254,7 @@ fn detect_parallel_interior(graph: &Graph) -> HashMap<String, String> {
|
|||
}
|
||||
|
||||
pub fn rewind(store: &Store, input: &RewindInput) -> Result<()> {
|
||||
let current_status = input.current_status.ok_or_else(|| {
|
||||
anyhow::anyhow!(
|
||||
"run {} current status is required before rewind",
|
||||
input.run_id
|
||||
)
|
||||
})?;
|
||||
ensure_not_archived(Some(current_status), &input.run_id)
|
||||
ensure_not_archived(Some(input.current_status), &input.run_id)
|
||||
.map_err(|err| anyhow::anyhow!("{err}"))?;
|
||||
let timeline = build_timeline(store, &input.run_id.to_string())?;
|
||||
let entry = timeline.resolve(&input.target)?;
|
||||
|
|
@ -525,7 +519,7 @@ mod tests {
|
|||
run_id: fixtures::RUN_1,
|
||||
target: RewindTarget::Ordinal(1),
|
||||
push: false,
|
||||
current_status: Some(RunStatus::Succeeded),
|
||||
current_status: RunStatus::Succeeded,
|
||||
})
|
||||
.unwrap();
|
||||
|
||||
|
|
@ -533,36 +527,6 @@ mod tests {
|
|||
assert_eq!(resolved, oid1);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn rewind_requires_current_status_to_enforce_archived_guard() {
|
||||
let (_dir, store) = temp_repo();
|
||||
let sig = test_sig();
|
||||
let branch = MetadataStore::branch_name(&fixtures::RUN_1.to_string());
|
||||
let bs = BranchStore::new(&store, &branch, &sig);
|
||||
bs.ensure_branch().unwrap();
|
||||
|
||||
bs.write_entry("run.json", b"{}", "init run").unwrap();
|
||||
bs.write_entry(
|
||||
"checkpoint.json",
|
||||
&make_checkpoint_json("start", 1, None),
|
||||
"checkpoint",
|
||||
)
|
||||
.unwrap();
|
||||
|
||||
let err = rewind(&store, &RewindInput {
|
||||
run_id: fixtures::RUN_1,
|
||||
target: RewindTarget::Ordinal(1),
|
||||
push: false,
|
||||
current_status: None,
|
||||
})
|
||||
.unwrap_err();
|
||||
|
||||
assert!(
|
||||
err.to_string().contains("current status"),
|
||||
"expected missing-status error, got: {err}"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn rewind_rejects_archived_runs() {
|
||||
let (_dir, store) = temp_repo();
|
||||
|
|
@ -571,7 +535,7 @@ mod tests {
|
|||
run_id: fixtures::RUN_1,
|
||||
target: RewindTarget::Ordinal(1),
|
||||
push: false,
|
||||
current_status: Some(RunStatus::Archived),
|
||||
current_status: RunStatus::Archived,
|
||||
})
|
||||
.unwrap_err();
|
||||
let message = err.to_string();
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue