diff --git a/crates/aura-registry/src/lib.rs b/crates/aura-registry/src/lib.rs index cd308df..520ddf5 100644 --- a/crates/aura-registry/src/lib.rs +++ b/crates/aura-registry/src/lib.rs @@ -1175,6 +1175,62 @@ mod tests { assert_eq!(ranked[0].metrics.total_pips, 3.0); // best-first within the family } + // Property: concurrent writers through the Registry append path leave the + // family store well-formed. When N threads share one registry handle (the + // #277 parallel cell loop's caller shape — exec.rs hands each cell the same + // `&Registry`) and each appends families to the shared `families.jsonl`, + // every member of every successful append must survive as its own complete, + // parseable JSON line — none torn, merged into a neighbour, or lost. Today's + // unsynchronized OpenOptions::append + writeln! interleaves the content and + // newline writes of concurrent appends, merging records onto one physical + // line so the store no longer parses. + #[test] + fn concurrent_appends_keep_the_family_store_well_formed() { + let path = temp_family_dir("concurrent"); + let reg = Registry::open(&path); + // Inline input: four distinct members per family, appended under a name + // unique to each (thread, iteration) so a clean store holds exactly + // `successful_appends * members` lines. + let members: Vec = (0..4).map(|k| report_with(k as f64, 0.5, 1)).collect(); + let per_append = members.len(); + let n_threads = 8usize; + let iters = 50usize; + + let total_ok: usize = std::thread::scope(|scope| { + let handles: Vec<_> = (0..n_threads) + .map(|tid| { + let reg = ® + let members = &members; + scope.spawn(move || { + let mut ok = 0usize; + for iter in 0..iters { + let name = format!("t{tid}-f{iter}"); + if reg.append_family(&name, FamilyKind::Sweep, members).is_ok() { + ok += 1; + } + } + ok + }) + }) + .collect(); + handles.into_iter().map(|h| h.join().expect("worker thread joined")).sum() + }); + + let loaded = reg.load_family_members(); + assert!( + loaded.is_ok(), + "concurrent appends must leave a parseable store; the writes corrupted it: {:?}", + loaded.err(), + ); + assert_eq!( + loaded.expect("store parses").len(), + total_ok * per_append, + "every member of every successful 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