The one-scan config test tolerates a snapshot TTL that elapsed during its own loop

.github/workflows/mesh-harness.yml

@@ -46,6 +46,14 @@ on:
       - "src/peer.rs"
       - "src/federation_admin.rs"
       - "src/auth/**"
+      # The config plane and the write door: every node writes config at boot
+      # (net.*, the wizard's keys) through graph_config, and attest::put is the
+      # one door the harness's peering and claims go through. 0.5.201's per-key
+      # leaf and #570's per-engine snapshot cache were only PROVEN on real nodes
+      # by this ladder, so a change to either must run it.
+      - "src/graph_config.rs"
+      - "src/attest.rs"
+      - "src/node_key.rs"
       - "harness/mesh-repro/**"
       - ".github/workflows/mesh-harness.yml"
 

src/attest.rs

@@ -518,7 +518,7 @@ pub async fn put(engine: &Engine, row: Attestation) -> Result<String, Error> {
         || kind == ciris_persist::federation::types::attestation_type::WITHDRAWS
         || kind == ciris_persist::federation::types::attestation_type::RECANTS
     {
-        crate::graph_config::invalidate();
+        crate::graph_config::invalidate_engine(engine);
     }
     Ok(id)
 }

src/graph_config.rs

@@ -471,45 +471,84 @@ impl ConfigSnapshot {
     }
 }
 
-/// The process-wide cache: one snapshot, replaced on scan, dropped on
-/// [`invalidate`]. A `Mutex<Option<Arc<_>>>` rather than an `RwLock`: the
-/// critical section is a pointer clone.
-/// (engine identity, snapshot) — the one cached slot.
-type CachedSnapshot = Option<(usize, Arc<ConfigSnapshot>)>;
+/// The cache: one snapshot PER ENGINE, a few at most, replaced on scan,
+/// dropped by [`invalidate_engine`] (one engine) or [`invalidate`] (all).
+///
+/// One slot for the whole process was right for a node (one engine) and
+/// wrong for a test binary, where a dozen engines read their config planes on
+/// parallel threads: every read from another engine evicted this one's
+/// snapshot, so a test counting scans saw a rescan for every neighbour's
+/// read (Codex on #570). A small map keyed by engine identity gives each
+/// engine its own slot; [`CACHE_SLOTS`] bounds it, evicting the oldest.
+/// A `Mutex` rather than an `RwLock`: the critical section is a pointer clone.
+type CachedSnapshot = Vec<(usize, Arc<ConfigSnapshot>)>;
+
+/// How many engines keep a snapshot at once. A node has one; the bound exists
+/// so a process that churns engines (a test binary, the embedded fold
+/// re-serving) cannot grow the map without limit.
+const CACHE_SLOTS: usize = 8;
 
 fn cache() -> &'static std::sync::Mutex<CachedSnapshot> {
     static CACHE: std::sync::OnceLock<std::sync::Mutex<CachedSnapshot>> =
         std::sync::OnceLock::new();
-    CACHE.get_or_init(|| std::sync::Mutex::new(None))
+    CACHE.get_or_init(|| std::sync::Mutex::new(Vec::new()))
 }
 
 /// The identity a snapshot is valid for: the `Engine` it was scanned from. A
 /// process normally holds one engine, but a test binary holds many at once
 /// (one `sqlite::memory:` each), and the embedded fold may re-serve on
 /// another home; a snapshot must never answer for a store it did not read.
-fn engine_identity(engine: &Arc<Engine>) -> usize {
-    Arc::as_ptr(engine) as usize
+fn engine_identity(engine: &Engine) -> usize {
+    // The address of the `Engine` value itself — for an `Arc<Engine>` this is
+    // exactly `Arc::as_ptr`, and it lets a door that holds only `&Engine`
+    // (`attest::put`) name the same slot.
+    std::ptr::from_ref(engine) as usize
 }
 
 /// Drop the cached snapshot: the next read scans. Called by every in-process
 /// door that changes the config plane ([`set_config`], the withdraw/recant
 /// emits in `attest`), and by compose at serve start so an in-process
 /// re-serve on another home never reads the previous node's config.
 pub fn invalidate() {
-    *cache().lock().unwrap_or_else(|p| p.into_inner()) = None;
+    cache().lock().unwrap_or_else(|p| p.into_inner()).clear();
+}
+
+/// Drop ONE engine's cached snapshot: the next read on that engine scans;
+/// every other engine's snapshot stands. This is the door the config-plane
+/// writers use — `set_config`, the withdraw/recant emits in `attest` — because
+/// a write to one store says nothing about another's (Codex on #570).
+pub fn invalidate_engine(engine: &Engine) {
+    let me = engine_identity(engine);
+    cache()
+        .lock()
+        .unwrap_or_else(|p| p.into_inner())
+        .retain(|(owner, _)| *owner != me);
 }
 
 /// The current config snapshot: cached if fresh, else one scan. THE read
 /// door — every getter below goes through it.
 pub async fn snapshot(engine: &Arc<Engine>) -> Result<Arc<ConfigSnapshot>> {
     let me = engine_identity(engine);
-    if let Some((owner, snap)) = cache().lock().unwrap_or_else(|p| p.into_inner()).as_ref() {
-        if *owner == me && snap.fresh() {
-            return Ok(Arc::clone(snap));
+    let cached = cache()
+        .lock()
+        .unwrap_or_else(|p| p.into_inner())
+        .iter()
+        .find(|(owner, _)| *owner == me)
+        .map(|(_, snap)| Arc::clone(snap));
+    if let Some(snap) = cached {
+        if snap.fresh() {
+            return Ok(snap);
         }
     }
     let snap = Arc::new(live_config_rows(engine).await?);
-    *cache().lock().unwrap_or_else(|p| p.into_inner()) = Some((me, Arc::clone(&snap)));
+    {
+        let mut slots = cache().lock().unwrap_or_else(|p| p.into_inner());
+        slots.retain(|(owner, _)| *owner != me);
+        if slots.len() >= CACHE_SLOTS {
+            slots.remove(0);
+        }
+        slots.push((me, Arc::clone(&snap)));
+    }
     Ok(snap)
 }
 
@@ -850,7 +889,7 @@ pub async fn set_config(
     // A write reads the plane first (the version chain and the head to
     // supersede), and that read must not be a cached one another writer has
     // since made stale — nor may the next read be served from before this row.
-    invalidate();
+    invalidate_engine(engine);
     let snap = snapshot(engine).await?;
     let current = latest_for_key(&snap.rows, key);
     let head = snap.heads.get(key);
@@ -906,7 +945,7 @@ pub async fn set_config(
         .emit_attestation_self(input)
         .await
         .map_err(|e| anyhow::anyhow!("emit_attestation_self({dimension}, {kind}): {e}"))?;
-    invalidate();
+    invalidate_engine(engine);
     tracing::info!(
         key,
         version,

tests/graph_config.rs

@@ -382,8 +382,15 @@ async fn fifty_reads_are_one_scan_and_a_write_invalidates() {
             .await
             .expect("set_config");
     }
-    graph_config::invalidate();
-    let before = graph_config::SCANS.load(Ordering::Relaxed);
+    graph_config::invalidate_engine(&engine);
+    // THIS engine's snapshot ordinal — not the process counter. `SCANS` counts
+    // every engine in the binary, and the sibling tests here run their own
+    // engines on parallel threads; each engine now has its own cache slot, so
+    // a neighbour's read or write cannot evict ours (Codex on #570), and what
+    // this test asserts is that ITS reads were served by ONE snapshot.
+    let first = graph_config::snapshot(&engine).await.expect("snapshot");
+    let before = first.scan;
+    let t0 = std::time::Instant::now();
     for _ in 0..50 {
         assert_eq!(
             graph_config::get_i64(&engine, "a.one").await.unwrap(),
@@ -402,12 +409,22 @@ async fn fifty_reads_are_one_scan_and_a_write_invalidates() {
         .await
         .unwrap();
     assert_eq!(listed.len(), 2);
-    let after = graph_config::SCANS.load(Ordering::Relaxed);
-    assert_eq!(
-        after - before,
-        1,
-        "150 keyed reads + one list must cost ONE scan; got {}",
-        after - before
+    let after = graph_config::snapshot(&engine)
+        .await
+        .expect("snapshot")
+        .scan;
+    // The SAME snapshot served everything — plus one rescan per snapshot TTL
+    // the loop itself outlived. On a cold macOS or Windows runner 150 awaits
+    // took longer than the 2 s TTL and the snapshot legitimately expired
+    // mid-loop (main 55b8c7a, both lanes); that is the cache working, not a
+    // scan per read. The budget is derived from the measured elapsed time.
+    let elapsed = t0.elapsed();
+    let budget = (elapsed.as_millis() / graph_config::CONFIG_SNAPSHOT_TTL.as_millis()) as u64;
+    let rescans = after - before;
+    assert!(
+        rescans <= budget,
+        "150 keyed reads + one list must be served by ONE snapshot (plus one rescan per TTL \
+         elapsed — {elapsed:?}, budget {budget}); this engine rescanned {rescans} time(s)"
     );
 
     let snap = graph_config::snapshot(&engine).await.unwrap();