event_graph: Validate static sync responses

static_sync() should reject incomplete or irrelevant batches

src/event_graph/mod.rs

@@ -474,6 +474,42 @@ pub(crate) fn filter_requested_event_rep(
     Ok((matched_events, matched_blobs, missing))
 }
 
+/// Merge one static-sync `EventRep` into the current batch state.
+///
+/// Returns the number of still-pending requested IDs satisfied by this
+/// response. Invalid responses are rejected before any state is mutated.
+pub(crate) fn merge_static_sync_event_rep(
+    requested: &[blake3::Hash],
+    pending: &mut HashSet<blake3::Hash>,
+    known: &mut HashSet<blake3::Hash>,
+    want: &mut HashSet<blake3::Hash>,
+    fetched: &mut Vec<(Event, Vec<u8>)>,
+    events: Vec<Event>,
+    blobs: Vec<Vec<u8>>,
+) -> Result<usize> {
+    let (matched_events, matched_blobs, _) = filter_requested_event_rep(requested, events, blobs)?;
+    let mut matched = 0;
+
+    for (ev, blob) in matched_events.into_iter().zip(matched_blobs) {
+        let eid = ev.id();
+        if !pending.remove(&eid) {
+            continue
+        }
+
+        matched += 1;
+        if known.insert(eid) {
+            for p in ev.header.parents.iter() {
+                if *p != NULL_ID && !known.contains(p) {
+                    want.insert(*p);
+                }
+            }
+            fetched.push((ev, blob));
+        }
+    }
+
+    Ok(matched)
+}
+
 /// The main Event Graph instance.
 ///
 /// Manages a rolling window of DAGs (one per rotation period), a
@@ -1149,6 +1185,11 @@ impl EventGraph {
         let mut fetched: Vec<(Event, Vec<u8>)> = vec![];
 
         while !want.is_empty() {
+            want.retain(|id| !known.contains(id));
+            if want.is_empty() {
+                break
+            }
+
             if fetched.len() >= SYNC_MAX_STATIC_EVENTS {
                 error!(
                     target: "event_graph::static_sync",
@@ -1159,50 +1200,50 @@ impl EventGraph {
             }
 
             let batch: Vec<blake3::Hash> = want.iter().copied().collect();
+            let mut pending: HashSet<blake3::Hash> = batch.iter().copied().collect();
             want.clear();
 
-            // Race the batch against all peers; first to respond
-            // with valid events wins. A peer that returns events we
-            // didn't ask for is striked via its protocol handler,
-            // not here - this is a best-effort pull.
+            // Ask every peer for the same batch. We keep consuming
+            // responses until the batch is complete or every peer has
+            // failed to help. Irrelevant, duplicate, or blob-misaligned
+            // replies do not satisfy the request.
             let mut req_futs = FuturesUnordered::new();
             for (i, ch) in channels.iter().enumerate() {
                 req_futs.push(request_event(ch.clone(), batch.clone(), i, timeout));
             }
 
-            let mut got_any = false;
-            while let Some((res, _, _)) = req_futs.next().await {
+            let mut made_progress = false;
+            while !pending.is_empty() {
+                let Some((res, _, _)) = req_futs.next().await else { break };
                 let Ok((evs, blobs)) = res else { continue };
                 if evs.is_empty() {
                     continue
                 }
-                got_any = true;
-                for (i, ev) in evs.into_iter().enumerate() {
-                    let eid = ev.id();
-                    if !batch.contains(&eid) {
-                        // Peer sent something we didn't ask for;
-                        // ignore the rest of this reply.
-                        break
-                    }
-                    if known.insert(eid) {
-                        // New parents to chase next round.
-                        for p in ev.header.parents.iter() {
-                            if *p != NULL_ID && !known.contains(p) {
-                                want.insert(*p);
-                            }
-                        }
-                        // Pair the event with its blob (or empty if
-                        // the peer didn't supply one - that's not an
-                        // error, see EventRep doc and the fall-through
-                        // in the apply loop below).
-                        let blob = blobs.get(i).cloned().unwrap_or_default();
-                        fetched.push((ev, blob));
-                    }
+
+                let Ok(matched) = merge_static_sync_event_rep(
+                    &batch,
+                    &mut pending,
+                    &mut known,
+                    &mut want,
+                    &mut fetched,
+                    evs,
+                    blobs,
+                ) else {
+                    continue
+                };
+
+                if matched > 0 {
+                    made_progress = true;
                 }
-                break
             }
 
-            if !got_any {
+            if !pending.is_empty() {
+                want.extend(pending.iter().copied().filter(|id| !known.contains(id)));
+            }
+
+            want.retain(|id| !known.contains(id));
+
+            if !made_progress {
                 // Nobody responded usefully. Give up so we don't
                 // loop forever on an unreachable ancestor.
                 error!(

src/event_graph/tests.rs

@@ -32,7 +32,7 @@ use crate::{
     event_graph::{
         compute_unreferenced_tips,
         event::Header,
-        filter_requested_event_rep,
+        filter_requested_event_rep, merge_static_sync_event_rep,
         proto::{EventPut, SyncDirection},
         test_helpers::{
             archive_config, bounded_dag_store_config, init_logger, make_eg, make_network,
@@ -87,6 +87,60 @@ fn evgr_event_rep_filter_matches_only_requested_ids() {
     assert!(filter_requested_event_rep(&requested, vec![event_a], Vec::new()).is_err());
 }
 
+#[test]
+fn evgr_static_sync_merge_tracks_partial_requested_batches() {
+    let event_a = test_event(b"static-requested-a", 11);
+    let event_b = test_event(b"static-requested-b", 12);
+    let unrelated = test_event(b"static-unrelated", 13);
+    let requested = vec![event_a.id(), event_b.id()];
+    let mut pending: HashSet<_> = requested.iter().copied().collect();
+    let mut known = HashSet::new();
+    let mut want = HashSet::new();
+    let mut fetched = Vec::new();
+
+    assert!(merge_static_sync_event_rep(
+        &requested,
+        &mut pending,
+        &mut known,
+        &mut want,
+        &mut fetched,
+        vec![unrelated],
+        vec![b"unrelated-blob".to_vec()],
+    )
+    .is_err());
+    assert_eq!(pending.len(), 2);
+    assert!(fetched.is_empty());
+
+    let matched = merge_static_sync_event_rep(
+        &requested,
+        &mut pending,
+        &mut known,
+        &mut want,
+        &mut fetched,
+        vec![event_b.clone()],
+        vec![b"blob-b".to_vec()],
+    )
+    .unwrap();
+    assert_eq!(matched, 1);
+    assert_eq!(pending, HashSet::from([event_a.id()]));
+
+    let matched = merge_static_sync_event_rep(
+        &requested,
+        &mut pending,
+        &mut known,
+        &mut want,
+        &mut fetched,
+        vec![event_a.clone()],
+        vec![b"blob-a".to_vec()],
+    )
+    .unwrap();
+    assert_eq!(matched, 1);
+    assert!(pending.is_empty());
+
+    let fetched_ids: HashSet<_> = fetched.iter().map(|(ev, _)| ev.id()).collect();
+    assert_eq!(fetched_ids, HashSet::from([event_a.id(), event_b.id()]));
+}
+
 #[test]
 fn evgr_time_index_queries_and_saturating_cursor() {
     // Forward, backward, newest, oldest queries plus the saturating

src/event_graph/tests_rln.rs

@@ -964,18 +964,16 @@ async fn static_sync_registration(ex: Arc<Executor<'static>>) {
     //
     // `static_sync` re-verifies historical RLN blobs (see the
     // `rln_verify_static_event` call in its body), so seeded
-    // nodes MUST persist a real blob - a missing blob causes
-    // the late-joiner to skip the event with a "no blob
-    // available" log. We build a real registration blob on
-    // node 0 (using the shared ZK keys, so the cost is amortized)
-    // and broadcast-equivalent it to the other three.
+    // nodes MUST persist the configured pregenerated guard blob -
+    // a missing blob causes the late-joiner to skip the event with
+    // a "no blob available" log. Free registration is disabled,
+    // so this test uses the app-configured pregenerated identity.
     let nodes = make_network(ex).await;
 
-    let id = TestIdentity::new();
-    let commitment = id.commitment();
-
-    let blob = id.create_registration(&nodes[0]).expect("build registration blob");
-    let blob_bytes = serialize_async(&blob).await;
+    let commitment = pallas::Base::from_repr(nodes[0].config.pregenerated_identity_commitments[0])
+        .into_option()
+        .unwrap();
+    let blob_bytes = GENESIS_BLOB_GUARD.to_vec();
 
     let rln_node = RLNNode::Registration(commitment);
     let content = serialize_async(&rln_node).await;