BFT-496: AttestationStatusWatch must have a value and AttestationStat…

…usRunner::init will set it

node/actors/executor/src/attestation.rs

@@ -58,12 +58,9 @@ impl AttesterRunner {
         self.status.mark_changed();
 
         loop {
-            let Some(batch_number) = sync::changed(ctx, &mut self.status)
+            let batch_number = sync::changed(ctx, &mut self.status)
                 .await?
-                .next_batch_to_attest
-            else {
-                continue;
-            };
+                .next_batch_to_attest;
 
             tracing::info!(%batch_number, "attestation status");
 

node/actors/executor/src/tests.rs

@@ -30,7 +30,7 @@ fn config(cfg: &network::Config) -> Config {
 /// The test executors below are not running with attesters, so we just create an [AttestationStatusWatch]
 /// that will never be updated.
 fn never_attest() -> Arc<AttestationStatusWatch> {
-    Arc::new(AttestationStatusWatch::default())
+    Arc::new(AttestationStatusWatch::new(attester::BatchNumber(0)))
 }
 
 fn validator(

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

@@ -10,9 +10,9 @@ use crate::watch::Watch;
 pub struct AttestationStatus {
     /// Next batch number where voting is expected.
     ///
-    /// Its value is `None` until the background process polling the main node
-    /// can establish a value to start from.
-    pub next_batch_to_attest: Option<attester::BatchNumber>,
+    /// The node is expected to poll the main node during initialization until
+    /// the batch to start from is established.
+    pub next_batch_to_attest: attester::BatchNumber,
 }
 
 /// The subscription over the attestation status which voters can monitor for change.
@@ -29,15 +29,14 @@ impl fmt::Debug for AttestationStatusWatch {
     }
 }
 
-impl Default for AttestationStatusWatch {
-    fn default() -> Self {
+impl AttestationStatusWatch {
+    /// Create a new watch going from a specific batch number.
+    pub fn new(next_batch_to_attest: attester::BatchNumber) -> Self {
         Self(Watch::new(AttestationStatus {
-            next_batch_to_attest: None,
+            next_batch_to_attest,
         }))
     }
-}
 
-impl AttestationStatusWatch {
     /// Subscribes to AttestationStatus updates.
     pub fn subscribe(&self) -> AttestationStatusReceiver {
         self.0.subscribe()
@@ -47,10 +46,10 @@ impl AttestationStatusWatch {
     pub async fn update(&self, next_batch_to_attest: attester::BatchNumber) {
         let this = self.0.lock().await;
         this.send_if_modified(|status| {
-            if status.next_batch_to_attest == Some(next_batch_to_attest) {
+            if status.next_batch_to_attest == next_batch_to_attest {
                 return false;
             }
-            status.next_batch_to_attest = Some(next_batch_to_attest);
+            status.next_batch_to_attest = next_batch_to_attest;
             true
         });
     }

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

@@ -171,12 +171,9 @@ impl Network {
 
         loop {
             // Wait until the status indicates that we're ready to sign the next batch.
-            let Some(next_batch_number) = sync::changed(ctx, &mut recv_status)
+            let next_batch_number = sync::changed(ctx, &mut recv_status)
                 .await?
-                .next_batch_to_attest
-            else {
-                continue;
-            };
+                .next_batch_to_attest;
 
             // Get rid of all previous votes. We don't expect this to go backwards without regenesis, which will involve a restart.
             self.batch_votes

node/actors/network/src/gossip/tests/fetch_blocks.rs

@@ -26,6 +26,7 @@ async fn test_simple() {
     scope::run!(ctx, |ctx, s| async {
         let store = TestMemoryStorage::new(ctx, &setup.genesis).await;
         s.spawn_bg(store.runner.run(ctx));
+
         let (_node, runner) = crate::testonly::Instance::new(
             cfg.clone(),
             store.blocks.clone(),

node/actors/network/src/testonly.rs

@@ -17,7 +17,7 @@ use zksync_concurrency::{
     ctx::{self, channel},
     io, limiter, net, scope, sync, time,
 };
-use zksync_consensus_roles::{node, validator};
+use zksync_consensus_roles::{attester, node, validator};
 use zksync_consensus_storage::{BatchStore, BlockStore};
 use zksync_consensus_utils::pipe;
 
@@ -199,7 +199,8 @@ impl Instance {
     ) -> (Self, InstanceRunner) {
         // Semantically we'd want this to be created at the same level as the stores,
         // but doing so would introduce a lot of extra cruft in setting up tests.
-        let attestation_status = Arc::new(AttestationStatusWatch::default());
+        let attestation_status =
+            Arc::new(AttestationStatusWatch::new(attester::BatchNumber::default()));
 
         let (actor_pipe, dispatcher_pipe) = pipe::new();
         let (net, net_runner) = Network::new(

node/tools/src/config.rs

@@ -264,8 +264,14 @@ impl Configs {
         let replica_store = store::RocksDB::open(self.app.genesis.clone(), &self.database).await?;
         let store = TestMemoryStorage::new(ctx, &self.app.genesis).await;
 
+        let next_batch = store
+            .batches
+            .next_batch_to_attest(ctx)
+            .await?
+            .unwrap_or_default();
+
         // We don't have an API to poll in this setup, we can only create a local store based attestation client.
-        let attestation_status = Arc::new(AttestationStatusWatch::default());
+        let attestation_status = Arc::new(AttestationStatusWatch::new(next_batch));
         let attestation_status_runner = AttestationStatusRunner::new_from_store(
             attestation_status.clone(),
             store.batches.clone(),