diff --git a/workshop-baker-params/src/notification.rs b/workshop-baker-params/src/notification.rs index 07f0487..62706ce 100644 --- a/workshop-baker-params/src/notification.rs +++ b/workshop-baker-params/src/notification.rs @@ -23,7 +23,7 @@ impl MergeParams for NotificationParams { retry_backoff: ParamVal::new(2.0), retry_delay_seconds: ParamVal::new(5.0), max_retry_delay_seconds: ParamVal::new(300.0), life_impact_factor: ParamVal::new(0.0), - batch_period: ParamVal::new(5), batch_watermark: ParamVal::new(5), + batch_period: ParamVal::new(5), batch_watermark: ParamVal::new(1), template_ref_depth: ParamVal::new(10), } } diff --git a/workshop-baker/Cargo.lock b/workshop-baker/Cargo.lock index 1336102..269854a 100644 --- a/workshop-baker/Cargo.lock +++ b/workshop-baker/Cargo.lock @@ -2,6 +2,12 @@ # It is not intended for manual editing. version = 4 +[[package]] +name = "adler2" +version = "2.0.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "320119579fcad9c21884f5c4861d16174d0e06250625266f50fe6898340abefa" + [[package]] name = "aho-corasick" version = "1.1.4" @@ -545,6 +551,17 @@ version = "1.11.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1e748733b7cbc798e1434b6ac524f0c1ff2ab456fe201501e6497c8417a4fc33" +[[package]] +name = "capctl" +version = "0.2.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4a6e71767585f51c2a33fed6d67147ec0343725fc3c03bf4b89fe67fede56aa5" +dependencies = [ + "bitflags 1.3.2", + "cfg-if", + "libc", +] + [[package]] name = "cast" version = "0.3.0" @@ -793,6 +810,15 @@ dependencies = [ "libc", ] +[[package]] +name = "crc32fast" +version = "1.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9481c1c90cbf2ac953f07c8d4a58aa3945c425b7185c9154d67a65e4230da511" +dependencies = [ + "cfg-if", +] + [[package]] name = "criterion" version = "0.5.1" @@ -1124,6 +1150,16 @@ version = "0.5.7" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1d674e81391d1e1ab681a28d99df07927c6d4aa5b027d7da16ba32d1d21ecd99" +[[package]] +name = "flate2" +version = "1.1.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "843fba2746e448b37e26a819579957415c8cef339bf08564fe8b7ddbd959573c" +dependencies = [ + "crc32fast", + "miniz_oxide", +] + [[package]] name = "fnv" version = "1.0.7" @@ -1969,6 +2005,16 @@ dependencies = [ "serde", ] +[[package]] +name = "miniz_oxide" +version = "0.8.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1fa76a2c86f704bdb222d66965fb3d63269ce38518b83cb0575fca855ebb6316" +dependencies = [ + "adler2", + "simd-adler32", +] + [[package]] name = "mio" version = "1.2.1" @@ -2463,6 +2509,12 @@ dependencies = [ "futures-io", ] +[[package]] +name = "pkg-config" +version = "0.3.33" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "19f132c84eca552bf34cab8ec81f1c1dcc229b811638f9d283dceabe58c5569e" + [[package]] name = "plain" version = "0.2.3" @@ -3264,6 +3316,12 @@ dependencies = [ "libc", ] +[[package]] +name = "simd-adler32" +version = "0.3.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "703d5c7ef118737c72f1af64ad2f6f8c5e1921f818cdcb97b8fe6fc69bf66214" + [[package]] name = "simd_cesu8" version = "1.1.1" @@ -4264,7 +4322,9 @@ dependencies = [ "anyhow", "async-trait", "axum 0.7.9", + "bitflags 2.10.0", "bollard", + "capctl", "cgroups-rs", "chrono", "clap", @@ -4272,6 +4332,7 @@ dependencies = [ "criterion", "duration-str", "env_logger", + "flate2", "futures-util", "glob", "globset", @@ -4292,12 +4353,14 @@ dependencies = [ "tempfile", "thiserror 2.0.18", "tokio", + "url", "uuid", "which", "workshop-baker-params", "workshop-engine", "workshop-getterurl", "workshop-schedule", + "zstd", ] [[package]] @@ -4545,6 +4608,34 @@ version = "1.0.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e9747e91771f56fd7893e1164abd78febd14a670ceec257caad15e051de35f06" +[[package]] +name = "zstd" +version = "0.13.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e91ee311a569c327171651566e07972200e76fcfe2242a4fa446149a3881c08a" +dependencies = [ + "zstd-safe", +] + +[[package]] +name = "zstd-safe" +version = "7.2.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8f49c4d5f0abb602a93fb8736af2a4f4dd9512e36f7f570d66e65ff867ed3b9d" +dependencies = [ + "zstd-sys", +] + +[[package]] +name = "zstd-sys" +version = "2.0.16+zstd.1.5.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "91e19ebc2adc8f83e43039e79776e3fda8ca919132d68a1fed6a5faca2683748" +dependencies = [ + "cc", + "pkg-config", +] + [[package]] name = "zvariant" version = "5.8.0" diff --git a/workshop-baker/Cargo.toml b/workshop-baker/Cargo.toml index 3f610c3..4534ffe 100644 --- a/workshop-baker/Cargo.toml +++ b/workshop-baker/Cargo.toml @@ -53,6 +53,11 @@ axum = "0.7" workshop-getterurl = { path = "../workshop-getterurl" } workshop-baker-params = { version = "0.1.0", path = "../workshop-baker-params" } +bitflags = "2" +url = "2.5.8" +zstd = "0.13.3" +flate2 = "1.1.9" +capctl = "0.2.4" [dev-dependencies] criterion = "0.5" diff --git a/workshop-baker/src/bake.rs b/workshop-baker/src/bake.rs index 44c6818..4cf7c43 100644 --- a/workshop-baker/src/bake.rs +++ b/workshop-baker/src/bake.rs @@ -1,3 +1,4 @@ +use crate::types::debug::DebugFeature; use crate::bake::builder::RenderMode; use crate::bake::constant::DEFAULT_WORKSPACE; use crate::bake::decorator::Decorator; @@ -29,6 +30,7 @@ pub async fn bake( _cli: &Cli, ctx: &mut ExecutionContext, event_tx: EventSender, + nevent_tx: Option, ) -> Result<(), BakeError> { log::debug!("Reading script: {:?}", script_path); let script_content = std::fs::read_to_string(script_path).map_err(|e| { @@ -98,12 +100,22 @@ pub async fn bake( let mut async_handles: Vec> = Vec::new(); while !dag.is_empty() { - log::debug!("[BAKE_SCHED] ready={} seq={} wild={} groups={:?}", - dag.ready_count(), - dag.ready_sequential().len(), - dag.ready_wildcards().len(), - dag.ready_groups(), - ); + let scheduler_debug = DebugFeature::from_bits_retain(ctx.debug_flags).contains(DebugFeature::Scheduler); + if scheduler_debug { + log::info!("[BAKE_SCHED] ready={} seq={} wild={} groups={:?}", + dag.ready_count(), + dag.ready_sequential().len(), + dag.ready_wildcards().len(), + dag.ready_groups(), + ); + } else { + log::debug!("[BAKE_SCHED] ready={} seq={} wild={} groups={:?}", + dag.ready_count(), + dag.ready_sequential().len(), + dag.ready_wildcards().len(), + dag.ready_groups(), + ); + } // Rule 1: sequential always goes first let seq_candidates = dag.ready_sequential(); if let Some(seq_name) = seq_candidates.first() { @@ -206,6 +218,8 @@ pub async fn bake( } } + crate::notify::types::send_stage_event(&nevent_tx, crate::types::buildstatus::StagePhase::Bake, "bake", crate::types::buildstatus::StageOutcome::Success).await; + Ok(()) } diff --git a/workshop-baker/src/bake/builder.rs b/workshop-baker/src/bake/builder.rs index 37c4f7d..1d82c55 100644 --- a/workshop-baker/src/bake/builder.rs +++ b/workshop-baker/src/bake/builder.rs @@ -1,3 +1,4 @@ +use crate::types::debug::DebugFeature; use crate::bake::constant::PIPELINE_SKIP_ERRORCODE; use crate::bake::decorator::Decorator; use crate::bake::error::BakeError; @@ -44,6 +45,7 @@ pub fn build_script( bake_base: &Path, target_dir: &Path, render_mode: RenderMode, + debug_flags: u32, ) -> Result { let name = &function.name; let body = &function.body; @@ -62,7 +64,12 @@ pub fn build_script( let template_content = match render_mode { RenderMode::Off => TRIVIAL_TEMPLATE.to_string(), RenderMode::Minimal | RenderMode::Full => { - log::debug!("Reading bake_base: {:?}", bake_base); + let script_debug = DebugFeature::from_bits_retain(debug_flags).contains(DebugFeature::Script); + if script_debug { + log::info!("Reading bake_base: {:?}", bake_base); + } else { + log::debug!("Reading bake_base: {:?}", bake_base); + } std::fs::read_to_string(bake_base).map_err(|e| BakeError::IoError { path: bake_base.display().to_string(), source: e, @@ -146,7 +153,7 @@ mod tests { body: "echo 'hello world'".to_string(), }; let result = build_script( - &function, &bake_base, temp_dir.path(), RenderMode::Full, + &function, &bake_base, temp_dir.path(), RenderMode::Full, 0, ); assert!(result.is_ok()); @@ -169,6 +176,7 @@ mod tests { Path::new("/nonexistent"), temp_dir.path(), RenderMode::Off, + 0, ); assert!(result.is_ok()); diff --git a/workshop-baker/src/bake/execute.rs b/workshop-baker/src/bake/execute.rs index c2869a6..a4a6ebb 100644 --- a/workshop-baker/src/bake/execute.rs +++ b/workshop-baker/src/bake/execute.rs @@ -1,3 +1,4 @@ +use crate::types::debug::DebugFeature; use crate::bake::builder::{self, RenderMode}; use crate::bake::constant::PIPELINE_SKIP_ERRORCODE; use crate::bake::decorator::{self, Decorator}; @@ -32,7 +33,7 @@ pub async fn execute_one( let mut modified = func.clone(); modified.body = remaining.to_string() + &func.body; let script = builder::build_script( - &modified, bake_base, workspace, RenderMode::Full, + &modified, bake_base, workspace, RenderMode::Full, ctx.debug_flags, )?; execute_with_combos(engine, &script, ctx, event_tx, &combos).await @@ -79,7 +80,12 @@ fn build_combinator_chain<'a>( Decorator::Loop(count) => Box::pin(async move { let mut last_stdout = String::new(); for i in 0..*count { - log::debug!("Loop {}/{}", i + 1, count); + let script_debug = DebugFeature::from_bits_retain(ctx.debug_flags).contains(DebugFeature::Script); + if script_debug { + log::info!("Loop {}/{}", i + 1, count); + } else { + log::debug!("Loop {}/{}", i + 1, count); + } last_stdout = build_combinator_chain( engine, script, ctx, event_tx, combos, idx + 1, ) @@ -279,7 +285,7 @@ pub async fn execute_daemon( let mut modified = func.clone(); modified.body = remaining.to_string() + &func.body; let script = builder::build_script( - &modified, bake_base, workspace, RenderMode::Full, + &modified, bake_base, workspace, RenderMode::Full, ctx.debug_flags, )?; // Spawn without waiting diff --git a/workshop-baker/src/bake/parser.rs b/workshop-baker/src/bake/parser.rs index 5fae56c..50554b5 100644 --- a/workshop-baker/src/bake/parser.rs +++ b/workshop-baker/src/bake/parser.rs @@ -174,7 +174,7 @@ pub fn parse_script(text: &str) -> Result { } continue; } - eprintln!( + log::warn!( "Warning: Dangling decorator(s) at line {}. Treating as comments.", line_number - 1 ); diff --git a/workshop-baker/src/bake/trivial.rs b/workshop-baker/src/bake/trivial.rs index cd9a984..8547ff8 100644 --- a/workshop-baker/src/bake/trivial.rs +++ b/workshop-baker/src/bake/trivial.rs @@ -25,7 +25,7 @@ pub async fn run_trivial( decorators: vec![], body: script_content.to_string(), }; - let script = build_script(&function, bake_base_path, workspace, render_mode)?; + let script = build_script(&function, bake_base_path, workspace, render_mode, ctx.debug_flags)?; engine.execute_script(&script, ctx, event_tx).await?; Ok(()) } diff --git a/workshop-baker/src/bare.rs b/workshop-baker/src/bare.rs index 4e21d8e..8f41ff9 100644 --- a/workshop-baker/src/bare.rs +++ b/workshop-baker/src/bare.rs @@ -1,15 +1,8 @@ -use std::path::PathBuf; - use crate::cli::{BareCommands, Cli}; use crate::error::CliError; use crate::types::resource::ResourceRegistry; use crate::{bake::bake, finalize::finalize, prebake::prebake}; use workshop_engine::{EventSender, ExecutionContext}; -fn or_workshop(path: &Option, name: &str) -> PathBuf { - path - .clone() - .unwrap_or_else(|| PathBuf::from(format!(".workshop/{}", name))) -} pub async fn run_bare( cli: &Cli, @@ -21,16 +14,16 @@ pub async fn run_bare( registry: &ResourceRegistry, ) -> Result<(), CliError> { let (stage, config) = match command { - BareCommands::Prebake { .. } => ("prebake", or_workshop(&cli.prebake, "prebake.yml")), - BareCommands::Bake { .. } => ("bake", or_workshop(&cli.bake, "bake.sh")), + BareCommands::Prebake { .. } => ("prebake", &cli.prebake), + BareCommands::Bake { .. } => ("bake", &cli.bake), BareCommands::Finalize { .. } => { - ("finalize", or_workshop(&cli.finalize, "finalize.yml")) + ("finalize", &cli.finalize) } }; log::debug!("Running bare command {} with config {}", stage, config.display()); - let prebake_path = or_workshop(&cli.prebake, "prebake.yml"); - let bake_base = or_workshop(&cli.bake_base, "bake_base.sh"); + let prebake_path = &cli.prebake; + let bake_base = &cli.bake_base; ctx.task_id = format!("{}-{}-{}", pipeline, build_id, stage); ctx.pipeline_name = pipeline.to_string(); @@ -41,7 +34,7 @@ pub async fn run_bare( prebake(&config, cli, None, ctx, event_tx, None).await?; } BareCommands::Bake { .. } => { - bake(&config, &bake_base, &prebake_path, cli, ctx, event_tx).await?; + bake(&config, &bake_base, &prebake_path, cli, ctx, event_tx, None).await?; } BareCommands::Finalize { .. } => { finalize(&config, cli, ctx, event_tx, registry).await?; diff --git a/workshop-baker/src/daemon.rs b/workshop-baker/src/daemon.rs new file mode 100644 index 0000000..e69de29 diff --git a/workshop-baker/src/prebake/env.rs b/workshop-baker/src/daemon/env.rs similarity index 100% rename from workshop-baker/src/prebake/env.rs rename to workshop-baker/src/daemon/env.rs diff --git a/workshop-baker/src/prebake/env/baremetal.rs b/workshop-baker/src/daemon/env/baremetal.rs similarity index 100% rename from workshop-baker/src/prebake/env/baremetal.rs rename to workshop-baker/src/daemon/env/baremetal.rs diff --git a/workshop-baker/src/prebake/env/container.rs b/workshop-baker/src/daemon/env/container.rs similarity index 100% rename from workshop-baker/src/prebake/env/container.rs rename to workshop-baker/src/daemon/env/container.rs diff --git a/workshop-baker/src/prebake/env/custom.rs b/workshop-baker/src/daemon/env/custom.rs similarity index 100% rename from workshop-baker/src/prebake/env/custom.rs rename to workshop-baker/src/daemon/env/custom.rs diff --git a/workshop-baker/src/prebake/env/firecracker.rs b/workshop-baker/src/daemon/env/firecracker.rs similarity index 100% rename from workshop-baker/src/prebake/env/firecracker.rs rename to workshop-baker/src/daemon/env/firecracker.rs diff --git a/workshop-baker/src/finalize.rs b/workshop-baker/src/finalize.rs index 4fd814d..2970472 100644 --- a/workshop-baker/src/finalize.rs +++ b/workshop-baker/src/finalize.rs @@ -2,6 +2,8 @@ pub mod config; pub mod constant; pub mod error; pub mod event; +pub mod cleanup; +pub mod artifact; pub mod plugin; pub mod stage; pub mod template; @@ -52,9 +54,17 @@ pub async fn finalize( // Determine if we should continue with artifacts/latehook // If prior stages failed, we still send notification but skip artifacts let _prior_failed = earlyhook_result.is_err(); - - // Prepare artifacts - // TODO: artifact stage implementation + // Artifact collection and publishing + if !_prior_failed { + log::info!("Processing {} artifacts", finalize.artifact.len()); + artifact::collect_and_publish( + &finalize.artifact, + ®istered_plugins, + ctx, + event_tx.clone(), +) + .await?; + } // Late hook stage::hook::hook( diff --git a/workshop-baker/src/finalize/artifact.rs b/workshop-baker/src/finalize/artifact.rs new file mode 100644 index 0000000..c1973e8 --- /dev/null +++ b/workshop-baker/src/finalize/artifact.rs @@ -0,0 +1,139 @@ +//! Artifact collection and publishing. +//! +//! Collects build artifacts (via glob patterns), optionally compresses them, +//! and dispatches each artifact through its configured publish plugins. +//! The publish plugin receives the artifact path as a system-injected +//! `artifact` field in its argument. + +use std::path::{Path, PathBuf}; +use crate::finalize::config::ArtifactDef; +use crate::finalize::error::FinalizeError; +use crate::finalize::plugin::PluginMap; +use crate::finalize::types::compression::CompressionMethod; +use workshop_engine::{EventSender, ExecutionContext}; + +/// Collect artifacts and execute their publish plugins. +/// +/// For each artifact: +/// 1. Resolve source files via glob +/// 2. Optionally compress (zstd, gzip, dir) +/// 3. Inject `artifact` key into each publish plugin's config +/// 4. Call the plugin +pub async fn collect_and_publish( + artifacts: &[ArtifactDef], + plugins: &PluginMap, + ctx: &ExecutionContext, + event_tx: EventSender, +) -> Result<(), FinalizeError> { + if artifacts.is_empty() { + return Ok(()); + } + + for artifact in artifacts { + let artifact_path = process_artifact(artifact).await?; + + if let Some(ref path) = artifact_path { + log::info!("Artifact processed: {} -> {}", artifact.path, path.display()); + } + + for (plugin_name, publish_config) in &artifact.publish { + let plugin = plugins.get(plugin_name).ok_or_else(|| { + FinalizeError::PluginExecutionFailed(format!( + "publish plugin '{}' not found for artifact '{}'", + plugin_name, artifact.path, + )) + })?; + + // Inject artifact path into the plugin argument + let argument = if let Some(ref path) = artifact_path { + let mut arg = publish_config.clone(); + if let Some(map) = arg.as_mapping_mut() { + map.insert( + serde_yaml::Value::String("artifact".into()), + serde_yaml::Value::String(path.to_string_lossy().into()), + ); + } + arg + } else { + publish_config.clone() + }; + + log::info!( + "Publishing artifact '{}' via plugin '{}'", + artifact.path, plugin_name, + ); + + plugin.call(argument, ctx, &event_tx).await?; + } + } + + Ok(()) +} + +/// Resolve and optionally compress an artifact. +/// Returns the local path to the processed artifact, or `None` if non-file artifact. +async fn process_artifact(artifact: &ArtifactDef) -> Result, FinalizeError> { + if artifact.path.is_empty() { + return Ok(None); // non-filesystem artifact (e.g. Docker image) + } + + let src = PathBuf::from(&artifact.path); + if !src.exists() { + if artifact.path.contains('*') || artifact.path.contains('?') { + return Err(FinalizeError::PluginExecutionFailed( + format!("glob patterns not supported: '{}'; use a directory path", artifact.path), + )); + } + return Err(FinalizeError::ArtifactNotFound(src)); + } + + match artifact.compression { + CompressionMethod::None => Ok(Some(src)), + CompressionMethod::Dir => Ok(Some(src)), + CompressionMethod::Gzip => { + let dst = std::env::temp_dir().join(format!("artifact-{}.tar.gz", artifact.id.as_deref().unwrap_or("unknown"))); + compress_tar_zstd_or_gzip(&src, &dst, false).await?; + Ok(Some(dst)) + } + CompressionMethod::Zstd => { + let dst = std::env::temp_dir().join(format!("artifact-{}.tar.zst", artifact.id.as_deref().unwrap_or("unknown"))); + compress_tar_zstd_or_gzip(&src, &dst, true).await?; + Ok(Some(dst)) + } + } +} + +async fn compress_tar_zstd_or_gzip( + src: &Path, + dst: &Path, + zstd: bool, +) -> Result<(), FinalizeError> { + let _archive = tokio::task::spawn_blocking({ + let src = src.to_path_buf(); + let dst = dst.to_path_buf(); + move || -> Result<(), String> { + let file = std::fs::File::create(&dst) + .map_err(|e| format!("cannot create {}: {}", dst.display(), e))?; + let writer: Box = if zstd { + Box::new(zstd::stream::write::Encoder::new(file, 0).map_err(|e| e.to_string())?) + } else { + Box::new(flate2::write::GzEncoder::new(file, flate2::Compression::default())) + }; + let mut tar = tar::Builder::new(writer); + if src.is_dir() { + tar.append_dir_all(".", &src) + .map_err(|e| format!("tar error: {}", e))?; + } else { + tar.append_path(&src) + .map_err(|e| format!("tar error: {}", e))?; + } + tar.into_inner().map_err(|e| e.to_string())?; + Ok(()) + } + }) + .await + .map_err(|e| FinalizeError::PluginExecutionFailed(format!("spawn_blocking: {}", e)))? + .map_err(FinalizeError::PluginExecutionFailed)?; + + Ok(()) +} diff --git a/workshop-baker/src/finalize/cleanup.rs b/workshop-baker/src/finalize/cleanup.rs new file mode 100644 index 0000000..e5bf845 --- /dev/null +++ b/workshop-baker/src/finalize/cleanup.rs @@ -0,0 +1,67 @@ +//! Pipeline cleanup and resource reclamation. +//! +//! Runs after all pipeline stages and notification dispatch complete. +//! Checks the configured [`CleanupPolicy`] and performs the following: +//! +//! 1. Temp file / cache directory removal +//! 2. Policy-based skip (never, manual, on_success) + +use crate::finalize::config::{CleanupConfig, CleanupPolicy}; +use crate::finalize::error::FinalizeError; +use workshop_engine::ExecutionContext; + +/// Execute cleanup according to the pipeline's [`CleanupPolicy`]. +/// +/// Returns `Ok(())` when cleanup completes (or is skipped per policy). +/// Returns `Err` only if a cleanup action itself fails (not if the +/// pipeline failed; pipeline outcome is checked via `pipeline_ok`). +pub fn cleanup( + config: &CleanupConfig, + _ctx: &ExecutionContext, + pipeline_ok: bool, +) -> Result<(), FinalizeError> { + match config.policy { + CleanupPolicy::Never => { + log::info!("[cleanup] policy=never, skipping all cleanup"); + return Ok(()); + } + CleanupPolicy::Manual => { + log::info!("[cleanup] policy=manual, hanging for bakerd signal"); + // TODO(daemon): hang on bakerd signal + // For now (standalone/bare) manual == Never + return Ok(()); + } + CleanupPolicy::OnSuccess if !pipeline_ok => { + log::info!("[cleanup] policy=on_success, but pipeline failed — skipping"); + // TODO(daemon): hang for bakerd + return Ok(()); + } + _ => {} + } + log::info!("[cleanup] starting"); + cleanup_temp_files(); + subprocess_reclaim(); + log::info!("[cleanup] complete"); + Ok(()) +} + +/// Reclaim child processes (placeholder for daemon-mode PID tracking). +/// In standalone mode, children are reaped by the kernel on process exit. +fn subprocess_reclaim() { + // TODO(daemon): track spawned PIDs during pipeline and signal them here. +} +/// Remove plugin cache and temporary directories. +fn cleanup_temp_files() { + let cache = std::path::Path::new("/tmp/baker"); + if cache.exists() { + match std::fs::remove_dir_all(cache) { + Ok(()) => log::debug!("[cleanup] removed cache: {}", cache.display()), + Err(e) => log::warn!("[cleanup] failed to remove {}: {}", cache.display(), e), + } + } + + let workshop_templates = std::env::temp_dir().join("workshop-templates"); + if workshop_templates.exists() { + let _ = std::fs::remove_dir_all(&workshop_templates); + } +} diff --git a/workshop-baker/src/finalize/config.rs b/workshop-baker/src/finalize/config.rs index 6816c69..b647480 100644 --- a/workshop-baker/src/finalize/config.rs +++ b/workshop-baker/src/finalize/config.rs @@ -342,8 +342,9 @@ pub struct ArtifactDef { /// Compression method: "zstd", "gzip", "none", "dir" #[serde(default)] pub compression: CompressionMethod, - // Publishing targets - // TODO: under heavy refactor + /// Publish targets: plugin name → configuration (artifact path injected by system) + #[serde(default)] + pub publish: HashMap } impl Default for ArtifactDef { @@ -353,6 +354,7 @@ impl Default for ArtifactDef { path: String::new(), retention_ms: default_retention_ms(), compression: CompressionMethod::default(), + publish: HashMap::new(), } } } diff --git a/workshop-baker/src/finalize/event.rs b/workshop-baker/src/finalize/event.rs index 163e491..7a576f8 100644 --- a/workshop-baker/src/finalize/event.rs +++ b/workshop-baker/src/finalize/event.rs @@ -1,4 +1,5 @@ use workshop_engine::{ExecutionEvent, StreamType}; +use crate::types::debug::DebugFeature; /// Receives and processes execution events for the finalize stage. /// Logs task lifecycle events (started, completed, failed) and streaming output (stdout/stderr) @@ -7,48 +8,62 @@ pub async fn event_receiver( mut rx: tokio::sync::mpsc::Receiver, pipeline_name: String, build_id: String, + debug_flags: u32, ) { + let events_debug = DebugFeature::from_bits_retain(debug_flags).contains(DebugFeature::Events); while let Some(event) = rx.recv().await { match event { ExecutionEvent::TaskStarted { task_id, timestamp } => { - log::debug!( - "[{}] [{}] [{}] Task started: {}", - pipeline_name, - build_id, - timestamp, - task_id - ); + if events_debug { + log::info!( + "[{}] [{}] [{}] Task started: {}", + pipeline_name, + build_id, + timestamp, + task_id + ); + } else { + log::debug!( + "Task started: {}", + task_id + ); + } } ExecutionEvent::OutputChunk { - task_id, + task_id: _, stream, data, } => match stream { - StreamType::Stdout => print!("[{}] [{}] [stdout] {}", pipeline_name, task_id, data), + StreamType::Stdout => print!("{}", data), StreamType::Stderr => { - eprint!("[{}] [{}] [stderr] {}", pipeline_name, task_id, data) + eprint!("{}", data) } }, ExecutionEvent::TaskCompleted { task_id, result } => { if result.success { - log::debug!( - "[{}] [{}] Task completed: {} (exit_code={}, duration={:?})", - pipeline_name, build_id, task_id, - result.exit_code, result.duration, - ); + if events_debug { + log::info!( + "[{}] [{}] Task completed: {} (exit_code={}, duration={:?})", + pipeline_name, build_id, task_id, + result.exit_code, result.duration, + ); + } else { + log::debug!( + "Task completed: {} (exit_code={}, duration={:?})", + task_id, result.exit_code, result.duration, + ); + } } else { log::warn!( - "[{}] [{}] Task completed with non-zero exit: {} (exit_code={}, duration={:?})", - pipeline_name, build_id, task_id, + "Task completed with non-zero exit: {} (exit_code={}, duration={:?})", + task_id, result.exit_code, result.duration, ); } } ExecutionEvent::TaskFailed { task_id, error } => { log::error!( - "[{}] [{}] Task failed: {} (error={})", - pipeline_name, - build_id, + "Task failed: {} (error={})", task_id, error ); diff --git a/workshop-baker/src/finalize/plugin.rs b/workshop-baker/src/finalize/plugin.rs index 9a52a96..6a7347a 100644 --- a/workshop-baker/src/finalize/plugin.rs +++ b/workshop-baker/src/finalize/plugin.rs @@ -10,7 +10,7 @@ use std::path::{Path, PathBuf}; use std::sync::Arc; mod dylib; -mod internal; +pub mod internal; mod metadata; mod rhai; mod shell; diff --git a/workshop-baker/src/finalize/plugin/internal.rs b/workshop-baker/src/finalize/plugin/internal.rs index b7c8fa5..68222ac 100644 --- a/workshop-baker/src/finalize/plugin/internal.rs +++ b/workshop-baker/src/finalize/plugin/internal.rs @@ -36,10 +36,12 @@ pub fn get_internal_plugin() -> Vec { ] } -pub fn get_dummy_plugin() -> Vec { +/// Returns dummy only (for `--debug notify.dummy` mode). +/// All real adapters and in-site-notify are excluded. +pub fn get_debug_plugin() -> Vec { vec![InternalPlugin { name: "dummy", entry: Box::new(dummy::Dummy), - renderer: NotificationRenderer::Server, + renderer: NotificationRenderer::Plugin, }] } diff --git a/workshop-baker/src/finalize/plugin/internal/dummy.rs b/workshop-baker/src/finalize/plugin/internal/dummy.rs index 47057cc..4bf8f4d 100644 --- a/workshop-baker/src/finalize/plugin/internal/dummy.rs +++ b/workshop-baker/src/finalize/plugin/internal/dummy.rs @@ -10,12 +10,12 @@ pub struct Dummy; impl AsyncPluginFn for Dummy { async fn call( &self, - _metadata: &PluginMetadata, - _argument: serde_yaml::Value, + metadata: &PluginMetadata, + argument: serde_yaml::Value, _ctx: &ExecutionContext, _event_tx: &EventSender, ) -> Result<(), PluginError> { - log::debug!("[NOTIFY_DEBUG] Successfully registered dummy plugin."); + log::info!("[Dummy] metadata={:?}, arg={:?}", metadata, argument); Ok(()) } } diff --git a/workshop-baker/src/finalize/plugin/metadata.rs b/workshop-baker/src/finalize/plugin/metadata.rs index 24fea29..51bc32b 100644 --- a/workshop-baker/src/finalize/plugin/metadata.rs +++ b/workshop-baker/src/finalize/plugin/metadata.rs @@ -4,7 +4,7 @@ use serde::Deserialize; use std::path::PathBuf; /// Plugin manifest metadata containing identification and configuration. -#[derive(Deserialize, Clone)] +#[derive(Deserialize, Clone, Debug)] pub struct PluginMetadata { /// Plugin name identifier. pub name: String, diff --git a/workshop-baker/src/lib.rs b/workshop-baker/src/lib.rs index f194f4a..05d88a6 100644 --- a/workshop-baker/src/lib.rs +++ b/workshop-baker/src/lib.rs @@ -3,9 +3,11 @@ pub mod bake; pub mod error; pub mod finalize; pub mod notify; +pub mod prefetch; pub mod prebake; pub mod types; pub mod utils; +pub mod daemon; pub use clap::{Parser, Subcommand}; pub use serde::Deserialize; @@ -21,37 +23,42 @@ pub mod cli { pub struct Cli { /// Username of the pipeline executor #[arg(short, long, env = "HBW_USERNAME")] - pub username: String, + pub username: Option, /// Pipeline ID of this pipeline #[arg(short, long, env = "HBW_PIPELINE")] - pub pipeline: String, + pub pipeline: Option, /// Build ID of this build #[arg(short, long, env = "HBW_BUILD_ID")] - pub build_id: String, + pub build_id: Option, /// Parse the script without executing #[arg(long, default_value = "false", env = "HBW_DRY_RUN")] pub dry_run: bool, /// Override prebake config path - #[arg(long)] - pub prebake: Option, + #[arg(long, default_value = ".workshop/prebake.yml")] + pub prebake: PathBuf, /// Override bake script path - #[arg(long)] - pub bake: Option, + #[arg(long, default_value = ".workshop/bake.sh")] + pub bake: PathBuf, /// Override bake_base.sh path - #[arg(long)] - pub bake_base: Option, + #[arg(long, default_value = ".workshop/bake_base.sh")] + pub bake_base: PathBuf, /// Override finalize config path - #[arg(long)] - pub finalize: Option, + #[arg(long, default_value = ".workshop/finalize.yml")] + pub finalize: PathBuf, - /// Debug features (comma-separated). "notify" enables notification debug logging. + /// Debug features. Use `--debug help` for details. #[arg(long, env = "HBW_DEBUG")] pub debug: Option, + /// JSON string of pre-resolved resources: `{"name":"/path",...}` + /// Only valid in `bare` mode. + #[arg(long, value_name = "JSON")] + pub resource_registry: Option, + #[command(subcommand)] pub command: Option, } diff --git a/workshop-baker/src/main.rs b/workshop-baker/src/main.rs index 2202011..0e92605 100644 --- a/workshop-baker/src/main.rs +++ b/workshop-baker/src/main.rs @@ -1,91 +1,164 @@ -use workshop_baker::types::resource::ResourceRegistry; -use std::path::PathBuf; - -use workshop_baker::{ - bake::bake, cli::{Cli, Commands, Parser}, - error::CliError, - finalize::finalize, notify::notify, prebake::prebake, -}; +use capctl; use workshop_baker::bare::run_bare; use workshop_baker::notify::queue::NotificationQueue; +use workshop_baker::types::debug::DebugFeature; +use workshop_baker::types::resource::ResourceRegistry; +use workshop_baker::{ + bake::bake, + cli::{Cli, Commands, Parser}, + error::CliError, + finalize::finalize, + notify::notify, + prebake::prebake, +}; use workshop_engine::{EventSender, ExecutionContext}; -fn or_workshop(path: &Option, name: &str) -> PathBuf { - path.clone().unwrap_or_else(|| PathBuf::from(format!(".workshop/{}", name))) -} - async fn worker_main(cli: Cli) -> Result<(), CliError> { - let prebake_path = cli.prebake.clone().unwrap_or_else(|| { - log::error!("--prebake is required in worker standalone mode"); - std::process::exit(1); - }); - let bake_path = cli.bake.clone().unwrap_or_else(|| { - log::error!("--bake is required in worker standalone mode"); - std::process::exit(1); - }); - let finalize_path = cli.finalize.clone().unwrap_or_else(|| { - log::error!("--finalize is required in worker standalone mode"); - std::process::exit(1); - }); + // a magic closure hack + let require = |name: &str, val: &Option| -> String { + val.clone().unwrap_or_else(|| { + log::error!("--{} is required", name); + std::process::exit(1); + }) + }; + // worker depends on these information to work + let pipeline = require("pipeline", &cli.pipeline); + let build_id = require("build-id", &cli.build_id); + let username = require("username", &cli.username); - log::info!("Worker standalone mode"); + println!("── worker standalone mode ───────────────────────"); + println!(" {:<14} {:?}", "prebake:", cli.prebake); + println!(" {:<14} {:?}", "bake:", cli.bake); + println!(" {:<14} {:?}", "bake-base:", cli.bake_base); + println!(" {:<14} {:?}", "finalize:", cli.finalize); + println!("──────────────────────────────────────────────────"); + + let debug_flags = DebugFeature::parse(cli.debug.as_deref()); + if !debug_flags.is_empty() { + log::info!("Debug features enabled: {:?}", debug_flags); + } let mut ctx = ExecutionContext { standalone: true, - task_id: format!("{}-{}-worker", cli.pipeline, cli.build_id), - pipeline_name: cli.pipeline.clone(), - username: cli.username.clone(), + task_id: format!("{}-{}-worker", pipeline, build_id), + pipeline_name: pipeline.clone(), + build_id: build_id.clone(), + username: username.clone(), dry_run: cli.dry_run, privileged: false, + debug_flags: debug_flags.bits(), ..Default::default() }; let (tx, event_rx) = tokio::sync::mpsc::channel(256); let event_tx: EventSender = Some(tx); let _event_handle = tokio::spawn(workshop_baker::prebake::event::event_receiver( - event_rx, cli.pipeline.clone(), cli.build_id.clone(), + event_rx, + pipeline.clone(), + build_id.clone(), + ctx.debug_flags, )); let (to_tx, to_rx) = tokio::sync::mpsc::channel(256); let (retry_tx, retry_rx) = tokio::sync::mpsc::channel(256); - // Resource registry — populated before stages, consumed by stages - let registry = ResourceRegistry::new(); - // TODO: pre-fetch plugins and templates into registry + // Resource registry — fetch plugins/templates before pipeline stages + let mut registry = ResourceRegistry::new(); + if let Ok(finalize_cfg) = workshop_baker::finalize::parse(&cli.finalize) { + let _ = workshop_baker::prefetch::prefetch_resources(&finalize_cfg, &mut registry).await; + } + let registry = registry; // freeze mut → immut - let notify_path = finalize_path.clone(); + let notify_finalize = cli.finalize.clone(); let notify_ctx = ctx.clone(); let notify_etx = event_tx.clone(); - let debug = cli.debug.clone(); let notify_registry = registry.clone(); + let notify_debug = ctx.debug_flags; let notify_handle = tokio::spawn(async move { - notify(¬ify_path, &mut notify_ctx.clone(), notify_etx, to_rx, retry_tx, debug.map(|d| d.contains("notify")).unwrap_or(false), ¬ify_registry).await + notify( + ¬ify_finalize, + &mut notify_ctx.clone(), + notify_etx, + to_rx, + retry_tx, + notify_debug, + ¬ify_registry, + ) + .await }); let nevent_tx = to_tx.clone(); - let mut queue = NotificationQueue::new(to_tx, retry_rx); + let mut queue = NotificationQueue::new(to_tx, retry_rx, notify_debug); let pipe_result: anyhow::Result<()> = async { - prebake(&prebake_path, &cli, None, &mut ctx, event_tx.clone(), Some(nevent_tx)).await?; - let bake_base_path = or_workshop(&cli.bake_base, "bake_base.sh"); - let pb_path = or_workshop(&cli.prebake, "prebake.yml"); - bake(&bake_path, &bake_base_path, &pb_path, &cli, &mut ctx, event_tx.clone()).await?; - finalize(&finalize_path, &cli, &mut ctx, event_tx.clone(), ®istry).await?; + prebake( + &cli.prebake, + &cli, + None, + &mut ctx, + event_tx.clone(), + Some(nevent_tx.clone()), + ) + .await?; + let bake_result = bake( + &cli.bake, + &cli.bake_base, + &cli.prebake, + &cli, + &mut ctx, + event_tx.clone(), + Some(nevent_tx.clone()), + ) + .await; + if bake_result.is_err() { + workshop_baker::notify::types::send_stage_event( + &Some(nevent_tx.clone()), + workshop_baker::types::buildstatus::StagePhase::Bake, + "bake", + workshop_baker::types::buildstatus::StageOutcome::Failure, + ) + .await; + } + bake_result?; + finalize(&cli.finalize, &cli, &mut ctx, event_tx.clone(), ®istry).await?; Ok(()) - }.await; + } + .await; if let Err(e) = pipe_result { + // Notify pipeline failure before exiting + workshop_baker::notify::types::send_stage_event( + &Some(nevent_tx.clone()), + workshop_baker::types::buildstatus::StagePhase::Bake, + "pipeline", + workshop_baker::types::buildstatus::StageOutcome::PipelineFailure, + ) + .await; log::error!("Pipeline stage failed: {}", e); return Err(CliError::General(e)); } + // Notify pipeline success + workshop_baker::notify::types::send_stage_event( + &Some(nevent_tx.clone()), + workshop_baker::types::buildstatus::StagePhase::Bake, + "pipeline", + workshop_baker::types::buildstatus::StageOutcome::PipelineSuccess, + ) + .await; log::info!("Pipeline complete. Draining notifications..."); - // Drop the sender so notify() can end and retry_tx is dropped + drop(nevent_tx); queue.tx = None; if let Err(e) = queue.drain().await { log::error!("Notification queue error: {}", e); } let _ = notify_handle.await; + + // Cleanup after notifications are fully drained + let pipeline_ok = pipe_result.is_ok(); + let _ = workshop_baker::finalize::parse(&cli.finalize).map(|cfg| { + let _ = workshop_baker::finalize::cleanup::cleanup(&cfg.cleanup, &ctx, pipeline_ok); + }); Ok(()) } @@ -93,32 +166,81 @@ async fn worker_main(cli: Cli) -> Result<(), CliError> { async fn main() { env_logger::init(); let cli = Cli::parse(); + if std::process::id() == 1 { + log::error!("Baker is not meant to be run as PID=1"); + std::process::exit(1); + } + if let Err(e) = capctl::prctl::set_subreaper(true){ + log::error!("Failed to set self as subreaper: {}", e); + log::error!("finalize.cleanup might not work properly"); + } + + // --debug help: print help and exit immediately + if let Some(ref raw) = cli.debug { + if DebugFeature::print_help(raw) { + return; + } + } + let debug_flags = DebugFeature::parse(cli.debug.as_deref()); if cli.dry_run { log::warn!("Dry run enabled, no changes will be made."); } - let pipeline = cli.pipeline.clone(); - let build_id = cli.build_id.clone(); - + // --resource-registry is only allowed in bare mode + if cli.resource_registry.is_some() && !matches!(cli.command, Some(Commands::Bare { .. })) { + log::error!("--resource-registry is only supported in bare mode"); + std::process::exit(1); + } match &cli.command { Some(Commands::Bare { command }) => { + // --pipeline, --build-id, --username are optional in bare mode + // and they will be defaulted to "BarePipeline", "0", "bare" respectively + let pipeline = cli + .pipeline + .as_deref() + .unwrap_or("BarePipeline") + .to_string(); + let build_id = cli.build_id.as_deref().unwrap_or("0").to_string(); + let username = cli.username.as_deref().unwrap_or("bare").to_string(); + let mut ctx = ExecutionContext { standalone: true, task_id: format!("{}-{}-bare", pipeline, build_id), pipeline_name: pipeline.clone(), - username: cli.username.clone(), + build_id: build_id.clone(), + username: username.clone(), dry_run: cli.dry_run, privileged: false, + debug_flags: debug_flags.bits(), ..Default::default() }; let (tx, event_rx) = tokio::sync::mpsc::channel(256); let event_tx: EventSender = Some(tx); let event_handle = tokio::spawn(workshop_baker::prebake::event::event_receiver( - event_rx, pipeline.clone(), build_id.clone(), + event_rx, + pipeline.clone(), + build_id.clone(), + ctx.debug_flags, )); - let registry = ResourceRegistry::new(); - let result = run_bare(&cli, command, &mut ctx, event_tx, &pipeline, &build_id, ®istry).await; + let registry = if let Some(ref json_str) = cli.resource_registry { + match workshop_baker::types::resource::ResourceRegistry::from_json_str(json_str) { + Ok(r) => { + log::info!("Loaded {} resources from --resource-registry", r.len()); + r + } + Err(e) => { + log::error!("Failed to parse --resource-registry: {}", e); + std::process::exit(1); + } + } + } else { + ResourceRegistry::new() + }; + let result = run_bare( + &cli, command, &mut ctx, event_tx, &pipeline, &build_id, ®istry, + ) + .await; drop(ctx); let _ = event_handle.await; diff --git a/workshop-baker/src/notify.rs b/workshop-baker/src/notify.rs index 7f5c888..5aea8b4 100644 --- a/workshop-baker/src/notify.rs +++ b/workshop-baker/src/notify.rs @@ -1,10 +1,12 @@ use crate::finalize::parse; use crate::finalize::plugin; +use crate::finalize::plugin::internal; use crate::notify::types::DEFAULT_NOTIFY_PARAMS; use crate::notify::types::NotificationEventReceiver; use crate::notify::types::NotificationEventSender; use crate::notify::handler::NotificationHandler; use crate::types::resource::ResourceRegistry; +use crate::types::debug::DebugFeature; use std::path::Path; use std::sync::Arc; use workshop_engine::{EventSender, ExecutionContext}; @@ -26,18 +28,23 @@ pub async fn notify( _event_tx: EventSender, mut rx: NotificationEventReceiver, retry_tx: NotificationEventSender, - _debug: bool, + debug_flags: u32, registry: &ResourceRegistry, ) -> anyhow::Result<()> { + let debug = DebugFeature::from_bits_retain(debug_flags); + let notify_debug = debug.contains(DebugFeature::NotifyDummy); + let mut finalize = parse(finalize_path)?; - // L18: 0 priorities → notification ends; L19: has priorities → inject in-site fallback + // L18: 0 priorities → notification ends; L19: has priorities → inject fallback + // In notify-debug mode, inject dummy instead of in-site-notify. if !finalize.notification.groups.is_empty() { - log::debug!("Injecting in-site-notify fallback"); + let fallback_name = if notify_debug { "dummy" } else { "in-site-notify" }; + log::debug!("Injecting {} fallback", fallback_name); let max_p = finalize.notification.groups.keys().max().copied().unwrap_or(0); let mut m = std::collections::HashMap::new(); m.insert( - "in-site-notify".to_string(), + fallback_name.to_string(), crate::finalize::NotificationMethod::default(), ); finalize.notification.groups.insert(max_p + 1, m); @@ -46,34 +53,64 @@ pub async fn notify( validate(&finalize)?; let plugin_list = extract_plugin(&finalize); - let notify_plugins = plugin::register( - finalize.plugin, - Some(&plugin_list), - &mut finalize.notification, - registry, - )?; + // Register plugins: use dummy-only set in notify-debug mode + let notify_plugins = if notify_debug { + log::info!("[DEBUG-NOTIFY] Replacing notification plugins with debug set"); + let mut plugins: crate::finalize::plugin::PluginMap = std::collections::HashMap::new(); + for i in internal::get_debug_plugin() { + log::info!("[DEBUG-NOTIFY] Registered: {}", i.name); + let mut plugin = plugin::FinalizePlugin::new(i.entry); + plugin.metadata.renderer = Some(i.renderer); + plugins.insert(i.name.to_string(), plugin); + } + plugins + } else { + plugin::register( + finalize.plugin, + Some(&plugin_list), + &mut finalize.notification, + registry, + )? + }; resolve_external_template_refs(&mut finalize.notification.templates, registry); flatten_template_refs(&mut finalize.notification.templates)?; let notification_rules = parse_rules(&finalize.notification); + // Debug mode: eliminate cooldown so bucket dispatches immediately + let (period, watermark) = if notify_debug { + (0, 0) + } else { + (DEFAULT_NOTIFY_PARAMS.batch_period.value, DEFAULT_NOTIFY_PARAMS.batch_watermark.value) + }; + let handler = Arc::new(NotificationHandler::new( Arc::new(notification_rules), Arc::new(notify_plugins), Arc::new(finalize.notification.templates), Arc::new(ctx.clone()), retry_tx, - DEFAULT_NOTIFY_PARAMS.batch_period.value, - DEFAULT_NOTIFY_PARAMS.batch_watermark.value, + period, + watermark, )); + let mut handles: Vec> = Vec::new(); loop { let Some(event) = rx.recv().await else { break }; + if notify_debug { + log::info!( + "[DEBUG-NOTIFY] event: id={} stage={:?} substage={} outcome={:?}", + event.id, event.stage, event.substage, event.outcome); + } let h = handler.clone(); - tokio::spawn(async move { + handles.push(tokio::spawn(async move { h.process_event(event).await; - }); + })); + } + // Wait for all bucket dispatches to complete before returning + for handle in handles { + let _ = handle.await; } Ok(()) } diff --git a/workshop-baker/src/notify/handler.rs b/workshop-baker/src/notify/handler.rs index 1e0c3a8..c8cc2a6 100644 --- a/workshop-baker/src/notify/handler.rs +++ b/workshop-baker/src/notify/handler.rs @@ -73,8 +73,10 @@ impl NotificationHandler { /// Process a single event: route to buckets, await completion, /// then handle retry/escalation for each bucket result. pub async fn process_event(self: Arc, event: NotificationEvent) { + log::info!("[TRACE] process_event: stage={:?} substage={} outcome={:?}", event.stage, event.substage, event.outcome); let entries = self.route_event(&event); if entries.is_empty() { + log::warn!("[TRACE] process_event: no matching rules, event dropped"); return; } @@ -142,6 +144,8 @@ impl NotificationHandler { ) -> Vec<((String, usize), Arc)> { let mut matches: Vec<(String, usize)> = Vec::new(); + let config_groups: Vec = self.config.values().map(|r| r.len()).collect(); + log::info!("[TRACE] route_event: groups={:?}", config_groups); for ruleset in self.config.values() { for (method, rules) in ruleset.iter() { if let Some(ref method_rules) = event.target.rules { diff --git a/workshop-baker/src/notify/queue.rs b/workshop-baker/src/notify/queue.rs index 3397e88..b4a61f8 100644 --- a/workshop-baker/src/notify/queue.rs +++ b/workshop-baker/src/notify/queue.rs @@ -1,3 +1,4 @@ +use crate::types::debug::DebugFeature; use crate::notify::types::{NotificationEvent, NotificationEventReceiver, NotificationEventSender}; use std::collections::BinaryHeap; @@ -10,14 +11,16 @@ pub struct NotificationQueue { pub pq: BinaryHeap, pub tx: Option, pub rx: NotificationEventReceiver, + pub debug_flags: u32, } impl NotificationQueue { - pub fn new(tx: NotificationEventSender, rx: NotificationEventReceiver) -> Self { + pub fn new(tx: NotificationEventSender, rx: NotificationEventReceiver, debug_flags: u32) -> Self { Self { pq: BinaryHeap::new(), tx: Some(tx), rx, + debug_flags, } } @@ -29,8 +32,12 @@ impl NotificationQueue { /// Non-blocking single round: recycle retry events into PQ, then /// pop and send all ready events (not_before ≤ now). pub async fn flush_now(&mut self) -> anyhow::Result<()> { + let events_debug = DebugFeature::from_bits_retain(self.debug_flags).contains(DebugFeature::Events); // Recycle retry events while let Ok(event) = self.rx.try_recv() { + if events_debug { + log::info!("[EVENTS] flush_now: recycle event id={} retrying", event.id); + } self.pq.push(event); } // Send ready events @@ -40,6 +47,9 @@ impl NotificationQueue { break; } let event = self.pq.pop().unwrap(); + if events_debug { + log::info!("[EVENTS] flush_now: sending event id={}", event.id); + } if let Some(ref tx) = self.tx { tx.send(event).await?; } @@ -51,9 +61,13 @@ impl NotificationQueue { /// For standalone mode: called after prebake/bake/finalize to wait for completion. /// For c/d mode: the main execution loop. pub async fn drain(&mut self) -> anyhow::Result<()> { + let events_debug = DebugFeature::from_bits_retain(self.debug_flags).contains(DebugFeature::Events); loop { // Drain retry channel while let Ok(event) = self.rx.try_recv() { + if events_debug { + log::info!("[EVENTS] drain: recycle event id={} retrying", event.id); + } self.pq.push(event); } // Determine sleep deadline @@ -70,7 +84,12 @@ impl NotificationQueue { tokio::select! { event = self.rx.recv() => { match event { - Some(event) => self.pq.push(event), + Some(event) => { + if events_debug { + log::info!("[EVENTS] drain: recv event id={}", event.id); + } + self.pq.push(event); + } None => break Ok(()), } } @@ -79,6 +98,9 @@ impl NotificationQueue { while let Some(peek) = self.pq.peek() { if peek.not_before > now { break; } let event = self.pq.pop().unwrap(); + if events_debug { + log::info!("[EVENTS] drain: sending event id={}", event.id); + } if let Some(ref tx) = self.tx { tx.send(event).await?; } diff --git a/workshop-baker/src/notify/types.rs b/workshop-baker/src/notify/types.rs index 7eb6976..e023534 100644 --- a/workshop-baker/src/notify/types.rs +++ b/workshop-baker/src/notify/types.rs @@ -1,4 +1,3 @@ -use crate::finalize::plugin::FinalizePlugin; use crate::types::buildstatus::StageOutcome; use crate::types::buildstatus::StagePhase; use chrono::{DateTime, Utc}; @@ -13,29 +12,21 @@ use std::time::Duration; use uuid::Uuid; use crate::finalize::NotificationTemplate; +use crate::finalize::plugin::FinalizePlugin; + +pub type NotificationEventReceiver = tokio::sync::mpsc::Receiver; +pub type NotificationEventSender = tokio::sync::mpsc::Sender; +pub type NotificationPluginMap = HashMap; /// Final runtime configuration for notification, restored and flattened from yaml pub struct NotificationRule { pub enabled: bool, - pub trigger: Vec, - pub template: NotificationTemplate, - - /// Whether notification failure should not trigger next priority group pub fallible: bool, - - /// Maximum number of attempts pub max_lives: u32, - - /// Timeout for notification execution pub timeout: Duration, - - /// Plugin-specific configuration pub config: serde_yaml::Value, - - /// Factor of exponential backoff for retry delay - // Not provided for external use pub retry_backoff: f64, } @@ -56,7 +47,6 @@ pub enum Matchable { impl Matchable { fn matches(&self, stage: &StagePhase, substage: &str) -> bool { let stagename = [stage.as_str(), ".", substage].concat(); - match self { Matchable::Glob(s) => s.is_match(stagename.as_str()), Matchable::Regex(r) => r.is_match(stagename.as_str()), @@ -69,7 +59,6 @@ impl ParsedTrigger { match self { ParsedTrigger::OnSuccess => outcome == &StageOutcome::PipelineSuccess, ParsedTrigger::OnFailure => outcome == &StageOutcome::PipelineFailure, - // TODO: Implement condition checking for OnFinishOf ParsedTrigger::OnFinishOf(m) => m.matches(stage, substage), } } @@ -78,12 +67,9 @@ impl ParsedTrigger { impl NotificationRule { pub fn accept(&self, result: &StageOutcome, stage: &StagePhase, substage: &str) -> bool { if self.trigger.is_empty() { - log::trace!("[ACCEPT] trigger empty, returning true"); return true; } - self.trigger - .iter() - .any(|t| t.accept(result, stage, substage)) + self.trigger.iter().any(|t| t.accept(result, stage, substage)) } } @@ -113,10 +99,6 @@ impl Default for NotificationTarget { pub type Priority = u32; pub type Method = String; pub type NotificationRulesMap = HashMap>>; -// pub type BundledNotificationEvent = Vec; -pub type NotificationEventReceiver = tokio::sync::mpsc::Receiver; -pub type NotificationEventSender = tokio::sync::mpsc::Sender; -pub type NotificationPluginMap = HashMap; #[derive(Debug, Clone, Serialize)] pub struct NotificationEvent { @@ -127,9 +109,9 @@ pub struct NotificationEvent { pub not_before: DateTime, pub priority: u32, pub effective_priority: u32, - pub retry_backoff: f64, // Exponential backoff factor - pub attempted: u32, // Cumulative cross-group retry count - pub target: NotificationTarget, // Current audience + pub retry_backoff: f64, + pub attempted: u32, + pub target: NotificationTarget, } impl Hash for NotificationEvent { @@ -188,23 +170,39 @@ impl Display for NotificationRenderer { } /// Notification batching configuration. -/// Controls how notification events are grouped before dispatch. #[derive(Debug, Clone)] pub struct BatchConfig { - /// Whether batching is enabled pub enabled: bool, - /// Group period in seconds (the maximum wait time for a batch) pub period: u32, - /// Group watermark threshold (the minimum number of events before batching) pub watermark: u32, } impl Default for BatchConfig { fn default() -> Self { - Self { - enabled: true, - period: 30, - watermark: 5, - } + Self { enabled: true, period: 30, watermark: 5 } + } +} + +/// Send a stage completion event to the notification queue. +pub async fn send_stage_event( + nevent_tx: &Option, + stage: StagePhase, + substage: &str, + outcome: StageOutcome, +) { + if let Some(tx) = nevent_tx { + let event = NotificationEvent { + id: Uuid::new_v4(), + stage, + substage: substage.to_string(), + outcome, + not_before: Utc::now(), + priority: 0, + effective_priority: 0, + retry_backoff: 1.0, + attempted: 0, + target: Default::default(), + }; + let _ = tx.send(event).await; } } diff --git a/workshop-baker/src/prebake.rs b/workshop-baker/src/prebake.rs index e4e51aa..95be612 100644 --- a/workshop-baker/src/prebake.rs +++ b/workshop-baker/src/prebake.rs @@ -22,20 +22,19 @@ pub mod config; pub mod constant; -pub mod env; pub mod error; pub mod event; pub mod security; pub mod stage; pub mod types; +use crate::notify::types::NotificationEventSender; +use crate::notify::types::send_stage_event; +use crate::types::buildstatus::{StageOutcome, StagePhase}; use crate::{ cli::Cli, prebake::{error::PrebakeError, security::get_drop_after, stage::PrebakeStage}, }; -use crate::notify::types::NotificationEventSender; -use crate::types::buildstatus::{StageOutcome, StagePhase}; -use uuid::Uuid; use chrono::Utc; use std::path::Path; use workshop_engine::{EventSender, ExecutionContext}; @@ -91,7 +90,7 @@ pub fn parse(yaml_content: &str) -> Result { /// - Any stage execution fails pub async fn prebake( prebake_path: &Path, - cli: &Cli, + _cli: &Cli, stage: Option, ctx: &mut ExecutionContext, event_tx: EventSender, @@ -130,19 +129,25 @@ pub async fn prebake( let _stage_start = Utc::now(); log::info!("Running PBStage: Bootstrap"); prebake_drop_privilege(PrebakeStage::Bootstrap, dropafter, target_user, ctx)?; - ctx.task_id = format!("{}-{}-bootstrap", cli.pipeline, cli.build_id); + ctx.task_id = format!("{}-{}-bootstrap", ctx.pipeline_name, ctx.build_id); let osinfo = os_info::get(); let result = stage::bootstrap::bootstrap(&prebake, &osinfo, ctx, &event_tx).await; result?; } - send_stage_event(&nevent_tx, StagePhase::Prebake, "bootstrap", StageOutcome::Success).await; + send_stage_event( + &nevent_tx, + StagePhase::Prebake, + "bootstrap", + StageOutcome::Success, + ) + .await; // EarlyHook stage if stage <= PrebakeStage::EarlyHook { let _stage_start = Utc::now(); log::info!("Running PBStage: EarlyHook"); prebake_drop_privilege(PrebakeStage::EarlyHook, dropafter, target_user, ctx)?; - ctx.task_id = format!("{}-{}-earlyhook", cli.pipeline, cli.build_id); + ctx.task_id = format!("{}-{}-earlyhook", ctx.pipeline_name, ctx.build_id); let earlyhook = match &prebake.hooks { Some(hooks) => hooks.early.clone(), None => None, @@ -150,7 +155,13 @@ pub async fn prebake( let result = stage::hook::hook(earlyhook, ctx, event_tx.clone()).await; result?; } - send_stage_event(&nevent_tx, StagePhase::Prebake, "earlyhook", StageOutcome::Success).await; + send_stage_event( + &nevent_tx, + StagePhase::Prebake, + "earlyhook", + StageOutcome::Success, + ) + .await; if let Some(dependencies) = prebake.dependencies { let endpoint = dependencies @@ -217,11 +228,17 @@ pub async fn prebake( let _stage_start = Utc::now(); log::info!("Running PBStage: DepsSystem"); prebake_drop_privilege(PrebakeStage::DepsSystem, dropafter, target_user, ctx)?; - ctx.task_id = format!("{}-{}-depssystem", cli.pipeline, cli.build_id); + ctx.task_id = format!("{}-{}-depssystem", ctx.pipeline_name, ctx.build_id); let result = stage::depssystem::depssystem(&system_config, ctx, event_tx.clone()).await; result?; } - send_stage_event(&nevent_tx, StagePhase::Prebake, "depssystem", StageOutcome::Success).await; + send_stage_event( + &nevent_tx, + StagePhase::Prebake, + "depssystem", + StageOutcome::Success, + ) + .await; if let Some(user) = dependencies.user && stage <= PrebakeStage::DepsUser @@ -229,56 +246,50 @@ pub async fn prebake( let _stage_start = Utc::now(); log::info!("Running PBStage: DepsUser"); prebake_drop_privilege(PrebakeStage::DepsUser, dropafter, target_user, ctx)?; - ctx.task_id = format!("{}-{}-depsuser", cli.pipeline, cli.build_id); + ctx.task_id = format!("{}-{}-depsuser", ctx.pipeline_name, ctx.build_id); let result = stage::depsuser::depsuser(&user, ctx, event_tx.clone(), &endpoint).await; result?; } } - send_stage_event(&nevent_tx, StagePhase::Prebake, "depsuser", StageOutcome::Success).await; + send_stage_event( + &nevent_tx, + StagePhase::Prebake, + "depsuser", + StageOutcome::Success, + ) + .await; // LateHook stage if stage <= PrebakeStage::LateHook { let _stage_start = Utc::now(); log::info!("Running PBStage: LateHook"); prebake_drop_privilege(PrebakeStage::LateHook, dropafter, target_user, ctx)?; - ctx.task_id = format!("{}-{}-latehook", cli.pipeline, cli.build_id); + ctx.task_id = format!("{}-{}-latehook", ctx.pipeline_name, ctx.build_id); let latehook = match &prebake.hooks { Some(hooks) => hooks.late.clone(), None => None, }; let result = stage::hook::hook(latehook, ctx, event_tx.clone()).await; result?; - send_stage_event(&nevent_tx, StagePhase::Prebake, "latehook", StageOutcome::Success).await; + send_stage_event( + &nevent_tx, + StagePhase::Prebake, + "latehook", + StageOutcome::Success, + ) + .await; } // Ready stage - ctx.task_id = format!("{}-{}-ready", cli.pipeline, cli.build_id); - send_stage_event(&nevent_tx, StagePhase::Prebake, "ready", StageOutcome::Success).await; + ctx.task_id = format!("{}-{}-ready", ctx.pipeline_name, ctx.build_id); + send_stage_event( + &nevent_tx, + StagePhase::Prebake, + "ready", + StageOutcome::Success, + ) + .await; log::info!("Prebake done!"); Ok(()) - } - -/// Send a stage completion event to the notification queue. -async fn send_stage_event( - nevent_tx: &Option, - stage: StagePhase, - substage: &str, - outcome: StageOutcome, -) { - if let Some(tx) = nevent_tx { - let event = crate::notify::types::NotificationEvent { - id: Uuid::new_v4(), - stage, - substage: substage.to_string(), - outcome, - not_before: chrono::Utc::now(), - priority: 0, - effective_priority: 0, - retry_backoff: 1.0, - attempted: 0, - target: Default::default(), - }; - let _ = tx.send(event).await; - } } diff --git a/workshop-baker/src/prebake/event.rs b/workshop-baker/src/prebake/event.rs index c79cac4..d3efa27 100644 --- a/workshop-baker/src/prebake/event.rs +++ b/workshop-baker/src/prebake/event.rs @@ -1,22 +1,33 @@ use workshop_engine::{ExecutionEvent, StreamType}; +use crate::types::debug::DebugFeature; use crate::bake::constant::PIPELINE_SKIP_ERRORCODE; pub async fn event_receiver( mut rx: tokio::sync::mpsc::Receiver, pipeline_name: String, build_id: String, + debug_flags: u32, ) { + let events_debug = DebugFeature::from_bits_retain(debug_flags).contains(DebugFeature::Events); while let Some(event) = rx.recv().await { match event { ExecutionEvent::TaskStarted { task_id, timestamp: _ } => { - log::debug!( - "[{}] [{}] Task started: {}", - pipeline_name, - build_id, - // timestamp, - task_id - ); + if events_debug { + log::info!( + "[{}] [{}] Task started: {}", + pipeline_name, + build_id, + task_id + ); + } else { + log::debug!( + "[{}] [{}] Task started: {}", + pipeline_name, + build_id, + task_id + ); + } } ExecutionEvent::OutputChunk { task_id: _, @@ -30,11 +41,19 @@ pub async fn event_receiver( }, ExecutionEvent::TaskCompleted { task_id, result } => { if result.success { - log::debug!( - "[{}] [{}] Task completed: {} (exit_code={}, duration={:?})", - pipeline_name, build_id, task_id, - result.exit_code, result.duration, - ); + if events_debug { + log::info!( + "[{}] [{}] Task completed: {} (exit_code={}, duration={:?})", + pipeline_name, build_id, task_id, + result.exit_code, result.duration, + ); + } else { + log::debug!( + "[{}] [{}] Task completed: {} (exit_code={}, duration={:?})", + pipeline_name, build_id, task_id, + result.exit_code, result.duration, + ); + } } else if result.exit_code == PIPELINE_SKIP_ERRORCODE { log::info!("[{}] [{}] Task skipped: {}", pipeline_name, build_id, task_id); } else { diff --git a/workshop-baker/src/prebake/security.rs b/workshop-baker/src/prebake/security.rs index 7437528..b4dacf9 100644 --- a/workshop-baker/src/prebake/security.rs +++ b/workshop-baker/src/prebake/security.rs @@ -231,6 +231,9 @@ pub fn prebake_drop_privilege( user ); ctx.privileged = false; + let home = format!("/home/{}", user); + //unsafe { std::env::set_var("HOME", &home); } + ctx.env_vars.insert("HOME".to_string(), home); } Ok(()) } diff --git a/workshop-baker/src/prebake/stage/bootstrap.rs b/workshop-baker/src/prebake/stage/bootstrap.rs index aec4778..b35fd05 100644 --- a/workshop-baker/src/prebake/stage/bootstrap.rs +++ b/workshop-baker/src/prebake/stage/bootstrap.rs @@ -79,7 +79,7 @@ pub async fn bootstrap( .await .map_err(BootstrapError::IOError)?; if ctx.dry_run { - log::info!("[DRY_RUN] Fetched bootstrap script:\n{}", content); + log::debug!("[DRY_RUN] Fetched bootstrap script:\n{}", content); return Ok(()); } write_bootstrap_content(&bootstrap_path, content)?; @@ -90,7 +90,7 @@ pub async fn bootstrap( // Inline content (not a valid getterURL pattern) else { if ctx.dry_run { - log::info!("[DRY_RUN] User provided inline bootstrap script:\n{}", raw); + log::debug!("[DRY_RUN] User provided inline bootstrap script:\n{}", raw); return Ok(()); } log::debug!("Using inline bootstrap script ({} bytes)", raw.len()); @@ -103,7 +103,7 @@ pub async fn bootstrap( } else { let script = generate_bootstrap(&bootstrap_config, config, osinfo, ctx.standalone)?; if ctx.dry_run { - log::info!("[DRY_RUN] Generated bootstrap script:\n{}", script); + log::debug!("[DRY_RUN] Generated bootstrap script:\n{}", script); return Ok(()); } log::debug!("Generated bootstrap script:\n{}", script); diff --git a/workshop-baker/src/prebake/stage/hook.rs b/workshop-baker/src/prebake/stage/hook.rs index b3dcea8..f1e4bd3 100644 --- a/workshop-baker/src/prebake/stage/hook.rs +++ b/workshop-baker/src/prebake/stage/hook.rs @@ -57,7 +57,6 @@ pub async fn hook( ctx: &ExecutionContext, event_tx: EventSender, ) -> Result<(), ExecutionError> { - // dbg!(ctx); let hook = match hook { Some(h) => h, None => { @@ -79,7 +78,6 @@ pub async fn hook( let result = engine .execute_command_with_delta(&hook_cmd.command, ctx, &event_tx, &deltactx) .await; - // dbg!(&result); if let Err(e) = result { log::error!("Hook {} failed: {:?}", index, e); return Err(e); diff --git a/workshop-baker/src/prefetch.rs b/workshop-baker/src/prefetch.rs new file mode 100644 index 0000000..5b6f884 --- /dev/null +++ b/workshop-baker/src/prefetch.rs @@ -0,0 +1,206 @@ +//! Resource prefetching for standalone mode. +//! +//! Populates a [`ResourceRegistry`] by fetching plugins and templates +//! from their external URLs **before** the pipeline runs. +//! +//! In worker mode this is the baker's responsibility; in daemon mode +//! the bakerd would call the same lower-level primitives (or replace +//! this module entirely). + +use std::path::{Path, PathBuf}; + +use crate::finalize::{FinalizeConfig, NotificationTemplateDef, PluginConfig, RawPluginMap}; +use crate::types::resource::ResourceRegistry; +use workshop_getterurl; + +/// Errors raised during resource prefetching. +#[derive(Debug, thiserror::Error)] +pub enum PrefetchError { + #[error("failed to construct resource URL for plugin '{name}': {detail}")] + UrlConstruction { name: String, detail: String }, + + #[error("failed to fetch '{url}': {source}")] + Fetch { + url: String, + #[source] + source: workshop_getterurl::GetterError, + }, + + #[error("file not found after resolving workspace path: {0}")] + FileNotFound(String), + + #[error(transparent)] + Io(#[from] std::io::Error), +} + +/// Fetch all external plugins and templates defined in [`FinalizeConfig`] +/// and register them into the given [`ResourceRegistry`]. +/// +/// * Plugins with the `late_binding` flag are **skipped** — they are +/// expected to be produced by the bake stage itself. +/// * Templates with a `use_` field (external URL) are fetched and +/// registered; inline templates are left as-is. +pub async fn prefetch_resources( + config: &FinalizeConfig, + registry: &mut ResourceRegistry, +) -> Result<(), PrefetchError> { + let _dir = tempfile::tempdir()?; + let target_base = _dir.path().to_path_buf(); + + prefetch_plugins(&config.plugin, registry, &target_base).await?; + prefetch_templates(&config.notification.templates, registry, &target_base).await?; + + Ok(()) +} + +// ── plugins ────────────────────────────────────────────────────────────── + +async fn prefetch_plugins( + plugins: &RawPluginMap, + registry: &mut ResourceRegistry, + cache: &Path, +) -> Result<(), PrefetchError> { + for (name, sources) in plugins { + let plugin_cfg = sources.values().next().ok_or_else(|| { + PrefetchError::UrlConstruction { + name: name.clone(), + detail: "no source URL configured".into(), + } + })?; + + if plugin_cfg.late_binding { + log::debug!("[prefetch] Skipping late-binding plugin '{}'", name); + continue; + } + + let source_url = sources.keys().next().unwrap(); // the URL string + let dest = fetch_plugin_resource(source_url, plugin_cfg, cache).await?; + registry.register(name, dest); + } + Ok(()) +} + +/// Build a full getter-url from a plugin source string and its associated +/// config, fetch it, and return the local path. +async fn fetch_plugin_resource( + source: &str, + cfg: &PluginConfig, + cache: &Path, +) -> Result { + // file:// — already local, just resolve and register + if let Some(path_str) = source.strip_prefix("file://") { + let abs = std::env::current_dir()?.join(path_str); + if !abs.exists() { + return Err(PrefetchError::FileNotFound(abs.display().to_string())); + } + return Ok(abs); + } + + // workspace:// — resolve relative to cwd then treat as file + if let Some(rel) = source.strip_prefix("workspace://") { + let abs = std::env::current_dir()?.join(rel); + if !abs.exists() { + return Err(PrefetchError::FileNotFound(abs.display().to_string())); + } + return Ok(abs); + } + + // Remote URL — add query params and fetch via getterurl + let url = enrich_url(source, cfg).map_err(|e| PrefetchError::UrlConstruction { + name: String::new(), + detail: e, + })?; + + workshop_getterurl::fetch(&url, cache) + .await + .map_err(|source| PrefetchError::Fetch { + url: url.clone(), + source, + }) +} + +// ── templates ──────────────────────────────────────────────────────────── + +async fn prefetch_templates( + templates: &std::collections::HashMap, + registry: &mut ResourceRegistry, + cache: &Path, +) -> Result<(), PrefetchError> { + for (name, tmpl) in templates { + let Some(ref url) = tmpl.use_ else { + continue; // inline template, no fetch needed + }; + + // file:// / workspace:// — resolve locally + if let Some(path_str) = url.strip_prefix("file://") { + let abs = std::env::current_dir()?.join(path_str); + if abs.exists() { + registry.register(name, abs); + } else { + log::warn!("[prefetch] template '{}' file not found: {}", name, abs.display()); + } + continue; + } + if let Some(rel) = url.strip_prefix("workspace://") { + let abs = std::env::current_dir()?.join(rel); + if abs.exists() { + registry.register(name, abs); + } else { + log::warn!("[prefetch] template '{}' workspace file not found: {}", name, abs.display()); + } + continue; + } + + // Remote template URL + match workshop_getterurl::fetch(url, cache).await { + Ok(path) => registry.register(name, path), + Err(e) => log::warn!("[prefetch] failed to fetch template '{}': {}", name, e), + } + } + Ok(()) +} + +// ── helpers ────────────────────────────────────────────────────────────── + +/// Append `?checksum=&depth=` to a plugin source URL when the +/// corresponding [`PluginConfig`] fields are set. +fn enrich_url(source: &str, cfg: &PluginConfig) -> Result { + // Validate the URL is parseable first + url::Url::parse(source).map_err(|e| format!("invalid URL '{}': {}", source, e))?; + + let mut result = source.to_string(); + let mut sep = if source.contains('?') { "&" } else { "?" }; + + if let Some(ref cs) = cfg.checksum { + result.push_str(sep); + result.push_str(&format!("checksum={}", cs)); + sep = "&"; + } + if let Some(true) = cfg.shallow { + result.push_str(sep); + result.push_str("depth=1"); + } + + Ok(result) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn test_enrich_url_checksum() { + let cfg = PluginConfig { + fallible: false, + retry: 0, + timeout_ms: 0, + late_binding: false, + checksum: Some("sha256:abc".into()), + shallow: None, + runtime: Default::default(), + config: serde_yaml::Value::Mapping(Default::default()), + }; + let url = enrich_url("https://example.com/plugin.tar.gz", &cfg).unwrap(); + assert_eq!(url, "https://example.com/plugin.tar.gz?checksum=sha256:abc"); + } +} diff --git a/workshop-baker/src/types.rs b/workshop-baker/src/types.rs index 30efcda..e9eab64 100644 --- a/workshop-baker/src/types.rs +++ b/workshop-baker/src/types.rs @@ -1,2 +1,3 @@ pub mod buildstatus; pub mod resource; +pub mod debug; diff --git a/workshop-baker/src/types/debug.rs b/workshop-baker/src/types/debug.rs new file mode 100644 index 0000000..543a0d3 --- /dev/null +++ b/workshop-baker/src/types/debug.rs @@ -0,0 +1,176 @@ +// Code in this module PARTIALLY or FULLY utilized AI Coding Agent +//! Debug mode feature flags for the HoneyBiscuitWorkshop pipeline. +//! +//! These flags are controlled by `--debug ` (comma-separated) +//! or the `HBW_DEBUG` environment variable. Each flag enables specific +//! behavioral changes that aid development and debugging. +//! +//! Unlike `RUST_LOG` (which controls log visibility), `DebugFeature` +//! controls **program behavior** — replacing real notifications with +//! dummy plugins, tracing internal event flow, etc. + +use bitflags::bitflags; + +bitflags! { + /// Bitmask of debug features enabled at runtime. + /// + /// Propagated through [`ExecutionContext`](workshop_engine::ExecutionContext) + /// so all pipeline stages can conditionally alter their behavior. + #[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] + pub struct DebugFeature: u32 { + /// Notification debug: replace real adapters with `Dummy` (no emails/webhooks). + const NotifyDummy = 1 << 0; + + /// Events: trace all `ExecutionEvent` (process lifecycle) and + /// `NotificationEvent` (stage outcomes) through the system. + const Events = 1 << 1; + + /// Scheduler: log DAG topology, batch selection, and scheduling + /// decisions at `info!` level (normally `debug!`/gated). + const Scheduler = 1 << 2; + + /// Script: display parsed script functions, generated final + /// scripts, and combinator chain details before execution. + const Script = 1 << 3; + } +} + +impl DebugFeature { + /// Parse a comma-separated `--debug` string into a `DebugFeature` bitmask. + /// + /// Unknown tokens are silently ignored. `"all"` enables all known features. + /// + /// # Examples + /// + /// ``` + /// use workshop_baker::types::debug::DebugFeature; + /// + /// let flags = DebugFeature::parse("notify.dummy,events"); + /// assert!(flags.contains(DebugFeature::NotifyDummy)); + /// assert!(flags.contains(DebugFeature::Events)); + /// assert!(!flags.contains(DebugFeature::Scheduler)); + /// + /// let all = DebugFeature::parse("all"); + /// assert!(all.contains(DebugFeature::all())); + /// ``` + pub fn _parse(raw: &str) -> Self { + if raw.contains("all") { + return Self::all(); + } + + let mut flags = Self::empty(); + for token in raw.split(',') { + let token = token.trim(); + match token { + "notify.dummy" => flags.insert(Self::NotifyDummy), + "events" => flags.insert(Self::Events), + "scheduler" => flags.insert(Self::Scheduler), + "script" => flags.insert(Self::Script), + "" => {} + unknown => log::warn!("Unknown debug feature: '{}'", unknown), + } + } + flags + } + + /// Print help for `--debug` features and return `true` if `help` was requested. + /// + /// Called before `env_logger::init()`, so output goes to stdout directly. + /// The caller should exit immediately when this returns `true`. + pub fn print_help(raw: &str) -> bool { + if raw.trim() != "help" { + return false; + } + println!("Debug features (comma-separated, use with --debug ):"); + println!(); + println!(" notify.dummy Replace real notification adapters with Dummy"); + println!(" (no emails/webhooks sent during debug)"); + println!(" events Trace all ExecutionEvent and NotificationEvent"); + println!(" through the event queue at info log level"); + println!(" scheduler Show DAG topology, batch selection, and scheduling"); + println!(" decisions each tick"); + println!(" script Display parsed bake.sh functions, generated final"); + println!(" scripts, and combinator chain details"); + println!(" all Enable all features above"); + println!(); + println!("Examples:"); + println!(" workshop-baker --debug notify.dummy,events ..."); + println!(" workshop-baker --debug all ..."); + println!(" HBW_DEBUG=scheduler,script workshop-baker ..."); + true + } + + /// Parse from an `Option<&str>`, returning empty flags for `None`. + pub fn parse(raw: Option<&str>) -> Self { + match raw { + Some(s) if !s.is_empty() => Self::_parse(s), + _ => Self::empty(), + } +} +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn parse_empty() { + assert_eq!(DebugFeature::_parse(""), DebugFeature::empty()); + } + + #[test] + fn parse_single() { + let f = DebugFeature::_parse("notify.dummy"); + assert!(f.contains(DebugFeature::NotifyDummy)); + assert!(!f.contains(DebugFeature::Events)); + assert!(!f.contains(DebugFeature::Scheduler)); + } + + #[test] + fn parse_multiple() { + let f = DebugFeature::_parse("notify.dummy,events"); + assert!(f.contains(DebugFeature::NotifyDummy)); + assert!(f.contains(DebugFeature::Events)); + assert!(!f.contains(DebugFeature::Scheduler)); + } + + #[test] + fn parse_all() { + let f = DebugFeature::_parse("all"); + assert!(f.contains(DebugFeature::NotifyDummy)); + assert!(f.contains(DebugFeature::Events)); + assert!(f.contains(DebugFeature::Scheduler)); + assert!(f.contains(DebugFeature::Script)); + } + + #[test] + fn parse_whitespace() { + let f = DebugFeature::_parse("notify.dummy , scheduler"); + assert!(f.contains(DebugFeature::NotifyDummy)); + assert!(f.contains(DebugFeature::Scheduler)); + } + + #[test] + fn parse_unknown_is_ignored() { + let f = DebugFeature::_parse("notify.dummy,foobar,events"); + assert!(f.contains(DebugFeature::NotifyDummy)); + assert!(f.contains(DebugFeature::Events)); + // unknown "foobar" just logged and skipped + } + + #[test] + fn parse_optional_none() { + assert_eq!(DebugFeature::parse(None), DebugFeature::empty()); + } + + #[test] + fn parse_optional_empty_str() { + assert_eq!(DebugFeature::parse(Some("")), DebugFeature::empty()); + } + + #[test] + fn parse_optional_some() { + let f = DebugFeature::parse(Some("events")); + assert!(f.contains(DebugFeature::Events)); + } +} diff --git a/workshop-baker/src/types/resource.rs b/workshop-baker/src/types/resource.rs index d7997b5..f2643a0 100644 --- a/workshop-baker/src/types/resource.rs +++ b/workshop-baker/src/types/resource.rs @@ -6,15 +6,15 @@ //! exclusively through `resolve()` — the registry is treated as read-only //! once the pipeline is running. +use serde::{Deserialize, Serialize}; use std::collections::HashMap; use std::path::PathBuf; - /// A locally resolved resource with a logical name and filesystem path. /// /// Created by the resource fetcher (bakerd in worker mode, or the baker /// itself in standalone mode) and consumed by pipeline stages via the /// [`ResourceRegistry`]. -#[derive(Debug, Clone, PartialEq, Eq)] +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] pub struct Resource { /// Logical name used for lookup (e.g., `"plugin/docker-push"`). pub name: String, @@ -36,7 +36,7 @@ impl Resource { /// /// Populated during the fetch phase (before prebake) and then treated as /// immutable for the remainder of the pipeline. -#[derive(Debug, Clone, Default)] +#[derive(Debug, Clone, Default, Serialize, Deserialize)] pub struct ResourceRegistry { map: HashMap, } @@ -78,6 +78,17 @@ impl ResourceRegistry { pub fn is_empty(&self) -> bool { self.map.is_empty() } + + /// Deserialize from a flat JSON: `{ "name": "/path", ... }` + pub fn from_json_str(json: &str) -> Result { + let flat: HashMap = serde_json::from_str(json) + .map_err(|e| format!("invalid resource registry JSON: {}", e))?; + let mut reg = ResourceRegistry::new(); + for (name, path) in flat { + reg.register(&name, path); + } + Ok(reg) + } } #[cfg(test)] diff --git a/workshop-engine/src/types.rs b/workshop-engine/src/types.rs index cc224d0..ea08aa9 100644 --- a/workshop-engine/src/types.rs +++ b/workshop-engine/src/types.rs @@ -11,6 +11,7 @@ pub struct ExecutionContext { pub task_id: String, pub pipeline_name: String, pub username: String, + pub build_id: String, pub working_dir: PathBuf, pub env_vars: HashMap, pub timeout: Option, @@ -21,6 +22,9 @@ pub struct ExecutionContext { /// stdin data piped by the scheduler (e.g. via `@pipe`). /// Consumed (taken) by engine during execution. pub stdin: Option>, + /// Bitmask of active debug features, decoded by `workshop-baker`. + /// 0 means no debug features. See `DebugFeature` for supported flags. + pub debug_flags: u32, } #[derive(Debug, Clone, Default)] @@ -36,6 +40,7 @@ impl Default for ExecutionContext { task_id: String::new(), pipeline_name: String::new(), username: String::new(), + build_id: String::new(), working_dir: std::env::current_dir().unwrap_or_else(|_| PathBuf::from("/")), env_vars: HashMap::new(), timeout: None, @@ -43,6 +48,7 @@ impl Default for ExecutionContext { dry_run: false, privileged: false, standalone: false, + debug_flags: 0, stdin: None, } }