pig2014/honey-biscuit-workshop
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(bake): support daemon, health and async decorator

Browse Source 
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.
This commit is contained in:
Catty Steve
2026-06-11 16:26:44 +08:00
parent 6d9ddbca56
commit 3629c33fc4
10 changed files with 456 additions and 103 deletions
Download Patch File Download Diff File Expand all files Collapse all files
workshop-baker-params/src/bake.rs
+27 -3
View File
@@ -6,19 +6,43 @@ use crate::{ParamVal, Writer, Overridden, traits::MergeParams};
#[serde(deny_unknown_fields)]
pub struct BakeParams {
#[serde(default)] pub workspace: ParamVal<String>,
/// Health probe interval in seconds (default 5).
#[serde(default)] pub health_period: ParamVal<u64>,
/// Health probe per-attempt timeout in seconds (default 10).
#[serde(default)] pub health_timeout: ParamVal<u64>,
}
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<BakeParams> for BakeSettings { fn from(p: BakeParams) -> Self { (&p).into() } }
workshop-baker/src/bake.rs
+72 -15
View File
@@ -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<String> = 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<String, String> = HashMap::new();
let mut async_handles: Vec<tokio::task::JoinHandle<()>> = Vec::new();
while !dag.is_empty() {
let ready: Vec<String> = 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(())
workshop-baker/src/bake/decorator.rs
+91 -2
View File
@@ -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<String>),
/// 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<Option<Decorator>,
}
"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<String> = 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), "@<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
// =====================================================================
workshop-baker/src/bake/error.rs
+4
View File
@@ -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,
}
}
}
workshop-baker/src/bake/execute.rs
+199 -34
View File
@@ -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<String, BakeError> {
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<String, BakeError> {
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<Box<dyn Future<Output = Result<(), BakeError>> + Send + 'a>> {
) -> Pin<Box<dyn Future<Output = Result<String, BakeError>> + 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<String, String>,
) -> Result<Vec<(String, String)>, 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<BakeError> = 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<String, String>,
) -> Result<String, BakeError> {
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(())
}
workshop-baker/src/bake/util.rs
+45
View File
@@ -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<String, String>,
) -> Option<Vec<u8>> {
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())
}
workshop-baker/src/prebake/stage/bootstrap.rs
-47
View File
@@ -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:
workshop-engine/src/executor.rs
+13 -2
View File
@@ -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,
}
}
workshop-engine/src/types.rs
+4
View File
@@ -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<Vec<u8>>,
}
#[derive(Debug, Clone, Default)]
@@ -40,6 +43,7 @@ impl Default for ExecutionContext {
dry_run: false,
privileged: false,
standalone: false,
stdin: None,
}
}
}
workshop-engine/tests/engine_integration.rs
+1
View File
@@ -27,6 +27,7 @@ fn create_context() -> ExecutionContext {
dry_run: false,
privileged: false,
standalone: false,
stdin: None,
}
}