diff --git a/Cargo.lock b/Cargo.lock index af35e8a..40fcb3f 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -114,6 +114,7 @@ dependencies = [ "aura-engine", "aura-registry", "aura-research", + "rayon", "serde", "serde_json", ] diff --git a/crates/aura-campaign/Cargo.toml b/crates/aura-campaign/Cargo.toml index 61f8792..8614f85 100644 --- a/crates/aura-campaign/Cargo.toml +++ b/crates/aura-campaign/Cargo.toml @@ -23,3 +23,8 @@ aura-research = { path = "../aura-research" } # realization payloads derive serde (the registry per-case policy, INDEX.md). serde = { workspace = true, features = ["derive"] } serde_json = { workspace = true } +# rayon is admitted under the amended C16 per-case dependency policy (INDEX.md): +# the campaign cell loop fans C1-disjoint cells into the same process-global +# work-stealing pool the engine's run_indexed already uses — parallelism across +# sims, never within one. Direct versioned dep, mirroring aura-engine. +rayon = "1" diff --git a/crates/aura-campaign/src/exec.rs b/crates/aura-campaign/src/exec.rs index 9eaca25..01db865 100644 --- a/crates/aura-campaign/src/exec.rs +++ b/crates/aura-campaign/src/exec.rs @@ -8,6 +8,7 @@ use std::collections::BTreeMap; use std::num::NonZeroUsize; +use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::Mutex; use aura_analysis::{FamilySelection, SelectionMode}; @@ -23,6 +24,7 @@ use aura_registry::{ PlateauMode, Registry, StageBootstrap, StageRealization, StageSelection, }; use aura_research::{Axis, CampaignDoc, DocRef, ProcessDoc, SelectRule, StageBlock, WfMode}; +use rayon::prelude::*; use crate::{ member_metric, predicate_holds, preflight, CellSpec, ExecFault, MemberFault, MemberRunner, @@ -73,10 +75,18 @@ pub struct CampaignOutcome { } /// Execute a validated campaign document's process pipeline once per -/// (strategy, instrument, window) cell, in doc order, sequentially (C1: -/// parallelism lives inside the engine `sweep`, across members, never across -/// cells). `campaign_id` is the campaign document's 64-hex content id (its -/// first 8 chars prefix family names); `strategies` is index-aligned with +/// (strategy, instrument, window) cell. Cells are grouped by instrument +/// ordinal and walked in sequential chunks of `parallel_instruments` groups; +/// within a chunk, every cell runs concurrently via `rayon::par_iter` (#277: +/// parallelism now also spans cells, bounded by `parallel_instruments` — C1 +/// still holds because cells are disjoint work, and RESULT order is rebuilt +/// in document order regardless of completion order). A non-containable +/// fault (see `contain`) latches a shared abort flag and short-circuits the +/// run deterministically: propagation picks the lowest document-order fault +/// among the cells that completed before the latch, which cell that is being +/// scheduling-dependent but acceptable for these external, non-input faults. +/// `campaign_id` is the campaign document's 64-hex content id (its first 8 +/// chars prefix family names); `strategies` is index-aligned with /// `campaign.strategies` as (blueprint content id, canonical blueprint json). /// Writes one family per family-producing stage plus one campaign-run record; /// a `MemberFault` aborts the whole run before any write of the faulted stage. @@ -117,6 +127,10 @@ pub fn execute( } let campaign_prefix = &campaign_id[..8]; + // A slot's per-cell outcome — the same type `run_cell` returns, named here + // so the chunk-walk bindings below don't nest it raw (clippy::type_complexity). + type CellResult = Result<(CellOutcome, CellRealization), ExecFault>; + // Flatten: the same 4-level nesting enumerates every cell with its // document-order index. Execution order below is instrument-major and // chunked; RESULT order stays this document order (C1). @@ -169,21 +183,47 @@ pub fn execute( for p in &planned { groups[p.instrument_ordinal].push(p.doc_index); } - let mut slots: Vec> = - (0..planned.len()).map(|_| None).collect(); + let abort = AtomicBool::new(false); + let mut slots: Vec> = (0..planned.len()).map(|_| None).collect(); for chunk in groups.chunks(parallel_instruments.get()) { - for &doc_index in chunk.iter().flatten() { - let p = &planned[doc_index]; - let pair = run_cell( - &p.cell, - process, - campaign_prefix, - p.window_ordinal, - campaign.seed, - runner, - registry, - )?; - slots[doc_index] = Some(pair); + let chunk_cells: Vec = chunk.iter().flatten().copied().collect(); + let results: Vec<(usize, Option)> = chunk_cells + .par_iter() + .map(|&doc_index| { + if abort.load(Ordering::Relaxed) { + // Never started: a fatal fault elsewhere already ends + // the run; this slot stays empty. + return (doc_index, None); + } + let p = &planned[doc_index]; + let result = run_cell( + &p.cell, + process, + campaign_prefix, + p.window_ordinal, + campaign.seed, + runner, + registry, + ); + if result.is_err() { + abort.store(true, Ordering::Relaxed); + } + (doc_index, Some(result)) + }) + .collect(); + for (doc_index, result) in results { + slots[doc_index] = result; + } + } + + // Non-containable faults stay run-fatal: propagate the lowest + // document-order fault among completed cells (deterministic attribution; + // which cells completed before the abort latched is scheduling-dependent, + // acceptable for external, non-input faults — see the spec). + for slot in slots.iter_mut() { + if matches!(slot, Some(Err(_))) { + let Some(Err(fault)) = slot.take() else { unreachable!() }; + return Err(fault); } } @@ -196,7 +236,12 @@ pub fn execute( type Nominee = Option<(Vec<(String, Scalar)>, RunReport)>; let mut nominees: BTreeMap<(usize, usize, usize), Vec<(String, Nominee)>> = BTreeMap::new(); for (p, slot) in planned.iter().zip(slots) { - let (outcome, realization) = slot.expect("the walk returned early on any fault"); + let (outcome, realization) = match slot { + Some(Ok(pair)) => pair, + // No fatal fault (checked above) means abort never latched, so + // every cell ran to Ok. + _ => unreachable!("no fatal fault, so every cell completed"), + }; nominees .entry((p.cell.strategy_ordinal, p.window_ordinal, p.cell.regime_ordinal)) .or_default() diff --git a/crates/aura-campaign/tests/execute.rs b/crates/aura-campaign/tests/execute.rs index 514c103..9831606 100644 --- a/crates/aura-campaign/tests/execute.rs +++ b/crates/aura-campaign/tests/execute.rs @@ -9,7 +9,7 @@ use aura_campaign::{ }; use aura_core::{Scalar, ScalarKind, Timestamp}; use aura_engine::{r_bootstrap, RMetrics, RunManifest, RunMetrics, RunReport, SelectionMode}; -use aura_registry::{generalization, FamilyKind, Registry, StageBootstrap}; +use aura_registry::{generalization, FamilyKind, Registry, RegistryError, StageBootstrap}; use aura_research::{ Axis, CampaignDoc, Cmp, DataSection, DocKind, DocRef, Predicate, Presentation, ProcessDoc, ProcessRef, SelectRule, StageBlock, StrategyEntry, SweepSelection, WfMode, Window, @@ -220,6 +220,22 @@ fn generalize_stage() -> StageBlock { StageBlock::Generalize { metric: "net_expectancy_r".to_string() } } +/// A registry whose family store path is pre-occupied by a directory: +/// `append_family`'s `OpenOptions::open` then fails with a real IO error +/// (`ErrorKind::IsADirectory`) — a `RegistryError::Io` that is neither a +/// `MemberFault` nor a `WindowFault`, so `run_cell`'s `?` on `append_family` +/// returns it as a non-containable `ExecFault` (never routed through +/// `contain`), unlike the `FakeRunner`-planted member faults used elsewhere +/// in this file. +fn temp_registry_with_blocked_family_store(name: &str) -> Registry { + let dir = std::path::Path::new(env!("CARGO_TARGET_TMPDIR")) + .join(format!("aura-campaign-exec-{name}")); + let _ = std::fs::remove_dir_all(&dir); + std::fs::create_dir_all(&dir).expect("create temp registry dir"); + std::fs::create_dir_all(dir.join("families.jsonl")).expect("occupy the family store path"); + Registry::open(dir.join("runs.jsonl")) +} + fn process(pipeline: Vec) -> ProcessDoc { ProcessDoc { format_version: 1, @@ -955,3 +971,209 @@ fn execute_generalize_keeps_regimes_separate() { assert!(g.generalization.is_some(), "two instruments -> graded"); } } + +/// C1 across the parallel cell loop: the same campaign under a 1-thread pool +/// with K=1 and an 8-thread pool with K=2 (fresh registry each) yields an +/// identical campaign-run record — the campaign-path sibling of the engine's +/// `*_is_deterministic_across_thread_counts` tests. +#[test] +fn campaign_outcome_is_deterministic_across_thread_counts_and_bounds() { + let doc = campaign(&["AAA", "BBB", "CCC"]); + let proc_doc = process(vec![sweep_stage(false)]); + let run_with = |threads: usize, k: usize, name: &str| { + let reg = temp_registry(name); + let pool = rayon::ThreadPoolBuilder::new() + .num_threads(threads) + .build() + .expect("build pool"); + pool.install(|| { + execute( + CAMPAIGN_ID, + &doc, + &proc_doc, + &strategies(), + &FakeRunner::clean(), + ®, + std::num::NonZeroUsize::new(k).expect("nonzero"), + ) + }) + .expect("campaign run") + }; + let serial = run_with(1, 1, "det_across_serial"); + let parallel = run_with(8, 2, "det_across_parallel"); + assert_eq!(serial.run, parallel.run); + assert_eq!(serial.record, parallel.record); +} + +/// #277: a non-containable fault (here, `append_family`'s registry-level IO +/// refusal, never routed through `contain` since it is neither a +/// `MemberFault` nor a `WindowFault`) still propagates as a global `Err` out +/// of the parallel abort-latch loop, and does so with IDENTICAL content +/// regardless of thread count or pool size. Every cell's sweep hits the SAME +/// blocked family-store path, so whichever completed cell's `Err` the +/// doc-order scan in `execute` happens to pick (scheduling-dependent per the +/// exec.rs doc comment) carries the same IO error kind — exercising the +/// abort-latch + lowest-doc-order-fault-scan path (and never the +/// `unreachable!` rebuild arm, since this run returns `Err` before reaching +/// the rebuild) that the fault-free determinism test above does not. +#[test] +fn execute_propagates_a_non_containable_fault_deterministically_under_parallel_chunks() { + let doc = campaign(&["AAA", "BBB", "CCC"]); + let proc_doc = process(vec![sweep_stage(false)]); + let run_with = |threads: usize, k: usize, name: &str| { + let reg = temp_registry_with_blocked_family_store(name); + let pool = rayon::ThreadPoolBuilder::new() + .num_threads(threads) + .build() + .expect("build pool"); + pool.install(|| { + execute( + CAMPAIGN_ID, + &doc, + &proc_doc, + &strategies(), + &FakeRunner::clean(), + ®, + std::num::NonZeroUsize::new(k).expect("nonzero"), + ) + }) + }; + let serial = run_with(1, 1, "fault_prop_serial") + .expect_err("a blocked family store is refused, not silently written"); + let parallel = run_with(8, 3, "fault_prop_parallel") + .expect_err("same refusal under real cross-cell parallelism"); + let kind = |e: &ExecFault| match e { + ExecFault::Registry(RegistryError::Io(io)) => io.kind(), + other => panic!("expected ExecFault::Registry(Io), got {other:?}"), + }; + assert_eq!(kind(&serial), std::io::ErrorKind::IsADirectory); + assert_eq!(kind(&serial), kind(¶llel), "same fault kind regardless of scheduling"); +} + +/// Wraps the clean fake and tracks how many DISTINCT instruments are inside +/// `run_member` at once (refcounted per instrument; high-water mark in +/// `max_seen`). The 5 ms hold makes overlap near-certain on an 8-thread pool +/// if the residency bound is absent. +struct CountingRunner { + inner: FakeRunner, + live: std::sync::Mutex>, + max_seen: std::sync::atomic::AtomicUsize, +} + +impl MemberRunner for CountingRunner { + fn run_member( + &self, + cell: &CellSpec, + params: &[(String, Scalar)], + window_ms: (i64, i64), + ) -> Result { + let distinct = { + let mut live = self.live.lock().expect("live instrument map"); + *live.entry(cell.instrument.clone()).or_insert(0) += 1; + live.len() + }; + self.max_seen.fetch_max(distinct, std::sync::atomic::Ordering::SeqCst); + std::thread::sleep(std::time::Duration::from_millis(5)); + let out = self.inner.run_member(cell, params, window_ms); + let mut live = self.live.lock().expect("live instrument map"); + let count = live.get_mut(&cell.instrument).expect("entered above"); + *count -= 1; + if *count == 0 { + live.remove(&cell.instrument); + } + out + } +} + +/// The structural residency bound: with K=1 on an 8-thread pool, no two +/// distinct instruments are ever inside `run_member` at the same time. +#[test] +fn parallel_cells_never_exceed_the_instrument_bound() { + let reg = temp_registry("instrument_bound"); + let doc = campaign(&["AAA", "BBB", "CCC"]); + let proc_doc = process(vec![sweep_stage(false)]); + let runner = CountingRunner { + inner: FakeRunner::clean(), + live: std::sync::Mutex::new(BTreeMap::new()), + max_seen: std::sync::atomic::AtomicUsize::new(0), + }; + let pool = rayon::ThreadPoolBuilder::new().num_threads(8).build().expect("build pool"); + pool.install(|| { + execute( + CAMPAIGN_ID, + &doc, + &proc_doc, + &strategies(), + &runner, + ®, + std::num::NonZeroUsize::new(1).expect("nonzero"), + ) + }) + .expect("campaign run"); + let max = runner.max_seen.load(std::sync::atomic::Ordering::SeqCst); + assert!(max <= 1, "K=1 must bound distinct live instruments to 1, saw {max}"); +} + +/// A registry-write fault is infrastructure, not a cell pathology: it stays +/// run-fatal in the parallel loop (no cell containment, no partial +/// campaign-run record). +#[test] +fn a_registry_fault_is_run_fatal_in_the_parallel_loop() { + let dir = std::path::Path::new(env!("CARGO_TARGET_TMPDIR")) + .join("aura-campaign-exec-registry_fatal"); + let _ = std::fs::remove_dir_all(&dir); + std::fs::create_dir_all(&dir).expect("create temp registry dir"); + // A directory squatting on the family store makes every append_family fail. + std::fs::create_dir(dir.join("families.jsonl")).expect("squat the family store"); + let reg = Registry::open(dir.join("runs.jsonl")); + let doc = campaign(&["AAA", "BBB"]); + let proc_doc = process(vec![sweep_stage(false)]); + let err = execute( + CAMPAIGN_ID, + &doc, + &proc_doc, + &strategies(), + &FakeRunner::clean(), + ®, + DEFAULT_PARALLEL_INSTRUMENTS, + ) + .expect_err("a dead family store is run-fatal"); + assert!(matches!(err, ExecFault::Registry(_))); + assert!( + !dir.join("campaign_runs.jsonl").exists(), + "no partial campaign-run record on the fatal path" + ); +} + +/// #272 under the parallel loop: member faults stay contained as failed +/// cells across concurrently-running cells — the run record persists with +/// every cell recorded, never a global abort. +#[test] +fn member_faults_stay_contained_in_the_parallel_loop() { + let reg = temp_registry("parallel_containment"); + let doc = campaign(&["AAA", "BBB", "CCC"]); + let proc_doc = process(vec![sweep_stage(false)]); + let point = |fast: i64, slow: i64| { + vec![("fast".to_string(), Scalar::i64(fast)), ("slow".to_string(), Scalar::i64(slow))] + }; + let runner = FakeRunner { + faults: vec![(point(2, 9), MemberFault::Run("boom".to_string()))], + }; + let pool = rayon::ThreadPoolBuilder::new().num_threads(8).build().expect("build pool"); + let out = pool + .install(|| { + execute( + CAMPAIGN_ID, + &doc, + &proc_doc, + &strategies(), + &runner, + ®, + std::num::NonZeroUsize::new(2).expect("nonzero"), + ) + }) + .expect("member faults are contained, not a global abort"); + assert_eq!(out.record.cells.len(), 3, "every cell is recorded"); + assert!(out.record.cells.iter().all(|c| c.fault.is_some())); + assert_eq!(reg.load_campaign_runs().expect("load runs").len(), 1); +} diff --git a/crates/aura-cli/tests/research_docs.rs b/crates/aura-cli/tests/research_docs.rs index 48fde37..05d5284 100644 --- a/crates/aura-cli/tests/research_docs.rs +++ b/crates/aura-cli/tests/research_docs.rs @@ -1727,6 +1727,60 @@ fn campaign_run_synthetic_e2e_cell_order_is_document_order_under_parallel_instru } } +/// #277 CLI-level containment under REAL cross-cell parallelism: with +/// `--parallel-instruments 2` both instruments land in the same chunk and run +/// genuinely concurrently on the ambient rayon pool (not the crate-level +/// `execute()` tests' `FakeRunner`, which never touches the real engine or +/// the CLI's exit-code/record-presentation layer). The fixture picks a window +/// that lies inside SYMA's synthetic span (2024-01..08) but strictly before +/// SYMB's (2024-03..06 — SYMB has no data in January), so SYMA's cell +/// succeeds while SYMB's cell hits a real per-cell data fault. A regression +/// that let the parallel chunk walk leak a fault out of `contain` (aborting +/// the whole run) or drop/misplace a completed cell would show up here as the +/// wrong exit code, a missing cell, or a fault on the wrong instrument. +/// Expected: exit 3 ("completed with failed cells"), both cells present in +/// document order, SYMA's carries no fault, SYMB's does. +#[test] +fn campaign_run_synthetic_e2e_parallel_instruments_contains_a_real_per_cell_fault() { + let (dir, _fixture) = fresh_project_with_data(); + let runs_dir = dir.join("runs"); + std::fs::remove_dir_all(&runs_dir).ok(); + let _cleanup = ScratchGuard(vec![ + ScratchPath::Dir(runs_dir.clone()), + ScratchPath::File(dir.join("parfault.process.json")), + ScratchPath::File(dir.join("parfault.campaign.json")), + ]); + let bp_id = seed_blueprint(&dir, "campaign-run-parfault-seed"); + let proc_id = register_process_doc(&dir, "parfault.process.json", SWEEP_ONLY_PROCESS_DOC); + // 2024-01-10 .. 2024-01-20: inside SYMA's Jan-Aug span, strictly before + // SYMB's Mar-Jun span. + let doc = campaign_doc_json_two( + ("SYMA", "SYMB"), + "", + &bp_id, + &proc_id, + (1704844800000, 1705708800000), + ); + write_doc(&dir, "parfault.campaign.json", &doc); + + let (out, code) = run_code_in( + &dir, + &["campaign", "run", "parfault.campaign.json", "--parallel-instruments", "2"], + ); + assert_eq!(code, Some(3), "one contained fault, one success: {out}"); + let line = out + .lines() + .find(|l| l.starts_with("{\"campaign_run\":")) + .expect("the always-on final campaign_run line"); + let v: serde_json::Value = serde_json::from_str(line).expect("record parses"); + let cells = v["campaign_run"]["cells"].as_array().expect("cells array"); + assert_eq!(cells.len(), 2, "one cell per instrument: {line}"); + assert_eq!(cells[0]["instrument"].as_str(), Some("SYMA"), "doc order preserved: {line}"); + assert_eq!(cells[1]["instrument"].as_str(), Some("SYMB"), "doc order preserved: {line}"); + assert!(cells[0]["fault"].is_null(), "SYMA has data in January, its cell succeeds: {line}"); + assert!(!cells[1]["fault"].is_null(), "SYMB has no January data, its cell is faulted: {line}"); +} + /// The v2 boundary, campaign side: a generalize-bearing process is an /// executable shape, but `std::generalize` needs >= 2 instruments in the /// campaign — a STATIC preflight fact (the referential gate has run, no data