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

feat(baker/finalize): notification system and consolidate status types

Browse Source 
- Rename types/build_status.rs to types/buildstatus.rs
- Rename StageInfo.sub_stage field to substage for consistency
- Move notification logic from finalize to dedicated notify module
- Remove notification module from finalize (moved to notify)
- Add notify module with event queuing, retry, and priority scheduling
- Add exponential backoff for temporary notification failures
- Implement plugin filtering by name list during registration
- Add fetch_plugins helper for batch plugin download
- Update finalize plugin registration to accept optional plugin filter
- Fix finalize config template indentation for notification policy
- Move all YAML templates to templates/ directory
This commit is contained in:
Catty Steve
2026-04-23 16:18:20 +08:00
parent 07d76d612d
commit 117660d5c4
13 changed files with 342 additions and 102 deletions
Download Patch File Download Diff File Expand all files Collapse all files
bake.sh.tmpl → templates/bake.sh.tmpl
View File
finalize.yml.tmpl → templates/finalize.yml.tmpl
+10 -5
View File
@@ -156,11 +156,16 @@ notification:
sms-notify:
to: "+8610012345678"
schema: "custom"
on_success: "[HBW Notify] Build Succeeded: {{ build.status }} ({{ build.duration }})"
on_failure: "[HBW Notify] Build Failed: {{ build.status }} ({{ build.duration }})"
on_finish_of:
stage: "bake.build"
content: "[HBW Notify] Build Progress: {{ build.status }} ({{ build.duration }})"
policy:
- on_success: "[HBW Notify] Build Succeeded: {{ build.status }} ({{ build.duration }})"
to: "+8610012345678"
- on_failure: "[HBW Notify] Build Failed: {{ build.status }} ({{ build.duration }})"
to: "+8610012345679"
- on_finish_of:
stage: "bake.build"
content: "[HBW Notify] Build Progress: {{ build.status }} ({{ build.duration }})"
to: "+8610012345677"
fallible: true
# =============================================================================
prebake.yml.tmpl → templates/prebake.yml.tmpl
View File
workshop-baker/src/bake.rs
+3 -3
View File
@@ -7,7 +7,7 @@ use crate::ExecutionContext;
use crate::prebake::PrebakeConfig;
use crate::prebake::security::get_drop_after;
use crate::prebake::stage::PrebakeStage;
use crate::types::build_status::{write_stage_status, StageInfo, StagePhase, StageResult};
use crate::types::buildstatus::{write_stage_status, StageInfo, StagePhase, StageResult};
use crate::{Engine, prebake};
use std::path::{PathBuf, Path};
use chrono::Utc;
@@ -71,7 +71,7 @@ pub async fn bake(
StageInfo {
name: "trivial".to_string(),
phase: StagePhase::Bake,
sub_stage: "trivial".to_string(),
substage: "trivial".to_string(),
result: stage_result,
duration_ms: Utc::now().signed_duration_since(stage_start).num_milliseconds() as u64,
started_at: stage_start,
@@ -101,7 +101,7 @@ pub async fn bake(
StageInfo {
name: func.name.clone(),
phase: StagePhase::Bake,
sub_stage: func.name.clone(),
substage: func.name.clone(),
result: stage_result,
duration_ms: Utc::now().signed_duration_since(stage_start).num_milliseconds() as u64,
started_at: stage_start,
workshop-baker/src/finalize.rs
+6 -64
View File
@@ -1,7 +1,6 @@
pub mod config;
pub mod error;
pub mod event;
pub mod notification;
pub mod plugin;
pub mod stage;
pub mod template;
@@ -12,35 +11,15 @@ use std::path::PathBuf;
use crate::finalize::plugin::fetch::{self, FetchArgument};
use crate::cli::Cli;
use crate::engine::EventSender;
use crate::types::build_status::{read_build_status, BuildStatus, StageResult};
use crate::types::build_status::{StageInfo, StagePhase};
use crate::types::buildstatus::{read_build_status, BuildStatus, StageResult};
use crate::types::buildstatus::{StageInfo, StagePhase};
use chrono::Utc;
// use crate::finalize::error::PluginError;
pub use config::*;
pub use error::FinalizeError;
use std::path::Path;
fn _extract_notification_plugins(
all_plugins: &plugin::PluginMap,
_notification_config: &config::NotificationConfig,
) -> Vec<String> {
use std::collections::HashSet;
let notification_names: HashSet<&str> = HashSet::from([
"mail",
"webhook",
"satori",
"in-site-notify",
]);
all_plugins
.iter()
.filter(|(name, _)| notification_names.contains(name.as_str()))
.map(|(name, _)| name.clone())
.collect()
}
fn parse(path: &Path) -> Result<FinalizeConfig, FinalizeError> {
pub fn parse(path: &Path) -> Result<FinalizeConfig, FinalizeError> {
let content = std::fs::read_to_string(path).map_err(FinalizeError::IoError)?;
let config: FinalizeConfig = serde_yaml::from_str(&content)?;
Ok(config)
@@ -86,7 +65,7 @@ pub async fn finalize(
// Register Plugins
log::info!("Registering {} plugins", finalize.plugin.len());
let registered_plugins = plugin::register(finalize.plugin)?;
let registered_plugins = plugin::register(finalize.plugin, None)?;
// Early hook
let earlyhook_result = stage::hook::hook(
@@ -106,43 +85,6 @@ pub async fn finalize(
// If prior stages failed, we still send notification but skip artifacts
let prior_failed = build_status.has_failures() || earlyhook_result.is_err();
// Send notification (before late hook, as per standalone flow)
if !finalize.notification.groups.is_empty() {
let core = notification::NotificationCore::new(registered_plugins.clone());
let request = notification::NotificationRequest {
pipeline_name: ctx.pipeline_name.clone(),
build_id: ctx.task_id.clone(),
build_status: build_status.clone(),
final_result: final_result.clone(),
config: finalize.notification,
};
if let Err(e) = core.dispatch(request, ctx, event_tx.clone()).await {
log::warn!("Notification dispatch failed: {}", e);
// Notification failure does not block finalize
}
} else {
log::debug!("No notification groups configured, skipping notification");
}
if prior_failed {
log::warn!("Prior stages failed, skipping artifacts and late hook");
// Write finalize stage status as failure
let _ = crate::types::build_status::write_stage_status(
STANDALONE_STATUS_DIR,
StageInfo {
name: "finalize".to_string(),
phase: StagePhase::Finalize,
sub_stage: "notify".to_string(),
result: StageResult::Failure,
duration_ms: 0,
started_at: Utc::now(),
finished_at: Utc::now(),
error_message: Some("Prior stage failed".to_string()),
},
);
return Ok(());
}
// Prepare artifacts
// TODO: artifact stage implementation
@@ -156,12 +98,12 @@ pub async fn finalize(
.await?;
// Write finalize success status
let _ = crate::types::build_status::write_stage_status(
let _ = crate::types::buildstatus::write_stage_status(
STANDALONE_STATUS_DIR,
StageInfo {
name: "finalize".to_string(),
phase: StagePhase::Finalize,
sub_stage: "clean".to_string(),
substage: "clean".to_string(),
result: StageResult::Success,
duration_ms: 0,
started_at: Utc::now(),
workshop-baker/src/finalize/plugin.rs
+49 -16
View File
@@ -3,16 +3,16 @@ use crate::finalize::RawPluginMap;
use crate::finalize::plugin::metadata::{PluginMetadata, PluginType};
use crate::{EventSender, ExecutionContext, finalize::error::PluginError};
use async_trait::async_trait;
use std::collections::HashMap;
use std::path::{PathBuf, Path};
use std::collections::{HashMap, HashSet};
use std::path::{Path, PathBuf};
use std::sync::Arc;
mod dylib;
pub mod fetch;
mod internal;
mod metadata;
mod rhai;
mod shell;
pub mod fetch;
#[derive(Clone)]
pub struct FinalizePlugin {
@@ -99,7 +99,7 @@ fn read_manifest(path: Option<PathBuf>, name: &str) -> Result<PluginMetadata, Pl
for ext in FINALIZE_PLUGIN_MANIFEST_EXTENSION {
match load_manifest(&path, name, ext) {
Ok(content) => break 'load content,
Err(PluginError::ManifestNotFound(_)) => {},
Err(PluginError::ManifestNotFound(_)) => {}
Err(e) => return Err(e),
}
}
@@ -107,26 +107,35 @@ fn read_manifest(path: Option<PathBuf>, name: &str) -> Result<PluginMetadata, Pl
};
log::debug!("[plugin:{}] Parsing manifest content", name);
let mut metadata: PluginMetadata = serde_yaml::from_str(manifest_content.as_str()).map_err(|e| PluginError::PluginUnavailable {
name: name.to_string(),
reason: format!("Failed to parse plugin manifest: {}", e),
})?;
let mut metadata: PluginMetadata =
serde_yaml::from_str(manifest_content.as_str()).map_err(|e| {
PluginError::PluginUnavailable {
name: name.to_string(),
reason: format!("Failed to parse plugin manifest: {}", e),
}
})?;
metadata.fspath = path;
Ok(metadata)
}
fn load_manifest(path: &Path, name: &str, suffix: &str) -> Result<String, PluginError> {
let manifest_path = path.join(format!("{}{}", FINALIZE_PLUGIN_MANIFEST,suffix));
log::debug!("[plugin:{}] Reading manifest from {}", name, manifest_path.display());
let manifest_path = path.join(format!("{}{}", FINALIZE_PLUGIN_MANIFEST, suffix));
log::debug!(
"[plugin:{}] Reading manifest from {}",
name,
manifest_path.display()
);
if !manifest_path.exists() {
return Err(PluginError::ManifestNotFound(manifest_path));
}
std::fs::read_to_string(&manifest_path).map_err(|e| {
PluginError::PluginUnavailable {
name: name.to_string(),
reason: format!("Failed to read plugin manifest from {}: {}", manifest_path.display(), e.to_string()),
}
std::fs::read_to_string(&manifest_path).map_err(|e| PluginError::PluginUnavailable {
name: name.to_string(),
reason: format!(
"Failed to read plugin manifest from {}: {}",
manifest_path.display(),
e.to_string()
),
})
}
@@ -202,17 +211,38 @@ fn get_entry(plugin_type: PluginType, name: &str) -> Result<Box<dyn AsyncPluginF
///
/// Returns a [`PluginMap`] containing all registered plugins, or a [`PluginError`] if
/// a plugin manifest cannot be read or parsed.
pub fn register(external_plugins: RawPluginMap) -> Result<PluginMap, PluginError> {
pub fn register(
external_plugins: RawPluginMap,
list: Option<&HashSet<String>>,
) -> Result<PluginMap, PluginError> {
let mut plugins: PluginMap = HashMap::new();
let mut count = 0;
// Register internal plugin
for i in internal::get_internal_plugin() {
let name = i.name.to_string();
match list {
Some(list) if !list.contains(&name) => {
log::debug!("Plugin {} not in list, skipping", name);
continue;
}
_ => {}
}
log::debug!("Registering internal plugin {}", i.name);
plugins.insert(i.name.to_string(), FinalizePlugin::new(i.entry));
count += 1;
}
// Register external plugin
for (name, data) in external_plugins.into_iter() {
match list {
Some(list) if !list.contains(&name) => {
log::debug!("Plugin {} not in list, skipping", name);
continue;
}
_ => {}
}
log::debug!("Registering external plugin {}", name);
let (_source, config) = data.into_iter().next().unwrap_or_else(|| {
panic!(
@@ -234,8 +264,11 @@ pub fn register(external_plugins: RawPluginMap) -> Result<PluginMap, PluginError
.with_metadata(metadata)
.with_argument(config.config),
);
count += 1;
}
log::info!("Registered {} plugins", count);
Ok(plugins)
}
workshop-baker/src/finalize/plugin/fetch.rs
+40 -3
View File
@@ -1,12 +1,17 @@
// Code in this module PARTIALLY OR FULLY utilized AI Coding Agent.
use crate::finalize::plugin::PluginError;
use std::collections::HashSet;
use crate::finalize::fetch;
use crate::finalize::{
PluginConfig,
plugin::{PluginError, RawPluginMap},
};
use std::path::Path;
mod types;
mod checksum;
mod extract;
mod http;
mod git;
mod http;
mod types;
pub use types::FetchArgument;
@@ -72,3 +77,35 @@ pub async fn fetch_plugin(
}
}
}
pub async fn fetch_plugins(
plugin_path: &Path,
plugin_config: &mut RawPluginMap,
list: Option<&HashSet<String>>,
) -> Result<(), PluginError> {
log::info!("Preparing to download {} plugins", plugin_config.len());
for (name, data) in plugin_config.iter_mut() {
match list {
Some(list) if !list.contains(name) => {
log::debug!("Plugin {} not in list, skipping", name);
continue;
}
_ => {}
}
log::debug!("Downloading plugin {}", name);
let (source, config) = data.iter_mut().next().ok_or_else(|| {
PluginError::PluginUnavailable {
name: name.clone(),
reason: "no source configured".to_string(),
}
})?;
let fetch_argument = FetchArgument {
checksum: config.checksum.clone(),
shallow: config.shallow,
};
fetch::fetch_plugin(&source, &name, &plugin_path, fetch_argument).await?;
log::debug!("Storing plugin {} to {}", name, plugin_path.display());
config.runtime.fspath = Some(plugin_path.join(name));
}
Ok(())
}
workshop-baker/src/lib.rs
+1
View File
@@ -11,6 +11,7 @@ pub mod bake;
pub mod socket;
pub mod types;
pub mod constant;
pub mod notify;
// External crates re-exports
pub use config;
workshop-baker/src/notify.rs
+203
View File
@@ -0,0 +1,203 @@
use crate::constant::FINALIZE_STANDALONE_PLUGIN_PATH;
use crate::finalize::FinalizeConfig;
use crate::finalize::NotificationConfig;
use crate::finalize::parse;
use crate::finalize::plugin;
use crate::finalize::plugin::FinalizePlugin;
use crate::finalize::plugin::fetch;
use crate::types::buildstatus::StagePhase;
use crate::types::buildstatus::StageResult;
use crate::{EventSender, ExecutionContext};
use chrono::{DateTime, Utc};
use serde::Serialize;
use std::collections::HashMap;
use std::collections::HashSet;
use std::hash::Hash;
use std::path::Path;
use std::path::PathBuf;
use uuid::Uuid;
mod error;
use error::NotifyError;
fn validate(finalize: &FinalizeConfig) -> anyhow::Result<()> {
let validate_result = finalize.validate();
if let Err(errors) = validate_result {
for error in &errors {
log::error!("Error: {}", error);
}
anyhow::bail!(NotifyError::FinalizeValidationError(errors))
};
Ok(())
}
pub type NotifyPluginMap = HashMap<String, FinalizePlugin>;
fn extract_plugin(finalize: &FinalizeConfig) -> HashSet<String> {
finalize
.notification
.groups
.iter()
.flat_map(|(_, methods)| methods.keys())
.cloned()
.collect()
}
#[derive(Debug, Clone, Serialize)]
pub struct NotifyEvent {
id: Uuid,
stage: StagePhase,
substage: String,
result: StageResult,
not_before: DateTime<Utc>,
priority: u32,
effective_priority: u32,
lives: u32,
max_lives: u32,
target: HashSet<String>, // Target audience
}
impl Hash for NotifyEvent {
fn hash<H: std::hash::Hasher>(&self, state: &mut H) {
self.id.hash(state);
}
}
impl PartialEq for NotifyEvent {
fn eq(&self, other: &Self) -> bool {
self.id == other.id
}
}
impl Eq for NotifyEvent {}
impl PartialOrd for NotifyEvent {
fn partial_cmp(&self, other: &Self) -> Option<std::cmp::Ordering> {
Some(
self.effective_priority
.cmp(&other.effective_priority)
.then(self.not_before.cmp(&other.not_before))
.reverse(),
)
}
}
impl Ord for NotifyEvent {
fn cmp(&self, other: &Self) -> std::cmp::Ordering {
self.partial_cmp(other).unwrap()
}
}
type BundledNotifyEvent = Vec<NotifyEvent>;
type NotifyEventReceiver = tokio::sync::mpsc::Receiver<BundledNotifyEvent>;
pub type NotifyEventSender = tokio::sync::mpsc::Sender<NotifyEvent>;
const NOTIFY_NOTREADY_DELAY: std::time::Duration = std::time::Duration::from_secs(5);
const NOTIFY_NOTREADY_PRIORITY: u32 = 99;
const NOTIFY_TEMPORARY_DELAY_FACTOR: f64 = 2.0;
const NOTIFY_TEMPORARY_DELAY_TIME: f64 = 5.0;
const NOTIFY_TEMPORARY_MAX_TIME: f64 = f64::MAX;
pub async fn notify(
finalize_path: &Path,
ctx: &mut ExecutionContext,
event_tx: EventSender,
notifyevent_rx: &mut NotifyEventReceiver,
notifyevent_tx: NotifyEventSender,
) -> anyhow::Result<()> {
let mut finalize = parse(finalize_path)?;
validate(&finalize)?;
let plugin_list = extract_plugin(&finalize);
if plugin_list.is_empty() {
log::warn!("No notification method provided, skipping notification");
// TODO: send event to event_tx
return Ok(());
}
if ctx.standalone {
fetch::fetch_plugins(
&PathBuf::from(FINALIZE_STANDALONE_PLUGIN_PATH),
&mut finalize.plugin,
Some(&plugin_list),
)
.await?;
}
let notify_plugins = plugin::register(finalize.plugin, Some(&plugin_list))?;
loop {
tokio::select! {
Some(events) = notifyevent_rx.recv() => {
match notify_handler(&notify_plugins, &finalize.notification, &events){
Ok(_) => {}, // Notification succeeded / failed with fallible
Err(e) => {
log::debug!("Failed to send notification: {:?}", e);
match e {
NotifyError::PluginNotAvailable(_) => {
// Should keep lives, add static delay and give less priority
for mut event in events {
event.not_before = Utc::now() + NOTIFY_NOTREADY_DELAY;
event.effective_priority = event.effective_priority.saturating_add(1);
notifyevent_tx.send(event).await?;
}
},
NotifyError::TemporaryError(_) => {
// Should cost 1 life, add dynamic (exponential) delay
for mut event in events {
event.lives -= 1;
event.not_before = Utc::now() + notify_delay(&event);
notifyevent_tx.send(event).await?;
}
},
NotifyError::PermanentError(_) => {
// Unexpected!
// Should be dropped
log::error!("None of the configured notification plugins works.");
log::error!("Message will be dropped.");
// TODO: Send audit/metric information
}
error => {
// Unexpected!
log::error!("Internal error: notify_handler returned unexpected error: {:?}", error);
}
};
}
};
}
}
}
}
fn notify_handler(
plugins: &NotifyPluginMap,
config: &NotificationConfig,
event: &Vec<NotifyEvent>,
) -> Result<(), NotifyError> {
todo!();
}
// calculate min(1, k/x)*t as notification grouping period
// to avoid excessive notification frequency
fn wait_time(group_period: u32, group_watermark: u32, count: u32) -> f32 {
if count == 0 {
log::warn!("Unexpected notification count: {}", count);
return group_period as f32;
}
f32::min(1.0, group_watermark as f32 / count as f32) * group_period as f32
}
// calculate p^(m-l)*t as notification delay
fn notify_delay(event: &NotifyEvent) -> std::time::Duration {
let exponent = (event.max_lives - event.lives) as i32;
if exponent < 0 {
log::warn!("Unexpected lives in notification: lives({}) > max_lives({})", event.lives, event.max_lives);
return std::time::Duration::from_secs_f64(NOTIFY_TEMPORARY_MAX_TIME);
}
let multiplier = NOTIFY_TEMPORARY_DELAY_FACTOR.powi(exponent);
let delay = multiplier * NOTIFY_TEMPORARY_DELAY_TIME;
if !delay.is_finite() || delay < 0.0 {
log::warn!("Unexpected calculated delay: {}, treating as infinite", delay);
return std::time::Duration::from_secs_f64(NOTIFY_TEMPORARY_MAX_TIME);
}
std::time::Duration::from_secs_f64(delay)
}
workshop-baker/src/notify/error.rs
+19
View File
@@ -0,0 +1,19 @@
use thiserror::Error;
#[derive(Error, Debug)]
pub enum NotifyError{
#[error("Finalize config validation failed: {0:?}")]
FinalizeValidationError(Vec<String>),
#[error("Notification plugin not available {0}")]
PluginNotAvailable(String),
#[error("Temporary failure: {0}")]
TemporaryError(String),
#[error("Permanent error: {0}")]
PermanentError(String),
#[error("Notification failed after {0} attempt(s)")]
Exhausted(u32),
}
workshop-baker/src/prebake.rs
+6 -6
View File
@@ -32,7 +32,7 @@ use crate::{
error::PrebakeError,
ExecutionContext,
prebake::{security::get_drop_after, stage::PrebakeStage},
types::build_status::{write_stage_status, StageInfo, StagePhase, StageResult},
types::buildstatus::{write_stage_status, StageInfo, StagePhase, StageResult},
constant::STANDALONE_STATUS_DIR,
};
use std::path::Path;
@@ -137,7 +137,7 @@ mut ctx: &mut ExecutionContext,
StageInfo {
name: "bootstrap".to_string(),
phase: StagePhase::Prebake,
sub_stage: "bootstrap".to_string(),
substage: "bootstrap".to_string(),
result: stage_result,
duration_ms: Utc::now().signed_duration_since(stage_start).num_milliseconds() as u64,
started_at: stage_start,
@@ -165,7 +165,7 @@ mut ctx: &mut ExecutionContext,
StageInfo {
name: "earlyhook".to_string(),
phase: StagePhase::Prebake,
sub_stage: "earlyhook".to_string(),
substage: "earlyhook".to_string(),
result: stage_result,
duration_ms: Utc::now().signed_duration_since(stage_start).num_milliseconds() as u64,
started_at: stage_start,
@@ -247,7 +247,7 @@ mut ctx: &mut ExecutionContext,
StageInfo {
name: "depssystem".to_string(),
phase: StagePhase::Prebake,
sub_stage: "depssystem".to_string(),
substage: "depssystem".to_string(),
result: stage_result,
duration_ms: Utc::now().signed_duration_since(stage_start).num_milliseconds() as u64,
started_at: stage_start,
@@ -272,7 +272,7 @@ mut ctx: &mut ExecutionContext,
StageInfo {
name: "depsuser".to_string(),
phase: StagePhase::Prebake,
sub_stage: "depsuser".to_string(),
substage: "depsuser".to_string(),
result: stage_result,
duration_ms: Utc::now().signed_duration_since(stage_start).num_milliseconds() as u64,
started_at: stage_start,
@@ -301,7 +301,7 @@ mut ctx: &mut ExecutionContext,
StageInfo {
name: "latehook".to_string(),
phase: StagePhase::Prebake,
sub_stage: "latehook".to_string(),
substage: "latehook".to_string(),
result: stage_result,
duration_ms: Utc::now().signed_duration_since(stage_start).num_milliseconds() as u64,
started_at: stage_start,
workshop-baker/src/types.rs
+1 -1
View File
@@ -1,5 +1,5 @@
pub mod architecture;
pub mod build_status;
pub mod buildstatus;
pub mod builderconfig;
pub mod cache;
pub mod command;
workshop-baker/src/types/build_status.rs → workshop-baker/src/types/buildstatus.rs
+4 -4
View File
@@ -41,7 +41,7 @@ impl std::fmt::Display for StageResult {
pub struct StageInfo {
pub name: String,
pub phase: StagePhase,
pub sub_stage: String,
pub substage: String,
pub result: StageResult,
pub duration_ms: u64,
pub started_at: DateTime<Utc>,
@@ -51,7 +51,7 @@ pub struct StageInfo {
impl StageInfo {
pub fn full_name(&self) -> String {
format!("{}.{}", self.phase, self.sub_stage)
format!("{}.{}", self.phase, self.substage)
}
}
@@ -143,7 +143,7 @@ mod tests {
StageInfo {
name: name.to_string(),
phase,
sub_stage: name.to_string(),
substage: name.to_string(),
result,
duration_ms: 1000,
started_at: now,
@@ -161,7 +161,7 @@ mod tests {
let status = read_build_status(&dir).unwrap();
assert_eq!(status.stages.len(), 1);
assert_eq!(status.stages[0].sub_stage, "bootstrap");
assert_eq!(status.stages[0].substage, "bootstrap");
assert_eq!(status.stages[0].result, StageResult::Success);
}