feat!: Use protocol version v22 as latest (#1432)

## What ❔

- Use protocol version v22 as latest
- Migration for syncing v22 batches/miniblock for ENs.

## Why ❔

- There was a protocol upgrade v22
- We had to update protocol version for already sealed batches in main
node DB. ENs should resync their data.

## Checklist

<!-- Check your PR fulfills the following items. -->
<!-- For draft PRs check the boxes as you complete them. -->

- [ ] PR title corresponds to the body of PR (we generate changelog
entries from PRs).
- [ ] Tests for the changes have been added / updated.
- [ ] Documentation comments have been added / updated.
- [ ] Code has been formatted via `zk fmt` and `zk lint`.
- [ ] Spellcheck has been run via `zk spellcheck`.
- [ ] Linkcheck has been run via `zk linkcheck`.

core/bin/external_node/src/main.rs

@@ -1,4 +1,4 @@
-use std::{sync::Arc, time::Duration};
+use std::{future, sync::Arc, time::Duration};
 
 use anyhow::Context as _;
 use clap::Parser;
@@ -51,6 +51,7 @@ mod config;
 mod helpers;
 mod init;
 mod metrics;
+mod version_sync_task;
 
 const RELEASE_MANIFEST: &str = include_str!("../../../../.github/release-please/manifest.json");
 
@@ -540,10 +541,24 @@ async fn main() -> anyhow::Result<()> {
     );
     // Start scraping Postgres metrics before store initialization as well.
     let metrics_pool = connection_pool.clone();
-    let mut task_handles = vec![tokio::spawn(async move {
-        PostgresMetrics::run_scraping(metrics_pool, Duration::from_secs(60)).await;
-        Ok(())
-    })];
+    let version_sync_task_pool = connection_pool.clone();
+    let version_sync_task_main_node_client = main_node_client.clone();
+    let mut task_handles = vec![
+        tokio::spawn(async move {
+            PostgresMetrics::run_scraping(metrics_pool, Duration::from_secs(60)).await;
+            Ok(())
+        }),
+        tokio::spawn(async move {
+            version_sync_task::sync_versions(
+                version_sync_task_pool,
+                version_sync_task_main_node_client,
+            )
+            .await?;
+            future::pending::<()>().await;
+            // ^ Since this is run as a task, we don't want it to exit on success (this would shut down the node).
+            Ok(())
+        }),
+    ];
 
     // Make sure that the node storage is initialized either via genesis or snapshot recovery.
     ensure_storage_initialized(

core/bin/external_node/src/version_sync_task.rs

@@ -0,0 +1,132 @@
+use std::cmp::Ordering;
+
+use anyhow::Context;
+use zksync_basic_types::{L1BatchNumber, MiniblockNumber};
+use zksync_dal::{ConnectionPool, Server, ServerDals};
+use zksync_types::ProtocolVersionId;
+use zksync_web3_decl::{
+    jsonrpsee::http_client::HttpClient,
+    namespaces::{EnNamespaceClient, ZksNamespaceClient},
+};
+
+pub async fn get_l1_batch_remote_protocol_version(
+    main_node_client: &HttpClient,
+    l1_batch_number: L1BatchNumber,
+) -> anyhow::Result<Option<ProtocolVersionId>> {
+    let Some((miniblock, _)) = main_node_client
+        .get_miniblock_range(l1_batch_number)
+        .await?
+    else {
+        return Ok(None);
+    };
+    let sync_block = main_node_client
+        .sync_l2_block(MiniblockNumber(miniblock.as_u32()), false)
+        .await?;
+    Ok(sync_block.map(|b| b.protocol_version))
+}
+
+// Synchronizes protocol version in `l1_batches` and `miniblocks` tables between EN and main node.
+pub async fn sync_versions(
+    connection_pool: ConnectionPool<Server>,
+    main_node_client: HttpClient,
+) -> anyhow::Result<()> {
+    tracing::info!("Starting syncing protocol version of blocks");
+
+    let mut connection = connection_pool.access_storage().await?;
+
+    // Load the first local batch number with version 22.
+    let Some(local_first_v22_l1_batch) = connection
+        .blocks_dal()
+        .get_first_l1_batch_number_for_version(ProtocolVersionId::Version22)
+        .await?
+    else {
+        return Ok(());
+    };
+    tracing::info!("First local v22 batch is #{local_first_v22_l1_batch}");
+
+    // Find the first remote batch with version 22, assuming it's less then or equal than local one.
+    // Uses binary search.
+
+    let mut left_bound = L1BatchNumber(0);
+    let mut right_bound = local_first_v22_l1_batch;
+
+    let right_bound_remote_version =
+        get_l1_batch_remote_protocol_version(&main_node_client, right_bound).await?;
+    if right_bound_remote_version != Some(ProtocolVersionId::Version22) {
+        anyhow::bail!("Remote protocol versions should be v22 for the first local v22 batch, got {right_bound_remote_version:?}");
+    }
+
+    while left_bound < right_bound {
+        let mid_batch = L1BatchNumber((left_bound.0 + right_bound.0) / 2);
+        let (mid_miniblock, _) = connection
+            .blocks_dal()
+            .get_miniblock_range_of_l1_batch(mid_batch)
+            .await
+            .with_context(|| format!("Failed to get miniblock range for L1 batch #{mid_batch}"))?
+            .with_context(|| {
+                format!("Postgres is inconsistent: missing miniblocks for L1 batch #{mid_batch}")
+            })?;
+        let mid_protocol_version = main_node_client
+            .sync_l2_block(mid_miniblock, false)
+            .await?
+            .with_context(|| format!("Main node missing data about miniblock #{mid_miniblock}"))?
+            .protocol_version;
+
+        match mid_protocol_version.cmp(&ProtocolVersionId::Version22) {
+            Ordering::Less => {
+                left_bound = mid_batch + 1;
+            }
+            Ordering::Equal => {
+                right_bound = mid_batch;
+            }
+            Ordering::Greater => {
+                anyhow::bail!("Unexpected remote protocol version: {mid_protocol_version:?} for miniblock #{mid_miniblock}");
+            }
+        }
+    }
+
+    let remote_first_v22_l1_batch = left_bound;
+    let (remote_first_v22_miniblock, _) = connection
+        .blocks_dal()
+        .get_miniblock_range_of_l1_batch(remote_first_v22_l1_batch)
+        .await
+        .with_context(|| format!("Failed to get miniblock range for L1 batch #{remote_first_v22_l1_batch}"))?
+        .with_context(|| {
+            format!("Postgres is inconsistent: missing miniblocks for L1 batch #{remote_first_v22_l1_batch}")
+        })?;
+
+    let mut transaction = connection.start_transaction().await?;
+
+    tracing::info!(
+        "Setting version 22 for batches {remote_first_v22_l1_batch}..={local_first_v22_l1_batch}"
+    );
+    transaction
+        .blocks_dal()
+        .reset_protocol_version_for_l1_batches(
+            remote_first_v22_l1_batch..=local_first_v22_l1_batch,
+            ProtocolVersionId::Version22,
+        )
+        .await?;
+
+    let (local_first_v22_miniblock, _) = transaction
+        .blocks_dal()
+        .get_miniblock_range_of_l1_batch(local_first_v22_l1_batch)
+        .await
+        .with_context(|| format!("Failed to get miniblock range for L1 batch #{local_first_v22_l1_batch}"))?
+        .with_context(|| {
+            format!("Postgres is inconsistent: missing miniblocks for L1 batch #{local_first_v22_l1_batch}")
+        })?;
+
+    tracing::info!("Setting version 22 for miniblocks {remote_first_v22_miniblock}..={local_first_v22_miniblock}");
+    transaction
+        .blocks_dal()
+        .reset_protocol_version_for_miniblocks(
+            remote_first_v22_miniblock..=local_first_v22_miniblock,
+            ProtocolVersionId::Version22,
+        )
+        .await?;
+
+    transaction.commit().await?;
+
+    Ok(())
+}

core/lib/basic_types/src/protocol_version.rs

@@ -44,15 +44,16 @@ pub enum ProtocolVersionId {
     Version20,
     Version21,
     Version22,
+    Version23,
 }
 
 impl ProtocolVersionId {
     pub fn latest() -> Self {
-        Self::Version21
+        Self::Version22
     }
 
     pub fn next() -> Self {
-        Self::Version22
+        Self::Version23
     }
 
     /// Returns VM version to be used by API for this protocol version.
@@ -82,6 +83,7 @@ impl ProtocolVersionId {
             ProtocolVersionId::Version20 => VmVersion::Vm1_4_1,
             ProtocolVersionId::Version21 => VmVersion::Vm1_4_2,
             ProtocolVersionId::Version22 => VmVersion::Vm1_4_2,
+            ProtocolVersionId::Version23 => VmVersion::Vm1_4_2,
         }
     }
 
@@ -170,15 +172,16 @@ pub enum FriProtocolVersionId {
     Version20,
     Version21,
     Version22,
+    Version23,
 }
 
 impl FriProtocolVersionId {
     pub fn latest() -> Self {
-        Self::Version21
+        Self::Version22
     }
 
     pub fn next() -> Self {
-        Self::Version22
+        Self::Version23
     }
 }
 
@@ -214,6 +217,7 @@ impl From<ProtocolVersionId> for FriProtocolVersionId {
             ProtocolVersionId::Version20 => FriProtocolVersionId::Version20,
             ProtocolVersionId::Version21 => FriProtocolVersionId::Version21,
             ProtocolVersionId::Version22 => FriProtocolVersionId::Version22,
+            ProtocolVersionId::Version23 => FriProtocolVersionId::Version23,
         }
     }
 }
@@ -280,6 +284,7 @@ impl From<ProtocolVersionId> for VmVersion {
             ProtocolVersionId::Version20 => VmVersion::Vm1_4_1,
             ProtocolVersionId::Version21 => VmVersion::Vm1_4_2,
             ProtocolVersionId::Version22 => VmVersion::Vm1_4_2,
+            ProtocolVersionId::Version23 => VmVersion::Vm1_4_2,
         }
     }
 }

core/lib/dal/src/blocks_dal.rs

@@ -2,6 +2,7 @@ use std::{
     collections::HashMap,
     convert::{Into, TryInto},
     ops,
+    ops::RangeInclusive,
 };
 
 use anyhow::Context as _;
@@ -2146,6 +2147,71 @@ impl BlocksDal<'_, '_> {
         .await?
         .map(|row| row.virtual_blocks as u32))
     }
+
+    pub async fn get_first_l1_batch_number_for_version(
+        &mut self,
+        protocol_version: ProtocolVersionId,
+    ) -> sqlx::Result<Option<L1BatchNumber>> {
+        Ok(sqlx::query!(
+            r#"
+            SELECT
+                MIN(number) AS "min?"
+            FROM
+                l1_batches
+            WHERE
+                protocol_version = $1
+            "#,
+            protocol_version as i32
+        )
+        .fetch_optional(self.storage.conn())
+        .await?
+        .and_then(|row| row.min)
+        .map(|min| L1BatchNumber(min as u32)))
+    }
+
+    pub async fn reset_protocol_version_for_l1_batches(
+        &mut self,
+        l1_batch_range: RangeInclusive<L1BatchNumber>,
+        protocol_version: ProtocolVersionId,
+    ) -> sqlx::Result<()> {
+        sqlx::query!(
+            r#"
+            UPDATE l1_batches
+            SET
+                protocol_version = $1
+            WHERE
+                number BETWEEN $2 AND $3
+            "#,
+            protocol_version as i32,
+            i64::from(l1_batch_range.start().0),
+            i64::from(l1_batch_range.end().0),
+        )
+        .execute(self.storage.conn())
+        .await?;
+        Ok(())
+    }
+
+    pub async fn reset_protocol_version_for_miniblocks(
+        &mut self,
+        miniblock_range: RangeInclusive<MiniblockNumber>,
+        protocol_version: ProtocolVersionId,
+    ) -> sqlx::Result<()> {
+        sqlx::query!(
+            r#"
+            UPDATE miniblocks
+            SET
+                protocol_version = $1
+            WHERE
+                number BETWEEN $2 AND $3
+            "#,
+            protocol_version as i32,
+            i64::from(miniblock_range.start().0),
+            i64::from(miniblock_range.end().0),
+        )
+        .execute(self.storage.conn())
+        .await?;
+        Ok(())
+    }
 }
 
 /// Temporary methods for migrating `fee_account_address`.

core/lib/zksync_core/src/api_server/tx_sender/mod.rs

@@ -92,7 +92,9 @@ impl MultiVMBaseSystemContracts {
             ProtocolVersionId::Version18 => self.post_boojum,
             ProtocolVersionId::Version19 => self.post_allowlist_removal,
             ProtocolVersionId::Version20 => self.post_1_4_1,
-            ProtocolVersionId::Version21 | ProtocolVersionId::Version22 => self.post_1_4_2,
+            ProtocolVersionId::Version21
+            | ProtocolVersionId::Version22
+            | ProtocolVersionId::Version23 => self.post_1_4_2,
         }
     }
 }