BFT-475: Add BatchVotePublisher and expose it to the Executor (#137)

## What ❔

Adds a `BatchVotePublisher` which will be used in the `Executor` to push
L1 batch votes to the gossip layer.

## Why ❔

So that we can insert the nodes attestations over L1 batches into vote
collection, which [automatically gets
gossiped](https://github.com/matter-labs/era-consensus/blob/f696992d77aa788fb228c4068a49f97868d8b531/node/actors/network/src/gossip/runner.rs#L208-L223)
to connected peers in `Network::run_stream`.

Co-authored-by: Bruno França <[email protected]>

node/actors/executor/src/lib.rs

@@ -132,6 +132,12 @@ impl Executor {
             net.register_metrics();
             s.spawn(async { runner.run(ctx).await.context("Network stopped") });
 
+            if let Some(_attester) = self.attester {
+                tracing::info!("Running the node in attester mode.");
+                let _publisher = net.batch_vote_publisher();
+                // TODO: Start a background task that polls the store for new L1 batches and publishes votes.
+            }
+
             if let Some(debug_config) = self.config.debug_page {
                 s.spawn(async {
                     http::DebugPageServer::new(debug_config, net)

node/actors/network/src/gossip/batch_votes.rs

@@ -100,3 +100,22 @@ impl BatchVotesWatch {
         Ok(())
     }
 }
+
+/// Wrapper around [BatchVotesWatch] to publish votes over batches signed by an attester key.
+pub struct BatchVotesPublisher(pub(crate) Arc<BatchVotesWatch>);
+
+impl BatchVotesPublisher {
+    /// Sign an L1 batch and push it into the batch, which should cause it to be gossiped by the network.
+    pub async fn publish(
+        &self,
+        attesters: &attester::Committee,
+        attester: &attester::SecretKey,
+        batch: Batch,
+    ) -> anyhow::Result<()> {
+        if !attesters.contains(&attester.public()) {
+            return Ok(());
+        }
+        let attestation = attester.sign_msg(batch);
+        self.0.update(attesters, &[Arc::new(attestation)]).await
+    }
+}

node/actors/network/src/gossip/mod.rs

@@ -12,6 +12,7 @@
 //! Static connections constitute a rigid "backbone" of the gossip network, which is insensitive to
 //! eclipse attack. Dynamic connections are supposed to improve the properties of the gossip
 //! network graph (minimize its diameter, increase connectedness).
+pub use self::batch_votes::BatchVotesPublisher;
 use self::batch_votes::BatchVotesWatch;
 use crate::{gossip::ValidatorAddrsWatch, io, pool::PoolWatch, Config, MeteredStreamStats};
 use anyhow::Context as _;
@@ -44,7 +45,7 @@ pub(crate) struct Network {
     /// Current state of knowledge about validators' endpoints.
     pub(crate) validator_addrs: ValidatorAddrsWatch,
     /// Current state of knowledge about batch votes.
-    pub(crate) batch_votes: BatchVotesWatch,
+    pub(crate) batch_votes: Arc<BatchVotesWatch>,
     /// Block store to serve `get_block` requests from.
     pub(crate) block_store: Arc<BlockStore>,
     /// Batch store to serve `get_batch` requests from.
@@ -79,7 +80,7 @@ impl Network {
             ),
             outbound: PoolWatch::new(cfg.gossip.static_outbound.keys().cloned().collect(), 0),
             validator_addrs: ValidatorAddrsWatch::default(),
-            batch_votes: BatchVotesWatch::default(),
+            batch_votes: Arc::new(BatchVotesWatch::default()),
             batch_qc: HashMap::new(),
             last_viewed_qc: None,
             cfg,

node/actors/network/src/lib.rs

@@ -1,5 +1,6 @@
 //! Network actor maintaining a pool of outbound and inbound connections to other nodes.
 use anyhow::Context as _;
+use gossip::BatchVotesPublisher;
 use std::sync::Arc;
 use tracing::Instrument as _;
 use zksync_concurrency::{
@@ -73,6 +74,11 @@ impl Network {
         metrics::NetworkGauges::register(Arc::downgrade(self));
     }
 
+    /// Create a batch vote publisher to push attestations to gossip.
+    pub fn batch_vote_publisher(&self) -> BatchVotesPublisher {
+        BatchVotesPublisher(self.gossip.batch_votes.clone())
+    }
+
     /// Handles a dispatcher message.
     async fn handle_message(
         &self,