rac_engine/
parallel_build.rs

1//! Parallel cold build of the derived read-model (ADR-107/ADR-108).
2//!
3//! Native shape of `services/parallel_build.py` + `parallel_merge.py`: the
4//! per-document work (parse, classify, index row, token vectors, live/scope
5//! projection) fans out across rayon workers as compact *fragments*; the
6//! parent merge runs only the cross-document steps (graph resolution,
7//! inbound fill, portfolio) — in sorted-path order, so the store bytes are
8//! worker-count-invariant by construction (no pickling boundary exists
9//! in-process; ADR-114).
10//!
11//! Correctness never depends on the parallel rung: below the file-count
12//! threshold, on a 1–2 core box, or on ANY worker fault (a panic in a
13//! fragment task — exercised by `DECIDED_PARALLEL_BUILD_FAULT`), the build
14//! falls back to the authoritative serial floor, whose partial results are
15//! never written.
16
17use std::path::PathBuf;
18
19use crate::derived::{build_derived_index_from_items, DerivedIndex, DECISION_TYPE};
20use crate::relationships::{relationships_from_corpus, CorpusItem};
21use crate::resolve::{entry_from_item, field_tokens_of, is_live_decision, IndexEntry};
22use crate::retrieve::{scope_rows_from_items, ScopeRow};
23
24const TIMING_ENV: &str = "DECIDED_TIMING";
25/// Fault-injection hook: when set, every fragment task panics — exercising
26/// the fault → serial-floor degrade in a real parallel run. Never set in
27/// production.
28const FAULT_ENV: &str = "DECIDED_PARALLEL_BUILD_FAULT";
29/// Below this file count the fan-out's coordination overhead outweighs the
30/// win, so the cold build stays on the serial floor (the oracle's measured
31/// crossover, kept for contract fidelity).
32pub const DEFAULT_MIN_PARALLEL_FILES: usize = 5_000;
33const MIN_FILES_ENV: &str = "DECIDED_PARALLEL_BUILD_MIN_FILES";
34
35/// Per-phase cold-build timings for the `DECIDED_TIMING` scorecard line.
36#[derive(Default)]
37pub struct BuildStats {
38    pub files: usize,
39    pub workers: usize,
40    pub parse_ms: f64,
41    pub derive_ms: f64,
42    pub write_ms: f64,
43}
44
45fn min_parallel_files() -> usize {
46    let Some(raw) = std::env::var_os(MIN_FILES_ENV) else {
47        return DEFAULT_MIN_PARALLEL_FILES;
48    };
49    match raw.to_string_lossy().trim().parse::<i64>() {
50        Ok(value) if value >= 0 => value as usize,
51        _ => DEFAULT_MIN_PARALLEL_FILES,
52    }
53}
54
55/// How many workers to use — 1 means the serial floor. An explicit count
56/// (the worker-invariance lever) is honoured up to the file count; the
57/// default policy stays serial on a small box or below the threshold.
58fn resolve_workers(workers: Option<usize>, n_files: usize) -> usize {
59    if n_files <= 1 {
60        return 1;
61    }
62    if let Some(w) = workers {
63        return w.max(1).min(n_files);
64    }
65    let cpu = std::thread::available_parallelism()
66        .map(std::num::NonZeroUsize::get)
67        .unwrap_or(1);
68    if cpu <= 2 || n_files < min_parallel_files() {
69        return 1;
70    }
71    cpu.min(n_files)
72}
73
74/// One document's compact derived projection — the unit workers emit.
75struct DocFragment {
76    item: CorpusItem,
77    index_entry: IndexEntry, // inbound stays 0; the merge fills it
78    field_tokens: crate::resolve::FieldTokens,
79    is_live_decision: bool,
80    scope_row: Option<ScopeRow>,
81}
82
83fn fragment_for(path_display: &str) -> DocFragment {
84    if std::env::var_os(FAULT_ENV).is_some() {
85        panic!("parallel-build worker fault (injected)");
86    }
87    let artifact = crate::parse::parse_file(path_display);
88    let spec = crate::spec::spec_for(&crate::classify::classify(&artifact).artifact_type);
89    let item = CorpusItem {
90        path: path_display.to_string(),
91        artifact,
92        spec,
93    };
94    let index_entry = entry_from_item(&item, 0);
95    let field_tokens = field_tokens_of(&index_entry);
96    let live = item.spec.map(|s| s.name == DECISION_TYPE).unwrap_or(false)
97        && is_live_decision(&item.artifact);
98    let scope_row = scope_rows_from_items(std::slice::from_ref(&item)).into_iter().next();
99    DocFragment {
100        item,
101        index_entry,
102        field_tokens,
103        is_live_decision: live,
104        scope_row,
105    }
106}
107
108/// Fan the parse + per-doc derive across `n_workers`, or None on any fault.
109fn fragments_parallel(paths: &[String], n_workers: usize) -> Option<Vec<DocFragment>> {
110    let pool = rayon::ThreadPoolBuilder::new()
111        .num_threads(n_workers)
112        .build()
113        .ok()?;
114    let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
115        pool.install(|| {
116            use rayon::prelude::*;
117            paths
118                .par_iter()
119                .map(|path| fragment_for(path))
120                .collect::<Vec<DocFragment>>()
121        })
122    }));
123    // A partially-derived result is discarded whole — the store is never
124    // written from a faulted fan-out.
125    result.ok()
126}
127
128/// Reproduce the derived read-model from per-document fragments: only the
129/// cross-document steps run here, in sorted-path order.
130fn reproduce(fragments: Vec<DocFragment>, directory: &str, recursive: bool) -> DerivedIndex {
131    let mut items = Vec::with_capacity(fragments.len());
132    let mut index_entries = Vec::with_capacity(fragments.len());
133    let mut field_tokens = Vec::with_capacity(fragments.len());
134    let mut live_decision_paths = Vec::new();
135    let mut scope_rows = Vec::new();
136    for fragment in fragments {
137        if fragment.is_live_decision {
138            live_decision_paths.push(fragment.index_entry.path.clone());
139        }
140        if let Some(row) = fragment.scope_row {
141            scope_rows.push(row);
142        }
143        items.push(fragment.item);
144        index_entries.push(fragment.index_entry);
145        field_tokens.push(fragment.field_tokens);
146    }
147    // The cross-document steps: graph resolution, inbound fill, portfolio.
148    let relationships = relationships_from_corpus(&items);
149    let mut inbound: std::collections::HashMap<&str, i64> = std::collections::HashMap::new();
150    for rel in &relationships {
151        if let Some(resolved) = &rel.resolved_path {
152            *inbound.entry(resolved.as_str()).or_insert(0) += 1;
153        }
154    }
155    for entry in &mut index_entries {
156        entry.inbound_count = inbound.get(entry.path.as_str()).copied().unwrap_or(0);
157    }
158    let summary = crate::portfolio::portfolio_from_corpus(directory, &items, recursive);
159    DerivedIndex {
160        index_entries,
161        field_tokens,
162        relationships,
163        live_decision_paths,
164        portfolio_summary: crate::output::portfolio_summary_value(&summary),
165        scope_rows,
166    }
167}
168
169/// Build the derived read-model with a parallel parse AND per-doc derive —
170/// byte-identical to `derived::build_derived_index` for any worker count.
171pub fn build_derived_index_parallel(
172    directory: &str,
173    recursive: bool,
174    workers: Option<usize>,
175) -> (DerivedIndex, BuildStats) {
176    let t0 = std::time::Instant::now();
177    let paths: Vec<String> = crate::walk::find_markdown_files(directory, recursive)
178        .into_iter()
179        .map(|e| e.display)
180        .collect();
181    let n_workers = resolve_workers(workers, paths.len());
182    let fragments = if n_workers > 1 {
183        fragments_parallel(&paths, n_workers)
184    } else {
185        None
186    };
187    if let Some(fragments) = fragments {
188        let used = n_workers;
189        let t1 = std::time::Instant::now();
190        let n_files = fragments.len();
191        let derived = reproduce(fragments, directory, recursive);
192        let t2 = std::time::Instant::now();
193        return (
194            derived,
195            BuildStats {
196                files: n_files,
197                workers: used,
198                parse_ms: (t1 - t0).as_secs_f64() * 1000.0,
199                derive_ms: (t2 - t1).as_secs_f64() * 1000.0,
200                write_ms: 0.0,
201            },
202        );
203    }
204    // Serial floor: the authoritative walk + derive; a fault above lands here.
205    let items = crate::relationships::corpus_items(directory, recursive);
206    let t1 = std::time::Instant::now();
207    let derived = build_derived_index_from_items(directory, &items, recursive);
208    let t2 = std::time::Instant::now();
209    (
210        derived,
211        BuildStats {
212            files: items.len(),
213            workers: 1,
214            parse_ms: (t1 - t0).as_secs_f64() * 1000.0,
215            derive_ms: (t2 - t1).as_secs_f64() * 1000.0,
216            write_ms: 0.0,
217        },
218    )
219}
220
221/// Write the cold-build scorecard line to stderr when `DECIDED_TIMING` is set —
222/// env-gated, stderr-only, byte-shaped like the oracle's (ADR-107).
223pub fn emit_build_timing(stats: &BuildStats) {
224    if std::env::var_os(TIMING_ENV).is_none() {
225        return;
226    }
227    eprintln!(
228        "decided-timing: build_parse_ms={:.3} build_derive_ms={:.3} build_write_ms={:.3} workers={} files={}",
229        stats.parse_ms, stats.derive_ms, stats.write_ms, stats.workers, stats.files
230    );
231}
232
233/// The paths type alias for the freshness tracker's explicit-list parse
234/// (INDEX-PLAN B6): parse a known path list through the one true per-file
235/// path, parallel when it pays; entries in list order.
236pub fn parallel_parse_paths(paths: &[PathBuf]) -> (Vec<CorpusItem>, usize) {
237    let displays: Vec<String> = paths
238        .iter()
239        .map(|p| p.to_string_lossy().into_owned())
240        .collect();
241    use rayon::prelude::*;
242    let items: Vec<CorpusItem> = displays
243        .par_iter()
244        .map(|path| {
245            let artifact = crate::parse::parse_file(path);
246            let spec =
247                crate::spec::spec_for(&crate::classify::classify(&artifact).artifact_type);
248            CorpusItem {
249                path: path.clone(),
250                artifact,
251                spec,
252            }
253        })
254        .collect();
255    let workers = std::thread::available_parallelism()
256        .map(std::num::NonZeroUsize::get)
257        .unwrap_or(1);
258    (items, workers)
259}
rac_engine/parallel_build.rs

rac_engine/
parallel_build.rs
