Merge branch 'main' into devnet4

Author: MegaRedHand

Cargo.lock

@@ -66,6 +66,7 @@ libssz-types = "0.2"
 # Build-time version info
 vergen-git2 = { version = "9", features = ["rustc"] }
 
+rayon = "1.11"
 rand = "0.9"
 rocksdb = "0.24"
 reqwest = { version = "0.12", default-features = false, features = ["rustls-tls"] }

Cargo.toml

@@ -23,6 +23,7 @@ spawned-concurrency.workspace = true
 
 tokio.workspace = true
 
+rayon.workspace = true
 thiserror.workspace = true
 tracing.workspace = true
 

crates/blockchain/Cargo.toml

@@ -198,6 +198,17 @@ impl BlockChainServer {
                 signature,
             };
 
+            // Self-deliver: store our own attestation locally for aggregation.
+            // Gossipsub does not deliver messages back to the sender, so without
+            // this the aggregator never sees its own validator's signature in
+            // gossip_signatures and it is excluded from aggregated proofs.
+            if self.is_aggregator {
+                let _ = store::on_gossip_attestation(&mut self.store, signed_attestation.clone())
+                    .inspect_err(|err| {
+                        warn!(%slot, %validator_id, %err, "Self-delivery of attestation failed")
+                    });
+            }
+
             // Publish to gossip network
             if let Some(ref p2p) = self.p2p {
                 let _ = p2p.publish_attestation(signed_attestation).inspect_err(

crates/blockchain/src/lib.rs

@@ -1271,40 +1271,58 @@ fn verify_signatures(
     let validators = &state.validators;
     let num_validators = validators.len() as u64;
 
-    // Verify each attestation's signature proof
+    // Verify each attestation's signature proof in parallel
     let aggregated_start = std::time::Instant::now();
-    for (attestation, aggregated_proof) in attestations.iter().zip(attestation_signatures) {
-        if attestation.aggregation_bits != aggregated_proof.participants {
-            return Err(StoreError::ParticipantsMismatch);
-        }
 
-        let slot: u32 = attestation.data.slot.try_into().expect("slot exceeds u32");
-        let message = attestation.data.hash_tree_root();
+    // Prepare verification inputs sequentially (cheap: bit checks + pubkey lookups)
+    let verification_inputs: Vec<_> = attestations
+        .iter()
+        .zip(attestation_signatures)
+        .map(|(attestation, aggregated_proof)| {
+            if attestation.aggregation_bits != aggregated_proof.participants {
+                return Err(StoreError::ParticipantsMismatch);
+            }
 
-        // Collect public keys with bounds check in a single pass
-        let public_keys: Vec<_> = validator_indices(&attestation.aggregation_bits)
-            .map(|vid| {
-                if vid >= num_validators {
-                    return Err(StoreError::InvalidValidatorIndex);
-                }
-                validators[vid as usize]
-                    .get_pubkey()
-                    .map_err(|_| StoreError::PubkeyDecodingFailed(vid))
-            })
-            .collect::<Result<_, _>>()?;
+            let slot: u32 = attestation.data.slot.try_into().expect("slot exceeds u32");
+            let message = attestation.data.hash_tree_root();
+
+            let public_keys: Vec<_> = validator_indices(&attestation.aggregation_bits)
+                .map(|vid| {
+                    if vid >= num_validators {
+                        return Err(StoreError::InvalidValidatorIndex);
+                    }
+                    validators[vid as usize]
+                        .get_pubkey()
+                        .map_err(|_| StoreError::PubkeyDecodingFailed(vid))
+                })
+                .collect::<Result<_, _>>()?;
 
-        let verification_result = {
-            let _timing = metrics::time_pq_sig_aggregated_signatures_verification();
-            verify_aggregated_signature(&aggregated_proof.proof_data, public_keys, &message, slot)
-        };
-        match verification_result {
-            Ok(()) => metrics::inc_pq_sig_aggregated_signatures_valid(),
-            Err(e) => {
-                metrics::inc_pq_sig_aggregated_signatures_invalid();
-                return Err(StoreError::AggregateVerificationFailed(e));
+            Ok((&aggregated_proof.proof_data, public_keys, message, slot))
+        })
+        .collect::<Result<_, StoreError>>()?;
+
+    // Run expensive signature verification in parallel.
+    // into_par_iter() moves each tuple, avoiding a clone of public_keys.
+    use rayon::prelude::*;
+    verification_inputs.into_par_iter().try_for_each(
+        |(proof_data, public_keys, message, slot)| {
+            let result = {
+                let _timing = metrics::time_pq_sig_aggregated_signatures_verification();
+                verify_aggregated_signature(proof_data, public_keys, &message, slot)
+            };
+            match result {
+                Ok(()) => {
+                    metrics::inc_pq_sig_aggregated_signatures_valid();
+                    Ok(())
+                }
+                Err(e) => {
+                    metrics::inc_pq_sig_aggregated_signatures_invalid();
+                    Err(StoreError::AggregateVerificationFailed(e))
+                }
             }
-        }
-    }
+        },
+    )?;
+
     let aggregated_elapsed = aggregated_start.elapsed();
 
     let proposer_start = std::time::Instant::now();

crates/blockchain/src/store.rs

@@ -142,8 +142,18 @@ pub async fn publish_attestation(server: &mut P2PServer, attestation: SignedAtte
         .cloned()
         .unwrap_or_else(|| attestation_subnet_topic(subnet_id));
 
-    // Publish to the attestation subnet topic
-    server.swarm_handle.publish(topic, compressed);
+    // Publish to the attestation subnet topic.
+    // Aggregators are subscribed to the subnet, so gossipsub uses mesh (not fanout).
+    // In small networks where no other node subscribes, this returns
+    // NoPeersSubscribedToTopic. Use best-effort to suppress the expected warning;
+    // the aggregator already self-delivered its attestation locally.
+    if server.is_aggregator {
+        server
+            .swarm_handle
+            .publish_ignore_no_peers(topic, compressed);
+    } else {
+        server.swarm_handle.publish(topic, compressed);
+    }
     info!(
         %slot,
         validator,

crates/net/p2p/src/gossipsub/handler.rs

@@ -92,6 +92,7 @@ pub struct BuiltSwarm {
     pub(crate) block_topic: libp2p::gossipsub::IdentTopic,
     pub(crate) aggregation_topic: libp2p::gossipsub::IdentTopic,
     pub(crate) bootnode_addrs: HashMap<PeerId, Multiaddr>,
+    pub(crate) is_aggregator: bool,
 }
 
 /// Build and configure the libp2p swarm, dial bootnodes, subscribe to topics.
@@ -257,6 +258,7 @@ pub fn build_swarm(
         block_topic,
         aggregation_topic,
         bootnode_addrs,
+        is_aggregator: config.is_aggregator,
     })
 }
 
@@ -280,6 +282,7 @@ impl P2P {
             attestation_committee_count: built.attestation_committee_count,
             block_topic: built.block_topic,
             aggregation_topic: built.aggregation_topic,
+            is_aggregator: built.is_aggregator,
             connected_peers: HashSet::new(),
             pending_requests: HashMap::new(),
             request_id_map: HashMap::new(),
@@ -314,6 +317,7 @@ pub struct P2PServer {
     pub(crate) attestation_committee_count: u64,
     pub(crate) block_topic: libp2p::gossipsub::IdentTopic,
     pub(crate) aggregation_topic: libp2p::gossipsub::IdentTopic,
+    pub(crate) is_aggregator: bool,
 
     pub(crate) connected_peers: HashSet<PeerId>,
     pub(crate) pending_requests: HashMap<H256, PendingRequest>,

crates/net/p2p/src/lib.rs

@@ -1,6 +1,7 @@
 use libp2p::{
     Multiaddr, PeerId, StreamProtocol,
     futures::StreamExt,
+    gossipsub::PublishError,
     request_response::{self, OutboundRequestId},
     swarm::SwarmEvent,
 };
@@ -13,6 +14,8 @@ pub enum SwarmCommand {
     Publish {
         topic: libp2p::gossipsub::IdentTopic,
         data: Vec<u8>,
+        /// When true, suppress NoPeersSubscribedToTopic errors (other errors still warn).
+        ignore_no_peers: bool,
     },
     Dial(Multiaddr),
     SendRequest {
@@ -37,7 +40,25 @@ impl SwarmHandle {
     pub fn publish(&self, topic: libp2p::gossipsub::IdentTopic, data: Vec<u8>) {
         let _ = self
             .cmd_tx
-            .send(SwarmCommand::Publish { topic, data })
+            .send(SwarmCommand::Publish {
+                topic,
+                data,
+                ignore_no_peers: false,
+            })
+            .inspect_err(|_| warn!("Swarm adapter closed, cannot publish"));
+    }
+
+    /// Publish, suppressing NoPeersSubscribedToTopic errors. Used when the sender
+    /// is also subscribed to the topic (e.g., aggregator publishing its own
+    /// attestation to a subnet it subscribes to) and no other peer subscribes.
+    pub fn publish_ignore_no_peers(&self, topic: libp2p::gossipsub::IdentTopic, data: Vec<u8>) {
+        let _ = self
+            .cmd_tx
+            .send(SwarmCommand::Publish {
+                topic,
+                data,
+                ignore_no_peers: true,
+            })
             .inspect_err(|_| warn!("Swarm adapter closed, cannot publish"));
     }
 
@@ -123,12 +144,17 @@ async fn swarm_loop(
 
 fn execute_command(swarm: &mut libp2p::Swarm<Behaviour>, cmd: SwarmCommand) {
     match cmd {
-        SwarmCommand::Publish { topic, data } => {
-            let _ = swarm
-                .behaviour_mut()
-                .gossipsub
-                .publish(topic, data)
-                .inspect_err(|err| warn!(%err, "Swarm adapter: publish failed"));
+        SwarmCommand::Publish {
+            topic,
+            data,
+            ignore_no_peers,
+        } => {
+            let result = swarm.behaviour_mut().gossipsub.publish(topic, data);
+            if let Err(err) = result
+                && !(ignore_no_peers && matches!(err, PublishError::NoPeersSubscribedToTopic))
+            {
+                warn!(%err, "Swarm adapter: publish failed");
+            }
         }
         SwarmCommand::Dial(addr) => {
             let _ = swarm