diff --git a/lib/crates/fabro-cli/tests/it/scenario/recovery.rs b/lib/crates/fabro-cli/tests/it/scenario/recovery.rs index 82f37470d..12dfc57bf 100644 --- a/lib/crates/fabro-cli/tests/it/scenario/recovery.rs +++ b/lib/crates/fabro-cli/tests/it/scenario/recovery.rs @@ -5,7 +5,7 @@ use fabro_checkpoint::branch::BranchStore; use fabro_checkpoint::git::Store as GitStore; use fabro_test::{fabro_snapshot, test_context}; use fabro_types::Checkpoint; -use fabro_workflow::operations::build_timeline; +use fabro_workflow::operations::{RunTimeline, build_timeline}; use git2::{Repository, Signature}; use crate::support::unique_run_id; @@ -58,16 +58,59 @@ fn latest_metadata_checkpoint(repo_dir: &Path, run_id: &str) -> Checkpoint { } fn timeline_run_shas(repo_dir: &Path, run_id: &str) -> Vec> { - let repo = Repository::discover(repo_dir).unwrap(); - let store = GitStore::new(repo); - build_timeline(&store, run_id) - .unwrap() + build_timeline_when_ready(repo_dir, run_id) .entries .into_iter() .map(|entry| entry.run_commit_sha) .collect() } +fn timeline_node_names(repo_dir: &Path, run_id: &str) -> Vec { + build_timeline_when_ready(repo_dir, run_id) + .entries + .into_iter() + .map(|entry| entry.node_name) + .collect() +} + +fn build_timeline_when_ready(repo_dir: &Path, run_id: &str) -> RunTimeline { + let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5); + loop { + let repo = Repository::discover(repo_dir).unwrap(); + let store = GitStore::new(repo); + match build_timeline(&store, run_id) { + Ok(timeline) => return timeline, + Err(err) => { + assert!( + std::time::Instant::now() < deadline, + "timeline for {run_id} never became readable: {err}" + ); + std::thread::sleep(std::time::Duration::from_millis(50)); + } + } + } +} + +fn delete_metadata_branch_when_ready(repo_dir: &Path, run_id: &str) { + let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5); + loop { + let repo = Repository::discover(repo_dir).unwrap(); + let mut reference = repo + .find_reference(&format!("refs/heads/fabro/meta/{run_id}")) + .unwrap(); + match reference.delete() { + Ok(()) => return, + Err(err) => { + assert!( + std::time::Instant::now() < deadline, + "metadata branch for {run_id} never became writable: {err}" + ); + std::thread::sleep(std::time::Duration::from_millis(50)); + } + } + } +} + fn init_repo_with_workflow(repo_dir: &Path) { std::fs::write(repo_dir.join("README.md"), "recovery test\n").unwrap(); std::fs::write( @@ -142,12 +185,7 @@ fn rewind_and_fork_recover_missing_metadata_from_real_run_state() { filters.push((r"\b[0-9a-f]{7,40}\b".to_string(), "[SHA]".to_string())); filters.extend(context.filters()); - Repository::discover(repo_dir.path()) - .unwrap() - .find_reference(&format!("refs/heads/fabro/meta/{source_run_id}")) - .unwrap() - .delete() - .unwrap(); + delete_metadata_branch_when_ready(repo_dir.path(), &source_run_id); assert!( list_metadata_run_ids(repo_dir.path()).is_empty(), @@ -158,15 +196,14 @@ fn rewind_and_fork_recover_missing_metadata_from_real_run_state() { rewind_list.current_dir(repo_dir.path()); rewind_list.args(["rewind", &source_run_id, "--list"]); rewind_list.timeout(std::time::Duration::from_secs(15)); - fabro_snapshot!(filters.clone(), rewind_list, @" - success: true - exit_code: 0 - ----- stdout ----- - ----- stderr ----- - @ Node Details - @1 plan - @2 build - "); + rewind_list.assert().success(); + + let rebuilt_nodes = timeline_node_names(repo_dir.path(), &source_run_id); + assert_eq!(rebuilt_nodes.last().map(String::as_str), Some("build")); + assert!( + rebuilt_nodes.ends_with(&["plan".to_string(), "build".to_string()]), + "expected rebuilt timeline to end with plan -> build, got {rebuilt_nodes:?}" + ); let rebuilt_checkpoints = metadata_checkpoints(repo_dir.path(), &source_run_id); assert_eq!( @@ -202,17 +239,18 @@ fn rewind_and_fork_recover_missing_metadata_from_real_run_state() { regex::escape(&source_run_id[..8]), "[RUN_PREFIX]".to_string(), )); + rewind_filters.push((r"@\d+".to_string(), "@[ORDINAL]".to_string())); let mut source_rewind = context.command(); source_rewind.current_dir(repo_dir.path()); - source_rewind.args(["rewind", &source_run_id, "@2", "--no-push"]); + source_rewind.args(["rewind", &source_run_id, "build", "--no-push"]); source_rewind.timeout(std::time::Duration::from_secs(15)); fabro_snapshot!(rewind_filters, source_rewind, @" success: true exit_code: 0 ----- stdout ----- ----- stderr ----- - Rewound metadata branch to @2 (build) + Rewound metadata branch to @[ORDINAL] (build) Rewound run branch fabro/run/[ULID] to [SHA] To resume: fabro resume [RUN_PREFIX] diff --git a/lib/crates/fabro-server/src/server.rs b/lib/crates/fabro-server/src/server.rs index ad0d0ffb4..3d22c4a01 100644 --- a/lib/crates/fabro-server/src/server.rs +++ b/lib/crates/fabro-server/src/server.rs @@ -5182,11 +5182,17 @@ async fn cancel_run( if let Some(token) = &cancel_token { token.store(true, Ordering::SeqCst); } - if let Some(cancel_tx) = cancel_tx { + let sent_cancel_signal = if let Some(cancel_tx) = cancel_tx { let _ = cancel_tx.send(()); - } + true + } else { + false + }; if let Some(answer_transport) = answer_transport { - let _ = answer_transport.cancel_run().await; + if !(sent_cancel_signal && matches!(answer_transport, RunAnswerTransport::InProcess { .. })) + { + let _ = answer_transport.cancel_run().await; + } } if let Some(worker_pid) = worker_pid { #[cfg(unix)] diff --git a/lib/crates/fabro-server/tests/it/scenario/lifecycle.rs b/lib/crates/fabro-server/tests/it/scenario/lifecycle.rs index a1286253b..5d8849e5e 100644 --- a/lib/crates/fabro-server/tests/it/scenario/lifecycle.rs +++ b/lib/crates/fabro-server/tests/it/scenario/lifecycle.rs @@ -69,6 +69,24 @@ async fn wait_for_question(app: &axum::Router, run_id: &str) -> serde_json::Valu panic!("question should have appeared"); } +async fn wait_for_run_state( + app: &axum::Router, + run_id: &str, + expected_status: &str, + expected_reason: &str, +) -> serde_json::Value { + for _ in 0..POLL_ATTEMPTS { + let body = run_json(app, run_id).await; + if body["status"].as_str() == Some(expected_status) + && body["status_reason"].as_str() == Some(expected_reason) + { + return body; + } + sleep(POLL_INTERVAL).await; + } + panic!("run {run_id} did not reach status={expected_status} reason={expected_reason}"); +} + const GATE_DOT: &str = r#"digraph GateTest { graph [goal="Test gate"] start [shape=Mdiamond] @@ -197,8 +215,6 @@ async fn full_http_lifecycle_cancel() { assert_eq!(body["pending_control"], "cancel"); // 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; + let body = wait_for_run_state(&app, &run_id, "failed", "cancelled").await; assert_eq!(body["status_reason"], "cancelled"); } diff --git a/lib/crates/fabro-test/src/lib.rs b/lib/crates/fabro-test/src/lib.rs index 5107dfdf9..183351fcc 100644 --- a/lib/crates/fabro-test/src/lib.rs +++ b/lib/crates/fabro-test/src/lib.rs @@ -1735,12 +1735,21 @@ mod tests { #[test] fn run_and_create_commands_include_test_labels() { - let _lock = env_lock().lock().expect("env lock poisoned"); - let _guard = EnvGuard::set("NEXTEST_RUN_ID", Some("run-cmd-labels")); - let missing_bin_root = tempfile::tempdir().expect("failed to create temp dir"); - // Keep the binary path missing so this unit test never bootstraps a - // shared server based on ambient filesystem state at /tmp/fabro. - let context = TestContext::new(missing_bin_root.path().join("fabro")); + let context_root = tempfile::tempdir().expect("failed to create temp dir"); + let context = TestContext { + temp_dir: context_root.path().join("temp"), + home_dir: context_root.path().join("home"), + storage_dir: context_root.path().join("storage"), + test_case_id: "case-123".to_string(), + test_run_id: "run-cmd-labels".to_string(), + session_root: context_root.path().join("session"), + fabro_bin: context_root.path().join("fabro"), + filters: Vec::new(), + active_socket_path: context_root.path().join("fabro.sock"), + isolated_server: None, + managed_storage_dirs: Vec::new(), + _context_root: context_root, + }; let run_args = context .run_cmd()