fix: self-deliver aggregator attestations to gossip_signatures (#270)

## Summary

- Aggregator now calls `on_gossip_attestation()` locally before
publishing to gossip, ensuring its own validator's attestation enters
`gossip_signatures` and is included in aggregated proofs
- Adds `publish_best_effort()` to suppress the expected
`NoPeersSubscribedToTopic` warning when the aggregator publishes to
attestation subnets it subscribes to
- Threads `is_aggregator` from `SwarmConfig` through `BuiltSwarm` into
`P2PServer` so the gossipsub handler can use best-effort publishing

## Context

Gossipsub does not deliver messages back to the sender. The aggregator
subscribes to the attestation subnet and publishes via mesh, but in
small devnets no other node subscribes to attestation subnets. This
causes the aggregator's own attestation publish to fail silently with
`NoPeersSubscribedToTopic`, meaning its vote never enters
`gossip_signatures`, is never aggregated, and is never included in
blocks.

The reference spec handles this explicitly in `_produce_attestations()`:

```python
# Process attestation locally before publishing.
#
# Gossipsub does not deliver messages back to the sender.
# Without local processing, the aggregator node never sees its own
# validator's attestation in gossip_signatures, reducing the
# aggregation count below the 2/3 safe-target threshold.
self.sync_service.store = self.sync_service.store.on_gossip_attestation(
    signed_attestation=signed_attestation,
    is_aggregator=is_aggregator_role,
)
```

## Test plan

- [x] `cargo build` passes
- [x] `cargo clippy --workspace -- -D warnings` passes
- [x] `cargo test --workspace --release --lib` passes (all 32 unit
tests)
- [ ] Deploy to devnet and verify:
- Validator 3's attestation appears in "Attestation processed" logs on
the aggregator
- Blocks from all proposers include aggregator's vote in aggregated
proofs (higher attestation counts)
  - No more `NoPeersSubscribedToTopic` warnings on the aggregator node
  - Finalization continues normally

Author: MegaRedHand

crates/blockchain/src/lib.rs

@@ -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/net/p2p/src/gossipsub/handler.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/lib.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/swarm_adapter.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