From 883401309a6da1bf5038c2d522eb02483ce67cdf Mon Sep 17 00:00:00 2001 From: claude Date: Thu, 16 Jul 2026 12:36:31 +0200 Subject: [PATCH] feat(registry): serialize the append write path for concurrent writers MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Registry::append_family and append_campaign_run did unsynchronized OpenOptions::append + writeln! to their shared JSONL stores; concurrent writers through one shared &Registry interleaved one append's content and newline writes with another's, merging records onto one physical line and corrupting the store (the #276 RED test failed 3/3 with a "trailing characters" parse error). The planned parallel campaign cell loop (#277) makes exactly that caller shape reachable, so this is its prerequisite. Mechanism: one internal append_write_lock: Mutex<()> on Registry, acquired by both append functions and held across the whole read-the-run-counter-then-write critical section — serializing the write also serializes the read-before-write counter derivation, so concurrent appends under one name cannot double-assign a run index. Mutual exclusion around a per-cell-per-stage microsecond write, not a barrier: cells never rendezvous, and the sim event loop stays lock-free (C1's "without locking" governs the sims; this lock sits at the results-persistence edge outside the hot path). A poisoned lock is entered anyway (PoisonError::into_inner): every store line is independent, so one panicked writer must not wedge all later appends. Scope, per the decision log on #276: in-process synchronization owned by the Registry type itself; cross-process locking stays out of contract (the corrected doc comments now state the actual guarantee). The serial per-cell-buffer alternative stays open to #277 as an ordering decision on top — this lock does not preclude it. Adds the sibling test pinning append_campaign_run under the same concurrent load. Verified: the #276 headline test green 3/3 (was red 3/3), sibling green, cargo test --workspace no failures, clippy --workspace --all-targets -D warnings clean. closes #276 --- crates/aura-registry/src/lib.rs | 60 ++++++++++++++++++++++++++++- crates/aura-registry/src/lineage.rs | 14 +++++-- 2 files changed, 69 insertions(+), 5 deletions(-) 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");