From b01a666cd94c2fdd94a5cfd59e448d7b14e7eb0d Mon Sep 17 00:00:00 2001 From: Bryan Helmkamp Date: Fri, 1 May 2026 20:55:56 -0400 Subject: [PATCH] refactor: simplify retry-related helpers and orphan dump scan Remove unused RunProjection::stage_mut, share decode_retry_and_filename between artifact_store decoders, reuse stage_visit() in the artifact lifecycle, and cache the dump.log entry index so RunDump::add_orphan_notice no longer rescans entries on every call. Co-Authored-By: Claude Opus 4.7 (1M context) --- lib/crates/fabro-dump/src/lib.rs | 22 +++++----- lib/crates/fabro-store/src/artifact_store.rs | 40 +++++++++---------- lib/crates/fabro-types/src/run_projection.rs | 4 -- .../fabro-workflow/src/lifecycle/artifact.rs | 5 +-- .../fabro-workflow/src/lifecycle/event.rs | 4 +- 5 files changed, 35 insertions(+), 40 deletions(-) diff --git a/lib/crates/fabro-dump/src/lib.rs b/lib/crates/fabro-dump/src/lib.rs index 16954bfe9..48f3d665c 100644 --- a/lib/crates/fabro-dump/src/lib.rs +++ b/lib/crates/fabro-dump/src/lib.rs @@ -23,8 +23,9 @@ pub type BlobReader = Box BoxFuture<'static, Result, - stage_ranks: HashMap, + entries: Vec, + stage_ranks: HashMap, + dump_log_index: Option, } #[derive(Debug, Clone)] @@ -151,6 +152,7 @@ impl RunDump { Ok(Self { entries, stage_ranks, + dump_log_index: None, }) } @@ -252,16 +254,15 @@ impl RunDump { fn add_orphan_notice(&mut self, stage_id: &StageId) { let line = format!("notice: artifact stage {stage_id} was not present in run projection\n"); - if let Some(entry) = self - .entries - .iter_mut() - .find(|entry| entry.path == "dump.log") - { - if let RunDumpContents::Text(text) = &mut entry.contents { + if let Some(index) = self.dump_log_index { + if let Some(RunDumpContents::Text(text)) = + self.entries.get_mut(index).map(|entry| &mut entry.contents) + { text.push_str(&line); return; } } + self.dump_log_index = Some(self.entries.len()); self.entries.push(RunDumpEntry::text("dump.log", line)); } @@ -779,11 +780,12 @@ mod tests { let blob_id = fabro_types::RunBlobId::new(&blob); let legacy_ref = format!("file:///sandbox/.fabro/artifacts/{blob_id}.json"); let mut dump = RunDump { - entries: vec![RunDumpEntry::json( + entries: vec![RunDumpEntry::json( "run.json", serde_json::json!({ "stdout": legacy_ref }), )], - stage_ranks: HashMap::new(), + stage_ranks: HashMap::new(), + dump_log_index: None, }; executor::block_on(async { diff --git a/lib/crates/fabro-store/src/artifact_store.rs b/lib/crates/fabro-store/src/artifact_store.rs index 8cc244131..25ba1a55d 100644 --- a/lib/crates/fabro-store/src/artifact_store.rs +++ b/lib/crates/fabro-store/src/artifact_store.rs @@ -294,24 +294,11 @@ fn decode_artifact_location( "artifact location {location} has an invalid visit number: {err}" )) })?; - let retry_part = parts.next().ok_or_else(|| { - Error::Other(format!( - "artifact location {location} is missing a retry segment" - )) - })?; - let retry = decode_retry_segment(location, retry_part.as_ref())?; - let filename_segments = parts - .map(|part| decode_path_segment("artifact filename segment", part.as_ref())) - .collect::>>()?; - if filename_segments.is_empty() { - return Err(Error::Other(format!( - "artifact location {location} is missing a filename" - ))); - } + let (retry, filename) = decode_retry_and_filename(location, &mut parts)?; Ok(NodeArtifact { node: StageId::new(node_id, visit), retry, - filename: filename_segments.join("/"), + filename, size, }) } @@ -326,6 +313,22 @@ fn decode_stage_artifact_entry( "artifact location {location} does not match expected prefix {prefix}" )) })?; + let (retry, filename) = decode_retry_and_filename(location, &mut parts)?; + Ok(StageArtifactEntry { + retry, + filename, + size, + }) +} + +fn decode_retry_and_filename<'a, I, P>( + location: &ObjectPath, + parts: &mut I, +) -> Result<(u32, String)> +where + I: Iterator, + P: AsRef + 'a, +{ let retry_part = parts.next().ok_or_else(|| { Error::Other(format!( "artifact location {location} is missing a retry segment" @@ -333,7 +336,6 @@ fn decode_stage_artifact_entry( })?; let retry = decode_retry_segment(location, retry_part.as_ref())?; let filename_segments = parts - .by_ref() .map(|part| decode_path_segment("artifact filename segment", part.as_ref())) .collect::>>()?; if filename_segments.is_empty() { @@ -341,11 +343,7 @@ fn decode_stage_artifact_entry( "artifact location {location} is missing a filename" ))); } - Ok(StageArtifactEntry { - retry, - filename: filename_segments.join("/"), - size, - }) + Ok((retry, filename_segments.join("/"))) } fn decode_retry_segment(location: &ObjectPath, segment: &str) -> Result { diff --git a/lib/crates/fabro-types/src/run_projection.rs b/lib/crates/fabro-types/src/run_projection.rs index bba87e156..67cf7de4c 100644 --- a/lib/crates/fabro-types/src/run_projection.rs +++ b/lib/crates/fabro-types/src/run_projection.rs @@ -113,10 +113,6 @@ impl RunProjection { &self.pending_interviews } - pub fn stage_mut(&mut self, node_id: &str, visit: u32) -> &mut StageState { - self.stage_entry(node_id, visit, 0) - } - pub fn stage_entry_id(&mut self, stage_id: &StageId, seq: u32) -> &mut StageState { self.stages .entry(stage_id.clone()) diff --git a/lib/crates/fabro-workflow/src/lifecycle/artifact.rs b/lib/crates/fabro-workflow/src/lifecycle/artifact.rs index 24f5f8c90..f65fcee86 100644 --- a/lib/crates/fabro-workflow/src/lifecycle/artifact.rs +++ b/lib/crates/fabro-workflow/src/lifecycle/artifact.rs @@ -18,7 +18,7 @@ use crate::artifact_snapshot::collect_artifacts; use crate::artifact_upload::ArtifactSink; use crate::event::{Emitter, Event, RunNoticeLevel}; use crate::graph::{WorkflowGraph, WorkflowNode}; -use crate::lifecycle::event::stage_scope_for; +use crate::lifecycle::event::{stage_scope_for, stage_visit}; use crate::outcome::BilledModelUsage; use crate::runtime_store::RunStoreHandle; @@ -95,8 +95,7 @@ impl RunLifecycle for ArtifactLifecycle { } let epoch = self.attempt_start_epoch.lock().unwrap().unwrap_or(0.0); let node_id = ctx.node.id(); - let visit_count = state.node_visits.get(node_id).copied().unwrap_or(1); - let visit = u32::try_from(visit_count.max(1)).unwrap_or(u32::MAX); + let visit = stage_visit(state, node_id); let node_slug = if visit <= 1 { node_id.to_string() } else { diff --git a/lib/crates/fabro-workflow/src/lifecycle/event.rs b/lib/crates/fabro-workflow/src/lifecycle/event.rs index 4dc95e512..d97d7f613 100644 --- a/lib/crates/fabro-workflow/src/lifecycle/event.rs +++ b/lib/crates/fabro-workflow/src/lifecycle/event.rs @@ -74,9 +74,9 @@ fn response_from_outcome(node_id: &str, outcome: &Outcome) -> Option { .and_then(|value| value.as_str().map(ToOwned::to_owned)) } -fn stage_visit(state: &WfRunState, node_id: &str) -> u32 { +pub(super) fn stage_visit(state: &WfRunState, node_id: &str) -> u32 { let visits = state.node_visits.get(node_id).copied().unwrap_or(1); - u32::try_from(visits.max(1)).unwrap_or(u32::MAX) + u32::try_from(visits).unwrap_or(u32::MAX) } pub(crate) fn stage_scope_for(state: &WfRunState, node_id: &str) -> StageScope {