diff --git a/crates/aura-registry/src/lib.rs b/crates/aura-registry/src/lib.rs index 520ddf5..03024ce 100644 --- a/crates/aura-registry/src/lib.rs +++ b/crates/aura-registry/src/lib.rs @@ -42,6 +42,17 @@ pub use trace_store::{FamilyMember, NameKind, RunTraces, TraceStore, TraceStoreE /// `RunReport`. pub struct Registry { path: PathBuf, + /// Serializes the family/campaign-run append write path (#276): one + /// `&Registry` shared across threads must never interleave two appends' + /// content and newline writes, which unsynchronized `OpenOptions::append` + /// plus `writeln!` permits at the OS level. Held across each append's + /// 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). + /// In-process only; cross-process locking stays out of contract. + append_write_lock: std::sync::Mutex<()>, } impl Registry { @@ -58,7 +69,7 @@ impl Registry { /// isolate runs (e.g. per-process or per-test), bind a distinct *directory*, /// not merely a distinct runs filename. pub fn open(path: impl AsRef) -> Registry { - Registry { path: path.as_ref().to_path_buf() } + Registry { path: path.as_ref().to_path_buf(), append_write_lock: std::sync::Mutex::new(()) } } /// Append one record as a single JSON line, creating the file (and its parent @@ -1995,6 +2006,53 @@ mod tests { ); } + /// Sibling of `concurrent_appends_keep_the_family_store_well_formed` + /// (#276): `append_campaign_run` shares `append_family`'s unsynchronized + /// `OpenOptions::append` + `writeln!` pattern at the same seam, so N + /// threads appending campaign runs through one shared `&Registry` must + /// leave `campaign_runs.jsonl` well-formed — every successful append's + /// record intact and parseable, none torn or lost. + #[test] + fn concurrent_campaign_run_appends_keep_the_store_well_formed() { + let path = temp_family_dir("concurrent_campaign_run"); + let reg = Registry::open(&path); + 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 = ® + scope.spawn(move || { + let mut ok = 0usize; + for iter in 0..iters { + let campaign = format!("t{tid}-c{iter}"); + if reg.append_campaign_run(&campaign_run_record(&campaign)).is_ok() { + ok += 1; + } + } + ok + }) + }) + .collect(); + handles.into_iter().map(|h| h.join().expect("worker thread joined")).sum() + }); + + let loaded = reg.load_campaign_runs(); + assert!( + loaded.is_ok(), + "concurrent campaign-run appends must leave a parseable store; the writes corrupted it: {:?}", + loaded.err(), + ); + assert_eq!( + loaded.expect("store parses").len(), + total_ok, + "every successful campaign-run append must survive as its own intact line", + ); + + let _ = fs::remove_dir_all(path.parent().expect("temp family dir")); + } + #[test] fn load_campaign_runs_missing_file_is_empty() { let path = temp_family_dir("campaign_runs_missing"); diff --git a/crates/aura-registry/src/lineage.rs b/crates/aura-registry/src/lineage.rs index ec97c24..6115a77 100644 --- a/crates/aura-registry/src/lineage.rs +++ b/crates/aura-registry/src/lineage.rs @@ -258,14 +258,17 @@ impl Registry { /// members carrying `name` + that `run`, and append them as N JSONL lines to /// the family store (a sibling of the flat runs store, `families.jsonl`, /// leaving the flat path untouched). Returns the derived id `"{name}-{run}"`. - /// Reads the store once to pick the run index (a read-before-write; - /// single-process CLI invocations do not race). + /// Reads the store once to pick the run index, then writes, the whole + /// read-then-write critical section serialized by `Registry`'s internal + /// `append_write_lock` (#276) — safe for concurrent writers sharing one + /// `&Registry`. pub fn append_family( &self, name: &str, kind: FamilyKind, reports: &[RunReport], ) -> Result { + let _guard = self.append_write_lock.lock().unwrap_or_else(std::sync::PoisonError::into_inner); let run = next_run(name, &self.load_family_members()?); let path = self.path.with_file_name("families.jsonl"); @@ -317,12 +320,15 @@ impl Registry { /// run. The input record's own `run` field is ignored; a `Some` /// `trace_name` (the executor's claim sentinel, cycle 0109) is replaced /// with the derived `"{campaign8}-{run}"` on the stored line, `None` - /// stays `None`. Reads the store once to pick the run index (a - /// read-before-write; single-process CLI invocations do not race). + /// stays `None`. Reads the store once to pick the run index, then writes, + /// the whole read-then-write critical section serialized by the same + /// `Registry` `append_write_lock` that `append_family` uses (#276) — safe + /// for concurrent writers sharing one `&Registry`. pub fn append_campaign_run( &self, record: &CampaignRunRecord, ) -> Result { + let _guard = self.append_write_lock.lock().unwrap_or_else(std::sync::PoisonError::into_inner); let run = next_campaign_run(&record.campaign, &self.load_campaign_runs()?); let path = self.path.with_file_name("campaign_runs.jsonl");