diff --git a/crates/oxo-flow-cli/src/commands/run.rs b/crates/oxo-flow-cli/src/commands/run.rs index 2202c3f..08706c4 100644 --- a/crates/oxo-flow-cli/src/commands/run.rs +++ b/crates/oxo-flow-cli/src/commands/run.rs @@ -157,6 +157,32 @@ pub async fn run_command( interpreter_map: config.workflow.interpreter_map.clone(), }; + // Fail fast if any rule's declared request can never fit an explicit + // --max-memory / --max-threads cap. Otherwise the run would execute earlier + // rules and only discover the impossible one mid-pipeline. + let scheduled: Vec<&oxo_flow_core::rule::Rule> = order + .iter() + .filter_map(|name| config.get_rule(name)) + .collect(); + let breaches = oxo_flow_core::scheduler::check_budget_feasibility( + &scheduled, + exec_config.max_threads, + exec_config.max_memory_mb, + ); + if !breaches.is_empty() { + progress.finish_and_clear(); + let detail = breaches + .iter() + .map(|b| format!(" - {b}")) + .collect::>() + .join("\n"); + return Err(anyhow::anyhow!( + "resource budget too small for {} rule(s); no rules were run:\n{}", + breaches.len(), + detail + )); + } + let executor = LocalExecutor::new(exec_config); let mut success_count = 0; let mut fail_count = 0; diff --git a/crates/oxo-flow-core/src/scheduler.rs b/crates/oxo-flow-core/src/scheduler.rs index c28a83d..e86059d 100644 --- a/crates/oxo-flow-core/src/scheduler.rs +++ b/crates/oxo-flow-core/src/scheduler.rs @@ -342,6 +342,43 @@ pub fn parse_memory_mb(memory: &str) -> Option { Some(mb_int) } +/// Pre-flight check: report rules whose declared request can never fit an +/// *explicitly configured* budget (`--max-threads` / `--max-memory`), so the +/// run can fail fast instead of dying mid-pipeline on an impossible rule. +/// A cap of `None` means "auto-detect" and is not treated as a hard limit. +/// Returns one message per breaching rule (empty if feasible). +pub fn check_budget_feasibility( + rules: &[&Rule], + max_threads: Option, + max_memory_mb: Option, +) -> Vec { + let mut breaches = Vec::new(); + for rule in rules { + if let Some(cap) = max_memory_mb { + let required = rule + .effective_memory() + .and_then(parse_memory_mb) + .unwrap_or(0); + if required > cap { + breaches.push(format!( + "rule '{}' requires {}MB memory but --max-memory caps the run at {}MB", + rule.name, required, cap + )); + } + } + if let Some(cap) = max_threads { + let required = rule.effective_threads(); + if required > cap { + breaches.push(format!( + "rule '{}' requires {} threads but --max-threads caps the run at {}", + rule.name, required, cap + )); + } + } + } + breaches +} + /// Validate that a rule's resource requirements don't exceed system capacity. /// Returns a list of warning messages (empty if all requirements are within limits). pub fn validate_resources_against_system( @@ -876,6 +913,57 @@ mod tests { assert!(warnings.is_empty()); } + #[test] + fn budget_breach_on_explicit_memory_cap() { + let rule = Rule { + name: "hungry".to_string(), + memory: Some("8G".to_string()), + ..Default::default() + }; + let breaches = check_budget_feasibility(&[&rule], None, Some(100)); + assert_eq!(breaches.len(), 1); + assert!(breaches[0].contains("hungry")); + assert!(breaches[0].contains("--max-memory")); + } + + #[test] + fn budget_breach_on_explicit_threads_cap() { + let rule = Rule { + name: "wide".to_string(), + threads: Some(16), + ..Default::default() + }; + let breaches = check_budget_feasibility(&[&rule], Some(4), None); + assert_eq!(breaches.len(), 1); + assert!(breaches[0].contains("wide")); + assert!(breaches[0].contains("--max-threads")); + } + + #[test] + fn budget_no_breach_within_caps() { + let rule = Rule { + name: "fits".to_string(), + threads: Some(2), + memory: Some("2G".to_string()), + ..Default::default() + }; + let breaches = check_budget_feasibility(&[&rule], Some(4), Some(4096)); + assert!(breaches.is_empty()); + } + + #[test] + fn budget_none_cap_is_not_a_hard_limit() { + // Auto-detect (None) must not be treated as a breach even for a huge request. + let rule = Rule { + name: "huge".to_string(), + threads: Some(1024), + memory: Some("1T".to_string()), + ..Default::default() + }; + let breaches = check_budget_feasibility(&[&rule], None, None); + assert!(breaches.is_empty()); + } + #[test] fn check_available_disk_mb_returns_value() { let dir = tempfile::tempdir().unwrap(); diff --git a/tests/cli_integration.rs b/tests/cli_integration.rs index 0a24f3b..35fdd28 100644 --- a/tests/cli_integration.rs +++ b/tests/cli_integration.rs @@ -493,6 +493,56 @@ shell = "cat mid.txt > output.txt" assert!(stderr.contains("Debugging 1 rules")); } +#[test] +fn cli_run_fails_fast_when_rule_exceeds_max_memory() { + // A rule declaring more memory than the explicit --max-memory cap can never + // be scheduled. The run must fail up front with a clear message and must NOT + // execute the earlier, feasible rule (no wasted work). + let dir = tempfile::tempdir().unwrap(); + let workflow = dir.path().join("test.oxoflow"); + fs::write( + &workflow, + r#" +[workflow] +name = "budget-test" +version = "1.0.0" + +[[rules]] +name = "cheap_first" +output = ["a.txt"] +shell = "echo did-real-work > a.txt" + +[[rules]] +name = "hungry_second" +input = ["a.txt"] +output = ["b.txt"] +shell = "echo hello > b.txt" +memory = "8G" +"#, + ) + .unwrap(); + + let output = oxo_flow_cmd() + .current_dir(dir.path()) + .args(["run", workflow.to_str().unwrap(), "--max-memory", "100"]) + .output() + .expect("failed to run command"); + + assert!( + !output.status.success(), + "run should fail when a rule exceeds the memory budget" + ); + let stderr = String::from_utf8_lossy(&output.stderr); + assert!( + stderr.contains("hungry_second") && stderr.contains("--max-memory"), + "error should name the breaching rule and the cap: {stderr}" + ); + assert!( + !dir.path().join("a.txt").exists(), + "no rules should have run; cheap_first must not have produced a.txt" + ); +} + // ─── Cluster CLI tests ────────────────────────────────────────────────────── #[test]