From 58c61c35c76b59e4ec3e07d16af9d5abdf52d45e Mon Sep 17 00:00:00 2001 From: Toni Bergholm Date: Wed, 21 Oct 2026 07:26:46 +0300 Subject: [PATCH 1/2] fix(repo): transcripts::reindex scans outside the index lock and merges concurrent ingest (#274) The clone's post-clone transcript reindex scanned and rewrote .sc/transcripts and .sc/transcript-drops without holding the index lock across the scan, so a serve-path ingest into the same destination during the scan lost its appended lines. Same pattern as the signature rebuild (#273): scan and tombstone signature checks run unlocked, up to four unlocked catch-up passes, then a final pass under the lock judges only what is new (transcripts whose snapshot landed late, tombstones not honored yet, tombstones whose dropped body is new) before the rewrite. Trusted set and signature index are re-read each pass; a changed trusted set discards earlier verdicts. The locked listing also reloads the pack index so packs another handle wrote since open count. ADR-0038 addendum updated. --- crates/repo/src/transcripts.rs | 348 +++++++++++++++++++-- docs/adr/0038-agent-session-transcripts.md | 16 + 2 files changed, 339 insertions(+), 25 deletions(-) diff --git a/crates/repo/src/transcripts.rs b/crates/repo/src/transcripts.rs index f4e6965e..d168a2d4 100644 --- a/crates/repo/src/transcripts.rs +++ b/crates/repo/src/transcripts.rs @@ -630,38 +630,154 @@ fn reevaluate_inner( /// that are `Object::Transcript` whose `snapshot` also resolves locally, /// and atomically rewrite the index to exactly that set (not an append — /// stale entries from a half-written prior index are dropped too). +/// +/// The scan and the tombstone signature checks run without the index lock. A +/// serve-path ingest into the same store stores its objects before it appends +/// (under the lock), so under the lock the rebuild lists the store again, +/// after a few unlocked catch-up passes, and judges only what appeared since, +/// plus the transcripts whose snapshot was missing and the tombstones that +/// were not honored yet (their signature or trust may have landed since) (#274): +/// an append made during the scan is covered, and one made after the rewrite +/// lands on top of it. The locked listing also reloads the pack index, so +/// packs another handle wrote since this store opened count too. pub(crate) fn reindex(layout: &Layout, store: &mut scl_core::Store) -> Result { - let mut entries: Vec<(ObjectId, ObjectId)> = Vec::new(); - let mut drops: Vec = Vec::new(); - let trusted = trusted_signers(layout); - let sigs = crate::signatures::load_sig_index(layout, None, &trusted)?; - for id in store.all_ids()? { - if let Object::Transcript(t) = store.get(&id)? { + reindex_with(layout, store, || {}, || {}) +} + +/// [`reindex`] with hooks so a test can interleave an ingest deterministically: +/// `after_scan` runs after the first scan, `before_lock` after the unlocked +/// catch-up passes, so only the locked pass can absorb what it does. +fn reindex_with( + layout: &Layout, + store: &mut scl_core::Store, + after_scan: impl FnOnce(), + before_lock: impl FnOnce(), +) -> Result { + let mut scan = Scan::default(); + scan.pass(layout, store)?; + after_scan(); + // Catch up on what arrived meanwhile without the lock, so the locked pass + // has little left to judge however large a concurrent push was. Bounded: a + // writer that keeps adding objects cannot postpone the write. + for _ in 0..CATCH_UP_PASSES { + if !scan.pass(layout, store)? { + break; + } + } + before_lock(); + with_index_lock(layout, || { + store.reload_packs()?; + scan.pass(layout, store)?; + let drops: Vec = scan + .honored + .values() + .copied() + .collect::>() + .into_iter() + .collect(); + let dropped: BTreeSet = drops.iter().map(|(_, _, d)| *d).collect(); + let entries: Vec<(ObjectId, ObjectId)> = scan + .bodies + .iter() + .filter(|(id, _)| !dropped.contains(id)) + .map(|(id, snap)| (*snap, *id)) + .collect::>() + .into_iter() + .collect(); + write_index(layout, &entries)?; + write_drops(layout, &drops)?; + Ok(entries.len()) + }) +} + +/// Unlocked catch-up passes `reindex` makes before taking the lock. +const CATCH_UP_PASSES: usize = 4; + +/// What [`reindex`] has learned about the store so far, across passes. +#[derive(Default)] +struct Scan { + seen: BTreeSet, + /// Transcripts (and tombstones) listed before their snapshot landed: + /// `(id, snapshot)`, retried each pass (an ingest stores a pack's objects + /// one by one). + waiting: Vec<(ObjectId, ObjectId)>, + /// Live-index candidates: transcript id to snapshot. + bodies: std::collections::BTreeMap, + /// Every tombstone with a held snapshot, honored or not: tombstone id to + /// the tombstone and the id it drops. + tombs: std::collections::BTreeMap, + /// The tombstones honored so far, as the drop line each would write. + honored: std::collections::BTreeMap, + /// Fingerprint of the trusted set `honored` was judged under. + fp: String, +} + +impl Scan { + /// One pass: decode the objects not seen yet, plus the waiting ones whose + /// snapshot is now held, then judge the tombstones again under the current + /// trusted set and signature index. Returns whether any new object was + /// listed. + fn pass(&mut self, layout: &Layout, store: &mut scl_core::Store) -> Result { + let new: Vec = store + .all_ids()? + .into_iter() + .filter(|id| self.seen.insert(*id)) + .collect(); + let (ready, waiting): (Vec<_>, Vec<_>) = std::mem::take(&mut self.waiting) + .into_iter() + .partition(|(_, snap)| store.contains(snap)); + self.waiting = waiting; + let retry = ready.into_iter().map(|(id, _)| id); + for id in new.iter().copied().chain(retry) { + let Object::Transcript(t) = store.get(&id)? else { + continue; + }; if !store.contains(&t.snapshot) { + self.waiting.push((id, t.snapshot)); + } else if let Some(dropped) = t.dropped_id() { + self.tombs.insert(id, (t, dropped)); + } else { + self.bodies.insert(id, t.snapshot); + } + } + let any_new = !new.is_empty(); + self.judge(layout, store, &new.into_iter().collect())?; + Ok(any_new) + } + + /// Judge the tombstones not honored yet (their signature or the signer's + /// trust may have landed since), and again any whose dropped transcript is + /// among the `new` objects, under the trusted set and signature index as + /// they are now. A changed trusted set invalidates every earlier verdict. + fn judge( + &mut self, + layout: &Layout, + store: &mut scl_core::Store, + new: &BTreeSet, + ) -> Result<()> { + if self.tombs.is_empty() { + return Ok(()); + } + let trusted = trusted_signers(layout); + let fp = fp_of(&trusted); + if fp != self.fp { + self.honored.clear(); + self.fp = fp; + } + let wanted: BTreeSet = self.tombs.keys().copied().collect(); + let sigs = crate::signatures::load_sig_index(layout, Some(&wanted), &trusted)?; + for (id, (t, dropped)) in &self.tombs { + if self.honored.contains_key(id) && !new.contains(dropped) { continue; } - match t.dropped_id() { - Some(dropped) => { - if tombstone_honored(store, &trusted, &sigs, id, &t, dropped)? { - drops.push((t.snapshot, id, dropped)); - } - } - None => entries.push((t.snapshot, id)), + if tombstone_honored(store, &trusted, &sigs, *id, t, *dropped)? { + self.honored.insert(*id, (t.snapshot, *id, *dropped)); + } else { + self.honored.remove(id); } } + Ok(()) } - drops.sort(); - drops.dedup(); - let dropped: BTreeSet = drops.iter().map(|(_, _, d)| *d).collect(); - entries.retain(|(_, tid)| !dropped.contains(tid)); - entries.sort(); - entries.dedup(); - let count = entries.len(); - with_index_lock(layout, || { - write_index(layout, &entries)?; - write_drops(layout, &drops) - })?; - Ok(count) } /// Sender-side seam (Task 3/4): the indexed transcript object ids (and drop @@ -2071,4 +2187,186 @@ pub(crate) mod tests { fn concurrent_serve_ingests_lose_no_index_entry_and_do_not_deadlock_with_update_ref() { concurrent_ingests_lose_nothing("conc-ref", false); } + + /// A pack of `objs`, as a serve-path ingest receives it. + fn pack_of(objs: &[Object]) -> Vec { + let enc: Vec<(ObjectId, Vec)> = objs.iter().map(|o| (o.id(), o.encode())).collect(); + scl_core::pack::build_pack(&enc).unwrap().0 + } + + /// A signature by `who` over the drop digest of `tomb`. + fn drop_sig(who: &scl_crypto::Identity, tomb: ObjectId) -> Object { + let k = who.signing.as_ref().unwrap(); + Object::Signature(SignatureObj { + snapshot: tomb, + signer: k.public().to_bytes(), + sig: scl_crypto::sign_transcript_id(k, &drop_digest(&tomb)), + }) + } + + fn put_pack_into(root: &std::path::Path, pack: &[u8]) { + use crate::transport::Transport; + crate::transport::LocalTransport::open(root) + .unwrap() + .put_pack(&mut &pack[..]) + .unwrap(); + } + + /// Run `reindex_with` on `dest`, calling `ingest` once, after the first + /// scan or (`just_before_lock`) only after the unlocked catch-up passes. + fn reindex_racing(dest: &Repo, just_before_lock: bool, ingest: impl FnOnce()) -> usize { + let ingest = std::cell::RefCell::new(Some(ingest)); + let arc = dest.store_arc(); + let mut store = arc.lock().unwrap(); + reindex_with( + dest.layout(), + &mut store, + || { + if !just_before_lock { + (ingest.borrow_mut().take().unwrap())(); + } + }, + || { + if just_before_lock { + (ingest.borrow_mut().take().unwrap())(); + } + }, + ) + .unwrap() + } + + /// A source repo with `snap`, a trusting `dest` that holds it, and the + /// identity `dest` trusts. + fn race_fixture(tag: &str) -> (Repo, ObjectId, Repo, scl_crypto::Identity) { + let src = Repo::init(tmp_root(&format!("{tag}-src"))).unwrap(); + std::fs::write(src.layout().root.join("f"), b"x").unwrap(); + let snap = src.commit("t", "c1").unwrap(); + let dest = Repo::init(tmp_root(&format!("{tag}-dst"))).unwrap(); + dest.store_arc() + .lock() + .unwrap() + .put(src.store_arc().lock().unwrap().get(&snap).unwrap()) + .unwrap(); + let (_s, owner) = scl_crypto::generate_identity_v2(); + trust_signer(&dest, &owner); + (src, snap, dest, owner) + } + + fn body_of(src: &Repo, snap: ObjectId, label: &str, who: &scl_crypto::Identity) -> Object { + let id = src + .attach_transcript(snap, "a", label, label.as_bytes(), &[who.enc.public()]) + .unwrap(); + src.store_arc().lock().unwrap().get(&id).unwrap() + } + + fn clean_up(repos: &[Repo]) { + let roots: Vec<_> = repos.iter().map(|r| r.layout().root.clone()).collect(); + for r in roots { + std::fs::remove_dir_all(&r).unwrap(); + assert!(!r.exists()); + } + } + + #[test] + fn reindex_keeps_an_ingest_that_lands_during_its_scan() { + // The ingest lands after the first scan, or only after the unlocked + // catch-up passes (so just the locked pass can absorb it). Without the + // locked relisting the rewrite replaces the index it appended to. + for just_before_lock in [false, true] { + let (src, snap, dest, owner) = race_fixture(&format!("race-{just_before_lock}")); + let base = body_of(&src, snap, "base", &owner); + let keep = body_of(&src, snap, "keep", &owner); + let doomed = body_of(&src, snap, "doomed", &owner); + let tomb = Object::Transcript(Transcript::drop_tombstone(snap, &doomed.id())); + dest.store_arc().lock().unwrap().put(base.clone()).unwrap(); + let pack = pack_of(&[ + keep.clone(), + doomed.clone(), + tomb.clone(), + drop_sig(&owner, tomb.id()), + ]); + let root = dest.layout().root.clone(); + let count = reindex_racing(&dest, just_before_lock, || put_pack_into(&root, &pack)); + assert_eq!(count, 2); + let mut live: Vec<_> = load(dest.layout()) + .unwrap() + .into_iter() + .map(|e| e.1) + .collect(); + live.sort(); + let mut want = vec![base.id(), keep.id()]; + want.sort(); + assert_eq!(live, want, "a transcript index line was lost"); + assert_eq!( + read_drops(dest.layout()).unwrap(), + vec![(snap, tomb.id(), doomed.id())], + "the tombstone was lost" + ); + clean_up(&[src, dest]); + } + } + + #[test] + fn reindex_judges_again_what_lands_after_the_first_scan() { + // A transcript listed before its snapshot, and a tombstone listed + // before its signature: both are settled by what arrives later, and + // a late signature by a key that is not trusted still honors nothing. + for (just_before_lock, trusted_signer) in + [(false, true), (true, true), (false, false), (true, false)] + { + let (src, snap, dest, owner) = + race_fixture(&format!("late-{just_before_lock}-{trusted_signer}")); + let (_s, stranger) = scl_crypto::generate_identity_v2(); + let signer = if trusted_signer { &owner } else { &stranger }; + std::fs::write(src.layout().root.join("f"), b"y").unwrap(); + let snap2 = src.commit("t", "c2").unwrap(); + let snap2_obj = src.store_arc().lock().unwrap().get(&snap2).unwrap(); + let orphan = body_of(&src, snap2, "orphan", &owner); + let doomed = body_of(&src, snap, "doomed", &owner); + let tomb = Object::Transcript(Transcript::drop_tombstone(snap, &doomed.id())); + for o in [&orphan, &doomed, &tomb] { + dest.store_arc().lock().unwrap().put(o.clone()).unwrap(); + } + let pack = pack_of(&[snap2_obj, drop_sig(signer, tomb.id())]); + let root = dest.layout().root.clone(); + let count = reindex_racing(&dest, just_before_lock, || put_pack_into(&root, &pack)); + let mut live: Vec<_> = load(dest.layout()) + .unwrap() + .into_iter() + .map(|e| e.1) + .collect(); + live.sort(); + let (mut want, drops) = if trusted_signer { + (vec![orphan.id()], vec![(snap, tomb.id(), doomed.id())]) + } else { + (vec![orphan.id(), doomed.id()], vec![]) + }; + want.sort(); + assert_eq!(count, want.len()); + assert_eq!(live, want, "the late snapshot or signature is mishandled"); + assert_eq!(read_drops(dest.layout()).unwrap(), drops); + clean_up(&[src, dest]); + } + } + + #[test] + fn reindex_sees_a_pack_another_handle_wrote_since_the_store_opened() { + for just_before_lock in [false, true] { + let (src, snap, dest, owner) = race_fixture(&format!("pack-{just_before_lock}")); + let late = body_of(&src, snap, "late", &owner); + let late_id = late.id(); + let objects = dest.layout().objects_dir(); + let count = reindex_racing(&dest, just_before_lock, || { + let mut s2 = + scl_core::Store::open_persistent(objects, crate::repo::default_budget()) + .unwrap(); + s2.put(late).unwrap(); + s2.write_pack(&[late_id]).unwrap(); + s2.delete(&late_id).unwrap(); + }); + assert_eq!(count, 1); + assert_eq!(load(dest.layout()).unwrap(), vec![(snap, late_id)]); + clean_up(&[src, dest]); + } + } } diff --git a/docs/adr/0038-agent-session-transcripts.md b/docs/adr/0038-agent-session-transcripts.md index 59fd4001..73e3323a 100644 --- a/docs/adr/0038-agent-session-transcripts.md +++ b/docs/adr/0038-agent-session-transcripts.md @@ -662,6 +662,22 @@ scan in `tombstone_honored` never reached it. Now: append is lost, and the lock is never held across the bulk of the verification. The relisting sees loose objects (what ingest writes), not packs another handle wrote since the store opened. +- **`transcripts::reindex` and a concurrent ingest (#274).** The same shape + for the transcript index and the drops file. The scan, and every tombstone + signature check, run without the index lock, followed by up to four + unlocked catch-up passes. Under the lock it lists the store once more and + judges only what is new: transcripts whose snapshot was not held when they + were listed, tombstones not honored yet (their signature or the signer's + trust may have landed since) and tombstones whose dropped body is new. Each + pass re-reads the trusted set and the signature index, and a changed set + discards the earlier verdicts, so only trusted signers are ever honored. + The rewrite of `.sc/transcripts` and `.sc/transcript-drops` then happens + under the lock, so an ingest that landed during the scan is relisted and one + that lands after appends on top; the pending ring is untouched. Unlike the + signature rebuild, the locked listing first reloads the pack index + (`Store::reload_packs`), so objects another handle packed since this store + opened count as well; it costs one read of the `.idx` files, once, under the + lock. (`signatures::reindex` still relists loose objects only.) ### 2. The index lock (#260) From a09993a8a6718afcd2b3964025f7220de4735dfa Mon Sep 17 00:00:00 2001 From: Toni Bergholm Date: Wed, 21 Oct 2026 07:42:53 +0300 Subject: [PATCH 2/2] fix(repo): cheaper tombstone re-judging in transcripts::reindex and trust-change tests (#274) Skip the signature index read when no trusted signer exists or no tombstone needs a verdict, refuse an unsigned tombstone before reading objects, and pin the trust-withdrawn and mismatched-body cases with tests. ADR addendum notes the pack reload is defensive. --- crates/repo/src/transcripts.rs | 84 +++++++++++++++++++--- docs/adr/0038-agent-session-transcripts.md | 6 +- 2 files changed, 81 insertions(+), 9 deletions(-) diff --git a/crates/repo/src/transcripts.rs b/crates/repo/src/transcripts.rs index d168a2d4..a58253dd 100644 --- a/crates/repo/src/transcripts.rs +++ b/crates/repo/src/transcripts.rs @@ -764,16 +764,30 @@ impl Scan { self.honored.clear(); self.fp = fp; } - let wanted: BTreeSet = self.tombs.keys().copied().collect(); + // Nothing is honorable without a trusted signer, and only the tombstones + // not honored yet (or whose dropped body just arrived) need a verdict: + // skip the signature index read when there are none. + let todo: Vec = self + .tombs + .iter() + .filter(|(id, (_, dropped))| !self.honored.contains_key(*id) || new.contains(dropped)) + .map(|(id, _)| *id) + .collect(); + if trusted.is_empty() || todo.is_empty() { + return Ok(()); + } + let wanted: BTreeSet = todo.iter().copied().collect(); let sigs = crate::signatures::load_sig_index(layout, Some(&wanted), &trusted)?; - for (id, (t, dropped)) in &self.tombs { - if self.honored.contains_key(id) && !new.contains(dropped) { - continue; - } - if tombstone_honored(store, &trusted, &sigs, *id, t, *dropped)? { - self.honored.insert(*id, (t.snapshot, *id, *dropped)); + for id in todo { + let (t, dropped) = &self.tombs[&id]; + // Cheap refusal first: no trusted signer has signed it, so the + // store reads in `tombstone_honored` cannot change the verdict. + if trusted.keys().all(|k| sigs.by(&id, k).is_none()) + || !tombstone_honored(store, &trusted, &sigs, id, t, *dropped)? + { + self.honored.remove(&id); } else { - self.honored.remove(id); + self.honored.insert(id, (t.snapshot, id, *dropped)); } } Ok(()) @@ -2369,4 +2383,58 @@ pub(crate) mod tests { clean_up(&[src, dest]); } } + + #[test] + fn reindex_drops_a_verdict_when_trust_is_withdrawn_between_passes() { + for just_before_lock in [false, true] { + let (src, snap, dest, owner) = race_fixture(&format!("untrust-{just_before_lock}")); + let doomed = body_of(&src, snap, "doomed", &owner); + let tomb = Object::Transcript(Transcript::drop_tombstone(snap, &doomed.id())); + let root = dest.layout().root.clone(); + put_pack_into( + &root, + &pack_of(&[doomed.clone(), tomb.clone(), drop_sig(&owner, tomb.id())]), + ); + assert_eq!(read_drops(dest.layout()).unwrap().len(), 1); + let recipients = dest.layout().dot_sc.join("recipients.toml"); + let count = reindex_racing(&dest, just_before_lock, || { + std::fs::write(recipients, "").unwrap(); + }); + assert_eq!(count, 1, "the signer is no longer trusted"); + assert!(read_drops(dest.layout()).unwrap().is_empty()); + clean_up(&[src, dest]); + } + } + + #[test] + fn reindex_refuses_a_drop_once_its_body_turns_out_to_be_on_another_snapshot() { + for just_before_lock in [false, true] { + let (src, snap, dest, owner) = race_fixture(&format!("mismatch-{just_before_lock}")); + std::fs::write(src.layout().root.join("f"), b"y").unwrap(); + let snap2 = src.commit("t", "c2").unwrap(); + let snap2_obj = src.store_arc().lock().unwrap().get(&snap2).unwrap(); + dest.store_arc().lock().unwrap().put(snap2_obj).unwrap(); + // The tombstone claims `snap`, but the body it names sits on `snap2`. + let body = body_of(&src, snap2, "elsewhere", &owner); + let tomb = Object::Transcript(Transcript::drop_tombstone(snap, &body.id())); + let root = dest.layout().root.clone(); + put_pack_into( + &root, + &pack_of(&[tomb.clone(), drop_sig(&owner, tomb.id())]), + ); + assert_eq!(read_drops(dest.layout()).unwrap().len(), 1, "held for now"); + let objects = dest.layout().objects_dir(); + let body_id = body.id(); + let count = reindex_racing(&dest, just_before_lock, || { + let mut s2 = + scl_core::Store::open_persistent(objects, crate::repo::default_budget()) + .unwrap(); + s2.put(body).unwrap(); + }); + assert_eq!(count, 1); + assert_eq!(load(dest.layout()).unwrap(), vec![(snap2, body_id)]); + assert!(read_drops(dest.layout()).unwrap().is_empty()); + clean_up(&[src, dest]); + } + } } diff --git a/docs/adr/0038-agent-session-transcripts.md b/docs/adr/0038-agent-session-transcripts.md index 73e3323a..f1df0950 100644 --- a/docs/adr/0038-agent-session-transcripts.md +++ b/docs/adr/0038-agent-session-transcripts.md @@ -677,7 +677,11 @@ scan in `tombstone_honored` never reached it. Now: signature rebuild, the locked listing first reloads the pack index (`Store::reload_packs`), so objects another handle packed since this store opened count as well; it costs one read of the `.idx` files, once, under the - lock. (`signatures::reindex` still relists loose objects only.) + lock. This is defensive: today only `gc` writes packs, and it needs the + repo lock the cloning `Repo` holds. (`signatures::reindex` still relists + loose objects only.) The locked pass reads the signature index only when a + trusted signer exists and some tombstone needs a verdict, and refuses a + tombstone no trusted signer has signed before any object read. ### 2. The index lock (#260)