diff --git a/crates/aura-registry/src/lib.rs b/crates/aura-registry/src/lib.rs index 03024ce..0b1f983 100644 --- a/crates/aura-registry/src/lib.rs +++ b/crates/aura-registry/src/lib.rs @@ -49,8 +49,9 @@ pub struct Registry { /// read-the-counter-then-write critical section (`append_family`, /// `append_campaign_run` in `lineage.rs`), so concurrent writers through /// one handle also see a consistent per-name run counter. Store-agnostic: - /// it guards BOTH `families.jsonl` and the separate `campaign_runs.jsonl` - /// store (they never contend on the same file, but share this one lock). + /// it guards all three JSONL append paths — the flat runs store + /// ([`Registry::append`]), `families.jsonl`, and `campaign_runs.jsonl` + /// (they never contend on the same file, but share this one lock). /// In-process only; cross-process locking stays out of contract. append_write_lock: std::sync::Mutex<()>, } @@ -73,8 +74,11 @@ impl Registry { } /// Append one record as a single JSON line, creating the file (and its parent - /// directory) if absent. + /// directory) if absent. The write is serialized by `Registry`'s internal + /// `append_write_lock` (#276) — safe for concurrent writers sharing one + /// `&Registry`. pub fn append(&self, report: &RunReport) -> Result<(), RegistryError> { + let _guard = self.append_write_lock.lock().unwrap_or_else(std::sync::PoisonError::into_inner); if let Some(parent) = self.path.parent().filter(|p| !p.as_os_str().is_empty()) { fs::create_dir_all(parent)?; } @@ -1242,6 +1246,55 @@ mod tests { let _ = fs::remove_dir_all(path.parent().expect("temp family dir")); } + /// Third sibling of the #276 concurrency property, for the flat runs store: + /// `Registry::append` (→ `runs.jsonl`) shares the same + /// `OpenOptions::append` + `writeln!` pattern as the family and + /// campaign-run appends, and it is public API with external consumers + /// (C18), so N threads appending through one shared `&Registry` must leave + /// `runs.jsonl` well-formed — every successful append's record intact and + /// parseable, none torn, merged, or lost. + #[test] + fn concurrent_flat_appends_keep_the_runs_store_well_formed() { + let path = temp_family_dir("concurrent_flat"); + let reg = Registry::open(&path); + let report = report_with(1.0, 0.5, 1); + let n_threads = 8usize; + let iters = 50usize; + + let total_ok: usize = std::thread::scope(|scope| { + let handles: Vec<_> = (0..n_threads) + .map(|_| { + let reg = ® + let report = &report; + scope.spawn(move || { + let mut ok = 0usize; + for _ in 0..iters { + if reg.append(report).is_ok() { + ok += 1; + } + } + ok + }) + }) + .collect(); + handles.into_iter().map(|h| h.join().expect("worker thread joined")).sum() + }); + + let loaded = reg.load(); + assert!( + loaded.is_ok(), + "concurrent flat appends must leave a parseable store; the writes corrupted it: {:?}", + loaded.err(), + ); + assert_eq!( + loaded.expect("store parses").len(), + total_ok, + "every successful flat append must survive as its own intact line", + ); + + let _ = fs::remove_dir_all(path.parent().expect("temp family dir")); + } + fn member(total_pips: f64, net_trade_rs: Vec) -> SweepPoint { // Every RMetrics field is derived from the R slice by `r_metrics_from_rs` // (the same arithmetic production members carry), then the per-trade