scriptoriumadmin/fabro
1
0
Fork 0
mirror of https://github.com/fabro-sh/fabro.git synced 2026-09-06 08:18:58 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 3

Register parallel handler to prevent dry-run hang on parallel pipelines

Browse source 
The parallel handler was never registered in default_registry() due to a
circular dependency (ParallelHandler::new needs Arc<HandlerRegistry>).
Break the cycle by storing ParallelHandler as a separate field on
PipelineEngine, created after Arc-wrapping the registry and emitter.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
Bryan Helmkamp 2026-02-23 16:04:16 -05:00
parent ceaa9bf560
commit 06480db417
1 changed files with 24 additions and 3 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

27
crates/attractor/src/engine.rs
View file

@ -15,6 +15,7 @@ use crate::context::Context;
use crate::error::{AttractorError, Result};
use crate::event::{EventEmitter, PipelineEvent};
use crate::graph::{Edge, Graph, Node};
use crate::handler::parallel::ParallelHandler;
use crate::handler::HandlerRegistry;
use crate::interviewer::Interviewer;
use crate::outcome::{Outcome, StageStatus};
@ -451,17 +452,23 @@ pub struct RunConfig {
/// The pipeline execution engine.
pub struct PipelineEngine {
pub registry: HandlerRegistry,
pub emitter: EventEmitter,
registry: Arc<HandlerRegistry>,
emitter: Arc<EventEmitter>,
parallel_handler: ParallelHandler,
pub interviewer: Option<Arc<dyn Interviewer>>,
}
impl PipelineEngine {
#[must_use]
pub fn new(registry: HandlerRegistry, emitter: EventEmitter) -> Self {
let registry = Arc::new(registry);
let emitter = Arc::new(emitter);
let parallel_handler =
ParallelHandler::new(Arc::clone(&registry), Arc::clone(&emitter));
Self {
registry,
emitter,
parallel_handler,
interviewer: None,
}
}
@ -473,13 +480,27 @@ impl PipelineEngine {
emitter: EventEmitter,
interviewer: Arc<dyn Interviewer>,
) -> Self {
let registry = Arc::new(registry);
let emitter = Arc::new(emitter);
let parallel_handler =
ParallelHandler::new(Arc::clone(&registry), Arc::clone(&emitter));
Self {
registry,
emitter,
parallel_handler,
interviewer: Some(interviewer),
}
}
/// Resolve the handler for a node, returning the parallel handler for
/// parallel nodes and delegating to the registry for everything else.
fn resolve_handler(&self, node: &Node) -> &dyn crate::handler::Handler {
if node.handler_type() == Some("parallel") {
return &self.parallel_handler;
}
self.registry.resolve(node)
}
/// Call inform on the interviewer, if one is configured.
fn inform(&self, message: &str, stage: &str) {
if let Some(ref interviewer) = self.interviewer {
@ -517,7 +538,7 @@ impl PipelineEngine {
policy: &RetryPolicy,
stage_index: usize,
) -> Result<(Outcome, u32)> {
let handler = self.registry.resolve(node);
let handler = self.resolve_handler(node);
let node_timeout = node.timeout();