From fdc00f0635022158396fe9dc0ebcb033ee7b80ee Mon Sep 17 00:00:00 2001 From: Catty Steve <4795515+Catty2014@user.noreply.gitee.com> Date: Wed, 17 Jun 2026 20:43:17 +0800 Subject: [PATCH] feat(baker): add debug feature flags and resource prefetching Add DebugFeature bitflags for controlling notify.dummy, events, scheduler, and script debug behaviors. Add prefetch module to fetch plugins and templates before pipeline runs. Add cleanup and artifact handling to finalize stage. Add build_id to ExecutionContext. --- workshop-baker-params/src/notification.rs | 2 +- workshop-baker/Cargo.lock | 91 +++++++ workshop-baker/Cargo.toml | 5 + workshop-baker/src/bake.rs | 26 +- workshop-baker/src/bake/builder.rs | 12 +- workshop-baker/src/bake/execute.rs | 12 +- workshop-baker/src/bake/parser.rs | 2 +- workshop-baker/src/bake/trivial.rs | 2 +- workshop-baker/src/bare.rs | 19 +- workshop-baker/src/daemon.rs | 0 workshop-baker/src/{prebake => daemon}/env.rs | 0 .../src/{prebake => daemon}/env/baremetal.rs | 0 .../src/{prebake => daemon}/env/container.rs | 0 .../src/{prebake => daemon}/env/custom.rs | 0 .../{prebake => daemon}/env/firecracker.rs | 0 workshop-baker/src/finalize.rs | 16 +- workshop-baker/src/finalize/artifact.rs | 139 +++++++++++ workshop-baker/src/finalize/cleanup.rs | 67 ++++++ workshop-baker/src/finalize/config.rs | 6 +- workshop-baker/src/finalize/event.rs | 55 +++-- workshop-baker/src/finalize/plugin.rs | 2 +- .../src/finalize/plugin/internal.rs | 6 +- .../src/finalize/plugin/internal/dummy.rs | 6 +- .../src/finalize/plugin/metadata.rs | 2 +- workshop-baker/src/lib.rs | 31 ++- workshop-baker/src/main.rs | 222 ++++++++++++++---- workshop-baker/src/notify.rs | 65 +++-- workshop-baker/src/notify/handler.rs | 4 + workshop-baker/src/notify/queue.rs | 26 +- workshop-baker/src/notify/types.rs | 70 +++--- workshop-baker/src/prebake.rs | 93 ++++---- workshop-baker/src/prebake/event.rs | 43 +++- workshop-baker/src/prebake/security.rs | 3 + workshop-baker/src/prebake/stage/bootstrap.rs | 6 +- workshop-baker/src/prebake/stage/hook.rs | 2 - workshop-baker/src/prefetch.rs | 206 ++++++++++++++++ workshop-baker/src/types.rs | 1 + workshop-baker/src/types/debug.rs | 176 ++++++++++++++ workshop-baker/src/types/resource.rs | 17 +- workshop-engine/src/types.rs | 6 + 40 files changed, 1207 insertions(+), 234 deletions(-) create mode 100644 workshop-baker/src/daemon.rs rename workshop-baker/src/{prebake => daemon}/env.rs (100%) rename workshop-baker/src/{prebake => daemon}/env/baremetal.rs (100%) rename workshop-baker/src/{prebake => daemon}/env/container.rs (100%) rename workshop-baker/src/{prebake => daemon}/env/custom.rs (100%) rename workshop-baker/src/{prebake => daemon}/env/firecracker.rs (100%) create mode 100644 workshop-baker/src/finalize/artifact.rs create mode 100644 workshop-baker/src/finalize/cleanup.rs create mode 100644 workshop-baker/src/prefetch.rs create mode 100644 workshop-baker/src/types/debug.rs 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, } }