From f5e4b4bb4e5eb4e94eb99a59f35c9085ba087854 Mon Sep 17 00:00:00 2001 From: Bryan Helmkamp Date: Sun, 29 Mar 2026 09:35:13 -0400 Subject: [PATCH] Make metadata rebuild recovery robust --- lib/crates/fabro-cli/tests/it/cli.rs | 316 +++++++++++++++++- .../fabro-workflows/src/operations/fork.rs | 31 +- .../src/operations/rebuild_meta.rs | 301 ++++++++++++++--- .../fabro-workflows/src/operations/rewind.rs | 36 +- 4 files changed, 608 insertions(+), 76 deletions(-) diff --git a/lib/crates/fabro-cli/tests/it/cli.rs b/lib/crates/fabro-cli/tests/it/cli.rs index db578952e..5e5e4eff3 100644 --- a/lib/crates/fabro-cli/tests/it/cli.rs +++ b/lib/crates/fabro-cli/tests/it/cli.rs @@ -1,8 +1,20 @@ +use std::collections::{BTreeSet, HashMap}; +use std::path::Path; +use std::sync::Arc; +use std::time::Duration; + use assert_cmd::Command; +use chrono::TimeZone; use fabro_config::FabroSettings; use fabro_config::mcp::McpTransport; -use fabro_store::RuntimeState; +use fabro_git_storage::branchstore::BranchStore; +use fabro_git_storage::gitobj::Store as GitStore; +use fabro_store::{NodeVisitRef, RuntimeState, SlateStore, Store as _}; +use fabro_types::{Checkpoint, Graph, RunRecord, StartRecord}; +use git2::{Repository, Signature}; +use object_store::local::LocalFileSystem; use predicates::prelude::*; +use tokio::runtime::Runtime; #[allow(deprecated)] fn arc() -> Command { @@ -205,6 +217,193 @@ commands = ["workflow-setup"] (home, project, storage_dir) } +fn init_cli_home(storage_dir: &Path) -> tempfile::TempDir { + let home = tempfile::tempdir().unwrap(); + let home_fabro = home.path().join(".fabro"); + std::fs::create_dir_all(&home_fabro).unwrap(); + let storage_dir = serde_json::to_string(&storage_dir.to_string_lossy().into_owned()).unwrap(); + std::fs::write( + home_fabro.join("cli.toml"), + format!("storage_dir = {storage_dir}\n"), + ) + .unwrap(); + home +} + +fn list_metadata_run_ids(repo_dir: &Path) -> BTreeSet { + let repo = Repository::discover(repo_dir).unwrap(); + repo.references() + .unwrap() + .flatten() + .filter_map(|reference| reference.name().map(ToOwned::to_owned)) + .filter_map(|name| { + name.strip_prefix("refs/heads/fabro/meta/") + .map(ToOwned::to_owned) + }) + .collect() +} + +fn seed_run_branch(repo_dir: &Path, run_id: &str, nodes: &[&str]) -> Vec { + let repo = Repository::discover(repo_dir).unwrap(); + let store = GitStore::new(repo); + let sig = Signature::now("Fabro", "noreply@fabro.sh").unwrap(); + let run_branch = format!("fabro/run/{run_id}"); + let empty_tree = store.write_empty_tree().unwrap(); + let mut shas = Vec::new(); + let mut parent = None; + + for node in nodes { + let parents = parent.into_iter().collect::>(); + let oid = store + .write_commit( + empty_tree, + &parents, + &format!("fabro({run_id}): {node} (completed)"), + &sig, + ) + .unwrap(); + store.update_ref(&run_branch, oid).unwrap(); + shas.push(oid.to_string()); + parent = Some(oid); + } + + shas +} + +fn checkpoint_record( + current_node: &str, + completed_nodes: &[&str], + node_visits: &[(&str, usize)], + git_commit_sha: Option<&str>, +) -> Checkpoint { + Checkpoint { + timestamp: chrono::Utc + .with_ymd_and_hms(2026, 1, 1, 0, 0, 0) + .single() + .unwrap(), + current_node: current_node.to_string(), + completed_nodes: completed_nodes + .iter() + .map(|node| (*node).to_string()) + .collect(), + node_retries: HashMap::new(), + context_values: HashMap::new(), + node_outcomes: HashMap::new(), + next_node_id: None, + git_commit_sha: git_commit_sha.map(ToOwned::to_owned), + loop_failure_signatures: HashMap::new(), + restart_failure_signatures: HashMap::new(), + node_visits: node_visits + .iter() + .map(|(node, visit)| ((*node).to_string(), *visit)) + .collect(), + } +} + +async fn seed_durable_run(storage_dir: &Path, repo_dir: &Path, run_id: &str) { + let store_path = storage_dir.join("store"); + std::fs::create_dir_all(&store_path).unwrap(); + let object_store = Arc::new(LocalFileSystem::new_with_prefix(&store_path).unwrap()); + let store = SlateStore::new(object_store, "", Duration::from_millis(5)); + let created_at = chrono::Utc + .with_ymd_and_hms(2026, 1, 1, 0, 0, 0) + .single() + .unwrap(); + let run_store = store.create_run(run_id, created_at, None).await.unwrap(); + + let run_record = RunRecord { + run_id: run_id.to_string(), + created_at, + settings: FabroSettings::default(), + graph: Graph::default(), + workflow_slug: None, + working_directory: repo_dir.to_path_buf(), + host_repo_path: Some(repo_dir.to_string_lossy().into_owned()), + base_branch: None, + labels: HashMap::new(), + }; + run_store.put_run(&run_record).await.unwrap(); + run_store + .put_start(&StartRecord { + run_id: run_id.to_string(), + start_time: created_at, + run_branch: Some(format!("fabro/run/{run_id}")), + base_sha: None, + }) + .await + .unwrap(); + + let start = NodeVisitRef { + node_id: "start", + visit: 1, + }; + run_store + .put_node_prompt(&start, "start prompt") + .await + .unwrap(); + let build = NodeVisitRef { + node_id: "build", + visit: 1, + }; + run_store + .put_node_prompt(&build, "build prompt") + .await + .unwrap(); + + run_store + .append_checkpoint(&checkpoint_record( + "start", + &["start"], + &[("start", 1)], + None, + )) + .await + .unwrap(); + run_store + .append_checkpoint(&checkpoint_record( + "build", + &["start", "build"], + &[("start", 1), ("build", 1)], + None, + )) + .await + .unwrap(); +} + +fn metadata_checkpoints(repo_dir: &Path, run_id: &str) -> Vec { + let repo = Repository::discover(repo_dir).unwrap(); + let store = GitStore::new(repo); + let sig = Signature::now("Fabro", "noreply@fabro.sh").unwrap(); + let branch = format!("fabro/meta/{run_id}"); + let bs = BranchStore::new(&store, &branch, &sig); + + bs.log(100) + .unwrap() + .iter() + .rev() + .filter(|commit| commit.message.starts_with("checkpoint")) + .map(|commit| { + serde_json::from_slice::( + &store + .read_blob_at(commit.oid, "checkpoint.json") + .unwrap() + .unwrap(), + ) + .unwrap() + }) + .collect() +} + +fn latest_metadata_checkpoint(repo_dir: &Path, run_id: &str) -> Checkpoint { + let repo = Repository::discover(repo_dir).unwrap(); + let store = GitStore::new(repo); + let tip = store + .resolve_ref(&format!("fabro/meta/{run_id}")) + .unwrap() + .unwrap(); + serde_json::from_slice(&store.read_blob_at(tip, "checkpoint.json").unwrap().unwrap()).unwrap() +} + // == LLM: prompt ============================================================== #[test] @@ -762,6 +961,121 @@ fn run_help_no_longer_shows_resume_or_run_branch() { .stdout(predicate::str::contains("--run-branch").not()); } +#[test] +fn rewind_and_fork_recover_missing_metadata_from_store() { + let storage_root = tempfile::tempdir().unwrap(); + let storage_dir = storage_root.path().join("fabro-data"); + let configured_home = init_cli_home(&storage_dir); + let repo_dir = tempfile::tempdir().unwrap(); + Repository::init(repo_dir.path()).unwrap(); + + let source_run_id = "run-recovery-source"; + let expected_shas = seed_run_branch(repo_dir.path(), source_run_id, &["start", "build"]); + Runtime::new().unwrap().block_on(seed_durable_run( + &storage_dir, + repo_dir.path(), + source_run_id, + )); + + assert!( + list_metadata_run_ids(repo_dir.path()).is_empty(), + "metadata branch should start missing" + ); + + let rewind_list = arc() + .env("HOME", configured_home.path()) + .env("NO_COLOR", "1") + .current_dir(repo_dir.path()) + .args(["rewind", source_run_id, "--list"]) + .timeout(Duration::from_secs(15)) + .assert() + .success() + .get_output() + .stderr + .clone(); + let rewind_list = String::from_utf8(rewind_list).unwrap(); + assert!( + rewind_list.contains("@1"), + "expected first checkpoint: {rewind_list}" + ); + assert!( + rewind_list.contains("@2"), + "expected second checkpoint: {rewind_list}" + ); + assert!( + !rewind_list.contains("no run commit"), + "rebuilt timeline should persist backfilled SHAs: {rewind_list}" + ); + + let rebuilt_checkpoints = metadata_checkpoints(repo_dir.path(), source_run_id); + assert_eq!(rebuilt_checkpoints.len(), 2); + assert_eq!( + rebuilt_checkpoints[0].git_commit_sha.as_deref(), + Some(expected_shas[0].as_str()) + ); + assert_eq!( + rebuilt_checkpoints[1].git_commit_sha.as_deref(), + Some(expected_shas[1].as_str()) + ); + + let before_child = list_metadata_run_ids(repo_dir.path()); + arc() + .env("HOME", configured_home.path()) + .env("NO_COLOR", "1") + .current_dir(repo_dir.path()) + .args(["fork", source_run_id, "--no-push"]) + .timeout(Duration::from_secs(15)) + .assert() + .success(); + let after_child = list_metadata_run_ids(repo_dir.path()); + let child_run_ids: Vec<_> = after_child.difference(&before_child).cloned().collect(); + assert_eq!(child_run_ids.len(), 1, "expected one child run"); + let child_run_id = &child_run_ids[0]; + + let child_checkpoint = latest_metadata_checkpoint(repo_dir.path(), child_run_id); + assert_eq!( + child_checkpoint.git_commit_sha.as_deref(), + Some(expected_shas[1].as_str()) + ); + + let child_rewind = arc() + .env("HOME", configured_home.path()) + .env("NO_COLOR", "1") + .current_dir(repo_dir.path()) + .args(["rewind", child_run_id, "@1", "--no-push"]) + .timeout(Duration::from_secs(15)) + .assert() + .success() + .get_output() + .stderr + .clone(); + let child_rewind = String::from_utf8(child_rewind).unwrap(); + assert!( + child_rewind.contains("Rewound run branch"), + "expected child rewind to move the run branch: {child_rewind}" + ); + assert!( + !child_rewind.contains("has no git_commit_sha"), + "child rewind should not lose git_commit_sha: {child_rewind}" + ); + + let before_grandchild = after_child; + arc() + .env("HOME", configured_home.path()) + .env("NO_COLOR", "1") + .current_dir(repo_dir.path()) + .args(["fork", child_run_id, "--no-push"]) + .timeout(Duration::from_secs(15)) + .assert() + .success(); + let after_grandchild = list_metadata_run_ids(repo_dir.path()); + let grandchild_run_ids: Vec<_> = after_grandchild + .difference(&before_grandchild) + .cloned() + .collect(); + assert_eq!(grandchild_run_ids.len(), 1, "expected one grandchild run"); +} + // == Bug regression: create/start/attach lifecycle ============================ /// Helper: create a minimal run directory that `resolve_run` can find. diff --git a/lib/crates/fabro-workflows/src/operations/fork.rs b/lib/crates/fabro-workflows/src/operations/fork.rs index 2bace5752..a7acd72f5 100644 --- a/lib/crates/fabro-workflows/src/operations/fork.rs +++ b/lib/crates/fabro-workflows/src/operations/fork.rs @@ -4,6 +4,7 @@ use fabro_git_storage::gitobj::Store; use git2::{Oid, Signature}; use crate::git::{MetadataStore, RUN_BRANCH_PREFIX, push_run_branches}; +use crate::records::Checkpoint; use crate::records::RunRecord; use crate::records::StartRecord; @@ -116,21 +117,25 @@ fn fork_from_entry( entry.metadata_commit_oid ) })?; + let mut checkpoint: Checkpoint = serde_json::from_slice(&checkpoint_bytes) + .context("failed to parse source checkpoint.json")?; + checkpoint.git_commit_sha.clone_from(&entry.run_commit_sha); + let checkpoint_bytes = + serde_json::to_vec_pretty(&checkpoint).context("failed to serialize checkpoint.json")?; - let mut file_entries: Vec<(&str, &[u8])> = vec![ - ("run.json", &new_run_record_bytes), - ("checkpoint.json", &checkpoint_bytes), - ]; + let mut init_entries: Vec<(&str, &[u8])> = vec![("run.json", &new_run_record_bytes)]; if let Some(ref start_record) = new_start_record_bytes { - file_entries.push(("start.json", start_record)); + init_entries.push(("start.json", start_record)); } if let Some(ref sandbox) = sandbox_bytes { - file_entries.push(("sandbox.json", sandbox)); + init_entries.push(("sandbox.json", sandbox)); } - let commit_msg = format!("fork from {} @{}", source_run_id, entry.ordinal); new_bs - .write_entries(&file_entries, &commit_msg) + .write_entries(&init_entries, "init run") + .map_err(|e| anyhow::anyhow!("failed to write init metadata entries: {e}"))?; + new_bs + .write_entry("checkpoint.json", &checkpoint_bytes, "checkpoint") .map_err(|e| anyhow::anyhow!("failed to write metadata entries: {e}"))?; if push { @@ -242,7 +247,7 @@ mod tests { fn fork_creates_new_run_and_metadata_branches() { let (_dir, store) = temp_repo(); let source_run_id = "run-source"; - let _run_oids = setup_source_run(&store, source_run_id, &["start", "build", "test"]); + let run_oids = setup_source_run(&store, source_run_id, &["start", "build", "test"]); let new_run_id = fork( &store, @@ -265,6 +270,14 @@ mod tests { let run_json = bs.read_entry("run.json").unwrap().unwrap(); let run_record: RunRecord = serde_json::from_slice(&run_json).unwrap(); assert_eq!(run_record.run_id, new_run_id); + + let timeline = build_timeline(&store, &new_run_id).unwrap(); + assert_eq!(timeline.entries.len(), 1); + assert_eq!(timeline.entries[0].node_name, "build"); + assert_eq!( + timeline.entries[0].run_commit_sha, + Some(run_oids[1].to_string()) + ); } #[test] diff --git a/lib/crates/fabro-workflows/src/operations/rebuild_meta.rs b/lib/crates/fabro-workflows/src/operations/rebuild_meta.rs index 991ae5b6c..0727d198c 100644 --- a/lib/crates/fabro-workflows/src/operations/rebuild_meta.rs +++ b/lib/crates/fabro-workflows/src/operations/rebuild_meta.rs @@ -9,9 +9,11 @@ use fabro_store::{ ListRunsQuery, NodeVisitRef, RunStore as DurableRunStore, Store as DurableStore, }; use git2::{Repository, Signature}; +use ulid::Ulid; -use super::rewind::{RunTimeline, build_timeline}; +use super::rewind::{self, RunTimeline, build_timeline}; use crate::git::MetadataStore; +use crate::records::Checkpoint; pub async fn rebuild_metadata_branch( git_store: &GitStore, @@ -29,69 +31,91 @@ pub async fn rebuild_metadata_branch( .ok_or_else(|| anyhow::anyhow!("run record not found for {run_id}"))?; let sig = Signature::now("Fabro", "noreply@fabro.sh")?; - let bs = BranchStore::new(git_store, &branch, &sig); - bs.ensure_branch()?; + let scratch_branch = format!("fabro/meta-rebuild/{run_id}/{}", Ulid::new()); + let bs = BranchStore::new(git_store, &scratch_branch, &sig); - let mut init_entries = Vec::new(); - init_entries.push(( - "run.json".to_string(), - serde_json::to_vec_pretty(&run_record)?, - )); - if let Some(start) = run_store.get_start().await? { - init_entries.push(("start.json".to_string(), serde_json::to_vec_pretty(&start)?)); - } - if let Some(sandbox) = run_store.get_sandbox().await? { + let result = async { + bs.ensure_branch()?; + + let mut init_entries = Vec::new(); init_entries.push(( - "sandbox.json".to_string(), - serde_json::to_vec_pretty(&sandbox)?, + "run.json".to_string(), + serde_json::to_vec_pretty(&run_record)?, )); - } - write_entries(&bs, &init_entries, "init run")?; + if let Some(start) = run_store.get_start().await? { + init_entries.push(("start.json".to_string(), serde_json::to_vec_pretty(&start)?)); + } + if let Some(sandbox) = run_store.get_sandbox().await? { + init_entries.push(( + "sandbox.json".to_string(), + serde_json::to_vec_pretty(&sandbox)?, + )); + } + write_entries(&bs, &init_entries, "init run")?; - for (_seq, checkpoint) in run_store.list_checkpoints().await? { - let mut entries = Vec::new(); - entries.push(( - "checkpoint.json".to_string(), - serde_json::to_vec_pretty(&checkpoint)?, - )); + let mut checkpoints = run_store.list_checkpoints().await?; + backfill_missing_checkpoint_shas(git_store, run_id, &mut checkpoints); - for node_id in &checkpoint.completed_nodes { - let max_visit = checkpoint.node_visits.get(node_id).copied().unwrap_or(1); - for visit in 1..=max_visit { - let visit = u32::try_from(visit) - .with_context(|| format!("visit {visit} for node {node_id} exceeds u32"))?; - let node = run_store.get_node(&NodeVisitRef { node_id, visit }).await?; + for (_seq, checkpoint) in checkpoints { + let mut entries = Vec::new(); + entries.push(( + "checkpoint.json".to_string(), + serde_json::to_vec_pretty(&checkpoint)?, + )); - if let Some(prompt) = node.prompt { - entries.push(( - node_file_path(node_id, visit, "prompt.md"), - prompt.into_bytes(), - )); - } - if let Some(response) = node.response { - entries.push(( - node_file_path(node_id, visit, "response.md"), - response.into_bytes(), - )); - } - if let Some(status) = node.status { - entries.push(( - node_file_path(node_id, visit, "status.json"), - serde_json::to_vec_pretty(&status)?, - )); + for node_id in &checkpoint.completed_nodes { + let max_visit = checkpoint.node_visits.get(node_id).copied().unwrap_or(1); + for visit in 1..=max_visit { + let visit = u32::try_from(visit) + .with_context(|| format!("visit {visit} for node {node_id} exceeds u32"))?; + let node = run_store.get_node(&NodeVisitRef { node_id, visit }).await?; + + if let Some(prompt) = node.prompt { + entries.push(( + node_file_path(node_id, visit, "prompt.md"), + prompt.into_bytes(), + )); + } + if let Some(response) = node.response { + entries.push(( + node_file_path(node_id, visit, "response.md"), + response.into_bytes(), + )); + } + if let Some(status) = node.status { + entries.push(( + node_file_path(node_id, visit, "status.json"), + serde_json::to_vec_pretty(&status)?, + )); + } } } + + write_entries(&bs, &entries, "checkpoint")?; } - write_entries(&bs, &entries, "checkpoint")?; - } + if let Some(retro) = run_store.get_retro().await? { + let entries = vec![("retro.json".to_string(), serde_json::to_vec_pretty(&retro)?)]; + write_entries(&bs, &entries, "finalize run")?; + } - if let Some(retro) = run_store.get_retro().await? { - let entries = vec![("retro.json".to_string(), serde_json::to_vec_pretty(&retro)?)]; - write_entries(&bs, &entries, "finalize run")?; + Ok::<(), anyhow::Error>(()) } + .await; - Ok(()) + let final_result = match result { + Ok(()) => { + let scratch_tip = git_store + .resolve_ref(&scratch_branch)? + .ok_or_else(|| anyhow::anyhow!("scratch metadata branch missing after rebuild"))?; + git_store.update_ref(&branch, scratch_tip)?; + Ok(()) + } + Err(err) => Err(err), + }; + + let _ = git_store.delete_ref(&scratch_branch); + final_result } pub async fn build_timeline_or_rebuild( @@ -177,6 +201,38 @@ fn write_entries( Ok(()) } +fn backfill_missing_checkpoint_shas( + git_store: &GitStore, + run_id: &str, + checkpoints: &mut [(u32, Checkpoint)], +) { + if !checkpoints + .iter() + .any(|(_, checkpoint)| checkpoint.git_commit_sha.is_none()) + { + return; + } + + let node_commits = rewind::run_commit_shas_by_node(git_store, run_id); + let mut node_indices: HashMap = HashMap::new(); + + for (_seq, checkpoint) in checkpoints.iter_mut() { + if checkpoint.git_commit_sha.is_some() { + continue; + } + + if let Some(shas) = node_commits.get(&checkpoint.current_node) { + let idx = node_indices + .entry(checkpoint.current_node.clone()) + .or_insert(0); + if *idx < shas.len() { + checkpoint.git_commit_sha = Some(shas[*idx].clone()); + *idx += 1; + } + } + } +} + fn node_file_path(node_id: &str, visit: u32, filename: &str) -> String { if visit <= 1 { format!("nodes/{node_id}/{filename}") @@ -339,6 +395,31 @@ mod tests { run_store } + fn seed_run_branch(git_store: &GitStore, run_id: &str, nodes: &[&str]) -> Vec { + let sig = test_sig(); + let run_branch = format!("fabro/run/{run_id}"); + let empty_tree = git_store.write_empty_tree().unwrap(); + let mut shas = Vec::new(); + let mut parent = None; + + for node in nodes { + let parents = parent.into_iter().collect::>(); + let oid = git_store + .write_commit( + empty_tree, + &parents, + &format!("fabro({run_id}): {node} (completed)"), + &sig, + ) + .unwrap(); + git_store.update_ref(&run_branch, oid).unwrap(); + shas.push(oid.to_string()); + parent = Some(oid); + } + + shas + } + #[tokio::test] async fn rebuild_metadata_branch_round_trips_timeline() { let (_dir, git_store) = temp_repo(); @@ -694,4 +775,122 @@ mod tests { .unwrap(); assert_eq!(from_refs, "abc-123"); } + + #[tokio::test] + async fn rebuild_metadata_branch_persists_backfilled_run_shas_in_checkpoint_blobs() { + let (_dir, git_store) = temp_repo(); + let durable_store = InMemoryStore::default(); + let run_store = create_run_store(&durable_store, "run-1", None).await; + + run_store + .append_checkpoint(&sample_checkpoint( + "start", + &["start"], + &[("start", 1)], + None, + )) + .await + .unwrap(); + run_store + .append_checkpoint(&sample_checkpoint( + "build", + &["start", "build"], + &[("start", 1), ("build", 1)], + None, + )) + .await + .unwrap(); + + let expected_shas = seed_run_branch(&git_store, "run-1", &["start", "build"]); + + rebuild_metadata_branch(&git_store, run_store.as_ref(), "run-1") + .await + .unwrap(); + + let sig = test_sig(); + let branch = MetadataStore::branch_name("run-1"); + let bs = BranchStore::new(&git_store, &branch, &sig); + let checkpoint_commits: Vec<_> = bs + .log(100) + .unwrap() + .iter() + .rev() + .filter(|commit| commit.message.starts_with("checkpoint")) + .map(|commit| commit.oid) + .collect(); + + let first: Checkpoint = serde_json::from_slice( + &git_store + .read_blob_at(checkpoint_commits[0], "checkpoint.json") + .unwrap() + .unwrap(), + ) + .unwrap(); + let second: Checkpoint = serde_json::from_slice( + &git_store + .read_blob_at(checkpoint_commits[1], "checkpoint.json") + .unwrap() + .unwrap(), + ) + .unwrap(); + + assert_eq!( + first.git_commit_sha.as_deref(), + Some(expected_shas[0].as_str()) + ); + assert_eq!( + second.git_commit_sha.as_deref(), + Some(expected_shas[1].as_str()) + ); + } + + #[tokio::test] + async fn rebuild_metadata_branch_is_atomic_on_failure() { + let (_dir, git_store) = temp_repo(); + let durable_store = InMemoryStore::default(); + let run_store = create_run_store(&durable_store, "run-1", None).await; + + let bad_node = "bad\0node"; + let bad_visit = NodeVisitRef { + node_id: bad_node, + visit: 1, + }; + run_store + .put_node_prompt(&bad_visit, "prompt") + .await + .unwrap(); + run_store + .append_checkpoint(&sample_checkpoint( + bad_node, + &[bad_node], + &[(bad_node, 1)], + None, + )) + .await + .unwrap(); + + let err = rebuild_metadata_branch(&git_store, run_store.as_ref(), "run-1") + .await + .unwrap_err(); + assert!(err.to_string().contains("nul") || err.to_string().contains("NUL")); + assert!( + git_store + .resolve_ref(&MetadataStore::branch_name("run-1")) + .unwrap() + .is_none() + ); + + let scratch_refs: Vec<_> = git_store + .repo() + .references() + .unwrap() + .flatten() + .filter_map(|reference| reference.name().map(ToOwned::to_owned)) + .filter(|name| name.starts_with("refs/heads/fabro/meta-rebuild/run-1/")) + .collect(); + assert!( + scratch_refs.is_empty(), + "leftover scratch refs: {scratch_refs:?}" + ); + } } diff --git a/lib/crates/fabro-workflows/src/operations/rewind.rs b/lib/crates/fabro-workflows/src/operations/rewind.rs index 9accd7455..4ce314bd9 100644 --- a/lib/crates/fabro-workflows/src/operations/rewind.rs +++ b/lib/crates/fabro-workflows/src/operations/rewind.rs @@ -166,13 +166,31 @@ fn backfill_run_shas(store: &Store, run_id: &str, timeline: &mut [TimelineEntry] return; } + let node_commits = run_commit_shas_by_node(store, run_id); + let mut node_indices: HashMap = HashMap::new(); + + for entry in timeline.iter_mut() { + if entry.run_commit_sha.is_some() { + continue; + } + if let Some(shas) = node_commits.get(&entry.node_name) { + let idx = node_indices.entry(entry.node_name.clone()).or_insert(0); + if *idx < shas.len() { + entry.run_commit_sha = Some(shas[*idx].clone()); + *idx += 1; + } + } + } +} + +pub(crate) fn run_commit_shas_by_node(store: &Store, run_id: &str) -> HashMap> { let run_branch = format!("{RUN_BRANCH_PREFIX}{run_id}"); let Ok(sig) = Signature::now("Fabro", "noreply@fabro.sh") else { - return; + return HashMap::new(); }; let bs = BranchStore::new(store, &run_branch, &sig); let Ok(run_commits) = bs.log(10_000) else { - return; + return HashMap::new(); }; let prefix = format!("fabro({run_id}): "); @@ -191,20 +209,8 @@ fn backfill_run_shas(store: &Store, run_id: &str, timeline: &mut [TimelineEntry] for shas in node_commits.values_mut() { shas.reverse(); } - let mut node_indices: HashMap = HashMap::new(); - for entry in timeline.iter_mut() { - if entry.run_commit_sha.is_some() { - continue; - } - if let Some(shas) = node_commits.get(&entry.node_name) { - let idx = node_indices.entry(entry.node_name.clone()).or_insert(0); - if *idx < shas.len() { - entry.run_commit_sha = Some(shas[*idx].clone()); - *idx += 1; - } - } - } + node_commits } fn detect_parallel_interior(graph: &Graph) -> HashMap {