feat: support for syncing pre-genesis blocks

.github/workflows/protobuf.yaml

@@ -72,4 +72,4 @@ jobs:
         with:
           github_token: ${{ github.token }}
       - name: buf breaking
-        run: buf breaking './after.binpb' --against './before.binpb' --config '{"version":"v1","breaking":{"use":["WIRE_JSON","WIRE"]}}' --error-format 'github-actions'
+        run: buf breaking './after.binpb' --against './before.binpb' --config '{"version":"v1","breaking":{"use":["WIRE_JSON"]}}' --error-format 'github-actions'

node/actors/bft/src/lib.rs

@@ -70,6 +70,13 @@ impl Config {
         anyhow::ensure!(genesis.protocol_version == validator::ProtocolVersion::CURRENT);
         genesis.verify().context("genesis().verify()")?;
 
+        if let Some(prev) = genesis.first_block.prev() {
+            tracing::info!("Waiting for the pre-genesis blocks to be persisted");
+            if let Err(ctx::Canceled) = self.block_store.wait_until_persisted(ctx, prev).await {
+                return Ok(());
+            }
+        }
+
         let cfg = Arc::new(self);
         let (leader, leader_send) = leader::StateMachine::new(ctx, cfg.clone(), pipe.send.clone());
         let (replica, replica_send) =

node/actors/bft/src/replica/block.rs

@@ -44,7 +44,7 @@ impl StateMachine {
         );
         self.config
             .block_store
-            .queue_block(ctx, block.clone())
+            .queue_block(ctx, block.clone().into())
             .await?;
         // For availability, replica should not proceed until it stores the block persistently.
         self.config

node/actors/bft/src/replica/tests.rs

@@ -224,7 +224,7 @@ async fn leader_prepare_invalid_payload() {
         util.replica
             .config
             .block_store
-            .queue_block(ctx, block)
+            .queue_block(ctx, block.into())
             .await
             .unwrap();
 

node/actors/bft/src/testonly/node.rs

@@ -34,7 +34,6 @@ pub(super) struct Node {
     pub(crate) net: network::Config,
     pub(crate) behavior: Behavior,
     pub(crate) block_store: Arc<storage::BlockStore>,
-    pub(crate) batch_store: Arc<storage::BatchStore>,
 }
 
 impl Node {

node/actors/bft/src/testonly/run.rs

@@ -15,7 +15,7 @@ use zksync_concurrency::{
     oneshot, scope,
 };
 use zksync_consensus_network::{self as network};
-use zksync_consensus_roles::validator;
+use zksync_consensus_roles::{validator, validator::testonly::Setup};
 use zksync_consensus_storage::{testonly::TestMemoryStorage, BlockStore};
 use zksync_consensus_utils::pipe;
 
@@ -114,26 +114,23 @@ impl Test {
     /// Run a test with the given parameters and a random network setup.
     pub(crate) async fn run(&self, ctx: &ctx::Ctx) -> Result<(), TestError> {
         let rng = &mut ctx.rng();
-        let setup = validator::testonly::Setup::new_with_weights(
-            rng,
-            self.nodes.iter().map(|(_, w)| *w).collect(),
-        );
+        let setup = Setup::new_with_weights(rng, self.nodes.iter().map(|(_, w)| *w).collect());
         let nets: Vec<_> = network::testonly::new_configs(rng, &setup, 1);
-        self.run_with_config(ctx, nets, &setup.genesis).await
+        self.run_with_config(ctx, nets, &setup).await
     }
 
     /// Run a test with the given parameters and network configuration.
     pub(crate) async fn run_with_config(
         &self,
         ctx: &ctx::Ctx,
         nets: Vec<Config>,
-        genesis: &validator::Genesis,
+        setup: &Setup,
     ) -> Result<(), TestError> {
         let mut nodes = vec![];
         let mut honest = vec![];
         scope::run!(ctx, |ctx, s| async {
             for (i, net) in nets.into_iter().enumerate() {
-                let store = TestMemoryStorage::new(ctx, genesis).await;
+                let store = TestMemoryStorage::new(ctx, setup).await;
                 s.spawn_bg(async { Ok(store.runner.run(ctx).await?) });
 
                 if self.nodes[i].0 == Behavior::Honest {
@@ -144,15 +141,14 @@ impl Test {
                     net,
                     behavior: self.nodes[i].0,
                     block_store: store.blocks,
-                    batch_store: store.batches,
                 });
             }
             assert!(!honest.is_empty());
             s.spawn_bg(async { Ok(run_nodes(ctx, &self.network, &nodes).await?) });
 
             // Run the nodes until all honest nodes store enough finalized blocks.
             assert!(self.blocks_to_finalize > 0);
-            let first = genesis.first_block;
+            let first = setup.genesis.first_block;
             let last = first + (self.blocks_to_finalize as u64 - 1);
             for store in &honest {
                 store.wait_until_queued(ctx, last).await?;
@@ -165,7 +161,7 @@ impl Test {
                 let want = honest[0].block(ctx, i).await?.context("missing block")?;
                 for store in &honest[1..] {
                     let got = store.block(ctx, i).await?.context("missing block")?;
-                    if want.payload != got.payload {
+                    if want.payload() != got.payload() {
                         return Err(TestError::BlockConflict);
                     }
                 }
@@ -189,11 +185,8 @@ async fn run_nodes_real(ctx: &ctx::Ctx, specs: &[Node]) -> anyhow::Result<()> {
     scope::run!(ctx, |ctx, s| async {
         let mut nodes = vec![];
         for (i, spec) in specs.iter().enumerate() {
-            let (node, runner) = network::testonly::Instance::new(
-                spec.net.clone(),
-                spec.block_store.clone(),
-                spec.batch_store.clone(),
-            );
+            let (node, runner) =
+                network::testonly::Instance::new(spec.net.clone(), spec.block_store.clone());
             s.spawn_bg(runner.run(ctx).instrument(tracing::info_span!("node", i)));
             nodes.push(node);
         }