Support for dynamic attester committee

node/actors/executor/src/io.rs

@@ -36,8 +36,8 @@ impl Dispatcher {
     }
 
     /// Method to start the IO dispatcher. It is simply a loop to receive messages from the actors and then forward them.
-    pub(super) async fn run(mut self, ctx: &ctx::Ctx) -> anyhow::Result<()> {
-        scope::run!(ctx, |ctx, s| async {
+    pub(super) async fn run(mut self, ctx: &ctx::Ctx) {
+        let _: ctx::OrCanceled<()> = scope::run!(ctx, |ctx, s| async {
             // Start a task to handle the messages from the consensus actor.
             s.spawn(async {
                 while let Ok(msg) = self.consensus_output.recv(ctx).await {
@@ -65,6 +65,6 @@ impl Dispatcher {
 
             Ok(())
         })
-        .await
+        .await;
     }
 }

node/actors/executor/src/lib.rs

@@ -2,20 +2,19 @@
 use crate::io::Dispatcher;
 use anyhow::Context as _;
 use network::http;
-pub use network::RpcConfig;
+pub use network::{gossip::attestation, RpcConfig};
 use std::{
     collections::{HashMap, HashSet},
     sync::Arc,
 };
 use zksync_concurrency::{ctx, limiter, net, scope, time};
 use zksync_consensus_bft as bft;
-use zksync_consensus_network::{self as network, gossip::AttestationStatusWatch};
-use zksync_consensus_roles::{attester, node, validator};
+use zksync_consensus_network as network;
+use zksync_consensus_roles::{node, validator};
 use zksync_consensus_storage::{BatchStore, BlockStore, ReplicaStore};
 use zksync_consensus_utils::pipe;
 use zksync_protobuf::kB;
 
-pub mod attestation;
 mod io;
 #[cfg(test)]
 mod tests;
@@ -31,13 +30,6 @@ pub struct Validator {
     pub payload_manager: Box<dyn bft::PayloadManager>,
 }
 
-/// Validator-related part of [`Executor`].
-#[derive(Debug)]
-pub struct Attester {
-    /// Consensus network configuration.
-    pub key: attester::SecretKey,
-}
-
 /// Config of the node executor.
 #[derive(Debug)]
 pub struct Config {
@@ -98,10 +90,9 @@ pub struct Executor {
     pub batch_store: Arc<BatchStore>,
     /// Validator-specific node data.
     pub validator: Option<Validator>,
-    /// Validator-specific node data.
-    pub attester: Option<Attester>,
-    /// Status showing where the main node wants attester to cast their votes.
-    pub attestation_status: Arc<AttestationStatusWatch>,
+    /// Attestation controller. Caller should actively configure the batch
+    /// for which the attestation votes should be collected.
+    pub attestation: Arc<attestation::Controller>,
 }
 
 impl Executor {
@@ -112,7 +103,6 @@ impl Executor {
             public_addr: self.config.public_addr.clone(),
             gossip: self.config.gossip(),
             validator_key: self.validator.as_ref().map(|v| v.key.clone()),
-            attester_key: self.attester.as_ref().map(|a| a.key.clone()),
             ping_timeout: Some(time::Duration::seconds(10)),
             max_block_size: self.config.max_payload_size.saturating_add(kB),
             max_batch_size: self.config.max_batch_size.saturating_add(kB),
@@ -137,35 +127,20 @@ impl Executor {
 
         tracing::debug!("Starting actors in separate threads.");
         scope::run!(ctx, |ctx, s| async {
-            s.spawn(async { dispatcher.run(ctx).await.context("IO Dispatcher stopped") });
+            s.spawn(async {
+                dispatcher.run(ctx).await;
+                Ok(())
+            });
             let (net, runner) = network::Network::new(
                 network_config,
                 self.block_store.clone(),
                 self.batch_store.clone(),
                 network_actor_pipe,
-                self.attestation_status.clone(),
+                self.attestation,
             );
             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 runner = attestation::AttesterRunner::new(
-                    self.block_store.clone(),
-                    self.batch_store.clone(),
-                    attester,
-                    net.batch_vote_publisher(),
-                    self.attestation_status.subscribe(),
-                    self.config.batch_poll_interval,
-                );
-                s.spawn(async {
-                    runner.run(ctx).await?;
-                    Ok(())
-                });
-            } else {
-                tracing::info!("Running the node in non-attester mode.");
-            }
-
             if let Some(debug_config) = self.config.debug_page {
                 s.spawn(async {
                     http::DebugPageServer::new(debug_config, net)