feat(bake): DAG scheduler, decorator execution, notify refactor, workshop-schedule crate

- Add workshop-schedule crate (petgraph-based DAG scheduling)
- Implement all bake decorators with combinator chain execution
- Refactor notify module: extract handler, rule, template, util, queue
- Add NotificationQueue with priority/drain semantics
- Enhance finalize plugin system with internal plugin registration
- Update ExecutionContext and event types
- Add per-crate AGENTS.md knowledge base files
- Remove inline AGENTS.md files (consolidated into per-crate docs)
This commit is contained in:
Catty Steve
2026-06-15 20:45:06 +08:00
parent 3629c33fc4
commit 26809df720
44 changed files with 1691 additions and 1032 deletions
+97 -62
View File
@@ -1,92 +1,127 @@
# PROJECT KNOWLEDGE BASE
**Generated:** 2026-05-19
**Commit:** 28abc5d
**Branch:** master
**Generated:** 2026-06-12T16:08:15Z
**Commit:** 3629c33
**Branch:** feat/bake-dag-tester
## OVERVIEW
HoneyBiscuitWorkshop ("蜜饼工坊") — Rust CI/CD system with decorator-driven shell pipeline DSL and LLM-powered configuration. 3 independent Cargo crates, no workspace.
HoneyBiscuitWorkshop (蜜饼工坊) — Rust-based CI/CD system inspired by Concourse CI. Pipeline: prebake.yml → bake.sh → finalize.yml. Multi-builder (Docker, Firecracker, bare metal). Plugin system.
## STRUCTURE
```
./
├── workshop-baker/ # Main orchestrator (CLI + pipeline stages)
├── workshop-engine/ # Execution engine (extracted from baker)
├── workshop-getterurl/ # URL fetching (go-getter style)
├── docs/ # mdBook documentation (zh-CN)
├── templates/ # Pipeline templates (bake.sh, finalize.yml, prebake.yml)
── archived/ # Deprecated crates (7 moved here)
├── workshop-baker/ # Main binary + lib — orchestrator, CLI, plugins
├── workshop-engine/ # Core execution engine — process mgmt, package managers
├── workshop-schedule/ # DAG scheduling library (petgraph-based)
├── workshop-getterurl/ # URL fetching with hash verification
├── workshop-baker-params/ # Config/parameter type definitions
── docs/ # mdBook documentation (Chinese, 3 volumes)
├── templates/ # Pipeline config templates (prebake/bake/finalize)
└── .github/workflows/ # CI (tests workshop-baker + workshop-engine)
```
**No root Cargo workspace** — crates linked by path deps. Build per-crate.
## WHERE TO LOOK
| Task | Location | Notes |
|------|----------|-------|
| Baker (CLI) | `workshop-baker/src/{lib,main}.rs` | Clap CLI + tokio entry |
| Pipeline stages | `workshop-baker/src/{bake,prebake,finalize}/` | Stage orchestration |
| Notifications | `workshop-baker/src/notify/` | Plugin-based event notifications |
|| Engine (execution) | `workshop-engine/src/` | Process mgmt, PM detection |
| URL fetching | `workshop-getterurl/src/` | Git/HTTP/File resource getter |
| CLI definition | `workshop-baker/src/lib.rs` | Clap Cli/Commands/BareCommands |
| Pipeline orchestration | `workshop-baker/src/main.rs` | prebake→bake→finalize→notify flow |
| Decorator parsing | `workshop-baker/src/bake/decorator.rs` | `# @decorator(args)` syntax |
| Build execution | `workshop-baker/src/bake/execute.rs` | Script execution with timeout/retry |
| Docker integration | `workshop-baker/src/prebake/env/container.rs` | Bollard-based |
| Package managers | `workshop-engine/src/pm/` | apt, pacman, dnf, apk |
| User package managers | `workshop-engine/src/upm/` | rustup, uv, go |
| Notification system | `workshop-baker/src/notify/` | Email, webhook, IM |
| Plugin system | `workshop-baker/src/finalize/plugin/` | shell, dylib, rhai, internal |
| Error types | `workshop-baker/src/error.rs` | CliError + HasExitCode |
| Exit codes | `workshop-engine/src/error.rs` | EXITCODE_* constants |
| DAG scheduling | `workshop-schedule/src/dag.rs` | Topological sort, batch picking |
| Pipeline templates | `templates/` | Reference config DSL |
## ENTRY POINTS
| Entry | Type | Path | Role |
|-------|------|------|------|
| `workshop-baker` | Rust bin | `workshop-baker/src/main.rs` | Primary CLI — dispatch: default (full pipeline), bare (single stage), daemon (unimplemented) |
| `workshop_baker` | Rust lib | `workshop-baker/src/lib.rs` | Baker library + CLI structs |
| `workshop_engine` | Rust lib | `workshop-engine/src/lib.rs` | Core engine — ExecutionContext, EventSender, Executor |
| `workshop_schedule` | Rust lib | `workshop-schedule/src/lib.rs` | Dag, DagNode, EdgeKind |
| `workshop_baker_params` | Rust lib | `workshop-baker-params/src/lib.rs` | Layered config: defaults → base → courier |
| `workshop_getterurl` | Rust lib | `workshop-getterurl/src/lib.rs` | GetterUrl, fetch() — http/git/file |
| docs | mdBook | `docs/book.toml` | Project documentation site |
## CODE MAP
| Symbol | Type | Location | Role |
|--------|------|----------|------|
| Cli | struct | workshop-baker/src/lib.rs:21 | CLI argument struct |
| ExecutionContext | struct | workshop-engine/src/types.rs:10 | Pipeline context |
| Engine | struct | workshop-engine/src/types.rs:99 | Build executor (placeholder) |
| PrebakeConfig | struct | workshop-baker/src/prebake/config.rs | Prebake YAML schema |
| FinalizeConfig | struct | workshop-baker/src/finalize/config.rs:21 | Finalize YAML schema |
| GetterUrl | struct | workshop-getterurl/src/lib.rs:84 | Parsed resource URL |
| Getter | trait | workshop-getterurl/src/getter.rs:18 | Resource fetch trait |
### Crate Dependency Graph
```
workshop-baker (bin)
├── workshop-engine
├── workshop-schedule
├── workshop-baker-params
└── workshop-getterurl
```
### workshop-baker Module Map
| Module | Submodules | Role |
|--------|-----------|------|
| `bake` | decorator, execute, parser, schedule, builder, trivial, util, constant, error | Script parsing + execution |
| `prebake` | config, stage (bootstrap, depssystem, depsuser, environment, hook), env (container, firecracker, baremetal, custom), security, event, types, constant, error | Environment setup |
| `finalize` | plugin (shell, dylib, rhai, internal), stage, template, types, config, constant, error, event | Post-build hooks |
| `notify` | handler, queue, rule, template, types, util, error | Async notification dispatch |
| `bare` | — | Standalone single-stage CLI mode |
| `types` | resource, buildstatus | Shared type definitions |
| `utils` | — | Cross-cutting utilities |
## CONVENTIONS
### Rust
- **Edition**: 2024 (requires Rust 1.85+), deprecates `mod.rs`
- **Naming**: `snake_case` files/modules, `PascalCase` types, `SCREAMING_SNAKE_CASE` consts
- **Error handling**: `thiserror` + `anyhow` combo; `HasExitCode` trait for exit codes
- **Async**: `#[tokio::main]` + `tokio` with "full" features
- **Imports**: `std``external_crate``crate::module`
- **Dual-mode**: CLI commands (prebake/bake/finalize) + daemon mode (unimplemented)
### Testing
- **Rust**: `#[test]` inline, `#[tokio::test]` async, `tests/` integration
- **Benchmarks**: Criterion (configured in Cargo.toml)
### CI
- **GitHub Actions**: rust-toolchain (stable), clippy + rustfmt, `cargo test --all-features`
- **Matrix**: workshop-baker, workshop-engine
- **Edition**: Rust 2024 (requires Rust 1.85+)
- **Formatting**: Default `rustfmt` — no `rustfmt.toml` / `.editorconfig`
- **Lint overrides** (workshop-baker only): `dead_code = "allow"`, `unreachable_code = "allow"`, `inherent_to_string = "allow"`, `non_canonical_partial_ord_impl = "allow"`
- **Import order**: `std` → external crates → `crate::` (blank line between groups)
- **Error handling**: `thiserror` for libraries, `anyhow` for binaries. Implement `HasExitCode` for exit code mapping.
- **Type system**: Newtype pattern preferred (e.g., `MemSize(u64)`). Shared types in `types/` module. `types/` must NOT depend on `bake/`, `engine/`, `prebake/`, `finalize/`.
- **Tests**: Inline `#[cfg(test)] mod tests` at bottom of source files. Async: `#[tokio::test]`. CI runs `cargo test --all-features`.
- **Commits**: `type(scope): description``feat|fix|docs|test|refactor|style`
- **AI annotation**: `// Code in this module PARTIALLY or FULLY utilized AI Coding Agent` — never delete if present
- **Docs language**: Chinese (zh-CN), mdBook
- **No `no_std`**: Explicitly unsupported
## ANTI-PATTERNS (THIS PROJECT)
1. **Hardcoded user** — "vulcan" username hardcoded throughout prebake
2. **Unsafe libc** — workshop-baker/src/prebake/security.rs has 5+ unsafe blocks
3. **Rust 2024 edition** — May not work with stable toolchains
4. **Empty daemon**`daemon/` dir is empty, Daemon command is `unimplemented!()`
## UNIQUE STYLES
- **Pipeline DSL**: Shell scripts with `# @decorator` annotations
- **Notification system**: Plugin-based with batching, retry, priority groups
- **LLM cookbook system**: YAML cookbooks for language-specific builds (archived)
- **Chinese docs** — Documentation primarily in zh-CN
- **Do NOT derive production behavior from bare mode** — bare mode is dev-only, ctx.privileged is fake
- **Do NOT use `detect()`/`adopt()` PM methods in production** — PM selection must come from env config
- **`drop_after < DepsUser` is FORBIDDEN** — will break system dependency installation (needs root)
- **Do NOT add magic URL inference** — URL is URL, behavior bound explicitly via `getter::` prefix
- **Do NOT use `@` for version locking in URLs** — it's an auth delimiter; use `?ref=`
- **Do NOT duplicate type definitions across modules** — single source of truth in `types/`
- **Careful with `unimplemented!()`** — Daemon mode, Rhai/Dylib plugins, and Custom builder are stubs
- **Security**: PIPELINE_CONTAINER_PRIVILEGED, PIPELINE_BAREMETAL_ELEVATE, PIPELINE_UNCOVER_SECRET require explicit approval + audit
## COMMANDS
```bash
# Build
cargo build --release -p workshop-baker
# Build (per crate, no workspace)
cd workshop-baker && cargo build --all-features
# Test
cargo test -p workshop-baker
cargo test -p workshop-engine
# Test (per crate)
cd workshop-baker && cargo test --all-features
cd workshop-engine && cargo test --all-features
# Lint
cargo clippy -- -D warnings
cargo fmt --all
# Lint (CI gate)
cargo fmt -- --check
cargo clippy --all-features -- -D warnings
# Run
cargo run --bin workshop-baker -- --pipeline demo --build-id 1 --prebake prebake.yml --bake bake.sh --finalize finalize.yml
# Docs
cd docs && mdbook build # output in docs/book/
cd docs && mdbook serve # local preview at localhost:3000
```
## NOTES
- **Docker required** for integration tests
- **No root workspace** — each crate independent, build separately
- **7 archived crates** workshop-agent, workshop-cert, workshop-deviceid, workshop-helper-mac, workshop-llm-detector, workshop-pipeline, workshop-vault
- **Python tests removed** — pytest.ini exists but test files no longer in workspace
- **No workspace Cargo.toml** at root — each crate has its own `Cargo.lock`. Build individually.
- **CI only tests** `workshop-baker` and `workshop-engine`; smaller crates tested transitively.
- **`quicktest.sh`** in workshop-baker/ is a dev smoke-test script (Docker-based, not CI).
- **Git submodules**: `rust-tongsuo`, `RustyVault` (crypto — not yet integrated into build).
- **`.secret/`** contains real credentials — never commit, never read into context.
- See `docs/src/zh-CN/vol3_dev/` for full developer documentation (architecture, design decisions, code style).
+48
View File
@@ -0,0 +1,48 @@
# DOCS KNOWLEDGE BASE
## OVERVIEW
mdBook project in zh-CN. Authoritative source for project conventions, coding style, and architecture docs. Build output at `docs/book/` (gitignored).
## STRUCTURE
```
docs/
├── book.toml # mdBook config (title, language=zh-CN)
├── src/SUMMARY.md # Sidebar nav — source of truth for page listing
└── src/zh-CN/
├── index.md # Landing
├── vol1_user/ # End-user documentation
├── vol2_admin/ # Admin reference: decorator cheatsheet, deployment,
│ # env vars, exit codes, plugin ref, security, troubleshooting
└── vol3_dev/ # Developer docs: architecture, code style, commit
# convention, testing, error handling, type system,
# design decisions, PR process
```
## WHERE TO LOOK
| Task | Location | Notes |
|------|----------|-------|
| Sidebar/nav editing | `src/SUMMARY.md` | Add entry here to surface a new page |
| Coding conventions | `vol3_dev/code_style.md` | Naming, imports, formatting |
| Commit rules | `vol3_dev/commit_convention.md` | feat/fix/docs/test/refactor/style |
| Error handling | `vol3_dev/error_handling_patterns.md` | thiserror+anyhow+HasExitCode |
| Type system | `vol3_dev/type_system_guide.md` | Newtypes, types/ module isolation |
| Testing strategy | `vol3_dev/testing_strategy.md` | 3 tiers: unit, integration, system |
| Anti-patterns | `vol3_dev/design_decisions.md` | Documented design rejections |
| Architecture | `vol3_dev/architecture.md` | Crate dependency graph, module map |
## CONVENTIONS
- **Language**: All prose in zh-CN. Code blocks in Rust (default) or bash.
- **Adding a page**: Create `.md` under `volN_*/`, then add to `SUMMARY.md`.
- **Formatting**: Standard Markdown. mdBook preprocessors handle TOC and links.
- **Cross-references**: Relative paths from source root (e.g., `../vol3_dev/code_style.md`).
- **Code blocks**: Fenced with language tag. No line numbers in docs.
- **File naming**: `snake_case.md`.
## COMMANDS
```bash
mdbook build # static site → docs/book/
mdbook serve # live preview at localhost:3000
```
+6
View File
@@ -10,6 +10,8 @@ pub struct BakeParams {
#[serde(default)] pub health_period: ParamVal<u64>,
/// Health probe per-attempt timeout in seconds (default 10).
#[serde(default)] pub health_timeout: ParamVal<u64>,
/// Max health probe retries before giving up (default 12).
#[serde(default)] pub health_max_retries: ParamVal<u32>,
}
impl MergeParams for BakeParams {
@@ -18,12 +20,14 @@ impl MergeParams for BakeParams {
workspace: ParamVal::new("/workspace".into()),
health_period: ParamVal::new(5),
health_timeout: ParamVal::new(10),
health_max_retries: ParamVal::new(12),
}
}
fn apply(&mut self, i: Self, w: Writer, ov: &mut Overridden, p: &str) {
apply_field!(self.workspace, i.workspace, w, ov, p, "workspace");
apply_field!(self.health_period, i.health_period, w, ov, p, "health_period");
apply_field!(self.health_timeout, i.health_timeout, w, ov, p, "health_timeout");
apply_field!(self.health_max_retries, i.health_max_retries, w, ov, p, "health_max_retries");
}
}
@@ -34,6 +38,7 @@ pub struct BakeSettings {
pub workspace: String,
pub health_period: u64,
pub health_timeout: u64,
pub health_max_retries: u32,
}
impl Default for BakeSettings { fn default() -> Self { BakeParams::defaults().into() } }
impl From<&BakeParams> for BakeSettings {
@@ -42,6 +47,7 @@ impl From<&BakeParams> for BakeSettings {
workspace: p.workspace.value.clone(),
health_period: p.health_period.value,
health_timeout: p.health_timeout.value,
health_max_retries: p.health_max_retries.value,
}
}
}
+1 -1
View File
@@ -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(30), batch_watermark: ParamVal::new(5),
batch_period: ParamVal::new(5), batch_watermark: ParamVal::new(5),
template_ref_depth: ParamVal::new(10),
}
}
+80
View File
@@ -0,0 +1,80 @@
# WORKSHOP-BAKER CRATE KNOWLEDGE BASE
## OVERVIEW
Primary crate — the only crate with a `[[bin]]` target. Contains the `workshop-baker` binary and `workshop_baker` library. Orchestrates the full CI/CD pipeline: prebake (env setup) → bake (script execution) → finalize (plugins, cleanup) → drain notification queue. ~69 .rs source files.
## STRUCTURE
```
src/
├── main.rs # Entry: default(full pipeline) | bare(dev-only) | daemon(unimplemented)
├── lib.rs # Clap Cli/Commands/BareCommands — CLI structs in lib, not bin
├── error.rs # CliError enum (wraps stage errors, delegates exit_code)
├── bare.rs # Dev-only single-stage execution, ctx.privileged always false
├── utils.rs # Cross-cutting helpers
├── bake/ # Script parsing + DAG scheduling + execution
│ ├── decorator.rs # @decorator(args) parsing, 81 tests — heaviest coverage in crate
│ ├── parser.rs # Step/function extraction
│ ├── schedule.rs # DAG build → topological sort
│ ├── execute.rs # Per-step timeout/retry
│ ├── builder.rs # Script assembly
│ ├── trivial.rs # Trivial step detection (skip in dry-run)
│ ├── constant.rs / error.rs / util.rs
├── prebake/ # Environment provisioning
│ ├── config.rs # PrebakeConfig deserialization
│ ├── stage.rs + stage/{bootstrap,depssystem,depsuser,environment,hook}.rs
│ ├── env.rs + env/{container,firecracker,baremetal,custom}.rs # Builder dispatch
│ ├── security.rs # Privilege/sandbox checks
│ ├── event.rs # Event sender/receiver channel
│ ├── types.rs + types/{memsize,cache,builderconfig,architecture}.rs
│ ├── constant.rs / error.rs
├── finalize/ # Post-build hooks + plugins
│ ├── config.rs # FinalizeConfig
│ ├── stage.rs + stage/hook.rs
│ ├── template.rs # Output templating (minijinja)
│ ├── plugin.rs + plugin/{shell,dylib,rhai,internal}.rs
│ │ └── plugin/internal/{dummy,mail,satori,webhook,insitenotify}.rs
│ ├── types.rs + types/compression.rs
│ ├── constant.rs / error.rs / event.rs
├── notify/ # Async notification dispatch
│ ├── queue.rs # NotificationQueue — drain on pipeline complete
│ ├── handler.rs / rule.rs / template.rs / types.rs / util.rs / error.rs
└── types/ # Shared types — ZERO deps on stage modules
├── resource.rs # ResourceRegistry (populated pre-pipeline, consumed read-only)
└── buildstatus.rs
```
## WHERE TO LOOK
| Task | Location | Notes |
|------|----------|-------|
| CLI args + subcommands | `src/lib.rs` | Cli, Commands(Bare/Daemon), BareCommands |
| Pipeline orchestration | `src/main.rs` | worker_main(): prebake→bake→finalize→queue.drain() |
| Decorator syntax | `src/bake/decorator.rs` | `# @decorator` or `# @decorator(args)`, line-by-line regex |
| Build execution | `src/bake/execute.rs` | timeout, retry, exit code capture |
| DAG scheduling | `src/bake/schedule.rs` | Uses workshop-schedule crate |
| Docker builder | `src/prebake/env/container.rs` | Bollard-based, buildkit support |
| Firecracker / Bare metal | `.../firecracker.rs`, `.../baremetal.rs` | MicroVM;
needs PIPELINE_BAREMETAL_ELEVATE |
| Plugin dispatch | `src/finalize/plugin.rs` | Dispatches shell / dylib(stub) / rhai(stub) / internal |
| Notification queue | `src/notify/queue.rs` | Spawned early, drained after pipeline, retry support |
| Error type | `src/error.rs` | CliError wraps PrebakeError/BakeError/FinalizeError/anyhow |
| Dev bare mode | `src/bare.rs` | ctx.privileged=false always; do not extrapolate to production |
## DEPENDENCIES
**Internal:** workshop-engine, workshop-schedule, workshop-getterurl, workshop-baker-params. **External:** bollard (Docker), clap (CLI), axum 0.7 (HTTP), lettre 0.11 (email), minijinja (templates), cgroups-rs (limits), privdrop.
## CONVENTIONS (CRATE-SPECIFIC)
- **Lint overrides** (this crate only): `dead_code=allow`, `unreachable_code=allow`, `inherent_to_string=allow`, `non_canonical_partial_ord_impl=allow`
- **CLI structs in lib.rs, not main.rs** — enables integration tests to import and construct Cli directly.
- **Pipeline stages return `anyhow::Result<()>`** — stages propagate with `?`; only main.rs converts to CliError.
- **ResourceRegistry pattern** — populated before pipeline stages, consumed read-only by stages. New resource types go in `types/resource.rs`.
- **Notification queue fire-and-forget + drain** — notify tasks spawned early, send during pipeline, drain on completion. `NotificationQueue::drain()` must be called.
- **Trivial step detection** — steps with no external deps can be marked trivial (`trivial.rs`), skipped in dry-run.
## ANTI-PATTERNS (THIS CRATE)
- **Do NOT derive production behavior from bare mode** — ctx.privileged is hardcoded false.
- **Do NOT add CLI args bypassing prebake→bake→finalize ordering** — the pipeline is linear. Parallelism lives in the DAG scheduler within bake.
- **Do NOT import types/ from bake/, prebake/, or finalize/** — `types/` must have zero crate-internal deps.
- **Do NOT add lint allows without justification** — the four existing overrides are intentional.
- **Do NOT spawn blocking work on tokio runtime** — use `spawn_blocking` for CPU-heavy tasks.
- **Decorator regex is line-by-line** — multi-line decorators need a parser rewrite, not a regex hack.
- **Careful with `unimplemented!()`** — Daemon, Rhai/Dylib plugins, Custom builder are stubs. Prefer error over crash.
+25 -7
View File
@@ -1118,6 +1118,12 @@ version = "0.1.6"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "645cbb3a84e60b7531617d5ae4e57f7e27308f6445f5abf653209ea76dec8dff"
[[package]]
name = "fixedbitset"
version = "0.5.7"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "1d674e81391d1e1ab681a28d99df07927c6d4aa5b027d7da16ba32d1d21ecd99"
[[package]]
name = "fnv"
version = "1.0.7"
@@ -2404,6 +2410,16 @@ dependencies = [
"sha2 0.10.9",
]
[[package]]
name = "petgraph"
version = "0.7.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "3672b37090dbd86368a4145bc067582552b29c27377cad4e0a306c97f9bd7772"
dependencies = [
"fixedbitset",
"indexmap 2.12.1",
]
[[package]]
name = "pin-project"
version = "1.1.10"
@@ -3625,12 +3641,6 @@ dependencies = [
"tonic",
]
[[package]]
name = "topological-sort"
version = "0.2.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "ea68304e134ecd095ac6c3574494fc62b909f416c4fca77e440530221e549d3d"
[[package]]
name = "tower"
version = "0.5.2"
@@ -4282,12 +4292,12 @@ dependencies = [
"tempfile",
"thiserror 2.0.18",
"tokio",
"topological-sort",
"uuid",
"which",
"workshop-baker-params",
"workshop-engine",
"workshop-getterurl",
"workshop-schedule",
]
[[package]]
@@ -4330,6 +4340,14 @@ dependencies = [
"url",
]
[[package]]
name = "workshop-schedule"
version = "0.1.0"
dependencies = [
"petgraph",
"thiserror 2.0.18",
]
[[package]]
name = "writeable"
version = "0.6.2"
+1 -1
View File
@@ -41,7 +41,7 @@ tar = "0.4.44"
tempfile = "3.24.0"
thiserror = "2.0.17"
tokio = { version = "1.48.0", features = ["full"] }
topological-sort = "0.2.2"
workshop-schedule = { path = "../workshop-schedule" }
async-trait = "0.1.87"
uuid = { version = "1.19.0", features = ["v4"] }
libc = "0.2"
-11
View File
@@ -1,11 +0,0 @@
[pytest]
testpaths = tests/integration
python_files = test_*.py
python_classes = Test*
python_functions = test_*
addopts = -v --tb=short
markers =
integration: marks tests as integration tests (deselect with '-m "not integration"')
slow: marks tests as slow running
filterwarnings =
ignore::DeprecationWarning
-57
View File
@@ -1,57 +0,0 @@
# workshop-baker/src
**Generated:** 2026-05-19
**Commit:** 28abc5d
CLI tool and runtime for executing CI/CD pipelines. Orchestrates prebake → bake → finalize stages with notification dispatch.
## STRUCTURE
```
src/
├── lib.rs # Module declarations + Cli struct + Commands enum
├── main.rs # tokio::main entry point
├── bare.rs # Bare mode (run individual stages)
├── error.rs # CliError (wraps stage errors)
├── bake.rs # Bake orchestration (script → execution)
├── bake/ # Script parsing, decorator DSL, scheduling
├── prebake.rs # Prebake orchestration (env setup)
├── prebake/ # Config, stages, security, env providers
├── finalize.rs # Finalize orchestration (post-build)
├── finalize/ # Config, plugins, hooks, templates
├── notify.rs # Notification system (batching, retry)
├── notify/ # Types, templates, interactive, queue
├── daemon/ # Empty (Daemon command = unimplemented!)
├── monitor/ # Resource usage monitoring (cgroups)
├── types.rs # Module re-exports
├── types/ # buildstatus, resource types
├── utils.rs # Module root (fetch submodule)
└── utils/ # fetch.rs
```
## WHERE TO LOOK
| Task | Location |
|------|----------|
| CLI args | `lib.rs:21` - `Cli` struct |
| Entry point | `main.rs:91` - `#[tokio::main]` |
| Pipeline orchestration | `main.rs:68-74` - prebake→bake→finalize chain |
| Bare mode (single stage) | `bare.rs:14` - `run_bare()` |
| Script parsing | `bake/parser.rs:56` - `parse_script()` |
| Decorator DSL | `bake/decorator.rs` - `Decorator` enum |
| Prebake stages | `prebake/stage/` - Bootstrap→DepsSystem→DepsUser→Hooks→Ready |
| Plugin system | `finalize/plugin.rs` + `finalize/plugin/` |
| Notification dispatch | `notify.rs:66` - `notify()` async loop |
| Notification types | `notify/types.rs` - `NotificationRule`, `ParsedTrigger` |
## CONVENTIONS
- **ExecutionContext**: Passed through all stages, carries task_id, username, env_vars, timeout
- **PrebakeStage**: Ordered enum Bootstrap→EarlyHook→DepsSystem→DepsUser→LateHook→Ready
- **Decorator**: `# @name(args)` comments above shell function declarations
- **Notification**: Plugin-based dispatch with batching, exponential backoff, priority groups
- **Engine**: Delegated to `workshop-engine` crate (formerly in-tree)
- **Dual-mode**: CLI commands (standalone) vs Daemon mode (unimplemented)
## ANTI-PATTERNS
1. **Empty daemon/** directory — Daemon command is `unimplemented!()`
2. **Dead code allowed**`lints.rust.dead_code = "allow"` in Cargo.toml
3. **Engine was extracted** — workshop-engine is its own crate now, src/engine/ removed
4. **Python test infra exists but files removed** — pytest.ini remains, test dir is empty
+74 -30
View File
@@ -2,7 +2,7 @@ use crate::bake::builder::RenderMode;
use crate::bake::constant::DEFAULT_WORKSPACE;
use crate::bake::decorator::Decorator;
use crate::bake::error::BakeError;
use crate::bake::util::{is_parallel, resolve_mode, FuncMode};
use crate::bake::util::{resolve_mode, FuncMode};
use crate::cli::Cli;
use crate::prebake;
use crate::prebake::PrebakeConfig;
@@ -30,11 +30,16 @@ pub async fn bake(
ctx: &mut ExecutionContext,
event_tx: EventSender,
) -> Result<(), BakeError> {
let script_content = std::fs::read_to_string(script_path)?;
let prebake_content = std::fs::read_to_string(prebake_path)?;
log::debug!("Reading script: {:?}", script_path);
let script_content = std::fs::read_to_string(script_path).map_err(|e| {
BakeError::IoError { path: script_path.display().to_string(), source: e }
})?;
log::debug!("Reading prebake.yml: {:?}", prebake_path);
let prebake_content = std::fs::read_to_string(prebake_path).map_err(|e| {
BakeError::IoError { path: prebake_path.display().to_string(), source: e }
})?;
let prebake = prebake::parse(prebake_content.as_str()).map_err(|e| {
log::error!("Failed to parse prebake.yml: {}", e);
log::error!("This is unexpected!");
BakeError::YamlParseError(e)
})?;
let workspace = get_workspace(&prebake);
@@ -58,6 +63,7 @@ pub async fn bake(
}
if !has_pipeline {
log::trace!("Running trivial script");
trivial::run_trivial(
&script_content,
bake_base_path,
@@ -69,6 +75,7 @@ pub async fn bake(
)
.await?;
} else {
log::trace!("Running pipeline script");
let remaining = functions.remaining_code;
let (mut dag, func_map) = schedule::build_dag(functions.pipeline_functions)?;
@@ -76,10 +83,6 @@ pub async fn bake(
let mut pipe_sources: HashSet<String> = HashSet::new();
for func in func_map.values() {
decorator::check_conflicts(func)?;
// Some simple optimization for @pipe
// Collect only output of specified function
// and exclude self-reference because it's handled by loop
for decorator in &func.decorators {
if let Decorator::Pipe(names) = decorator {
for source in names {
@@ -95,29 +98,31 @@ pub async fn bake(
let mut async_handles: Vec<tokio::task::JoinHandle<()>> = Vec::new();
while !dag.is_empty() {
let ready: Vec<String> = dag.peek_all().into_iter().cloned().collect();
if ready.is_empty() {
let remaining_nodes: Vec<String> = dag.clone().into_iter().collect();
return Err(BakeError::CircularDependency(remaining_nodes));
}
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
if let Some(seq_name) = ready.iter().find(|n| !is_parallel(&func_map[n.as_str()])).cloned() {
let func = &func_map[seq_name.as_str()];
let seq_candidates = dag.ready_sequential();
if let Some(seq_name) = seq_candidates.first() {
let seq_name = seq_name.to_string();
log::info!("[BAKE_SCHED] seq {}", seq_name);
let func = &func_map[&seq_name];
let mode = resolve_mode(func);
match mode {
FuncMode::Daemon => {
dag.pop();
execute::execute_daemon(
&engine, func, ctx, &event_tx,
&workspace, &remaining, bake_base_path,
Duration::from_secs(bake_settings.health_period),
Duration::from_secs(bake_settings.health_timeout),
bake_settings.health_max_retries,
).await?;
}
FuncMode::Async => {
dag.pop();
let handle = execute::execute_async(
func, ctx, &event_tx,
&workspace, &remaining, bake_base_path,
@@ -125,35 +130,74 @@ pub async fn bake(
async_handles.push(handle);
}
FuncMode::Normal => {
let stdout = execute::execute_normal(
let result = execute::execute_normal(
&engine, func, ctx, &event_tx,
&workspace, &remaining, bake_base_path, &stdout_cache,
).await?;
if pipe_sources.contains(&seq_name) {
stdout_cache.insert(seq_name.clone(), stdout);
).await;
match result {
Ok(stdout) => {
if pipe_sources.contains(&seq_name) {
stdout_cache.insert(seq_name.clone(), stdout);
}
}
Err(e) if dag.is_fallible(&seq_name) => {
log::warn!("'{}' failed (fallible): {}", seq_name, e);
}
Err(e) => return Err(e),
}
dag.pop();
}
}
// Pop AFTER execution (Daemon/Async pop at start)
if !matches!(mode, FuncMode::Daemon | FuncMode::Async) {
dag.pop(&seq_name).map_err(|e| BakeError::IncompatibleDecorators {
function: seq_name.clone(),
reason: e.to_string(),
})?;
} else {
dag.pop(&seq_name).map_err(|e| BakeError::IncompatibleDecorators {
function: seq_name.clone(),
reason: e.to_string(),
})?;
}
continue;
}
// All remaining are parallel — group and pick a batch
let (star_nodes, groups) = schedule::partition_by_group(&ready, &func_map);
if let Some(batch) = schedule::pick_batch(&star_nodes, &groups) {
// All remaining are parallel — pick a batch
if let Some(batch) = dag.pick_batch() {
log::info!("[BAKE_SCHED] par {:?}", batch);
let results = execute::fire_parallel_batch(
&batch, &engine, &func_map, ctx, &event_tx,
&workspace, &remaining, bake_base_path, &stdout_cache,
).await?;
for (name, stdout) in results {
if pipe_sources.contains(&name) {
stdout_cache.insert(name, stdout);
stdout_cache.insert(name.clone(), stdout);
}
if let Err(e) = dag.pop(&name) {
log::warn!("Failed to pop '{}' from DAG: {}", name, e);
}
dag.pop();
}
} else if dag.is_stalled() {
// Deadlock: named parallel groups exist but can't complete
let stalled = dag.stalled_groups();
let remaining = dag.remaining();
return Err(BakeError::IncompatibleDecorators {
function: remaining.join(", "),
reason: format!(
"parallel group(s) {:?} can never complete — blocked by @after chain",
stalled
),
});
} else {
// Ready set is empty — cycle
let remaining = dag.remaining();
if !remaining.is_empty() {
return Err(BakeError::CircularDependency(
remaining.into_iter().map(|s| s.to_string()).collect(),
));
}
break;
}
// No complete group and no sequential — nothing fireable
}
// Wait for all @async tasks to finish
-48
View File
@@ -1,48 +0,0 @@
# workshop-baker/src/bake
**Generated:** 2026-04-22
**Commit:** 7b3b71a
Script parsing and build orchestration with decorator DSL.
## STRUCTURE
```
bake/
├── parser.rs # @decorator-aware shell script parser
├── decorator.rs # Decorator enum + 100+ unit tests
├── builder.rs # Script template assembly
├── schedule.rs # Function ordering via @after dependencies
├── error.rs # BakeError enum
└── constant.rs # DEFAULT_WORKSPACE, TRIVIAL_SCRIPT_NAME
```
## WHERE TO LOOK
| Task | Location |
|------|----------|
Parse script | `parser.rs:56` - `parse_script()` |
| Decorator types | `decorator.rs` - `Decorator` enum variants |
| Build script | `builder.rs` - template builder |
| Schedule deps | `schedule.rs` - @after resolution |
## DECORATORS
| Decorator | Purpose |
|-----------|---------|
| `@pipeline` | Marks main build function |
| `@timeout(N)` | Execution timeout seconds |
| `@retry(N, M)` | Retry N times, M sec delay |
| `@parallel` | Run concurrently |
| `@fallible` | Failure doesn't fail pipeline |
| `@after(func)` | Dependency ordering |
| `@export` | Export vars to env |
| `@loop(N)` | Repeat N times |
| `@if(COND)` | Conditional execution |
## CONVENTIONS
- **Decorator Syntax**: `# @name(args)` comment above function
- **Parsing**: Shell grammar with decorator annotation extraction
- **Scheduling**: DAG resolution for @after dependencies
- **Tests**: `decorator.rs:175-829` has 100+ inline unit tests
+13 -8
View File
@@ -43,7 +43,6 @@ pub fn build_script(
function: &Function,
bake_base: &Path,
target_dir: &Path,
temp_path: Option<PathBuf>,
render_mode: RenderMode,
) -> Result<PathBuf, BakeError> {
let name = &function.name;
@@ -55,24 +54,27 @@ pub fn build_script(
None
}
});
let temp_path = temp_path.unwrap_or_else(|| PathBuf::from("/tmp"));
// let temp_path = temp_path.unwrap_or_else(|| PathBuf::from("/tmp"));
let temp_path = target_dir;
let filename = format!("bakefn_{}.sh", name);
let target = target_dir.join(filename);
let template_content = match render_mode {
RenderMode::Off => TRIVIAL_TEMPLATE.to_string(),
RenderMode::Minimal | RenderMode::Full => {
std::fs::read_to_string(bake_base).map_err(|e| BakeError::ScriptGenerationFailed {
script: bake_base.display().to_string(),
reason: e.to_string(),
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,
})?
}
};
let condition_code = match condition {
Some(content) => {
// with some bash workaround
format!(
"if [ ! {} ]; then\n exit {}\nfi\n",
"if [[ ! {} ]]; then\n exit {}\nfi\n",
content, PIPELINE_SKIP_ERRORCODE
)
}
@@ -99,6 +101,7 @@ pub fn build_script(
let mut temp_file =
NamedTempFile::new_in(&temp_path).map_err(|e| BakeError::ScriptGenerationFailed {
script: temp_path.display().to_string(),
action: "creating temp file".to_string(),
reason: e.to_string(),
})?;
@@ -107,17 +110,20 @@ pub fn build_script(
.set_permissions(Permissions::from_mode(0o700))
.map_err(|e| BakeError::ScriptGenerationFailed {
script: temp_path.display().to_string(),
action: "setting permissions".to_string(),
reason: e.to_string(),
})?;
std::fs::write(&temp_file, rendered).map_err(|e| BakeError::ScriptGenerationFailed {
script: temp_path.display().to_string(),
action: "writing to temp file".to_string(),
reason: e.to_string(),
})?;
temp_file
.persist(&target)
.map_err(|e| BakeError::ScriptGenerationFailed {
script: target.display().to_string(),
action: "persisting temp file".to_string(),
reason: e.error.to_string(),
})?;
Ok(target)
@@ -140,7 +146,7 @@ mod tests {
body: "echo 'hello world'".to_string(),
};
let result = build_script(
&function, &bake_base, temp_dir.path(), None, RenderMode::Full,
&function, &bake_base, temp_dir.path(), RenderMode::Full,
);
assert!(result.is_ok());
@@ -162,7 +168,6 @@ mod tests {
&function,
Path::new("/nonexistent"),
temp_dir.path(),
None,
RenderMode::Off,
);
+1 -1
View File
@@ -1,4 +1,4 @@
pub const DEFAULT_WORKSPACE: &str = "/workspace";
pub const BAKE_SCRIPT_PATH: &str = "bake.sh";
pub const TRIVIAL_SCRIPT_NAME: &str = "BAKE_TRIVIAL";
pub const PIPELINE_SKIP_ERRORCODE: u8 = 233;
pub const PIPELINE_SKIP_ERRORCODE: i32 = 233;
+5 -5
View File
@@ -7,8 +7,8 @@ use workshop_engine::{
#[derive(Error, Debug)]
pub enum BakeError {
#[error("Failed to generate staged script {script}: {reason}")]
ScriptGenerationFailed { script: String, reason: String },
#[error("Failed to generate staged script {script} when {action}: {reason}")]
ScriptGenerationFailed { script: String, action: String, reason: String },
#[error("Template error: {0}")]
TemplateError(#[from] minijinja::Error),
@@ -19,8 +19,8 @@ pub enum BakeError {
#[error("Unknown @after of function {0}: {1}")]
UnknownDependency(String, String),
#[error("IO Error: {0}")]
IoError(#[from] std::io::Error),
#[error("IO error on {path}: {source}")]
IoError { path: String, source: std::io::Error },
#[error("YAML parse error: {0}")]
YamlParseError(#[from] serde_yaml::Error),
@@ -49,7 +49,7 @@ impl HasExitCode for BakeError {
BakeError::TemplateError(_) => EXITCODE_PARSE_ERROR,
BakeError::CircularDependency(_) => EXITCODE_PARSE_ERROR,
BakeError::UnknownDependency(..) => EXITCODE_PARSE_ERROR,
BakeError::IoError(_) => EXITCODE_IO_ERROR,
BakeError::IoError { .. } => EXITCODE_IO_ERROR,
BakeError::YamlParseError(_) => EXITCODE_PARSE_ERROR,
BakeError::DecoratorParseError { .. } => EXITCODE_PARSE_ERROR,
BakeError::ScriptExecutionError(e) => e.exit_code(),
+35 -19
View File
@@ -1,4 +1,5 @@
use crate::bake::builder::{self, RenderMode};
use crate::bake::constant::PIPELINE_SKIP_ERRORCODE;
use crate::bake::decorator::{self, Decorator};
use crate::bake::error::BakeError;
use crate::bake::parser::Function;
@@ -31,7 +32,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, None, RenderMode::Full,
&modified, bake_base, workspace, RenderMode::Full,
)?;
execute_with_combos(engine, &script, ctx, event_tx, &combos).await
@@ -60,11 +61,17 @@ fn build_combinator_chain<'a>(
) -> Pin<Box<dyn Future<Output = Result<String, BakeError>> + Send + 'a>> {
if idx >= combos.len() {
return Box::pin(async move {
let result = engine
.execute_script(script, ctx, event_tx)
.await
.map_err(|e| BakeError::ScriptExecutionError(e))?;
Ok(result.stdout)
let result = engine.execute_script(script, ctx, event_tx).await;
match result {
Ok(r) => Ok(r.stdout),
Err(ExecutionError::ExecutionFailed { exit_code, .. })
if exit_code == PIPELINE_SKIP_ERRORCODE as i32 =>
{
log::debug!("Function skipped (@if condition)");
Ok(String::new())
}
Err(e) => Err(BakeError::ScriptExecutionError(e)),
}
});
}
@@ -187,9 +194,10 @@ pub async fn fire_parallel_batch(
let mut results: Vec<(String, String)> = Vec::with_capacity(handles.len());
for handle in handles {
let (name, result) = handle.await.map_err(|_| {
BakeError::ScriptExecutionError(ExecutionError::ExecutionFailed(
"parallel task panicked".to_string(),
))
BakeError::ScriptExecutionError(ExecutionError::ExecutionFailed {
task_id: "parallel".to_string(),
exit_code: -1,
})
})?;
match result {
Ok(stdout) => results.push((name, stdout)),
@@ -264,13 +272,14 @@ pub async fn execute_daemon(
bake_base: &Path,
health_period: Duration,
health_timeout: Duration,
health_max_retries: u32,
) -> Result<(), BakeError> {
let fallible = is_fallible(func);
let mut modified = func.clone();
modified.body = remaining.to_string() + &func.body;
let script = builder::build_script(
&modified, bake_base, workspace, None, RenderMode::Full,
&modified, bake_base, workspace, RenderMode::Full,
)?;
// Spawn without waiting
@@ -292,18 +301,20 @@ pub async fn execute_daemon(
log::warn!("Failed to spawn daemon '{}' (fallible): {}", func.name, e);
return Ok(());
}
Err(e) => {
return Err(BakeError::ScriptExecutionError(ExecutionError::ExecutionFailed(
format!("Failed to spawn daemon '{}': {}", func.name, e),
)));
Err(_) => {
return Err(BakeError::ScriptExecutionError(ExecutionError::ExecutionFailed {
task_id: func.name.clone(),
exit_code: -1,
}));
}
};
// Periodic health probe — block until healthy or fatal error
// Periodic health probe — block until healthy, retry exhausted, or fatal
if let Some(health_cmd) = func.decorators.iter().find_map(|d| match d {
Decorator::Health(cmd) => Some(cmd.as_str()),
_ => None,
}) {
let mut retries: u32 = 0;
loop {
let probe_result = tokio::time::timeout(
health_timeout,
@@ -312,13 +323,18 @@ pub async fn execute_daemon(
.await;
match probe_result {
Ok(Ok(())) => break,
_ if fallible => {
Ok(Ok(())) => break, // healthy — done
_ if retries + 1 < health_max_retries => {
retries += 1;
log::warn!(
"Health probe for '{}' failed, retrying in {:?}...",
func.name, health_period
"Health probe for '{}' failed ({}/{}), retrying in {:?}...",
func.name, retries, health_max_retries, health_period,
);
}
_ if fallible => {
log::warn!("Health probe for '{}' exhausted (fallible), skipping", func.name);
break; // exhausted but fallible → skip health check
}
Ok(Err(e)) => return Err(e),
Err(_) => {
return Err(BakeError::HealthProbeFailed {
+2
View File
@@ -224,6 +224,7 @@ pub fn parse_script(text: &str) -> Result<ParsedScript, BakeError> {
{
return Err(BakeError::ScriptGenerationFailed {
script: "bake.sh".to_string(),
action: "parsing function body".to_string(),
reason: format!(
"Unexpected EOF while parsing function '{}'. Unmatched braces (remaining: {})?",
name, brace_depth
@@ -234,6 +235,7 @@ pub fn parse_script(text: &str) -> Result<ParsedScript, BakeError> {
let last_decorator = pending_decorators.last().unwrap();
return Err(BakeError::ScriptGenerationFailed {
script: "bake.sh".to_string(),
action: "parsing decorators".to_string(),
reason: format!(
"Dangling decorator {} before EOF. Missing function definition?",
last_decorator
+127 -525
View File
@@ -1,47 +1,124 @@
use super::{decorator::Decorator, decorator::parallel_group, parser::Function};
// Code in this module PARTIALLY or FULLY utilized AI Coding Agent
use super::decorator::Decorator;
use super::parser::Function;
use crate::bake::error::BakeError;
use std::collections::HashMap;
use topological_sort::TopologicalSort;
use workshop_schedule::{Dag, DagNode, EdgeKind, NodeColor};
/// Builds a DAG from functions using `@after` dependencies and implicit ordering
/// for consecutive `@pipeline` functions. Returns the sorted DAG and a name→Function map.
/// The caller should use `peek_all()` (non-destructive) to get ready nodes, execute them,
/// then `pop()` each completed node to advance the DAG.
/// Thin wrapper so `Function` implements `DagNode`.
#[derive(Clone)]
pub(crate) struct BakeNode {
name: String,
color: NodeColor,
fallible: bool,
/// Original function body + decorators for execution
func: Function,
}
impl DagNode for BakeNode {
fn name(&self) -> &str {
&self.name
}
fn color(&self) -> NodeColor {
self.color.clone()
}
fn fallible(&self) -> bool {
self.fallible
}
}
/// Convert a `Function` into a `BakeNode` by extracting scheduling metadata
/// from decorators.
fn function_to_node(func: &Function) -> BakeNode {
let color = if func
.decorators
.iter()
.any(|d| matches!(d, Decorator::Parallel(_)))
{
let group = func
.decorators
.iter()
.find_map(|d| {
if let Decorator::Parallel(Some(g)) = d {
Some(g.clone())
} else {
None
}
})
.unwrap_or("*".to_string());
if group == "*" {
NodeColor::Wildcard
} else {
NodeColor::Named(group)
}
} else {
NodeColor::Sequential
};
let fallible = func
.decorators
.iter()
.any(|d| matches!(d, Decorator::Fallible));
BakeNode {
name: func.name.clone(),
color,
fallible,
func: func.clone(),
}
}
/// Builds a DAG from functions using `@after` dependencies and implicit
/// ordering for consecutive `@pipeline` functions.
pub fn build_dag(
functions: Vec<Function>,
) -> Result<(TopologicalSort<String>, HashMap<String, Function>), BakeError> {
let function_mapped = functions
) -> Result<(Dag<BakeNode>, HashMap<String, Function>), BakeError> {
let function_mapped: HashMap<&str, &Function> = functions
.iter()
.map(|f| (f.name.as_str(), f))
.collect::<HashMap<&str, &Function>>();
.collect();
let mut ts = TopologicalSort::<String>::new();
// Track last pipeline function for implicit @after
let mut dag = Dag::new();
let mut last_pipeline: Option<String> = None;
for function in &functions {
ts.insert(function.name.clone());
// Determine whether this function is a pipeline node
let is_pipeline = function
.decorators
.iter()
.any(|d| matches!(d, Decorator::Pipeline));
if !is_pipeline {
continue;
}
let has_after = function
.decorators
.iter()
.any(|d| matches!(d, Decorator::After(_)));
// Add implicit @after for pipeline functions that don't have explicit @after
if is_pipeline {
if !has_after {
if let Some(ref last) = last_pipeline {
ts.add_dependency(last, function.name.clone());
}
let node = function_to_node(function);
dag.add_node(node).map_err(|e| {
BakeError::IncompatibleDecorators {
function: function.name.clone(),
reason: e.to_string(),
}
last_pipeline = Some(function.name.clone());
}
})?;
// Add implicit @after for consecutive @pipeline functions
if !has_after {
if let Some(ref last) = last_pipeline {
dag.add_edge(last, &function.name, EdgeKind::Implicit)
.map_err(|e| BakeError::IncompatibleDecorators {
function: function.name.clone(),
reason: e.to_string(),
})?;
}
}
last_pipeline = Some(function.name.clone());
// Add explicit @after edges
for decorator in &function.decorators {
if let Decorator::After(dependency) = decorator {
if !function_mapped.contains_key(dependency.as_str()) {
@@ -55,531 +132,56 @@ pub fn build_dag(
dependency.clone(),
));
}
ts.add_dependency(dependency, function.name.clone());
dag.add_edge(dependency, &function.name, EdgeKind::Explicit)
.map_err(|e| BakeError::IncompatibleDecorators {
function: function.name.clone(),
reason: e.to_string(),
})?;
}
}
}
dag.freeze();
let owned_map: HashMap<String, Function> = functions
.into_iter()
.map(|f| (f.name.clone(), f))
.collect();
Ok((ts, owned_map))
Ok((dag, owned_map))
}
/// Group parallel nodes by their declared group.
/// Group ready parallel nodes by their declared group.
/// Returns `(star_nodes, named_groups)`.
/// Star nodes (`group == "*"`) are compatible with any group.
pub fn partition_by_group(
names: &[String],
func_map: &HashMap<String, Function>,
///
/// Deprecated: use `Dag::ready_wildcards()` and `Dag::ready_groups()` instead.
pub fn partition_by_group<'a>(
ready: &[String],
dag: &Dag<BakeNode>,
) -> (Vec<String>, HashMap<String, Vec<String>>) {
let mut star_nodes = Vec::new();
let mut named_groups: HashMap<String, Vec<String>> = HashMap::new();
for name in names {
let func = &func_map[name.as_str()];
let group = func
.decorators
.iter()
.find_map(|d| {
if let Decorator::Parallel(_) = d {
Some(parallel_group(d))
} else {
None
}
})
.unwrap_or("*");
if group == "*" {
for name in ready {
if dag.is_wildcard(name) {
star_nodes.push(name.clone());
} else {
named_groups
.entry(group.to_string())
.or_default()
.push(name.clone());
} else if let Some(g) = dag.group_of(name) {
named_groups.entry(g).or_default().push(name.clone());
}
// Sequential nodes: not included (caller handles them)
}
(star_nodes, named_groups)
}
/// Pick a complete group to execute.
/// - If any named group has ready nodes, pick the first (star nodes join).
/// - If only star nodes exist, fire them together.
/// - Otherwise, return None.
///
/// Deprecated: use `Dag::pick_batch()` instead.
pub fn pick_batch(
star_nodes: &[String],
groups: &HashMap<String, Vec<String>>,
_star_nodes: &[String],
_groups: &HashMap<String, Vec<String>>,
_group_totals: &HashMap<String, usize>,
) -> Option<Vec<String>> {
for nodes in groups.values() {
let mut batch = nodes.clone();
batch.extend_from_slice(star_nodes);
return Some(batch);
}
if !star_nodes.is_empty() && groups.is_empty() {
return Some(star_nodes.to_vec());
}
None
}
#[cfg(test)]
mod tests {
use super::*;
fn create_function(name: &str, decorators: Vec<Decorator>, body: &str) -> Function {
Function {
name: name.to_string(),
decorators,
body: body.to_string(),
}
}
// Helper: drain the DAG and return sorted order (for backward-compatible tests)
fn drain_dag(functions: Vec<Function>) -> Result<Vec<Function>, BakeError> {
let (mut dag, func_map) = build_dag(functions)?;
let mut result = Vec::new();
while !dag.is_empty() {
let batch = dag.pop_all();
if batch.is_empty() {
// Cycle detected: remaining nodes form a circular dependency
let remaining: Vec<String> = dag.into_iter().collect();
return Err(BakeError::CircularDependency(remaining));
}
for name in batch {
result.push(func_map[&name].clone());
}
}
Ok(result)
}
#[test]
fn test_empty_functions() {
let functions = vec![];
let result = drain_dag(functions).unwrap();
assert!(result.is_empty());
}
#[test]
fn test_single_function_no_dependencies() {
let function = create_function("func1", vec![], "body1");
let functions = vec![function.clone()];
let result = drain_dag(functions).unwrap();
assert_eq!(result.len(), 1);
assert_eq!(result[0].name, "func1");
}
#[test]
fn test_multiple_functions_no_dependencies() {
let function1 = create_function("func1", vec![], "body1");
let function2 = create_function("func2", vec![], "body2");
let function3 = create_function("func3", vec![], "body3");
let functions = vec![function1.clone(), function2.clone(), function3.clone()];
let result = drain_dag(functions).unwrap();
assert_eq!(result.len(), 3);
let names: Vec<String> = result.iter().map(|f| f.name.clone()).collect();
assert!(names.contains(&"func1".to_string()));
assert!(names.contains(&"func2".to_string()));
assert!(names.contains(&"func3".to_string()));
}
#[test]
fn test_simple_dependency() {
let function1 = create_function("func1", vec![], "body1");
let function2 = create_function(
"func2",
vec![Decorator::After("func1".to_string())],
"body2",
);
let functions = vec![function1.clone(), function2.clone()];
let result = drain_dag(functions).unwrap();
assert_eq!(result.len(), 2);
assert_eq!(result[0].name, "func1");
assert_eq!(result[1].name, "func2");
}
#[test]
fn test_multiple_dependencies() {
let function1 = create_function("func1", vec![], "body1");
let function2 = create_function(
"func2",
vec![Decorator::After("func1".to_string())],
"body2",
);
let function3 = create_function(
"func3",
vec![Decorator::After("func2".to_string())],
"body3",
);
let functions = vec![function1.clone(), function2.clone(), function3.clone()];
let result = drain_dag(functions).unwrap();
assert_eq!(result.len(), 3);
assert_eq!(result[0].name, "func1");
assert_eq!(result[1].name, "func2");
assert_eq!(result[2].name, "func3");
}
#[test]
fn test_complex_dependencies() {
let function1 = create_function("func1", vec![], "body1");
let function2 = create_function(
"func2",
vec![
Decorator::After("func1".to_string()),
Decorator::After("func3".to_string()),
],
"body2",
);
let function3 = create_function(
"func3",
vec![Decorator::After("func1".to_string())],
"body3",
);
let function4 = create_function("func4", vec![], "body4");
let functions = vec![
function1.clone(),
function2.clone(),
function3.clone(),
function4.clone(),
];
let result = drain_dag(functions).unwrap();
assert_eq!(result.len(), 4);
let positions: std::collections::HashMap<String, usize> = result
.iter()
.enumerate()
.map(|(i, f)| (f.name.clone(), i))
.collect();
assert!(positions.get("func1").unwrap() < positions.get("func2").unwrap());
assert!(positions.get("func1").unwrap() < positions.get("func3").unwrap());
assert!(positions.get("func3").unwrap() < positions.get("func2").unwrap());
}
#[test]
fn test_mixed_decorators() {
let function1 = create_function(
"func1",
vec![Decorator::Pipeline, Decorator::Timeout(1000)],
"body1",
);
let function2 = create_function(
"func2",
vec![
Decorator::After("func1".to_string()),
Decorator::Retry(3, 100),
Decorator::Fallible,
],
"body2",
);
let functions = vec![function1.clone(), function2.clone()];
let result = drain_dag(functions).unwrap();
assert_eq!(result.len(), 2);
assert_eq!(result[0].name, "func1");
assert_eq!(result[1].name, "func2");
}
#[test]
fn test_function_with_multiple_after_decorators() {
let function1 = create_function("func1", vec![], "body1");
let function2 = create_function("func2", vec![], "body2");
let function3 = create_function(
"func3",
vec![
Decorator::After("func1".to_string()),
Decorator::After("func2".to_string()),
],
"body3",
);
let functions = vec![function1.clone(), function2.clone(), function3.clone()];
let result = drain_dag(functions).unwrap();
assert_eq!(result.len(), 3);
let positions: std::collections::HashMap<String, usize> = result
.iter()
.enumerate()
.map(|(i, f)| (f.name.clone(), i))
.collect();
assert!(positions.get("func1").unwrap() < positions.get("func3").unwrap());
assert!(positions.get("func2").unwrap() < positions.get("func3").unwrap());
}
#[test]
fn test_ignores_non_after_decorators() {
let function1 = create_function(
"func1",
vec![
Decorator::Pipeline,
Decorator::If("condition".to_string()),
Decorator::Fallible,
],
"body1",
);
let function2 = create_function(
"func2",
vec![
Decorator::Loop(5),
Decorator::Timeout(1000),
Decorator::Retry(3, 100),
Decorator::Export,
],
"body2",
);
let functions = vec![function1.clone(), function2.clone()];
let result = drain_dag(functions).unwrap();
assert_eq!(result.len(), 2);
let names: Vec<String> = result.iter().map(|f| f.name.clone()).collect();
assert!(names.contains(&"func1".to_string()));
assert!(names.contains(&"func2".to_string()));
}
#[test]
fn test_implicit_after_pipeline() {
let func1 = create_function("func1", vec![Decorator::Pipeline], "body1");
let func2 = create_function("func2", vec![Decorator::Pipeline], "body2");
let func3 = create_function("func3", vec![Decorator::Pipeline], "body3");
let functions = vec![func1.clone(), func2.clone(), func3.clone()];
let result = drain_dag(functions).unwrap();
assert_eq!(result.len(), 3);
assert_eq!(result[0].name, "func1");
assert_eq!(result[1].name, "func2");
assert_eq!(result[2].name, "func3");
}
#[test]
fn test_first_pipeline_no_implicit_after() {
let func1 = create_function("func1", vec![Decorator::Pipeline], "body1");
let functions = vec![func1.clone()];
let result = drain_dag(functions).unwrap();
assert_eq!(result.len(), 1);
assert_eq!(result[0].name, "func1");
}
#[test]
fn test_explicit_after_overrides_implicit() {
let func1 = create_function("func1", vec![Decorator::Pipeline], "body1");
let func2 = create_function(
"func2",
vec![Decorator::Pipeline, Decorator::After("func1".to_string())],
"body2",
);
let func3 = create_function("func3", vec![Decorator::Pipeline], "body3");
let functions = vec![func1.clone(), func2.clone(), func3.clone()];
let result = drain_dag(functions).unwrap();
assert_eq!(result.len(), 3);
let positions: std::collections::HashMap<String, usize> = result
.iter()
.enumerate()
.map(|(i, f)| (f.name.clone(), i))
.collect();
assert!(positions.get("func1").unwrap() < positions.get("func2").unwrap());
assert!(positions.get("func2").unwrap() < positions.get("func3").unwrap());
}
#[test]
fn test_mixed_pipeline_and_non_pipeline() {
let func1 = create_function("func1", vec![], "body1");
let func2 = create_function("func2", vec![Decorator::Pipeline], "body2");
let func3 = create_function("func3", vec![], "body3");
let func4 = create_function("func4", vec![Decorator::Pipeline], "body4");
let functions = vec![func1.clone(), func2.clone(), func3.clone(), func4.clone()];
let result = drain_dag(functions).unwrap();
assert_eq!(result.len(), 4);
let positions: std::collections::HashMap<String, usize> = result
.iter()
.enumerate()
.map(|(i, f)| (f.name.clone(), i))
.collect();
assert!(positions.get("func2").unwrap() < positions.get("func4").unwrap());
}
#[test]
fn test_single_pipeline_works() {
let func1 = create_function("func1", vec![Decorator::Pipeline], "body1");
let functions = vec![func1.clone()];
let result = drain_dag(functions).unwrap();
assert_eq!(result.len(), 1);
assert_eq!(result[0].name, "func1");
}
#[test]
fn test_two_pipelines_implicit_chain() {
let func1 = create_function("func1", vec![Decorator::Pipeline], "body1");
let func2 = create_function("func2", vec![Decorator::Pipeline], "body2");
let functions = vec![func1.clone(), func2.clone()];
let result = drain_dag(functions).unwrap();
assert_eq!(result.len(), 2);
let positions: std::collections::HashMap<String, usize> = result
.iter()
.enumerate()
.map(|(i, f)| (f.name.clone(), i))
.collect();
assert!(positions.get("func1").unwrap() < positions.get("func2").unwrap());
}
#[test]
fn test_pipeline_explicit_after_skips_implicit() {
let func1 = create_function("func1", vec![Decorator::Pipeline], "body1");
let func2 = create_function("func2", vec![Decorator::Pipeline], "body2");
let func3 = create_function(
"func3",
vec![Decorator::Pipeline, Decorator::After("func1".to_string())],
"body3",
);
let functions = vec![func1.clone(), func2.clone(), func3.clone()];
let result = drain_dag(functions).unwrap();
assert_eq!(result.len(), 3);
let positions: std::collections::HashMap<String, usize> = result
.iter()
.enumerate()
.map(|(i, f)| (f.name.clone(), i))
.collect();
assert!(positions.get("func1").unwrap() < positions.get("func3").unwrap());
// TODO: Our design spec indicates order-based sorting, however toposort crate does not guarantee order of independent nodes. We should consider whether we want to enforce order-based sorting for pipelines, or if we are okay with any valid topological order.
// assert!(positions.get("func2").unwrap() < positions.get("func3").unwrap());
}
#[test]
fn test_circular_dependency_detected() {
let func1 = create_function(
"func1",
vec![Decorator::After("func2".to_string())],
"body1",
);
let func2 = create_function(
"func2",
vec![Decorator::After("func1".to_string())],
"body2",
);
let functions = vec![func1.clone(), func2.clone()];
let result = drain_dag(functions);
assert!(result.is_err());
assert!(matches!(
result.unwrap_err(),
BakeError::CircularDependency(_)
));
}
#[test]
fn test_unknown_dependency_error() {
let func1 = create_function(
"func1",
vec![Decorator::After("nonexistent".to_string())],
"body1",
);
let functions = vec![func1.clone()];
let result = drain_dag(functions);
assert!(result.is_err());
assert!(matches!(
result.unwrap_err(),
BakeError::UnknownDependency(_, _)
));
}
#[test]
fn test_pipeline_with_multiple_decorators() {
let func1 = create_function(
"func1",
vec![
Decorator::Pipeline,
Decorator::Timeout(60),
Decorator::Fallible,
],
"body1",
);
let func2 = create_function(
"func2",
vec![
Decorator::Pipeline,
Decorator::Retry(3, 5),
Decorator::After("func1".to_string()),
],
"body2",
);
let functions = vec![func1.clone(), func2.clone()];
let result = drain_dag(functions).unwrap();
assert_eq!(result.len(), 2);
let positions: std::collections::HashMap<String, usize> = result
.iter()
.enumerate()
.map(|(i, f)| (f.name.clone(), i))
.collect();
assert!(positions.get("func1").unwrap() < positions.get("func2").unwrap());
}
#[test]
fn test_long_pipeline_chain() {
let func1 = create_function("func1", vec![Decorator::Pipeline], "body1");
let func2 = create_function("func2", vec![Decorator::Pipeline], "body2");
let func3 = create_function("func3", vec![Decorator::Pipeline], "body3");
let func4 = create_function("func4", vec![Decorator::Pipeline], "body4");
let func5 = create_function("func5", vec![Decorator::Pipeline], "body5");
let functions = vec![
func1.clone(),
func2.clone(),
func3.clone(),
func4.clone(),
func5.clone(),
];
let result = drain_dag(functions).unwrap();
assert_eq!(result.len(), 5);
let positions: std::collections::HashMap<String, usize> = result
.iter()
.enumerate()
.map(|(i, f)| (f.name.clone(), i))
.collect();
assert!(positions.get("func1").unwrap() < positions.get("func2").unwrap());
assert!(positions.get("func2").unwrap() < positions.get("func3").unwrap());
assert!(positions.get("func3").unwrap() < positions.get("func4").unwrap());
assert!(positions.get("func4").unwrap() < positions.get("func5").unwrap());
}
#[test]
fn test_non_pipeline_between_pipelines() {
let func1 = create_function("func1", vec![Decorator::Pipeline], "body1");
let func2 = create_function("func2", vec![], "body2");
let func3 = create_function("func3", vec![Decorator::Pipeline], "body3");
let functions = vec![func1.clone(), func2.clone(), func3.clone()];
let result = drain_dag(functions).unwrap();
assert_eq!(result.len(), 3);
let positions: std::collections::HashMap<String, usize> = result
.iter()
.enumerate()
.map(|(i, f)| (f.name.clone(), i))
.collect();
assert!(positions.get("func1").unwrap() < positions.get("func3").unwrap());
}
// This function is replaced by Dag::pick_batch() which has all the
// group metadata internally. Keeping the signature for backward compat
// but redirect callers to use Dag directly.
unimplemented!("use dag.pick_batch() instead")
}
+2 -3
View File
@@ -1,5 +1,4 @@
use crate::bake::builder::{build_script, RenderMode};
use crate::bake::constant::TRIVIAL_SCRIPT_NAME;
use crate::bake::error::BakeError;
use crate::bake::parser::Function;
use std::path::Path;
@@ -22,11 +21,11 @@ pub async fn run_trivial(
render_mode: RenderMode,
) -> Result<(), BakeError> {
let function = Function {
name: TRIVIAL_SCRIPT_NAME.to_string(),
name: String::new(),
decorators: vec![],
body: script_content.to_string(),
};
let script = build_script(&function, bake_base_path, workspace, None, render_mode)?;
let script = build_script(&function, bake_base_path, workspace, render_mode)?;
engine.execute_script(&script, ctx, event_tx).await?;
Ok(())
}
+1
View File
@@ -27,6 +27,7 @@ pub async fn run_bare(
("finalize", or_workshop(&cli.finalize, "finalize.yml"))
}
};
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");
-52
View File
@@ -1,52 +0,0 @@
# workshop-baker/src/finalize
**Generated:** 2026-05-19
**Commit:** 28abc5d
Artifact packaging, plugin-based notifications, and post-build hooks.
## STRUCTURE
```
finalize/
├── config.rs # FinalizeConfig YAML (379 lines, plugins, notifications)
├── stage.rs # FinalizeStage enum
├── stage/hook.rs # Post-build hook execution
├── event.rs # Event propagation
├── template.rs # Variable substitution {{VARIABLE}}
├── plugin.rs # Plugin trait + registration
├── plugin/
│ ├── metadata.rs # Plugin manifest parsing
│ ├── dylib.rs # Dynamic library plugins
│ ├── rhai.rs # Rhai scripting (todo!())
│ ├── shell.rs # Shell command plugins
│ └── internal.rs + internal/ # Built-in plugins (mail, webhook, satori, insitenotify)
├── types.rs
├── types/compression.rs
├── error.rs # FinalizeError
└── constant.rs
```
## WHERE TO LOOK
| Task | Location |
|------|----------|
| Config loading | `config.rs:17` - `FinalizeConfig` + `parse()` |
| Plugin registration | `plugin.rs` - `register()` |
| Email notifications | `plugin/internal/mail.rs` - SMTP |
| Webhooks | `plugin/internal/webhook.rs` |
| Satori integration | `plugin/internal/satori.rs` |
| Shell plugins | `plugin/shell.rs` |
| Template rendering | `template.rs` - `{{VARIABLE}}` handlebars-style |
## CONVENTIONS
- **Plugin System**: `FinalizePlugin` trait + registry with both internal and dynamic plugins
- **Template Syntax**: `{{VARIABLE}}` handlebars-style substitution via `template.rs`
- **Stages**: EarlyHook → (Artifacts) → LateHook → Notifications
- **Notification**: `notify.rs` in parent module drives plugin-based notification dispatch
- **ResourceRegistry**: Shared resource cache passed through to plugins
## ANTI-PATTERNS
1. **fetch/ subdirectory removed** — was `plugin/fetch/{git,http,extract,checksum}`, no longer exists
2. **SHA1 deprecated** — workshop-getterurl handles checksums now
3. **Unimplemented**`rhai.rs:17` - "todo!()" for Rhai scripting
4. **TODO markers** — Multiple TODOs in config.rs (lines 305-306)
5. **Artifact stage not implemented**`finalize.rs:57` — "TODO: artifact stage implementation"
+11 -5
View File
@@ -185,15 +185,21 @@ impl<'de> serde::Deserialize<'de> for NotificationConfig {
#[serde(default)]
templates: HashMap<String, NotificationTemplateDef>,
#[serde(flatten)]
groups_raw: HashMap<String, HashMap<String, NotificationMethod>>,
groups_raw: HashMap<String, Option<HashMap<String, NotificationMethod>>>,
}
let h = Helper::deserialize(deserializer)?;
let groups: HashMap<u32, _> = h.groups_raw.into_iter()
.map(|(k, v)| {
k.parse::<u32>().map(|n| (n, v))
.map_err(|_| serde::de::Error::custom(format!("priority key must be integer, got {:?}", k)))
.filter_map(|(k, v)| {
let parsed = k.parse::<u32>().map(|n| (n, v.unwrap_or_default()));
match parsed {
Ok(pair) => Some(pair),
Err(e) => {
log::warn!("priority key must be integer, got {:?}: {}", k, e);
None
}
}
})
.collect::<Result<_, _>>()?;
.collect();
Ok(NotificationConfig { templates: h.templates, groups })
}
}
+13 -8
View File
@@ -30,14 +30,19 @@ pub async fn event_receiver(
}
},
ExecutionEvent::TaskCompleted { task_id, result } => {
log::debug!(
"[{}] [{}] Task completed: {} (exit_code={}, duration={:?})",
pipeline_name,
build_id,
task_id,
result.exit_code,
result.duration
);
if result.success {
log::debug!(
"[{}] [{}] Task completed: {} (exit_code={}, duration={:?})",
pipeline_name, build_id, 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,
result.exit_code, result.duration,
);
}
}
ExecutionEvent::TaskFailed { task_id, error } => {
log::error!(
+3 -1
View File
@@ -232,7 +232,9 @@ pub fn register(
_ => {}
}
log::debug!("Registering internal plugin {}", i.name);
plugins.insert(i.name.to_string(), FinalizePlugin::new(i.entry));
let mut plugin = FinalizePlugin::new(i.entry);
plugin.metadata.renderer = Some(i.renderer);
plugins.insert(i.name.to_string(), plugin);
count += 1;
}
@@ -1,4 +1,5 @@
use crate::finalize::plugin::AsyncPluginFn;
use crate::notify::types::NotificationRenderer;
pub mod dummy;
mod insitenotify;
mod mail;
@@ -7,6 +8,7 @@ mod webhook;
pub struct InternalPlugin {
pub name: &'static str,
pub entry: Box<dyn AsyncPluginFn>,
pub renderer: NotificationRenderer,
}
pub fn get_internal_plugin() -> Vec<InternalPlugin> {
@@ -14,18 +16,22 @@ pub fn get_internal_plugin() -> Vec<InternalPlugin> {
InternalPlugin {
name: "in-site-notify",
entry: Box::new(insitenotify::InsiteNotify),
renderer: NotificationRenderer::Plugin,
},
InternalPlugin {
name: "mail",
entry: Box::new(mail::Mail),
renderer: NotificationRenderer::Server,
},
InternalPlugin {
name: "satori",
entry: Box::new(satori::Satori),
renderer: NotificationRenderer::Server,
},
InternalPlugin {
name: "webhook",
entry: Box::new(webhook::Webhook),
renderer: NotificationRenderer::Server,
},
]
}
@@ -34,5 +40,6 @@ pub fn get_dummy_plugin() -> Vec<InternalPlugin> {
vec![InternalPlugin {
name: "dummy",
entry: Box::new(dummy::Dummy),
renderer: NotificationRenderer::Server,
}]
}
+1 -1
View File
@@ -51,7 +51,7 @@ pub async fn hook(
.await
.map_err(|e| {
log::error!("Plugin hook {} failed: {:?}", index, e);
ExecutionError::ExecutionFailed(e.to_string())
ExecutionError::ExecutionFailed { task_id: format!("hook-{}", index), exit_code: -1 }
})?;
continue;
}
+2
View File
@@ -80,6 +80,8 @@ async fn worker_main(cli: Cli) -> Result<(), CliError> {
}
log::info!("Pipeline complete. Draining notifications...");
// Drop the sender so notify() can end and retry_tx is dropped
queue.tx = None;
if let Err(e) = queue.drain().await {
log::error!("Notification queue error: {}", e);
}
+4 -3
View File
@@ -31,8 +31,9 @@ pub async fn notify(
) -> anyhow::Result<()> {
let mut finalize = parse(finalize_path)?;
// Always inject in-site-notify as the lowest priority fallback
{
// L18: 0 priorities → notification ends; L19: has priorities → inject in-site fallback
if !finalize.notification.groups.is_empty() {
log::debug!("Injecting in-site-notify fallback");
let max_p = finalize.notification.groups.keys().max().copied().unwrap_or(0);
let mut m = std::collections::HashMap::new();
m.insert(
@@ -50,7 +51,7 @@ pub async fn notify(
Some(&plugin_list),
&mut finalize.notification,
registry,
)?;
)?;
resolve_external_template_refs(&mut finalize.notification.templates, registry);
flatten_template_refs(&mut finalize.notification.templates)?;
+2 -1
View File
@@ -336,7 +336,8 @@ fn build_plugin_argument(
Some(argument)
}
NotificationRenderer::Plugin => {
unimplemented!("Plugin renderer is not yet implemented")
// PSR: pass config directly to plugin (plugin handles own rendering)
Some(rule.config.clone())
}
NotificationRenderer::Interactive => {
unimplemented!("Interactive renderer is not yet implemented")
+8 -4
View File
@@ -8,7 +8,7 @@ use std::collections::BinaryHeap;
/// - `rx` receives retry events back from notify()
pub struct NotificationQueue {
pub pq: BinaryHeap<NotificationEvent>,
pub tx: NotificationEventSender,
pub tx: Option<NotificationEventSender>,
pub rx: NotificationEventReceiver,
}
@@ -16,7 +16,7 @@ impl NotificationQueue {
pub fn new(tx: NotificationEventSender, rx: NotificationEventReceiver) -> Self {
Self {
pq: BinaryHeap::new(),
tx,
tx: Some(tx),
rx,
}
}
@@ -40,7 +40,9 @@ impl NotificationQueue {
break;
}
let event = self.pq.pop().unwrap();
self.tx.send(event).await?;
if let Some(ref tx) = self.tx {
tx.send(event).await?;
}
}
Ok(())
}
@@ -77,7 +79,9 @@ impl NotificationQueue {
while let Some(peek) = self.pq.peek() {
if peek.not_before > now { break; }
let event = self.pq.pop().unwrap();
self.tx.send(event).await?;
if let Some(ref tx) = self.tx {
tx.send(event).await?;
}
}
}
}
+2
View File
@@ -109,6 +109,7 @@ pub fn resolve_external_template_refs(
templates: &mut HashMap<String, NotificationTemplateDef>,
registry: &ResourceRegistry,
) {
log::debug!("Resolving external template refs");
for (name, tmpl_def) in templates.iter_mut() {
if tmpl_def.use_.is_none() {
continue;
@@ -186,6 +187,7 @@ pub fn resolve_template_by_schema(
pub fn flatten_template_refs(
templates: &mut HashMap<String, NotificationTemplateDef>,
) -> Result<(), NotifyError> {
log::debug!("Flattening template refs");
let names: Vec<String> = templates.keys().cloned().collect();
for name in names {
let mut visited = std::collections::HashSet::new();
+1 -1
View File
@@ -78,7 +78,7 @@ impl ParsedTrigger {
impl NotificationRule {
pub fn accept(&self, result: &StageOutcome, stage: &StagePhase, substage: &str) -> bool {
if self.trigger.is_empty() {
eprintln!("[ACCEPT] trigger empty, returning true");
log::trace!("[ACCEPT] trigger empty, returning true");
return true;
}
self.trigger
+1
View File
@@ -5,6 +5,7 @@ use crate::notify::types::{DEFAULT_NOTIFY_PARAMS, NotificationEvent, Notificatio
use std::collections::HashSet;
pub fn validate(finalize: &FinalizeConfig) -> anyhow::Result<()> {
log::debug!("Validating finalize config");
let validate_result = finalize.validate();
if let Err(errors) = validate_result {
for error in &errors {
-54
View File
@@ -1,54 +0,0 @@
# workshop-baker/src/prebake
**Generated:** 2026-04-22
**Commit:** 7b3b71a
Build environment setup: bootstrap, dependencies, security, hooks.
## STRUCTURE
```
prebake/
├── config.rs # PrebakeConfig YAML schema (environment, deps, cache)
├── stage.rs # PrebakeStage enum (Bootstrap→EarlyHook→DepsSystem→DepsUser→LateHook→Ready)
├── stage/
│ ├── bootstrap.rs # User creation (vulcan), sudo/doas setup
│ ├── environment.rs # Env var injection
│ ├── depssystem.rs # System package installation
│ ├── depsuser.rs # User package installation (cargo, npm, etc.)
│ └── hook.rs # Hook execution (pre/post stage)
├── env/ # Environment providers
│ ├── mod.rs # EnvProvider trait
│ ├── container.rs # Docker/container support
│ ├── firecracker.rs # Firecracker microVM
│ ├── baremetal.rs # Direct host execution
│ └── custom.rs # Custom env (todo!())
├── security.rs # Privilege dropping (unsafe libc calls)
├── event.rs # Event logging
├── error.rs # PrebakeError + BootstrapError
├── types.rs # Module re-exports
├── types/ # memsize, cache, builderconfig, architecture
├── constant.rs # Module constants
└── env.rs # Environment module root
```
## WHERE TO LOOK
| Task | Location |
|------|----------|
| Config loading | `config.rs:PrebakeConfig::from_yaml()` |
| Stage ordering | `stage.rs:PrebakeStage` enum |
| User setup | `stage/bootstrap.rs` - vulcan user creation |
| Security | `security.rs:72` - privilege dropping |
| Hooks | `stage/hook.rs` - pre/post stage hooks |
## CONVENTIONS
- **PrebakeStage**: Ordered enum Bootstrap→EarlyHook→DepsSystem→DepsUser→LateHook→Ready
- **ExecutionContext**: Carries task_id, username, env_vars through all stages
- **Target User**: "vulcan" (hardcoded in bootstrap.rs:6-24)
## ANTI-PATTERNS
1. **Hardcoded user** — "vulcan" username hardcoded throughout
2. **Unsafe blocks**`security.rs` has 5+ unsafe libc calls
3. **Incomplete** — custom.rs and apt.rs have `todo!()` stubs
4. **doas.conf** — bootstrap.rs:183: "We do not know how to write doas.conf lol"
+23 -14
View File
@@ -1,5 +1,7 @@
use workshop_engine::{ExecutionEvent, StreamType};
use crate::bake::constant::PIPELINE_SKIP_ERRORCODE;
pub async fn event_receiver(
mut rx: tokio::sync::mpsc::Receiver<ExecutionEvent>,
pipeline_name: String,
@@ -7,34 +9,41 @@ pub async fn event_receiver(
) {
while let Some(event) = rx.recv().await {
match event {
ExecutionEvent::TaskStarted { task_id, timestamp } => {
ExecutionEvent::TaskStarted { task_id, timestamp: _ } => {
log::debug!(
"[{}] [{}] [{}] Task started: {}",
"[{}] [{}] Task started: {}",
pipeline_name,
build_id,
timestamp,
// timestamp,
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 } => {
log::debug!(
"[{}] [{}] Task completed: {} (exit_code={}, duration={:?})",
pipeline_name,
build_id,
task_id,
result.exit_code,
result.duration
);
if result.success {
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 {
log::warn!(
"[{}] [{}] Task completed with non-zero exitcode: {} (exit_code={}, duration={:?})",
pipeline_name, build_id, task_id,
result.exit_code, result.duration,
);
}
}
ExecutionEvent::TaskFailed { task_id, error } => {
log::error!(
+39
View File
@@ -0,0 +1,39 @@
# WORKSHOP-ENGINE KNOWLEDGE BASE
## OVERVIEW
Core execution layer for HoneyBiscuitWorkshop. Library-only crate (no binary). Provides Engine executor, process abstraction, package manager interfaces, duration parsing, and typed error/event systems. Used by workshop-baker; depends on no other workspace crates.
## WHERE TO LOOK
| Task | File | Notes |
|------|------|-------|
| Script execution | `src/executor.rs` | `Engine::execute_script()` — spawn, stream, timeout, kill |
| Process abstraction | `src/child.rs` | `ManagedChild` trait; `TokioChild` (real), `MockChild` (hand-rolled, no mockall) |
| Configurable commands | `src/command.rs` | `CustomCommand` — name, command, env, working_dir, timeout |
| System PMs | `src/pm.rs` + `pm/{apt,pacman,dnf,apk}.rs` | `PackageManager` trait; `all_managers()`, `select(name)` |
| User PMs | `src/upm.rs` + `upm/{rust,go,python,custom}.rs` | `UserPackageManager` trait (async); `UPMSysDeps` |
| Repology endpoint | `src/repology.rs` + `repology/local.rs` | `RepologyEndpoint` enum; local instance support |
| Duration parsing | `src/time.rs` | `parse_duration_to_ms()`, serde helpers |
| Core types | `src/types.rs` | `ExecutionContext`, `ExecutionResult`, `ExecutionEvent`, `EventSender`, `StreamType`, `Engine` |
| Error types | `src/error.rs` | `ExecutionError`, `DependencyError`, `HasExitCode` trait, `EXITCODE_*` constants |
| Integration tests | `tests/engine_integration.rs` | 12 async tests; run with `cargo test --test engine_integration` |
## CONVENTIONS
- **Edition**: Rust 2024
- **Key deps**: tokio (full), thiserror, serde+serde_yaml, async-trait, chrono, os_info, duration-str, libc, lazy_static
- **Process mgmt**: `ManagedChild` trait abstracts over real/mock children. `TokioChild` wraps `tokio::process::Child`. `MockChild` is hand-rolled (no mockall dep); supports hanging, custom output, exit codes.
- **Package manager selection**: Use `pm::select("apt")` with explicit name — NEVER auto-detect by probing the filesystem.
- **Error mapping**: `HasExitCode` trait maps errors to numeric exit codes. `ExecutionError` maps to specific `EXITCODE_*`; `DependencyError` always maps to 1.
- **Re-exports**: `lib.rs` re-exports key types for convenience (`Engine`, `ExecutionContext`, `PackageManager`, `ExecutionError`, etc.)
- **Tests**: Inline `#[cfg(test)]` blocks in `pm.rs` (8 tests). Integration tests in `tests/engine_integration.rs` (12 async tests, separate binary). CI runs both.
- **Kill semantics**: `TokioChild::kill()` sends `SIGKILL` to entire process group via `libc::kill(-PGID, SIGKILL)` with `ESRCH` silently ignored.
## ANTI-PATTERNS
- **Do NOT auto-detect the host package manager** — always use `pm::select()` with an explicitly configured name. Production PM selection comes from environment config, not probing.
- **Do NOT use `detect()`/`adopt()` patterns** for PM selection — these don't exist in this crate, and callers must not invent them.
- **Do NOT depend on workshop-baker or other workspace crates** — engine is a leaf dependency. Types needed by external callers should flow through `ExecutionContext`, not reverse imports.
- **Do NOT use mockall or other mocking frameworks** — `MockChild` is the only mock mechanism. Keep it that way.
- **Do NOT use `unimplemented!()`** — all package manager implementations are complete. Stubs belong in workshop-baker, not here.
- **Do NOT add magic platform detection** — `os_info` is used only for mirror config mapping in PM impls. Do not extend its use to auto-detect behavior.
-51
View File
@@ -1,51 +0,0 @@
# workshop-engine/src
**Generated:** 2026-05-19
**Commit:** 28abc5d
Execution engine extracted from workshop-baker. Process management, package detection, resource isolation.
## STRUCTURE
```
src/
├── lib.rs # Module declarations + 10+ convenience re-exports
├── error.rs # ExecutionError + DependencyError + HasExitCode trait
├── types.rs # ExecutionContext, ExecutionResult, Engine, etc.
├── child.rs # ManagedChild + TokioChild process wrappers
├── command.rs # CustomCommand builder
├── executor.rs # Script execution
├── time.rs # Duration parsing helpers
├── pm.rs # PackageManager trait + detection
├── pm/ # Backends: apt, apk, dnf, pacman
├── upm.rs # UserPackageManager trait
├── upm/ # Backends: rust (cargo), python (pip), go, custom
├── repology.rs # RepologyEndpoint enum
└── repology/ # Package name resolution (local/)
```
## KEY EXPORTS
| Symbol | Type | Role |
|--------|------|------|
| ExecutionContext | struct | Pipeline context (task_id, username, env, etc.) |
| Engine | struct | Script executor (placeholder) |
| ExecutionError | enum | Error types with exit codes |
| HasExitCode | trait | Exit code protocol for all error types |
| ManagedChild | struct | Child process wrapper with managed lifecycle |
| TokioChild | struct | Tokio-based child process wrapper |
| CustomCommand | struct | Builder for command execution |
| PackageManager | trait | System package manager abstraction |
| UserPackageManager | trait | User-level package manager abstraction |
| RepologyEndpoint | enum | Package name resolution endpoint |
| ExecutionEvent | enum | Event stream (TaskStarted, OutputChunk, etc.) |
| ExecutionResult | struct | Command execution result (exit_code, stdout, stderr) |
| EventSender | type alias | `Option<mpsc::Sender<ExecutionEvent>>` |
## CONVENTIONS
- **Exit codes**: Centralized in `error.rs` — EXITCODE_OK=0 through EXITCODE_PRIV_DROP_FAILED=201
- **ExecutionContext**: Defaults to current directory, bash shell, non-privileged
- **PackageManagers**: Auto-detected via `os_info`, fallback chain (apt→apk→dnf→pacman)
- **Engine**: Currently a unit struct with `execute_script()` method
- **All errors implement HasExitCode**: Enables uniform error propagation up to CLI
## ANTI-PATTERNS
1. **Engine is a stub**`types.rs:99``pub struct Engine;` with no fields, minimal implementation
+3 -3
View File
@@ -21,8 +21,8 @@ pub enum ExecutionError {
#[error("Invalid command: {0}")]
InvalidCommand(String),
#[error("Execution failed: {0}")]
ExecutionFailed(String),
#[error("Task {task_id} failed with exit code {exit_code}")]
ExecutionFailed { task_id: String, exit_code: i32 },
#[error("Permission denied: uid={euid}, gid={egid}: {reason}")]
PermissionDenied {
@@ -48,7 +48,7 @@ impl HasExitCode for ExecutionError {
fn exit_code(&self) -> i32 {
match self {
ExecutionError::InvalidCommand(_) => EXITCODE_EXECUTION_ERROR,
ExecutionError::ExecutionFailed(_) => EXITCODE_EXECUTION_ERROR,
ExecutionError::ExecutionFailed { .. } => EXITCODE_EXECUTION_ERROR,
ExecutionError::PermissionDenied { .. } => EXITCODE_GENERAL_ERROR,
ExecutionError::Timeout => EXITCODE_TIMEOUT,
ExecutionError::IoError(_) => EXITCODE_IO_ERROR,
+71 -56
View File
@@ -13,7 +13,7 @@ use crate::types::{
DeltaExecutionContext, EventSender, ExecutionContext, ExecutionError, ExecutionEvent,
ExecutionResult, StreamType,
};
const STDOUT_BUF_CAP: usize = 2 * 1024 * 1024;
impl crate::types::Engine {
pub fn new() -> Self {
Self
@@ -159,11 +159,8 @@ impl crate::types::Engine {
.await;
}
let task_id = ctx.task_id.clone();
let (stdout_tx, mut stdout_rx) = mpsc::channel::<String>(1024);
let (stderr_tx, mut stderr_rx) = mpsc::channel::<String>(1024);
let mut stdout_buf = String::new();
let mut stderr_buf = String::new();
let mut child_stdout = child.take_stdout();
let mut child_stderr = child.take_stderr();
@@ -208,6 +205,32 @@ impl crate::types::Engine {
}
});
// Forwarder: drains channels in real-time, emits OutputChunk events,
// and accumulates output into 2 MiB ring buffers.
let task_id_ev = ctx.task_id.clone();
let event_tx_fwd = event_tx.clone();
let forwarder = tokio::spawn(async move {
let mut stdout_buf: Vec<u8> = Vec::new();
let mut stderr_buf: Vec<u8> = Vec::new();
loop {
tokio::select! {
biased;
Some(data) = stdout_rx.recv() => {
emit_chunk(&event_tx_fwd, &task_id_ev, StreamType::Stdout, &data).await;
stdout_buf.extend_from_slice(data.as_bytes());
truncate_buf(&mut stdout_buf);
}
Some(data) = stderr_rx.recv() => {
emit_chunk(&event_tx_fwd, &task_id_ev, StreamType::Stderr, &data).await;
stderr_buf.extend_from_slice(data.as_bytes());
truncate_buf(&mut stderr_buf);
}
else => break,
}
}
(stdout_buf, stderr_buf)
});
let exit_code = if let Some(timeout_duration) = ctx.timeout {
match timeout(timeout_duration, child.wait()).await {
Ok(Ok(code)) => code,
@@ -236,12 +259,7 @@ impl crate::types::Engine {
drop(stderr_tx);
let _ = stdout_handle.await;
let _ = stderr_handle.await;
while let Some(data) = stdout_rx.recv().await {
stdout_buf.push_str(&data);
}
while let Some(data) = stderr_rx.recv().await {
stderr_buf.push_str(&data);
}
let _ = forwarder.await;
return Err(ExecutionError::Timeout);
}
}
@@ -256,31 +274,9 @@ impl crate::types::Engine {
let _ = stdout_handle.await;
let _ = stderr_handle.await;
while let Some(data) = stdout_rx.recv().await {
stdout_buf.push_str(&data);
if let Some(tx) = event_tx {
let _ = tx
.send(ExecutionEvent::OutputChunk {
task_id: task_id.clone(),
stream: StreamType::Stdout,
data: data.clone(),
})
.await;
}
}
while let Some(data) = stderr_rx.recv().await {
stderr_buf.push_str(&data);
if let Some(tx) = event_tx {
let _ = tx
.send(ExecutionEvent::OutputChunk {
task_id: task_id.clone(),
stream: StreamType::Stderr,
data: data.clone(),
})
.await;
}
}
let (stdout_buf, stderr_buf) = forwarder
.await
.map_err(|e| ExecutionError::IoError(format!("Forwarder task panicked: {}", e)))?;
let duration = start_time.elapsed();
@@ -289,32 +285,26 @@ impl crate::types::Engine {
exit_code,
success: exit_code == 0,
duration,
stdout: stdout_buf,
stderr: stderr_buf,
stdout: String::from_utf8_lossy(&stdout_buf).into_owned(),
stderr: String::from_utf8_lossy(&stderr_buf).into_owned(),
};
if let Some(tx) = event_tx {
let event = if result.success {
ExecutionEvent::TaskCompleted {
let _ = tx
.send(ExecutionEvent::TaskCompleted {
task_id: ctx.task_id.clone(),
result: result.clone(),
}
} else {
ExecutionEvent::TaskFailed {
task_id: ctx.task_id.clone(),
error: format!("Exit code: {}", exit_code),
}
};
let _ = tx.send(event).await;
})
.await;
}
if result.success {
Ok(result)
} else {
Err(ExecutionError::ExecutionFailed(format!(
"Task {} failed with exit code {}",
ctx.task_id, exit_code
)))
Err(ExecutionError::ExecutionFailed {
task_id: ctx.task_id.clone(),
exit_code,
})
}
}
@@ -352,6 +342,30 @@ impl crate::types::Engine {
Ok(combined_result)
}
}
/// Truncate a buffer to [`STDOUT_BUF_CAP`] bytes, discarding the oldest data.
fn truncate_buf(buf: &mut Vec<u8>) {
if buf.len() > STDOUT_BUF_CAP {
buf.drain(0..buf.len() - STDOUT_BUF_CAP);
}
}
/// Emit an `OutputChunk` event if the channel exists.
async fn emit_chunk(
tx: &EventSender,
task_id: &str,
stream: StreamType,
data: &str,
) {
if let Some(tx) = tx {
let _ = tx
.send(ExecutionEvent::OutputChunk {
task_id: task_id.to_string(),
stream,
data: data.to_string(),
})
.await;
}
}
impl Default for crate::types::Engine {
fn default() -> Self {
@@ -436,12 +450,12 @@ mod tests {
assert!(result.is_err());
assert!(matches!(
result.unwrap_err(),
ExecutionError::ExecutionFailed(_)
ExecutionError::ExecutionFailed { .. }
));
}
#[tokio::test]
async fn exit_nonzero_sends_task_failed_event() {
async fn exit_nonzero_sends_task_completed_event() {
let engine = Engine::new();
let (tx, mut rx) = mpsc::channel(64);
let child = MockChild::with_exit_code(42);
@@ -451,9 +465,10 @@ mod tests {
// Skip TaskStarted
let _ = rx.recv().await;
let event = rx.recv().await;
assert!(matches!(event, Some(ExecutionEvent::TaskFailed { .. })));
if let Some(ExecutionEvent::TaskFailed { error, .. }) = event {
assert!(error.contains("42"), "error should mention exit code 42");
assert!(matches!(event, Some(ExecutionEvent::TaskCompleted { .. })));
if let Some(ExecutionEvent::TaskCompleted { result, .. }) = event {
assert_eq!(result.exit_code, 42);
assert!(!result.success);
}
}
+104
View File
@@ -0,0 +1,104 @@
# This file is automatically @generated by Cargo.
# It is not intended for manual editing.
version = 4
[[package]]
name = "equivalent"
version = "1.0.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "877a4ace8713b0bcf2a4e7eec82529c029f1d0619886d18145fea96c3ffe5c0f"
[[package]]
name = "fixedbitset"
version = "0.5.7"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "1d674e81391d1e1ab681a28d99df07927c6d4aa5b027d7da16ba32d1d21ecd99"
[[package]]
name = "hashbrown"
version = "0.17.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "ed5909b6e89a2db4456e54cd5f673791d7eca6732202bbf2a9cc504fe2f9b84a"
[[package]]
name = "indexmap"
version = "2.14.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "d466e9454f08e4a911e14806c24e16fba1b4c121d1ea474396f396069cf949d9"
dependencies = [
"equivalent",
"hashbrown",
]
[[package]]
name = "petgraph"
version = "0.7.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "3672b37090dbd86368a4145bc067582552b29c27377cad4e0a306c97f9bd7772"
dependencies = [
"fixedbitset",
"indexmap",
]
[[package]]
name = "proc-macro2"
version = "1.0.106"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "8fd00f0bb2e90d81d1044c2b32617f68fcb9fa3bb7640c23e9c748e53fb30934"
dependencies = [
"unicode-ident",
]
[[package]]
name = "quote"
version = "1.0.45"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "41f2619966050689382d2b44f664f4bc593e129785a36d6ee376ddf37259b924"
dependencies = [
"proc-macro2",
]
[[package]]
name = "syn"
version = "2.0.117"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "e665b8803e7b1d2a727f4023456bbbbe74da67099c585258af0ad9c5013b9b99"
dependencies = [
"proc-macro2",
"quote",
"unicode-ident",
]
[[package]]
name = "thiserror"
version = "2.0.18"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "4288b5bcbc7920c07a1149a35cf9590a2aa808e0bc1eafaade0b80947865fbc4"
dependencies = [
"thiserror-impl",
]
[[package]]
name = "thiserror-impl"
version = "2.0.18"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "ebc4ee7f67670e9b64d05fa4253e753e016c6c95ff35b89b7941d6b856dec1d5"
dependencies = [
"proc-macro2",
"quote",
"syn",
]
[[package]]
name = "unicode-ident"
version = "1.0.24"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "e6e4313cd5fcd3dad5cafa179702e2b244f760991f45397d14d4ebf38247da75"
[[package]]
name = "workshop-schedule"
version = "0.1.0"
dependencies = [
"petgraph",
"thiserror",
]
+9
View File
@@ -0,0 +1,9 @@
[package]
name = "workshop-schedule"
version = "0.1.0"
edition = "2024"
description = "Semi-general DAG scheduling library for CI/CD pipelines"
[dependencies]
petgraph = "0.7"
thiserror = "2"
+780
View File
@@ -0,0 +1,780 @@
// Code in this module PARTIALLY or FULLY utilized AI Coding Agent
use petgraph::stable_graph::{NodeIndex, StableGraph};
use petgraph::Direction;
use petgraph::visit::EdgeRef;
use std::collections::{BTreeMap, BTreeSet, HashMap};
use thiserror::Error;
use crate::edge::EdgeKind;
use crate::node::{DagNode, NodeColor};
/// Errors that can occur during DAG operations.
#[derive(Debug, Error)]
pub enum DagError {
#[error("duplicate node name: {0}")]
DuplicateNode(String),
#[error("node not found: {0}")]
NodeNotFound(String),
#[error("node not ready: {0}")]
NotReady(String),
#[error("already popped: {0}")]
AlreadyPopped(String),
#[error("circular dependency detected among: {0:?}")]
Cycle(Vec<String>),
#[error("parallel group can never complete: {0:?}")]
GroupDeadlock(Vec<String>),
}
/// Internal per-node metadata.
enum NodeState {
/// Has unsatisfied dependencies.
Pending,
/// All dependencies satisfied, in the ready set.
Ready,
/// Has been popped — done executing.
Done,
}
/// Internal per-group metadata.
#[derive(Debug)]
struct GroupMeta {
/// Node indices belonging to this group.
members: Vec<NodeIndex>,
/// Total number of members (constant after construction).
total: usize,
}
/// A DAG (Directed Acyclic Graph) for scheduling pipeline nodes.
///
/// Nodes implement [`DagNode`] and carry a [`NodeColor`] that determines
/// sequential vs parallel semantics. Edges carry an [`EdgeKind`] distinguishing
/// explicit `@after` from implicit pipeline chains.
///
/// The DAG maintains a ready set (nodes with in-degree 0) in deterministic
/// BTreeSet order, and provides methods to pop nodes **by name** — fixing the
/// limitation of `topological-sort` that could only pop the first ready node.
pub struct Dag<N: DagNode> {
graph: StableGraph<N, EdgeKind>,
name_to_idx: HashMap<String, NodeIndex>,
state: HashMap<NodeIndex, NodeState>,
in_degree: HashMap<NodeIndex, usize>,
/// Ready nodes, sorted by name for deterministic iteration.
ready_set: BTreeSet<(String, NodeIndex)>,
/// Named groups and their members.
groups: BTreeMap<String, GroupMeta>,
node_count: usize,
}
impl<N: DagNode> Dag<N> {
/// Create an empty DAG.
pub fn new() -> Self {
Self {
graph: StableGraph::new(),
name_to_idx: HashMap::new(),
state: HashMap::new(),
in_degree: HashMap::new(),
ready_set: BTreeSet::new(),
groups: BTreeMap::new(),
node_count: 0,
}
}
/// Number of nodes (including pending, ready, and done).
pub fn node_count(&self) -> usize {
self.node_count
}
/// True if no nodes remain in the DAG (all popped).
pub fn is_empty(&self) -> bool {
self.ready_set.is_empty()
&& self.state.values().all(|s| matches!(s, NodeState::Done))
}
// ═══════════════════════════════════════════════════════════════════
// Construction
// ═══════════════════════════════════════════════════════════════════
/// Add a node to the DAG.
///
/// Returns an error if a node with the same name already exists.
pub fn add_node(&mut self, node: N) -> Result<(), DagError> {
let name = node.name().to_string();
if self.name_to_idx.contains_key(&name) {
return Err(DagError::DuplicateNode(name));
}
let color = node.color(); // extract before move
let idx = self.graph.add_node(node);
self.name_to_idx.insert(name.clone(), idx);
self.state.insert(idx, NodeState::Ready);
self.in_degree.insert(idx, 0);
self.ready_set.insert((name.clone(), idx));
self.node_count += 1;
// Register named groups
if let NodeColor::Named(group) = &color {
self.groups
.entry(group.clone())
.or_insert_with(|| GroupMeta {
members: Vec::new(),
total: 0,
})
.members
.push(idx);
}
Ok(())
}
/// Freeze group totals (call after all nodes added, before any pop).
///
/// Locks in `total` for each group. Named groups whose `total` is less
/// than 2 are harmless but wasteful — they'll fire solo like sequential.
pub fn freeze(&mut self) {
for meta in self.groups.values_mut() {
meta.total = meta.members.len();
}
}
/// Add a dependency edge from `from` to `to`.
///
/// `to` depends on `from` — `to` won't be ready until `from` is popped.
pub fn add_edge(
&mut self,
from: &str,
to: &str,
kind: EdgeKind,
) -> Result<(), DagError> {
let from_idx = self
.name_to_idx
.get(from)
.copied()
.ok_or_else(|| DagError::NodeNotFound(from.to_string()))?;
let to_idx = self
.name_to_idx
.get(to)
.copied()
.ok_or_else(|| DagError::NodeNotFound(to.to_string()))?;
self.graph.add_edge(from_idx, to_idx, kind);
// Increment in-degree of `to`
let deg = self.in_degree.get_mut(&to_idx).unwrap();
*deg += 1;
// Remove `to` from ready set if it was there
let to_name = self.graph[to_idx].name().to_string();
self.ready_set.remove(&(to_name.clone(), to_idx));
*self.state.get_mut(&to_idx).unwrap() = NodeState::Pending;
// Sanity: check for self-loop (cycle detection lite)
if from == to {
return Err(DagError::Cycle(vec![from.to_string()]));
}
Ok(())
}
// ═══════════════════════════════════════════════════════════════════
// Queries
// ═══════════════════════════════════════════════════════════════════
/// Name → color lookup. O(1).
pub fn color_of(&self, name: &str) -> Option<NodeColor> {
let idx = self.name_to_idx.get(name)?;
Some(self.graph[*idx].color())
}
/// True if the node is sequential (not parallel). O(1).
pub fn is_sequential(&self, name: &str) -> bool {
self.color_of(name)
.map(|c| c.is_sequential())
.unwrap_or(false)
}
/// True if the node is a wildcard parallel. O(1).
pub fn is_wildcard(&self, name: &str) -> bool {
self.color_of(name)
.map(|c| c.is_wildcard())
.unwrap_or(false)
}
/// True if the node is parallel (wildcard or named). O(1).
pub fn is_parallel(&self, name: &str) -> bool {
self.color_of(name)
.map(|c| c.is_parallel())
.unwrap_or(false)
}
/// Group name for a named parallel node, or `None`. O(1).
pub fn group_of(&self, name: &str) -> Option<String> {
let idx = self.name_to_idx.get(name)?;
self.graph[*idx].color().group_name().map(|s| s.to_string())
}
/// In-degree of a node. O(1).
pub fn in_degree(&self, name: &str) -> Option<usize> {
let idx = self.name_to_idx.get(name)?;
self.in_degree.get(idx).copied()
}
/// Get the edge kind between two nodes. O(e).
pub fn edge_kind(&self, from: &str, to: &str) -> Option<EdgeKind> {
let from_idx = self.name_to_idx.get(from)?;
let to_idx = self.name_to_idx.get(to)?;
self.graph
.find_edge(*from_idx, *to_idx)
.map(|e| self.graph[e])
}
/// Reference to the node data. O(1).
pub fn node(&self, name: &str) -> Option<&N> {
let idx = self.name_to_idx.get(name)?;
Some(&self.graph[*idx])
}
/// True if the node is fallible.
pub fn is_fallible(&self, name: &str) -> bool {
self.node(name).map(|n| n.fallible()).unwrap_or(false)
}
/// Number of ready nodes (in-degree 0, not yet popped).
pub fn ready_count(&self) -> usize {
self.ready_set.len()
}
// ═══════════════════════════════════════════════════════════════════
// Ready set queries
// ═══════════════════════════════════════════════════════════════════
/// All ready node names, in deterministic BTreeSet order.
pub fn ready_names(&self) -> Vec<&str> {
self.ready_set
.iter()
.map(|(name, _)| name.as_str())
.collect()
}
/// Ready sequential node names.
pub fn ready_sequential(&self) -> Vec<&str> {
self.ready_set
.iter()
.filter(|(name, _)| self.is_sequential(name))
.map(|(name, _)| name.as_str())
.collect()
}
/// Ready wildcard node names.
pub fn ready_wildcards(&self) -> Vec<&str> {
self.ready_set
.iter()
.filter(|(name, _)| self.is_wildcard(name))
.map(|(name, _)| name.as_str())
.collect()
}
/// Ready nodes grouped by their parallel group name.
///
/// Returns `(group_name, ready_count, total_count)` for each group that
/// has at least one ready member.
pub fn ready_groups(&self) -> Vec<(&str, usize, usize)> {
let mut result = Vec::new();
for (gname, meta) in &self.groups {
let ready_count = meta
.members
.iter()
.filter(|idx| matches!(self.state.get(idx), Some(NodeState::Ready)))
.count();
if ready_count > 0 {
result.push((gname.as_str(), ready_count, meta.total));
}
}
result
}
/// Pick a fireable batch per bake scheduling rules:
///
/// 1. Named groups: flood-fill from ready members through same-color
/// implicit edges. If the whole group is reachable, fire it.
/// 2. Wildcards join any fireable group.
/// 3. Otherwise, wildcards form their own batch.
/// 4. Returns `None` only when truly stuck.
pub fn pick_batch(&self) -> Option<Vec<String>> {
use std::collections::{HashSet, VecDeque};
let wildcards: Vec<String> = self
.ready_set
.iter()
.filter(|(name, _)| self.is_wildcard(name))
.map(|(name, _)| name.clone())
.collect();
for (_gname, meta) in &self.groups {
if meta.total == 0 {
continue;
}
// Flood-fill from ready members through same-color implicit edges
let mut reached: HashSet<NodeIndex> = HashSet::new();
let mut queue: VecDeque<NodeIndex> = VecDeque::new();
for idx in &meta.members {
if matches!(self.state.get(idx), Some(NodeState::Ready)) {
reached.insert(*idx);
queue.push_back(*idx);
}
}
while let Some(from) = queue.pop_front() {
let from_color = self.graph[from].color();
if from_color.is_sequential() {
continue;
}
for edge in self.graph.edges_directed(from, Direction::Outgoing) {
if !edge.weight().is_implicit() {
continue;
}
let to = edge.target();
let to_color = self.graph[to].color();
if to_color.is_sequential() {
continue;
}
let same = match (&from_color, &to_color) {
(NodeColor::Wildcard, NodeColor::Wildcard) => true,
(NodeColor::Named(a), NodeColor::Named(b)) => a.as_str() == b.as_str(),
_ => false,
};
if !same {
continue;
}
if !meta.members.contains(&to) {
continue;
}
if reached.insert(to) {
queue.push_back(to);
}
}
}
if reached.len() == meta.total {
let batch: Vec<String> = meta
.members
.iter()
.map(|idx| self.graph[*idx].name().to_string())
.collect();
let mut batch = batch;
batch.extend(wildcards);
return Some(batch);
}
}
// No fireable named group — wildcards fire alone
if !wildcards.is_empty() {
return Some(wildcards);
}
None
}
// ═══════════════════════════════════════════════════════════════════
// Scheduling operations
// ═══════════════════════════════════════════════════════════════════
/// Pop a specific node by name.
///
/// Decrements in-degree of all successors. Those reaching 0 become ready.
/// This is the key fix over `topological-sort` — we pop by name, not by
/// arbitrary HashMap iteration order.
pub fn pop(&mut self, name: &str) -> Result<(), DagError> {
let idx = self
.name_to_idx
.get(name)
.copied()
.ok_or_else(|| DagError::NodeNotFound(name.to_string()))?;
match self.state.get(&idx) {
None => return Err(DagError::NodeNotFound(name.to_string())),
Some(NodeState::Done) => return Err(DagError::AlreadyPopped(name.to_string())),
Some(NodeState::Pending) => return Err(DagError::NotReady(name.to_string())),
Some(NodeState::Ready) => {}
}
// Remove from ready set
let owned_name = name.to_string();
self.ready_set.remove(&(owned_name, idx));
// Decrement in-degree of all successors
let successors: Vec<NodeIndex> = self
.graph
.neighbors_directed(idx, Direction::Outgoing)
.collect();
for succ in successors {
let deg = self.in_degree.get_mut(&succ).unwrap();
*deg -= 1;
if *deg == 0 {
// Successor becomes ready
let succ_name = self.graph[succ].name().to_string();
self.ready_set.insert((succ_name, succ));
*self.state.get_mut(&succ).unwrap() = NodeState::Ready;
}
}
// Mark as done
*self.state.get_mut(&idx).unwrap() = NodeState::Done;
Ok(())
}
/// Pop a sequential node from the ready set.
///
/// Returns the name of the popped node, or `None` if no sequential
/// node is ready. Convenience wrapper: finds first ready sequential,
/// pops it by name.
pub fn pop_sequential(&mut self) -> Result<Option<String>, DagError> {
let name = self
.ready_set
.iter()
.find(|(name, _)| self.is_sequential(name))
.map(|(name, _)| name.clone());
if let Some(ref name) = name {
self.pop(name)?;
}
Ok(name)
}
// ═══════════════════════════════════════════════════════════════════
// Deadlock / cycle detection
// ═══════════════════════════════════════════════════════════════════
/// True if ready nodes exist but no batch can fire (potential deadlock).
pub fn is_stalled(&self) -> bool {
!self.ready_set.is_empty() && self.pick_batch().is_none()
}
/// Names of groups that are causing a stall (incomplete groups,
/// including those with zero ready members but non-Done pending members).
pub fn stalled_groups(&self) -> Vec<&str> {
let mut result = Vec::new();
for (gname, meta) in &self.groups {
let ready_count = meta
.members
.iter()
.filter(|idx| matches!(self.state.get(idx), Some(NodeState::Ready)))
.count();
let pending_count = meta
.members
.iter()
.filter(|idx| matches!(self.state.get(idx), Some(NodeState::Pending)))
.count();
if meta.total > 0 && (ready_count > 0 || pending_count > 0) && ready_count < meta.total {
result.push(gname.as_str());
}
}
result
}
/// All nodes that haven't been popped yet (for cycle diagnostics).
pub fn remaining(&self) -> Vec<&str> {
self.state
.iter()
.filter(|(_, s)| !matches!(s, NodeState::Done))
.map(|(idx, _)| self.graph[*idx].name())
.collect()
}
}
impl<N: DagNode> Default for Dag<N> {
fn default() -> Self {
Self::new()
}
}
// ── Tests ──────────────────────────────────────────────────────────────────
#[cfg(test)]
mod tests {
use super::*;
#[derive(Clone)]
struct TestNode {
name: String,
color: NodeColor,
fallible: bool,
}
impl DagNode for TestNode {
fn name(&self) -> &str {
&self.name
}
fn color(&self) -> NodeColor {
self.color.clone()
}
fn fallible(&self) -> bool {
self.fallible
}
}
fn seq(name: &str) -> TestNode {
TestNode {
name: name.into(),
color: NodeColor::Sequential,
fallible: false,
}
}
fn wild(name: &str) -> TestNode {
TestNode {
name: name.into(),
color: NodeColor::Wildcard,
fallible: false,
}
}
fn group(name: &str, g: &str) -> TestNode {
TestNode {
name: name.into(),
color: NodeColor::Named(g.into()),
fallible: false,
}
}
// ── Basic construction ────────────────────────────────────────────
#[test]
fn empty_dag() {
let dag = Dag::<TestNode>::new();
assert!(dag.is_empty());
assert_eq!(dag.node_count(), 0);
}
#[test]
fn add_single_node() {
let mut dag = Dag::new();
dag.add_node(seq("a")).unwrap();
dag.freeze();
assert!(!dag.is_empty());
assert_eq!(dag.ready_count(), 1);
assert_eq!(dag.ready_names(), vec!["a"]);
assert!(dag.is_sequential("a"));
}
#[test]
fn duplicate_node_rejected() {
let mut dag = Dag::new();
dag.add_node(seq("a")).unwrap();
assert!(dag.add_node(seq("a")).is_err());
}
// ── Edges & in-degree ─────────────────────────────────────────────
#[test]
fn simple_dependency() {
let mut dag = Dag::new();
dag.add_node(seq("a")).unwrap();
dag.add_node(seq("b")).unwrap();
dag.add_edge("a", "b", EdgeKind::Explicit).unwrap();
dag.freeze();
assert!(dag.is_sequential("a"));
assert_eq!(dag.ready_names(), vec!["a"]);
assert_eq!(dag.in_degree("a"), Some(0));
assert_eq!(dag.in_degree("b"), Some(1));
}
#[test]
fn pop_advances_dependents() {
let mut dag = Dag::new();
dag.add_node(seq("a")).unwrap();
dag.add_node(seq("b")).unwrap();
dag.add_node(seq("c")).unwrap();
dag.add_edge("a", "b", EdgeKind::Explicit).unwrap();
dag.add_edge("b", "c", EdgeKind::Explicit).unwrap();
dag.freeze();
assert_eq!(dag.ready_names(), vec!["a"]);
dag.pop("a").unwrap();
assert_eq!(dag.ready_names(), vec!["b"]);
dag.pop("b").unwrap();
assert_eq!(dag.ready_names(), vec!["c"]);
dag.pop("c").unwrap();
assert!(dag.is_empty());
}
#[test]
fn pop_not_ready_rejected() {
let mut dag = Dag::new();
dag.add_node(seq("a")).unwrap();
dag.add_node(seq("b")).unwrap();
dag.add_edge("a", "b", EdgeKind::Explicit).unwrap();
dag.freeze();
assert!(dag.pop("b").is_err()); // b not ready
}
#[test]
fn pop_twice_rejected() {
let mut dag = Dag::new();
dag.add_node(seq("a")).unwrap();
dag.freeze();
dag.pop("a").unwrap();
assert!(dag.pop("a").is_err()); // already done
}
// ── Node coloring ─────────────────────────────────────────────────
#[test]
fn color_queries() {
let mut dag = Dag::new();
dag.add_node(seq("s")).unwrap();
dag.add_node(wild("w")).unwrap();
dag.add_node(group("g", "build")).unwrap();
dag.freeze();
assert!(dag.is_sequential("s"));
assert!(dag.is_wildcard("w"));
assert_eq!(dag.group_of("g"), Some("build".to_string()));
assert!(dag.is_parallel("w"));
assert!(dag.is_parallel("g"));
assert!(!dag.is_parallel("s"));
}
#[test]
fn fallible_query() {
let mut dag = Dag::new();
dag.add_node(TestNode {
name: "f".into(),
color: NodeColor::Sequential,
fallible: true,
})
.unwrap();
dag.freeze();
assert!(dag.is_fallible("f"));
}
// ── Ready set queries ─────────────────────────────────────────────
#[test]
fn ready_filtering() {
let mut dag = Dag::new();
dag.add_node(seq("s1")).unwrap();
dag.add_node(seq("s2")).unwrap();
dag.add_node(wild("w1")).unwrap();
dag.add_node(wild("w2")).unwrap();
dag.add_node(group("g1", "build")).unwrap();
dag.add_node(group("g2", "build")).unwrap();
dag.freeze();
assert_eq!(dag.ready_sequential(), vec!["s1", "s2"]);
assert_eq!(dag.ready_wildcards(), vec!["w1", "w2"]);
let groups = dag.ready_groups();
assert_eq!(groups.len(), 1);
assert_eq!(groups[0], ("build", 2, 2));
}
// ── Batch picking ─────────────────────────────────────────────────
#[test]
fn pick_batch_complete_group_with_wildcards() {
let mut dag = Dag::new();
dag.add_node(seq("s")).unwrap(); // not ready (has dep)
dag.add_node(group("g1", "build")).unwrap();
dag.add_node(group("g2", "build")).unwrap();
dag.add_node(wild("w")).unwrap();
dag.add_edge("s", "g1", EdgeKind::Explicit).unwrap(); // block g1
dag.freeze();
// g1 blocked by s, build group incomplete
// BUT wildcard w is ready → fires as its own batch
let batch = dag.pick_batch().unwrap();
assert_eq!(batch.len(), 1);
assert!(batch.contains(&"w".to_string()));
dag.pop("w").unwrap();
dag.pop("s").unwrap();
// Now g1, g2 both ready; build group complete
let batch = dag.pick_batch().unwrap();
assert!(batch.contains(&"g1".to_string()));
assert!(batch.contains(&"g2".to_string()));
}
#[test]
fn pick_batch_wildcards_only_no_groups() {
let mut dag = Dag::new();
dag.add_node(wild("w1")).unwrap();
dag.add_node(wild("w2")).unwrap();
dag.freeze();
let batch = dag.pick_batch().unwrap();
assert_eq!(batch.len(), 2);
assert!(batch.contains(&"w1".to_string()));
assert!(batch.contains(&"w2".to_string()));
}
#[test]
fn pick_batch_none_when_sequential_only() {
let mut dag = Dag::new();
dag.add_node(seq("s1")).unwrap();
dag.add_node(seq("s2")).unwrap();
dag.freeze();
// pick_batch only picks parallel batches; sequential nodes
// should be picked individually via pop_sequential()
assert!(dag.pick_batch().is_none());
}
// ── pop_sequential convenience ────────────────────────────────────
#[test]
fn pop_sequential_picks_first_seq() {
let mut dag = Dag::new();
dag.add_node(wild("w")).unwrap();
dag.add_node(seq("s1")).unwrap();
dag.add_node(seq("s2")).unwrap();
dag.freeze();
let popped = dag.pop_sequential().unwrap().unwrap();
assert_eq!(popped, "s1"); // BTreeSet order: s1 before s2
assert!(!dag.ready_names().contains(&"s1"));
assert!(dag.ready_names().contains(&"s2"));
}
// ── Deadlock detection ────────────────────────────────────────────
#[test]
fn stall_when_group_incomplete() {
let mut dag = Dag::new();
dag.add_node(seq("s")).unwrap();
dag.add_node(group("g1", "build")).unwrap();
dag.add_node(group("g2", "build")).unwrap();
dag.add_edge("s", "g2", EdgeKind::Explicit).unwrap();
dag.freeze();
// g1 ready, g2 not (depends on s). pick_batch → none.
assert!(dag.is_stalled());
assert_eq!(dag.stalled_groups(), vec!["build"]);
}
#[test]
fn remaining_after_partial_pop() {
let mut dag = Dag::new();
dag.add_node(seq("a")).unwrap();
dag.add_node(seq("b")).unwrap();
dag.add_node(seq("c")).unwrap();
dag.freeze();
dag.pop("a").unwrap();
let rem = dag.remaining();
assert_eq!(rem.len(), 2);
assert!(rem.contains(&"b"));
assert!(rem.contains(&"c"));
}
}
+20
View File
@@ -0,0 +1,20 @@
// Code in this module PARTIALLY or FULLY utilized AI Coding Agent
/// Edge kind — distinguishes explicit `@after` from implicit pipeline chaining.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
pub enum EdgeKind {
/// Explicit dependency: `@after(func)` in the script.
Explicit,
/// Implicit dependency: consecutive `@pipeline` functions without `@after`.
Implicit,
}
impl EdgeKind {
pub fn is_explicit(&self) -> bool {
matches!(self, EdgeKind::Explicit)
}
pub fn is_implicit(&self) -> bool {
matches!(self, EdgeKind::Implicit)
}
}
+9
View File
@@ -0,0 +1,9 @@
// Code in this module PARTIALLY or FULLY utilized AI Coding Agent
pub mod dag;
pub mod edge;
pub mod node;
pub use dag::{Dag, DagError};
pub use edge::EdgeKind;
pub use node::{DagNode, NodeColor};
+60
View File
@@ -0,0 +1,60 @@
// Code in this module PARTIALLY or FULLY utilized AI Coding Agent
/// Node coloring — mirrors the bake DAG generator's color semantics.
///
/// | Color | Probability | Bake semantics |
/// |-------------|------------|-----------------------------|
/// | Sequential | p_seq | Runs alone, one at a time |
/// | Wildcard | p_wild | Joins any complete group |
/// | Named(G) | p_named/K | Fires when group complete |
#[derive(Debug, Clone, PartialEq, Eq, Hash)]
pub enum NodeColor {
/// Sequential node — executes alone.
Sequential,
/// Wildcard parallel — compatible with any complete named group.
Wildcard,
/// Named parallel group member — fires when all members of this group are ready.
Named(String),
}
impl NodeColor {
/// Returns true for `Sequential`.
pub fn is_sequential(&self) -> bool {
matches!(self, NodeColor::Sequential)
}
/// Returns true for `Wildcard`.
pub fn is_wildcard(&self) -> bool {
matches!(self, NodeColor::Wildcard)
}
/// Returns the group name if `Named`, or `None`.
pub fn group_name(&self) -> Option<&str> {
match self {
NodeColor::Named(g) => Some(g.as_str()),
_ => None,
}
}
/// Returns true if this is a parallel node (Wildcard or Named).
pub fn is_parallel(&self) -> bool {
!self.is_sequential()
}
}
/// Trait that nodes must implement to be managed by [`Dag`](super::dag::Dag).
///
/// Only `name()` and `color()` are required. Override `fallible()` if the
/// node can fail without halting the pipeline.
pub trait DagNode {
/// Unique name of this node within the DAG.
fn name(&self) -> &str;
/// Scheduling color — determines sequential/parallel/group semantics.
fn color(&self) -> NodeColor;
/// Whether failure of this node is non-fatal to the pipeline.
fn fallible(&self) -> bool {
false
}
}