diff --git a/crates/arc-workflows/src/graph/types.rs b/crates/arc-workflows/src/graph/types.rs index 218a1fe2c..1f353233e 100644 --- a/crates/arc-workflows/src/graph/types.rs +++ b/crates/arc-workflows/src/graph/types.rs @@ -101,6 +101,7 @@ pub fn shape_to_handler_type(shape: &str) -> Option<&'static str> { "tripleoctagon" => Some("parallel.fan_in"), "parallelogram" => Some("script"), "house" => Some("stack.manager_loop"), + "insulator" => Some("wait.timer"), _ => None, } } @@ -502,6 +503,7 @@ mod tests { ); assert_eq!(shape_to_handler_type("parallelogram"), Some("script")); assert_eq!(shape_to_handler_type("house"), Some("stack.manager_loop")); + assert_eq!(shape_to_handler_type("insulator"), Some("wait.timer")); assert_eq!(shape_to_handler_type("unknown"), None); } diff --git a/crates/arc-workflows/src/handler/mod.rs b/crates/arc-workflows/src/handler/mod.rs index dadbe3715..a468b900d 100644 --- a/crates/arc-workflows/src/handler/mod.rs +++ b/crates/arc-workflows/src/handler/mod.rs @@ -7,6 +7,7 @@ pub mod parallel; pub mod script; pub mod start; pub mod wait_human; +pub mod wait_timer; use std::collections::HashMap; use std::path::Path; @@ -140,6 +141,7 @@ pub fn default_registry( "stack.manager_loop", Box::new(manager_loop::SubWorkflowHandler), ); + registry.register("wait.timer", Box::new(wait_timer::WaitTimerHandler)); registry } diff --git a/crates/arc-workflows/src/handler/wait_timer.rs b/crates/arc-workflows/src/handler/wait_timer.rs new file mode 100644 index 000000000..b2f6e42fc --- /dev/null +++ b/crates/arc-workflows/src/handler/wait_timer.rs @@ -0,0 +1,91 @@ +use std::path::Path; + +use async_trait::async_trait; + +use crate::context::Context; +use crate::error::ArcError; +use crate::graph::{AttrValue, Graph, Node}; +use crate::outcome::Outcome; + +use super::{EngineServices, Handler}; + +/// Sleeps for a configured duration before proceeding. +pub struct WaitTimerHandler; + +#[async_trait] +impl Handler for WaitTimerHandler { + async fn execute( + &self, + node: &Node, + _context: &Context, + _graph: &Graph, + _logs_root: &Path, + _services: &EngineServices, + ) -> Result { + let duration = node + .attrs + .get("duration") + .and_then(AttrValue::as_duration) + .ok_or_else(|| { + ArcError::Validation(format!( + "wait.timer node {:?} is missing a valid `duration` attribute", + node.id + )) + })?; + tokio::time::sleep(duration).await; + Ok(Outcome::success()) + } +} + +#[cfg(test)] +mod tests { + use std::time::Duration; + + use super::*; + use crate::event::EventEmitter; + use crate::handler::HandlerRegistry; + + fn make_services() -> EngineServices { + EngineServices { + registry: std::sync::Arc::new(HandlerRegistry::new(Box::new( + crate::handler::start::StartHandler, + ))), + emitter: std::sync::Arc::new(EventEmitter::new()), + sandbox: std::sync::Arc::new(arc_agent::LocalSandbox::new( + std::env::current_dir().unwrap_or_else(|_| std::path::PathBuf::from(".")), + )), + git_state: std::sync::RwLock::new(None), + } + } + + #[tokio::test] + async fn wait_timer_success_with_short_duration() { + let handler = WaitTimerHandler; + let mut node = Node::new("wait60"); + node.attrs.insert( + "duration".to_string(), + AttrValue::Duration(Duration::from_millis(1)), + ); + let context = Context::new(); + let graph = Graph::new("test"); + let logs_root = Path::new("/tmp/test"); + let outcome = handler + .execute(&node, &context, &graph, logs_root, &make_services()) + .await + .unwrap(); + assert_eq!(outcome.status, crate::outcome::StageStatus::Success); + } + + #[tokio::test] + async fn wait_timer_errors_without_duration() { + let handler = WaitTimerHandler; + let node = Node::new("wait_no_dur"); + let context = Context::new(); + let graph = Graph::new("test"); + let logs_root = Path::new("/tmp/test"); + let result = handler + .execute(&node, &context, &graph, logs_root, &make_services()) + .await; + assert!(result.is_err()); + } +} diff --git a/crates/arc-workflows/src/validation/rules.rs b/crates/arc-workflows/src/validation/rules.rs index 8dcaf1101..5350f6998 100644 --- a/crates/arc-workflows/src/validation/rules.rs +++ b/crates/arc-workflows/src/validation/rules.rs @@ -392,6 +392,7 @@ const KNOWN_HANDLER_TYPES: &[&str] = &[ "script", "tool", "stack.manager_loop", + "wait.timer", ]; impl LintRule for TypeKnownRule { diff --git a/crates/arc-workflows/tests/integration.rs b/crates/arc-workflows/tests/integration.rs index 3994ae838..4b4304a14 100644 --- a/crates/arc-workflows/tests/integration.rs +++ b/crates/arc-workflows/tests/integration.rs @@ -18,6 +18,7 @@ use arc_workflows::handler::manager_loop::SubWorkflowHandler; use arc_workflows::handler::script::ScriptHandler; use arc_workflows::handler::start::StartHandler; use arc_workflows::handler::wait_human::WaitHumanHandler; +use arc_workflows::handler::wait_timer::WaitTimerHandler; use arc_workflows::handler::{Handler, HandlerRegistry}; use arc_workflows::interviewer::auto_approve::AutoApproveInterviewer; use arc_workflows::interviewer::queue::QueueInterviewer; @@ -1251,6 +1252,7 @@ fn make_full_registry(interviewer: Arc) -> HandlerRegistry { registry.register("conditional", Box::new(ConditionalHandler)); registry.register("script", Box::new(ScriptHandler)); registry.register("wait.human", Box::new(WaitHumanHandler::new(interviewer))); + registry.register("wait.timer", Box::new(WaitTimerHandler)); registry.register("stack.manager_loop", Box::new(SubWorkflowHandler)); registry } @@ -11504,3 +11506,45 @@ async fn asset_collection_docker_sandbox() { sandbox.cleanup().await.unwrap(); } + +#[tokio::test] +async fn wait_timer_e2e() { + let mut graph = make_graph_with_start_exit("WaitTimerTest"); + let mut wait_node = Node::new("wait60"); + wait_node.attrs.insert( + "shape".to_string(), + AttrValue::String("insulator".to_string()), + ); + wait_node.attrs.insert( + "label".to_string(), + AttrValue::String("Wait 1ms".to_string()), + ); + wait_node.attrs.insert( + "duration".to_string(), + AttrValue::Duration(std::time::Duration::from_millis(1)), + ); + graph.nodes.insert("wait60".to_string(), wait_node); + graph.edges.push(Edge::new("start", "wait60")); + graph.edges.push(Edge::new("wait60", "exit")); + + let dir = tempfile::tempdir().unwrap(); + let interviewer = Arc::new(AutoApproveInterviewer); + let engine = WorkflowRunEngine::new( + make_full_registry(interviewer), + Arc::new(EventEmitter::new()), + local_env(), + ); + let config = RunConfig { + logs_root: dir.path().to_path_buf(), + cancel_token: None, + dry_run: false, + run_id: "test-run".into(), + git_checkpoint: None, + base_sha: None, + run_branch: None, + meta_branch: None, + labels: std::collections::HashMap::new(), + }; + let outcome = engine.run(&graph, &config).await.expect("run"); + assert_eq!(outcome.status, StageStatus::Success); +}