From 3629c33fc4f07562d5a182e73a2d442829c8ce54 Mon Sep 17 00:00:00 2001 From: Catty Steve <4795515+Catty2014@user.noreply.gitee.com> Date: Thu, 11 Jun 2026 16:26:44 +0800 Subject: [PATCH] feat(bake): support daemon, health and async decorator Add `@async` decorator for fire-and-forget tasks, `@pipe` for stdout routing between functions, and `@daemon` with configurable health probes. Functions now resolve execution mode and track stdout cache for inter-function communication. --- workshop-baker-params/src/bake.rs | 30 ++- workshop-baker/src/bake.rs | 87 +++++-- workshop-baker/src/bake/decorator.rs | 93 ++++++- workshop-baker/src/bake/error.rs | 4 + workshop-baker/src/bake/execute.rs | 233 +++++++++++++++--- workshop-baker/src/bake/util.rs | 45 ++++ workshop-baker/src/prebake/stage/bootstrap.rs | 47 ---- workshop-engine/src/executor.rs | 15 +- workshop-engine/src/types.rs | 4 + workshop-engine/tests/engine_integration.rs | 1 + 10 files changed, 456 insertions(+), 103 deletions(-) diff --git a/workshop-baker-params/src/bake.rs b/workshop-baker-params/src/bake.rs index 1eb790d..7bfc61b 100644 --- a/workshop-baker-params/src/bake.rs +++ b/workshop-baker-params/src/bake.rs @@ -6,19 +6,43 @@ use crate::{ParamVal, Writer, Overridden, traits::MergeParams}; #[serde(deny_unknown_fields)] pub struct BakeParams { #[serde(default)] pub workspace: ParamVal, + /// Health probe interval in seconds (default 5). + #[serde(default)] pub health_period: ParamVal, + /// Health probe per-attempt timeout in seconds (default 10). + #[serde(default)] pub health_timeout: ParamVal, } impl MergeParams for BakeParams { - fn defaults() -> Self { Self { workspace: ParamVal::new("/workspace".into()) } } + fn defaults() -> Self { + Self { + workspace: ParamVal::new("/workspace".into()), + health_period: ParamVal::new(5), + health_timeout: ParamVal::new(10), + } + } fn apply(&mut self, i: Self, w: Writer, ov: &mut Overridden, p: &str) { apply_field!(self.workspace, i.workspace, w, ov, p, "workspace"); + apply_field!(self.health_period, i.health_period, w, ov, p, "health_period"); + apply_field!(self.health_timeout, i.health_timeout, w, ov, p, "health_timeout"); } } impl Default for BakeParams { fn default() -> Self { Self::defaults() } } #[derive(Debug, Clone)] -pub struct BakeSettings { pub workspace: String } +pub struct BakeSettings { + pub workspace: String, + pub health_period: u64, + pub health_timeout: u64, +} impl Default for BakeSettings { fn default() -> Self { BakeParams::defaults().into() } } -impl From<&BakeParams> for BakeSettings { fn from(p: &BakeParams) -> Self { Self { workspace: p.workspace.value.clone() } } } +impl From<&BakeParams> for BakeSettings { + fn from(p: &BakeParams) -> Self { + Self { + workspace: p.workspace.value.clone(), + health_period: p.health_period.value, + health_timeout: p.health_timeout.value, + } + } +} impl From for BakeSettings { fn from(p: BakeParams) -> Self { (&p).into() } } diff --git a/workshop-baker/src/bake.rs b/workshop-baker/src/bake.rs index 73bf36d..9eb3b16 100644 --- a/workshop-baker/src/bake.rs +++ b/workshop-baker/src/bake.rs @@ -1,11 +1,15 @@ use crate::bake::builder::RenderMode; use crate::bake::constant::DEFAULT_WORKSPACE; +use crate::bake::decorator::Decorator; use crate::bake::error::BakeError; -use crate::bake::util::is_parallel; +use crate::bake::util::{is_parallel, resolve_mode, FuncMode}; use crate::cli::Cli; use crate::prebake; use crate::prebake::PrebakeConfig; +use std::collections::{HashMap, HashSet}; use std::path::{Path, PathBuf}; +use std::time::Duration; +use workshop_baker_params::BakeSettings; use workshop_engine::{Engine, EventSender, ExecutionContext}; mod builder; @@ -45,6 +49,8 @@ pub async fn bake( let render_mode = RenderMode::Full; let engine = Engine::new(); + // TODO: resolve from merged param layers (PIP integration) + let bake_settings = BakeSettings::default(); if let Some(env) = &prebake.envvars && let Some(prebake_env) = &env.bake { @@ -66,6 +72,28 @@ pub async fn bake( let remaining = functions.remaining_code; let (mut dag, func_map) = schedule::build_dag(functions.pipeline_functions)?; + // Pre-scan: pipe sources + decorator conflict checks. + let mut pipe_sources: HashSet = HashSet::new(); + for func in func_map.values() { + decorator::check_conflicts(func)?; + + // Some simple optimization for @pipe + // Collect only output of specified function + // and exclude self-reference because it's handled by loop + for decorator in &func.decorators { + if let Decorator::Pipe(names) = decorator { + for source in names { + if *source != func.name { + pipe_sources.insert(source.clone()); + } + } + } + } + } + + let mut stdout_cache: HashMap = HashMap::new(); + let mut async_handles: Vec> = Vec::new(); + while !dag.is_empty() { let ready: Vec = dag.peek_all().into_iter().cloned().collect(); if ready.is_empty() { @@ -75,34 +103,63 @@ pub async fn bake( // Rule 1: sequential always goes first if let Some(seq_name) = ready.iter().find(|n| !is_parallel(&func_map[n.as_str()])).cloned() { - execute::execute_one( - &engine, - &func_map[seq_name.as_str()], - ctx, - &event_tx, - &workspace, - &remaining, - bake_base_path, - ) - .await?; - dag.pop(); + let func = &func_map[seq_name.as_str()]; + let mode = resolve_mode(func); + + match mode { + FuncMode::Daemon => { + dag.pop(); + execute::execute_daemon( + &engine, func, ctx, &event_tx, + &workspace, &remaining, bake_base_path, + Duration::from_secs(bake_settings.health_period), + Duration::from_secs(bake_settings.health_timeout), + ).await?; + } + FuncMode::Async => { + dag.pop(); + let handle = execute::execute_async( + func, ctx, &event_tx, + &workspace, &remaining, bake_base_path, + ); + async_handles.push(handle); + } + FuncMode::Normal => { + let stdout = execute::execute_normal( + &engine, func, ctx, &event_tx, + &workspace, &remaining, bake_base_path, &stdout_cache, + ).await?; + if pipe_sources.contains(&seq_name) { + stdout_cache.insert(seq_name.clone(), stdout); + } + dag.pop(); + } + } continue; } // All remaining are parallel — group and pick a batch let (star_nodes, groups) = schedule::partition_by_group(&ready, &func_map); if let Some(batch) = schedule::pick_batch(&star_nodes, &groups) { - execute::fire_parallel_batch( + let results = execute::fire_parallel_batch( &batch, &engine, &func_map, ctx, &event_tx, - &workspace, &remaining, bake_base_path, + &workspace, &remaining, bake_base_path, &stdout_cache, ).await?; - for _name in &batch { + for (name, stdout) in results { + if pipe_sources.contains(&name) { + stdout_cache.insert(name, stdout); + } dag.pop(); } } // No complete group and no sequential — nothing fireable } + + // Wait for all @async tasks to finish + for handle in async_handles { + let _ = handle.await; + } } Ok(()) diff --git a/workshop-baker/src/bake/decorator.rs b/workshop-baker/src/bake/decorator.rs index 656f93e..f22ba82 100644 --- a/workshop-baker/src/bake/decorator.rs +++ b/workshop-baker/src/bake/decorator.rs @@ -4,6 +4,7 @@ use std::fmt::Display; use std::fmt::Formatter; use crate::bake::error::BakeError; +use crate::bake::parser::Function; lazy_static! { // "# @decorator" or "# @decorator(parameters)" @@ -43,8 +44,10 @@ pub enum Decorator { Parallel(Option), /// Run as a background daemon process. Daemon, - /// Health check endpoint for daemon verification. + /// Health check command for daemon verification (shell statement). Health(String), + /// Fire-and-forget execution; DAG pop happens at start, not end. + Async, } /// Returns true if the decorator is a flag (position-independent, consumed after execution). @@ -58,6 +61,7 @@ pub fn is_flag(d: &Decorator) -> bool { | Decorator::After(_) | Decorator::Parallel(_) | Decorator::Health(_) + | Decorator::Async | Decorator::Unknown ) } @@ -69,7 +73,7 @@ pub fn is_combinator(d: &Decorator) -> bool { /// Returns true if the decorator is a mode (at most one, alters what gets executed). pub fn is_mode(d: &Decorator) -> bool { - matches!(d, Decorator::Pipe(_) | Decorator::Daemon) + matches!(d, Decorator::Pipe(_) | Decorator::Daemon | Decorator::Async) } /// Extract the parallel group name. Returns "*" for wildcard (no group specified). @@ -80,6 +84,78 @@ pub fn parallel_group(d: &Decorator) -> &str { } } +/// Check for incompatible decorator combinations on a function. +/// Returns `Err(IncompatibleDecorators)` on conflict, `Ok(())` otherwise. +pub fn check_conflicts(func: &Function) -> Result<(), BakeError> { + let has_daemon = func.decorators.iter().any(|d| matches!(d, Decorator::Daemon)); + let has_async = func.decorators.iter().any(|d| matches!(d, Decorator::Async)); + let has_pipe = func.decorators.iter().any(|d| matches!(d, Decorator::Pipe(_))); + let has_parallel = func.decorators.iter().any(|d| matches!(d, Decorator::Parallel(_))); + let has_health = func.decorators.iter().any(|d| matches!(d, Decorator::Health(_))); + let has_export = func.decorators.iter().any(|d| matches!(d, Decorator::Export)); + let has_combinator = func.decorators.iter().any(|d| is_combinator(d)); + + let mode_count = [has_daemon, has_async, has_pipe].iter().filter(|&&x| x).count(); + if mode_count > 1 { + return Err(BakeError::IncompatibleDecorators { + function: func.name.clone(), + reason: "@daemon, @async, and @pipe are mutually exclusive".into(), + }); + } + + if has_daemon { + if has_parallel { + return Err(BakeError::IncompatibleDecorators { + function: func.name.clone(), + reason: "@daemon cannot be combined with @parallel".into(), + }); + } + if has_combinator { + return Err(BakeError::IncompatibleDecorators { + function: func.name.clone(), + reason: "@daemon cannot be combined with @loop/@retry/@timeout".into(), + }); + } + if has_export { + return Err(BakeError::IncompatibleDecorators { + function: func.name.clone(), + reason: "@daemon cannot be combined with @export".into(), + }); + } + } + + if has_health && !has_daemon { + return Err(BakeError::IncompatibleDecorators { + function: func.name.clone(), + reason: "@health requires @daemon".into(), + }); + } + + if has_async { + if has_parallel { + return Err(BakeError::IncompatibleDecorators { + function: func.name.clone(), + reason: "@async cannot be combined with @parallel".into(), + }); + } + if has_export { + return Err(BakeError::IncompatibleDecorators { + function: func.name.clone(), + reason: "@async cannot be combined with @export".into(), + }); + } + } + + if has_pipe && has_parallel { + return Err(BakeError::IncompatibleDecorators { + function: func.name.clone(), + reason: "@pipe cannot be combined with @parallel".into(), + }); + } + + Ok(()) +} + impl Display for Decorator { fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result { match self { @@ -99,6 +175,7 @@ impl Display for Decorator { }, Decorator::Daemon => write!(f, "@daemon"), Decorator::Health(check) => write!(f, "@health({})", check), + Decorator::Async => write!(f, "@async"), } } } @@ -208,6 +285,7 @@ pub fn parse_decorator(line: &str, line_num: usize) -> Result, } "daemon" => decorator_noarg(Decorator::Daemon), "health" => required_arg().map(|a| Some(Decorator::Health(a.to_string()))), + "async" => decorator_noarg(Decorator::Async), "pipe" => required_arg().and_then(|a| { let functions: Vec = a .split(',') @@ -358,6 +436,11 @@ mod tests { ); } + #[test] + fn test_decorator_display_async() { + assert_eq!(format!("{}", Decorator::Async), "@async"); + } + #[test] fn test_decorator_display_unknown() { assert_eq!(format!("{}", Decorator::Unknown), "@"); @@ -403,6 +486,12 @@ mod tests { assert_eq!(result, Some(Decorator::Daemon)); } + #[test] + fn test_parse_decorator_async() { + let result = parse_decorator("# @async", 1).unwrap(); + assert_eq!(result, Some(Decorator::Async)); + } + // ===================================================================== // Tests for parse_decorator - Valid single-argument decorators // ===================================================================== diff --git a/workshop-baker/src/bake/error.rs b/workshop-baker/src/bake/error.rs index 1aad016..719be35 100644 --- a/workshop-baker/src/bake/error.rs +++ b/workshop-baker/src/bake/error.rs @@ -37,6 +37,9 @@ pub enum BakeError { #[error("Incompatible decorators on function '{function}': {reason}")] IncompatibleDecorators { function: String, reason: String }, + + #[error("Health probe failed for daemon '{command}': {error}")] + HealthProbeFailed { command: String, error: String }, } impl HasExitCode for BakeError { @@ -51,6 +54,7 @@ impl HasExitCode for BakeError { BakeError::DecoratorParseError { .. } => EXITCODE_PARSE_ERROR, BakeError::ScriptExecutionError(e) => e.exit_code(), BakeError::IncompatibleDecorators { .. } => EXITCODE_PARSE_ERROR, + BakeError::HealthProbeFailed { .. } => EXITCODE_IO_ERROR, } } } diff --git a/workshop-baker/src/bake/execute.rs b/workshop-baker/src/bake/execute.rs index 309e68b..cdf9f42 100644 --- a/workshop-baker/src/bake/execute.rs +++ b/workshop-baker/src/bake/execute.rs @@ -2,15 +2,17 @@ use crate::bake::builder::{self, RenderMode}; use crate::bake::decorator::{self, Decorator}; use crate::bake::error::BakeError; use crate::bake::parser::Function; -use crate::bake::util::is_fallible; +use crate::bake::util::{is_fallible, resolve_pipe_stdin}; use std::collections::HashMap; use std::future::Future; use std::path::{Path, PathBuf}; use std::pin::Pin; +use std::process::Stdio; use std::time::Duration; use workshop_engine::{Engine, EventSender, ExecutionContext, ExecutionError}; /// Execute a single function with combinator wrapping. +/// Returns the captured stdout (for `@pipe` consumers). pub async fn execute_one( engine: &Engine, func: &Function, @@ -19,7 +21,7 @@ pub async fn execute_one( workspace: &Path, remaining: &str, bake_base: &Path, -) -> Result<(), BakeError> { +) -> Result { let combos: Vec<&Decorator> = func .decorators .iter() @@ -38,40 +40,49 @@ pub async fn execute_one( async fn execute_with_combos( engine: &Engine, script: &PathBuf, - ctx: &ExecutionContext, + ctx: &mut ExecutionContext, event_tx: &EventSender, combos: &[&Decorator], -) -> Result<(), BakeError> { +) -> Result { build_combinator_chain(engine, script, ctx, event_tx, combos, 0).await } -/// Builds a combinator execution chain using `Box::pin` to avoid -/// infinite-sized async fn recursion. +/// Builds a combinator execution chain. Returns the captured stdout. +/// For `@loop` + `@pipe(self)`, each iteration feeds its stdout back as +/// the next iteration's stdin via `ctx.stdin`. fn build_combinator_chain<'a>( engine: &'a Engine, script: &'a PathBuf, - ctx: &'a ExecutionContext, + ctx: &'a mut ExecutionContext, event_tx: &'a EventSender, combos: &'a [&'a Decorator], idx: usize, -) -> Pin> + Send + 'a>> { +) -> Pin> + Send + 'a>> { if idx >= combos.len() { return Box::pin(async move { - engine.execute_script(script, ctx, event_tx).await?; - Ok(()) + let result = engine + .execute_script(script, ctx, event_tx) + .await + .map_err(|e| BakeError::ScriptExecutionError(e))?; + Ok(result.stdout) }); } match combos[idx] { Decorator::Loop(count) => Box::pin(async move { + let mut last_stdout = String::new(); for i in 0..*count { log::debug!("Loop {}/{}", i + 1, count); - build_combinator_chain( + last_stdout = build_combinator_chain( engine, script, ctx, event_tx, combos, idx + 1, ) .await?; + // Feed stdout as stdin for the next iteration (@pipe(self)) + if i + 1 < *count { + ctx.stdin = Some(last_stdout.clone().into_bytes()); + } } - Ok(()) + Ok(last_stdout) }), Decorator::Timeout(secs) => Box::pin(async move { tokio::time::timeout( @@ -81,9 +92,7 @@ fn build_combinator_chain<'a>( ), ) .await - .map_err(|_| { - BakeError::ScriptExecutionError(ExecutionError::Timeout) - })? + .map_err(|_| BakeError::ScriptExecutionError(ExecutionError::Timeout))? }), Decorator::Retry(retries, delay_secs) => Box::pin(async move { let max_attempts = *retries as usize + 1; @@ -95,7 +104,7 @@ fn build_combinator_chain<'a>( ) .await { - Ok(()) => return Ok(()), + Ok(stdout) => return Ok(stdout), Err(e) => { if matches!( &e, @@ -128,9 +137,12 @@ fn build_combinator_chain<'a>( /// Fire a batch of parallel nodes concurrently. /// -/// All nodes in the batch are waited on before returning — matching the -/// GitHub/GitLab/Concourse convention of collecting full failure information. -/// A non-`@fallible` failure is surfaced only after all nodes complete. +/// All nodes in the batch are waited on before returning. A non-`@fallible` +/// failure is surfaced only after all nodes complete (GitHub/GitLab/Concourse +/// convention). +/// +/// Returns a `(name, stdout)` pair for each completed node so the caller can +/// populate the stdout cache for `@pipe`. pub async fn fire_parallel_batch( batch: &[String], _engine: &Engine, @@ -140,12 +152,15 @@ pub async fn fire_parallel_batch( workspace: &Path, remaining: &str, bake_base: &Path, -) -> Result<(), BakeError> { + stdout_cache: &HashMap, +) -> Result, BakeError> { let handles: Vec<_> = batch .iter() .map(|name| { let function = func_map[name.as_str()].clone(); let mut task_ctx = ctx.clone(); + // Resolve per-task stdin from @pipe + stdout_cache + task_ctx.stdin = resolve_pipe_stdin(&function, stdout_cache); let task_events = event_tx.clone(); let task_workspace = workspace.to_path_buf(); let task_remaining = remaining.to_string(); @@ -153,24 +168,23 @@ pub async fn fire_parallel_batch( let task_name = name.clone(); tokio::spawn(async move { let task_engine = Engine::new(); - ( - task_name, - execute_one( - &task_engine, - &function, - &mut task_ctx, - &task_events, - &task_workspace, - &task_remaining, - &task_bake_base, - ) - .await, + let result = execute_one( + &task_engine, + &function, + &mut task_ctx, + &task_events, + &task_workspace, + &task_remaining, + &task_bake_base, ) + .await; + (task_name, result) }) }) .collect(); let mut first_fatal: Option = None; + let mut results: Vec<(String, String)> = Vec::with_capacity(handles.len()); for handle in handles { let (name, result) = handle.await.map_err(|_| { BakeError::ScriptExecutionError(ExecutionError::ExecutionFailed( @@ -178,8 +192,8 @@ pub async fn fire_parallel_batch( )) })?; match result { - Ok(()) => {} - Err(ref err) if is_fallible(&func_map[&name]) => { + Ok(stdout) => results.push((name, stdout)), + Err(err) if is_fallible(&func_map[&name]) => { log::warn!("'{}' failed (fallible, non-fatal): {}", name, err); } Err(err) => { @@ -192,5 +206,156 @@ pub async fn fire_parallel_batch( if let Some(err) = first_fatal { return Err(err); } + Ok(results) +} + +/// Spawn an `@async` function in the background. +/// Returns a `JoinHandle` that the caller must await before returning. +pub fn execute_async( + func: &Function, + ctx: &ExecutionContext, + event_tx: &EventSender, + workspace: &Path, + remaining: &str, + bake_base: &Path, +) -> tokio::task::JoinHandle<()> { + let mut task_ctx = ctx.clone(); + let task_events = event_tx.clone(); + let task_workspace = workspace.to_path_buf(); + let task_remaining = remaining.to_string(); + let task_bake_base = bake_base.to_path_buf(); + let task_func = func.clone(); + tokio::spawn(async move { + let task_engine = Engine::new(); + let _ = execute_one( + &task_engine, &task_func, &mut task_ctx, + &task_events, &task_workspace, + &task_remaining, &task_bake_base, + ).await; + }) +} + +/// Execute a Normal-mode function, resolving pipe stdin first. +pub async fn execute_normal( + engine: &Engine, + func: &Function, + ctx: &mut ExecutionContext, + event_tx: &EventSender, + workspace: &Path, + remaining: &str, + bake_base: &Path, + stdout_cache: &HashMap, +) -> Result { + ctx.stdin = resolve_pipe_stdin(func, stdout_cache); + execute_one(engine, func, ctx, event_tx, workspace, remaining, bake_base).await +} + +// ── Daemon / Health ───────────────────────────────────────────────────────── + +/// Execute a `@daemon` function: build script, spawn, periodic health-probe +/// (blocking until healthy), then return. +pub async fn execute_daemon( + engine: &Engine, + func: &Function, + ctx: &ExecutionContext, + event_tx: &EventSender, + workspace: &Path, + remaining: &str, + bake_base: &Path, + health_period: Duration, + health_timeout: Duration, +) -> Result<(), BakeError> { + let fallible = is_fallible(func); + + let mut modified = func.clone(); + modified.body = remaining.to_string() + &func.body; + let script = builder::build_script( + &modified, bake_base, workspace, None, RenderMode::Full, + )?; + + // Spawn without waiting + let mut command = tokio::process::Command::new(ctx.shell.as_str()); + command.arg("-c"); + command.arg(script.to_str().unwrap()); + command.current_dir(&ctx.working_dir); + command.envs(&ctx.env_vars); + command.stdin(Stdio::null()); + command.stdout(Stdio::null()); + command.stderr(Stdio::null()); + #[cfg(unix)] + command.process_group(0); + + //let child = // Necessary? + match command.spawn() { + Ok(c) => c, + Err(e) if fallible => { + log::warn!("Failed to spawn daemon '{}' (fallible): {}", func.name, e); + return Ok(()); + } + Err(e) => { + return Err(BakeError::ScriptExecutionError(ExecutionError::ExecutionFailed( + format!("Failed to spawn daemon '{}': {}", func.name, e), + ))); + } + }; + + // Periodic health probe — block until healthy or fatal error + if let Some(health_cmd) = func.decorators.iter().find_map(|d| match d { + Decorator::Health(cmd) => Some(cmd.as_str()), + _ => None, + }) { + loop { + let probe_result = tokio::time::timeout( + health_timeout, + health_probe_one(engine, health_cmd, remaining, ctx, event_tx), + ) + .await; + + match probe_result { + Ok(Ok(())) => break, + _ if fallible => { + log::warn!( + "Health probe for '{}' failed, retrying in {:?}...", + func.name, health_period + ); + } + Ok(Err(e)) => return Err(e), + Err(_) => { + return Err(BakeError::HealthProbeFailed { + command: health_cmd.to_string(), + error: "timeout".to_string(), + }); + } + } + tokio::time::sleep(health_period).await; + } + } + + // Cleanup is handled by finalize. + Ok(()) +} + +/// Single health-probe attempt. Exit 0 = healthy. +async fn health_probe_one( + engine: &Engine, + cmd: &str, + remaining: &str, + ctx: &ExecutionContext, + event_tx: &EventSender, +) -> Result<(), BakeError> { + let full_cmd = format!("{}\n{}", remaining, cmd); + let result = engine + .execute_command(&full_cmd, ctx, event_tx) + .await + .map_err(|e| BakeError::HealthProbeFailed { + command: cmd.to_string(), + error: format!("{}", e), + })?; + if !result.success { + return Err(BakeError::HealthProbeFailed { + command: cmd.to_string(), + error: format!("exit code {}", result.exit_code), + }); + } Ok(()) } diff --git a/workshop-baker/src/bake/util.rs b/workshop-baker/src/bake/util.rs index e5b0323..a869607 100644 --- a/workshop-baker/src/bake/util.rs +++ b/workshop-baker/src/bake/util.rs @@ -1,5 +1,27 @@ use crate::bake::decorator::Decorator; use crate::bake::parser::Function; +use std::collections::HashMap; + +/// Execution mode for a pipeline function. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum FuncMode { + Normal, + Daemon, + Async, +} + +/// Determine the execution mode from decorators. +/// `@daemon`, `@async`, `@pipe` are mutually exclusive modes. +pub fn resolve_mode(func: &Function) -> FuncMode { + for d in &func.decorators { + match d { + Decorator::Daemon => return FuncMode::Daemon, + Decorator::Async => return FuncMode::Async, + _ => {} + } + } + FuncMode::Normal +} /// True if the function is decorated with `@parallel`. pub fn is_parallel(func: &Function) -> bool { @@ -14,3 +36,26 @@ pub fn is_fallible(func: &Function) -> bool { .iter() .any(|d| matches!(d, Decorator::Fallible)) } + +/// Resolve stdin for a `@pipe`-decorated function from the stdout cache. +/// +/// Rule 1: self-references are skipped (handled by loop combinator). +/// Rule 3: sources present in the cache are concatenated. +/// Returns `None` if any source is missing (caller decides based on `@fallible`). +pub fn resolve_pipe_stdin( + func: &Function, + cache: &HashMap, +) -> Option> { + let sources = func.decorators.iter().find_map(|d| match d { + Decorator::Pipe(names) => Some(names.as_slice()), + _ => None, + })?; + let mut combined = String::new(); + for source in sources { + if *source == func.name { + continue; // Rule 1: skip self + } + combined.push_str(cache.get(source)?); // Rule 3 + } + Some(combined.into_bytes()) +} diff --git a/workshop-baker/src/prebake/stage/bootstrap.rs b/workshop-baker/src/prebake/stage/bootstrap.rs index 31b945b..aec4778 100644 --- a/workshop-baker/src/prebake/stage/bootstrap.rs +++ b/workshop-baker/src/prebake/stage/bootstrap.rs @@ -133,53 +133,6 @@ fn write_bootstrap_content(bootstrap_path: &Path, content: String) -> Result<(), Ok(()) } -/// Source type for a custom bootstrap script. -/// -/// Mirrors the scheme-based resolution in `finalize::plugin::fetch` URL parsing, -/// and supports the schemes defined in the prebake.yml.tmpl spec: -/// `workspace://`, `file://`, `server://`, `https?://`, plus inline content. -enum BootstrapSource { - /// Inline script content (no scheme match). - Inline(String), - /// Project-relative file: `workspace://path/to/bootstrap.sh` - Workspace(String), - /// Script provided by bake server: `server://script-name` - Server(String), - /// HTTP(S) download: `https://example.com/bootstrap.sh` - Http(String), - /// Local file path: `file:///bin/bootstrap.sh` - File(String), -} - -impl BootstrapSource { - /// Parses a raw custom bootstrap string into its source type. - /// - /// Scheme detection follows the template spec: - /// - `workspace://` → Workspace - /// - `file://` → File (local path) - /// - `server://` → Server (bakerd registry) - /// - `http://`/`https://`→ Http - /// - No recognized scheme → Inline - fn parse(raw: &str) -> Self { - if let Some(rest) = raw.strip_prefix("workspace://") { - return BootstrapSource::Workspace(rest.to_string()); - } - if let Some(path) = raw.strip_prefix("file://") { - return BootstrapSource::File(path.to_string()); - } - if let Some(name) = raw.strip_prefix("server://") { - return BootstrapSource::Server(name.to_string()); - } - if let Some(url) = raw - .strip_prefix("https://") - .or_else(|| raw.strip_prefix("http://")) - { - return BootstrapSource::Http(url.to_string()); - } - BootstrapSource::Inline(raw.to_string()) - } -} - /// Generates a bootstrap script based on configuration and target OS. /// /// The generated script performs the following setup tasks: diff --git a/workshop-engine/src/executor.rs b/workshop-engine/src/executor.rs index b27b0e3..e7a781e 100644 --- a/workshop-engine/src/executor.rs +++ b/workshop-engine/src/executor.rs @@ -2,6 +2,7 @@ use chrono::Utc; use std::path::PathBuf; use std::process::Stdio; use tokio::io::AsyncReadExt; +use tokio::io::AsyncWriteExt; use tokio::process::Command; use tokio::sync::mpsc; use tokio::time::timeout; @@ -87,7 +88,7 @@ impl crate::types::Engine { .env("HBW_TASKID", &ctx.task_id) .env("HBW_PIPELINE", &ctx.pipeline_name) .env("HBW_USERNAME", &ctx.username) - .stdin(Stdio::null()) + .stdin(if ctx.stdin.is_some() { Stdio::piped() } else { Stdio::null() }) .stdout(Stdio::piped()) .stderr(Stdio::piped()); } @@ -121,10 +122,19 @@ impl crate::types::Engine { #[cfg(unix)] command.process_group(0); - let child = command.spawn().map_err(|e| { + let mut child = command.spawn().map_err(|e| { ExecutionError::InvalidCommand(format!("Failed to spawn command: {}", e)) })?; + // Write piped stdin (e.g. from @pipe). Consumed once per execution. + if let Some(ref data) = ctx.stdin { + if let Some(mut stdin_pipe) = child.stdin.take() { + stdin_pipe.write_all(data).await.map_err(|e| { + ExecutionError::IoError(format!("Failed to write stdin: {}", e)) + })?; + } + } + self.execute_child(TokioChild(child), ctx, event_tx).await } @@ -369,6 +379,7 @@ mod tests { dry_run: false, privileged: false, standalone: false, + stdin: None, } } diff --git a/workshop-engine/src/types.rs b/workshop-engine/src/types.rs index c1a77df..cc224d0 100644 --- a/workshop-engine/src/types.rs +++ b/workshop-engine/src/types.rs @@ -18,6 +18,9 @@ pub struct ExecutionContext { pub dry_run: bool, pub privileged: bool, pub standalone: bool, + /// stdin data piped by the scheduler (e.g. via `@pipe`). + /// Consumed (taken) by engine during execution. + pub stdin: Option>, } #[derive(Debug, Clone, Default)] @@ -40,6 +43,7 @@ impl Default for ExecutionContext { dry_run: false, privileged: false, standalone: false, + stdin: None, } } } diff --git a/workshop-engine/tests/engine_integration.rs b/workshop-engine/tests/engine_integration.rs index d6aba7e..813041c 100644 --- a/workshop-engine/tests/engine_integration.rs +++ b/workshop-engine/tests/engine_integration.rs @@ -27,6 +27,7 @@ fn create_context() -> ExecutionContext { dry_run: false, privileged: false, standalone: false, + stdin: None, } }