From 127e0211cceb52d568b4fb9dbc1b78fb6854716b Mon Sep 17 00:00:00 2001 From: Bryan Helmkamp Date: Sat, 28 Mar 2026 00:50:35 -0400 Subject: [PATCH] Deduplicate timeline resolution and test helpers in operations Move resolve_target into RunTimeline::resolve() method and extract shared test helpers (temp_repo, test_sig, make_checkpoint_json) into a test_support module used by both fork.rs and rewind.rs. Co-Authored-By: Claude Opus 4.6 (1M context) --- .../fabro-workflows/src/operations/fork.rs | 87 +----------- .../fabro-workflows/src/operations/mod.rs | 2 + .../fabro-workflows/src/operations/rewind.rs | 132 +++++++----------- .../src/operations/test_support.rs | 34 +++++ 4 files changed, 93 insertions(+), 162 deletions(-) create mode 100644 lib/crates/fabro-workflows/src/operations/test_support.rs diff --git a/lib/crates/fabro-workflows/src/operations/fork.rs b/lib/crates/fabro-workflows/src/operations/fork.rs index 4db57ab05..d45d796a7 100644 --- a/lib/crates/fabro-workflows/src/operations/fork.rs +++ b/lib/crates/fabro-workflows/src/operations/fork.rs @@ -7,7 +7,7 @@ use crate::git::MetadataStore; use crate::records::RunRecord; use crate::records::StartRecord; -use super::rewind::{build_timeline, RewindTarget, RunTimeline, TimelineEntry}; +use super::rewind::{build_timeline, RewindTarget, TimelineEntry}; #[derive(Debug, Clone)] pub struct ForkRunInput { @@ -22,7 +22,7 @@ pub struct ForkRunInput { pub fn fork(store: &Store, input: ForkRunInput) -> Result { let timeline = build_timeline(store, &input.source_run_id)?; let entry = match input.target.as_ref() { - Some(target) => resolve_timeline_entry(&timeline, target)?, + Some(target) => timeline.resolve(target)?, None => timeline.entries.last().ok_or_else(|| { anyhow::anyhow!("no checkpoints found for run {}", input.source_run_id) })?, @@ -30,59 +30,6 @@ pub fn fork(store: &Store, input: ForkRunInput) -> Result { fork_from_entry(store, &input.source_run_id, entry, input.push) } -fn resolve_timeline_entry<'a>( - timeline: &'a RunTimeline, - target: &RewindTarget, -) -> Result<&'a TimelineEntry> { - match target { - RewindTarget::Ordinal(n) => timeline - .entries - .iter() - .find(|e| e.ordinal == *n) - .ok_or_else(|| { - anyhow::anyhow!( - "ordinal @{n} out of range (max @{})", - timeline.entries.len() - ) - }), - RewindTarget::LatestVisit(name) => { - let effective_name = timeline.parallel_map.get(name).unwrap_or(name); - timeline - .entries - .iter() - .rev() - .find(|e| e.node_name == *effective_name) - .ok_or_else(|| { - if effective_name != name { - anyhow::anyhow!( - "node '{name}' is inside parallel '{effective_name}'; \ - no checkpoint found for '{effective_name}'" - ) - } else { - anyhow::anyhow!("no checkpoint found for node '{name}'") - } - }) - } - RewindTarget::SpecificVisit(name, visit) => { - let effective_name = timeline.parallel_map.get(name).unwrap_or(name); - timeline - .entries - .iter() - .find(|e| e.node_name == *effective_name && e.visit == *visit) - .ok_or_else(|| { - if effective_name != name { - anyhow::anyhow!( - "node '{name}' is inside parallel '{effective_name}'; \ - no visit {visit} found for '{effective_name}'" - ) - } else { - anyhow::anyhow!("no visit {visit} found for node '{name}'") - } - }) - } - } -} - fn fork_from_entry( store: &Store, source_run_id: &str, @@ -217,40 +164,14 @@ fn fork_from_entry( #[cfg(test)] mod tests { - use std::collections::HashMap; use std::str::FromStr; + use super::super::test_support::*; use super::*; - use git2::Repository; + use git2::Oid; use crate::operations::find_run_id_by_prefix; - fn temp_repo() -> (tempfile::TempDir, Store) { - let dir = tempfile::TempDir::new().unwrap(); - let repo = Repository::init(dir.path()).unwrap(); - (dir, Store::new(repo)) - } - - fn test_sig() -> Signature<'static> { - Signature::now("Test", "test@example.com").unwrap() - } - - fn make_checkpoint_json(current_node: &str, visit: usize, git_sha: Option<&str>) -> Vec { - let mut node_visits = HashMap::new(); - node_visits.insert(current_node.to_string(), visit); - let cp = serde_json::json!({ - "timestamp": "2025-01-01T00:00:00Z", - "current_node": current_node, - "completed_nodes": [current_node], - "node_retries": {}, - "context_values": {}, - "logs": [], - "node_visits": node_visits, - "git_commit_sha": git_sha, - }); - serde_json::to_vec(&cp).unwrap() - } - fn make_run_record_json(run_id: &str) -> Vec { let record = serde_json::json!({ "run_id": run_id, diff --git a/lib/crates/fabro-workflows/src/operations/mod.rs b/lib/crates/fabro-workflows/src/operations/mod.rs index 4ab837971..2bdc2f048 100644 --- a/lib/crates/fabro-workflows/src/operations/mod.rs +++ b/lib/crates/fabro-workflows/src/operations/mod.rs @@ -4,6 +4,8 @@ mod resume; mod rewind; mod source; mod start; +#[cfg(test)] +mod test_support; mod validate; pub use crate::pipeline::{DevcontainerSpec, LlmSpec, SandboxEnvSpec, SandboxSpec}; diff --git a/lib/crates/fabro-workflows/src/operations/rewind.rs b/lib/crates/fabro-workflows/src/operations/rewind.rs index a96235eb4..73030f17f 100644 --- a/lib/crates/fabro-workflows/src/operations/rewind.rs +++ b/lib/crates/fabro-workflows/src/operations/rewind.rs @@ -61,6 +61,54 @@ pub struct RunTimeline { pub parallel_map: HashMap, } +impl RunTimeline { + pub fn resolve(&self, target: &RewindTarget) -> Result<&TimelineEntry> { + match target { + RewindTarget::Ordinal(n) => { + self.entries + .iter() + .find(|e| e.ordinal == *n) + .ok_or_else(|| { + anyhow::anyhow!("ordinal @{n} out of range (max @{})", self.entries.len()) + }) + } + RewindTarget::LatestVisit(name) => { + let effective_name = self.parallel_map.get(name).unwrap_or(name); + self.entries + .iter() + .rev() + .find(|e| e.node_name == *effective_name) + .ok_or_else(|| { + if effective_name != name { + anyhow::anyhow!( + "node '{name}' is inside parallel '{effective_name}'; \ + no checkpoint found for '{effective_name}'" + ) + } else { + anyhow::anyhow!("no checkpoint found for node '{name}'") + } + }) + } + RewindTarget::SpecificVisit(name, visit) => { + let effective_name = self.parallel_map.get(name).unwrap_or(name); + self.entries + .iter() + .find(|e| e.node_name == *effective_name && e.visit == *visit) + .ok_or_else(|| { + if effective_name != name { + anyhow::anyhow!( + "node '{name}' is inside parallel '{effective_name}'; \ + no visit {visit} found for '{effective_name}'" + ) + } else { + anyhow::anyhow!("no visit {visit} found for node '{name}'") + } + }) + } + } + } +} + #[derive(Debug, Clone)] pub struct RewindInput { pub run_id: String, @@ -193,55 +241,9 @@ fn detect_parallel_interior(graph: &Graph) -> HashMap { interior_map } -fn resolve_target<'a>( - timeline: &'a [TimelineEntry], - target: &RewindTarget, - parallel_map: &HashMap, -) -> Result<&'a TimelineEntry> { - match target { - RewindTarget::Ordinal(n) => timeline - .iter() - .find(|e| e.ordinal == *n) - .ok_or_else(|| anyhow::anyhow!("ordinal @{n} out of range (max @{})", timeline.len())), - RewindTarget::LatestVisit(name) => { - let effective_name = parallel_map.get(name).unwrap_or(name); - timeline - .iter() - .rev() - .find(|e| e.node_name == *effective_name) - .ok_or_else(|| { - if effective_name != name { - anyhow::anyhow!( - "node '{name}' is inside parallel '{effective_name}'; \ - no checkpoint found for '{effective_name}'" - ) - } else { - anyhow::anyhow!("no checkpoint found for node '{name}'") - } - }) - } - RewindTarget::SpecificVisit(name, visit) => { - let effective_name = parallel_map.get(name).unwrap_or(name); - timeline - .iter() - .find(|e| e.node_name == *effective_name && e.visit == *visit) - .ok_or_else(|| { - if effective_name != name { - anyhow::anyhow!( - "node '{name}' is inside parallel '{effective_name}'; \ - no visit {visit} found for '{effective_name}'" - ) - } else { - anyhow::anyhow!("no visit {visit} found for node '{name}'") - } - }) - } - } -} - pub fn rewind(store: &Store, input: RewindInput) -> Result<()> { let timeline = build_timeline(store, &input.run_id)?; - let entry = resolve_target(&timeline.entries, &input.target, &timeline.parallel_map)?; + let entry = timeline.resolve(&input.target)?; rewind_to_entry(store, &input.run_id, entry, input.push) } @@ -371,34 +373,9 @@ fn load_parallel_map(store: &Store, run_id: &str) -> HashMap { #[cfg(test)] mod tests { + use super::super::test_support::*; use super::*; - fn temp_repo() -> (tempfile::TempDir, Store) { - let dir = tempfile::TempDir::new().unwrap(); - let repo = Repository::init(dir.path()).unwrap(); - (dir, Store::new(repo)) - } - - fn test_sig() -> Signature<'static> { - Signature::now("Test", "test@example.com").unwrap() - } - - fn make_checkpoint_json(current_node: &str, visit: usize, git_sha: Option<&str>) -> Vec { - let mut node_visits = HashMap::new(); - node_visits.insert(current_node.to_string(), visit); - let cp = serde_json::json!({ - "timestamp": "2025-01-01T00:00:00Z", - "current_node": current_node, - "completed_nodes": [current_node], - "node_retries": {}, - "context_values": {}, - "logs": [], - "node_visits": node_visits, - "git_commit_sha": git_sha, - }); - serde_json::to_vec(&cp).unwrap() - } - #[test] fn parse_target_ordinal() { assert_eq!( @@ -466,12 +443,9 @@ mod tests { parallel_map: HashMap::new(), }; - let entry = resolve_target( - &timeline.entries, - &RewindTarget::LatestVisit("build".to_string()), - &timeline.parallel_map, - ) - .unwrap(); + let entry = timeline + .resolve(&RewindTarget::LatestVisit("build".to_string())) + .unwrap(); assert_eq!(entry.ordinal, 3); } diff --git a/lib/crates/fabro-workflows/src/operations/test_support.rs b/lib/crates/fabro-workflows/src/operations/test_support.rs new file mode 100644 index 000000000..f686bb8b9 --- /dev/null +++ b/lib/crates/fabro-workflows/src/operations/test_support.rs @@ -0,0 +1,34 @@ +use std::collections::HashMap; + +use fabro_git_storage::gitobj::Store; +use git2::{Repository, Signature}; + +pub(super) fn temp_repo() -> (tempfile::TempDir, Store) { + let dir = tempfile::TempDir::new().unwrap(); + let repo = Repository::init(dir.path()).unwrap(); + (dir, Store::new(repo)) +} + +pub(super) fn test_sig() -> Signature<'static> { + Signature::now("Test", "test@example.com").unwrap() +} + +pub(super) fn make_checkpoint_json( + current_node: &str, + visit: usize, + git_sha: Option<&str>, +) -> Vec { + let mut node_visits = HashMap::new(); + node_visits.insert(current_node.to_string(), visit); + let cp = serde_json::json!({ + "timestamp": "2025-01-01T00:00:00Z", + "current_node": current_node, + "completed_nodes": [current_node], + "node_retries": {}, + "context_values": {}, + "logs": [], + "node_visits": node_visits, + "git_commit_sha": git_sha, + }); + serde_json::to_vec(&cp).unwrap() +}