From 3e431cf295c764c389e11e34317cc2999abaa7b3 Mon Sep 17 00:00:00 2001 From: Dave Nagoda Date: Tue, 30 Jun 2026 15:57:01 -0700 Subject: [PATCH] Refine batch failure handling Keep --batch-continue-on-error (opt-in; batch stops on the first error by default) but fix the success/failure accounting. Previously every run that returned Ok was counted as a success, even when the function itself returned success: false. Now batch mode tracks processed/successful/failed counts from FunctionRunResult.success, and a function-level failure is treated like an error: it stops the batch by default and is skipped (still recorded) under --batch-continue-on-error. The stderr summary reports processed, successful, and failed counts. Add integration coverage for default stop-on-failure behavior and for --batch-continue-on-error processing every row with accurate counts. Also switch integration tests to assert_cmd::cargo::cargo_bin! instead of the deprecated Command::cargo_bin helper. Verified with: - cargo test batch_ - cargo test Assisted-By: devx/c659e918-9568-4750-b122-e3890447348a --- src/main.rs | 34 +++++++++++++----- tests/integration_tests.rs | 70 ++++++++++++++++++++++++++++++++++++++ 2 files changed, 95 insertions(+), 9 deletions(-) diff --git a/src/main.rs b/src/main.rs index 453254c1..cd8dea94 100644 --- a/src/main.rs +++ b/src/main.rs @@ -227,7 +227,7 @@ fn run_batch_mode( )); }; - // Pre-calculate scale factor (constant across all inputs in batch) + // Load schema/query once; scale factor is computed per input. let schema_string = opts.read_schema_to_string().transpose()?; let query_string = opts.read_query_to_string().transpose()?; @@ -235,8 +235,9 @@ fn run_batch_mode( let profile_opts = None; let mut line_num = 0; + let mut processed_count = 0; let mut success_count = 0; - let mut error_count = 0; + let mut failed_count = 0; for line_result in input_reader.lines() { line_num += 1; @@ -244,7 +245,7 @@ fn run_batch_mode( let line = match line_result { Ok(l) => l, Err(e) => { - error_count += 1; + failed_count += 1; if opts.batch_continue_on_error { eprintln!("Error reading line {}: {}", line_num, e); println!( @@ -263,11 +264,13 @@ fn run_batch_mode( continue; } + processed_count += 1; + // Parse input let input = match BytesContainer::new(BytesContainerType::Input, codec, line.into_bytes()) { Ok(i) => i, Err(e) => { - error_count += 1; + failed_count += 1; if opts.batch_continue_on_error { eprintln!("Error parsing line {}: {}", line_num, e); println!(r#"{{"success":false,"error":"Invalid JSON input: {}"}}"#, e); @@ -292,7 +295,7 @@ fn run_batch_mode( ) { Ok(sf) => sf, Err(e) => { - error_count += 1; + failed_count += 1; if opts.batch_continue_on_error { eprintln!("Error analyzing schema for line {}: {}", line_num, e); println!( @@ -323,14 +326,27 @@ fn run_batch_mode( // Output result immediately (streaming JSONL - compact format for line-by-line parsing) match result { Ok(function_result) => { - success_count += 1; + let function_succeeded = function_result.success; + if function_succeeded { + success_count += 1; + } else { + failed_count += 1; + } + // Use compact JSON (not pretty-printed) for JSONL format let compact_json = serde_json::to_string(&function_result) .unwrap_or_else(|error| error.to_string()); println!("{}", compact_json); + + if !function_succeeded && !opts.batch_continue_on_error { + anyhow::bail!( + "Function execution failed on line {}. Review the logs for more information.", + line_num + ); + } } Err(e) => { - error_count += 1; + failed_count += 1; if opts.batch_continue_on_error { eprintln!("Error executing line {}: {}", line_num, e); println!(r#"{{"success":false,"error":"Execution failed: {}"}}"#, e); @@ -343,8 +359,8 @@ fn run_batch_mode( // Log summary to stderr (so it doesn't interfere with JSONL output on stdout) eprintln!( - "Batch complete: {} inputs processed, {} successful, {} errors", - line_num, success_count, error_count + "Batch complete: {} inputs processed, {} successful, {} failed", + processed_count, success_count, failed_count ); Ok(()) diff --git a/tests/integration_tests.rs b/tests/integration_tests.rs index 5baa864a..bf18c63b 100644 --- a/tests/integration_tests.rs +++ b/tests/integration_tests.rs @@ -64,6 +64,69 @@ mod tests { Ok(()) } + #[test] + fn batch_continue_on_error_processes_all_and_counts_function_failures() -> Result<()> { + let mut cmd = Command::new(cargo_bin!()); + let input_file = temp_batch_input("{\"code\":0}\n{\"code\":1}\n{\"code\":0}\n")?; + + cmd.args(["--function", "tests/fixtures/build/exit_code.wasm"]) + .arg("--batch") + .arg("--batch-continue-on-error") + .arg("--input") + .arg(input_file.as_os_str()); + + let output = cmd.output()?; + + assert!(output.status.success()); + + let stdout = String::from_utf8(output.stdout)?; + let results = stdout + .lines() + .map(serde_json::from_str::) + .collect::, _>>()?; + assert_eq!(results.len(), 3); + assert_eq!(results[0]["success"], true); + assert_eq!(results[1]["success"], false); + assert_eq!(results[2]["success"], true); + + assert_eq!( + String::from_utf8(output.stderr)?, + "Batch complete: 3 inputs processed, 2 successful, 1 failed\n" + ); + + Ok(()) + } + + #[test] + fn batch_stops_on_function_failure_by_default() -> Result<()> { + let mut cmd = Command::new(cargo_bin!()); + let input_file = temp_batch_input("{\"code\":0}\n{\"code\":1}\n{\"code\":0}\n")?; + + cmd.args(["--function", "tests/fixtures/build/exit_code.wasm"]) + .arg("--batch") + .arg("--input") + .arg(input_file.as_os_str()); + + let output = cmd.output()?; + + assert!(!output.status.success()); + + let stdout = String::from_utf8(output.stdout)?; + let results = stdout + .lines() + .map(serde_json::from_str::) + .collect::, _>>()?; + assert_eq!(results.len(), 2); + assert_eq!(results[0]["success"], true); + assert_eq!(results[1]["success"], false); + + assert!(String::from_utf8(output.stderr)?.contains( + "Function execution failed on line 2. Review the logs for more information." + )); + + Ok(()) + } + #[test] fn run_no_opts() -> Result<()> { let mut cmd = Command::new(cargo_bin!()); @@ -265,6 +328,13 @@ mod tests { Ok(file) } + fn temp_batch_input(jsonl: &str) -> Result { + let file = assert_fs::NamedTempFile::new("input.jsonl")?; + file.write_str(jsonl)?; + + Ok(file) + } + #[test] fn test_scale_limits_analyzer_use_defaults_when_query_and_schema_not_provided() -> Result<()> { let mut cmd = Command::new(cargo_bin!());