feat: add a flag for stage1->stage2 validium migration

core/bin/zksync_server/src/node_builder.rs

@@ -607,6 +607,7 @@ impl MainNodeBuilder {
         self.node.add_layer(DataAvailabilityDispatcherLayer::new(
             state_keeper_config,
             da_config,
+            self.contracts_config.clone(),
         ));
 
         Ok(self)

core/lib/config/src/configs/contracts.rs

@@ -57,6 +57,7 @@ pub struct ContractsConfig {
 
     pub chain_admin_addr: Option<Address>,
     pub l2_da_validator_addr: Option<Address>,
+    pub no_da_validium_l1_validator_addr: Option<Address>,
 }
 
 impl ContractsConfig {
@@ -82,6 +83,7 @@ impl ContractsConfig {
             ecosystem_contracts: Some(EcosystemContracts::for_tests()),
             chain_admin_addr: Some(Address::repeat_byte(0x18)),
             l2_da_validator_addr: Some(Address::repeat_byte(0x1a)),
+            no_da_validium_l1_validator_addr: Some(Address::repeat_byte(0x1b)),
         }
     }
 }

core/lib/config/src/configs/da_dispatcher.rs

@@ -11,6 +11,8 @@ pub const DEFAULT_MAX_ROWS_TO_DISPATCH: u32 = 3;
 pub const DEFAULT_MAX_RETRIES: u16 = 5;
 /// Use dummy value as inclusion proof instead of getting it from the client.
 pub const DEFAULT_USE_DUMMY_INCLUSION_DATA: bool = false;
+/// The default value for the inclusion_verification_transition_enabled flag.
+pub const DEFAULT_INCLUSION_VERIFICATION_TRANSITION_ENABLED: bool = false;
 
 #[derive(Debug, Clone, PartialEq, Deserialize)]
 pub struct DADispatcherConfig {
@@ -21,9 +23,11 @@ pub struct DADispatcherConfig {
     /// The maximum number of retries for the dispatch of a blob.
     pub max_retries: Option<u16>,
     /// Use dummy value as inclusion proof instead of getting it from the client.
-    // TODO: run a verification task to check if the L1 contract expects the inclusion proofs to
-    // avoid the scenario where contracts expect real proofs, and server is using dummy proofs.
     pub use_dummy_inclusion_data: Option<bool>,
+    /// This flag is used to signal that the transition to the full DA validators is in progress.
+    /// It will make the dispatcher stop polling for inclusion data and ensure all the old batches
+    /// have at least dummy inclusion data.
+    pub inclusion_verification_transition_enabled: Option<bool>,
 }
 
 impl DADispatcherConfig {
@@ -33,6 +37,9 @@ impl DADispatcherConfig {
             max_rows_to_dispatch: Some(DEFAULT_MAX_ROWS_TO_DISPATCH),
             max_retries: Some(DEFAULT_MAX_RETRIES),
             use_dummy_inclusion_data: Some(DEFAULT_USE_DUMMY_INCLUSION_DATA),
+            inclusion_verification_transition_enabled: Some(
+                DEFAULT_INCLUSION_VERIFICATION_TRANSITION_ENABLED,
+            ),
         }
     }
 
@@ -56,4 +63,9 @@ impl DADispatcherConfig {
         self.use_dummy_inclusion_data
             .unwrap_or(DEFAULT_USE_DUMMY_INCLUSION_DATA)
     }
+
+    pub fn inclusion_verification_transition_enabled(&self) -> bool {
+        self.inclusion_verification_transition_enabled
+            .unwrap_or(DEFAULT_INCLUSION_VERIFICATION_TRANSITION_ENABLED)
+    }
 }

core/lib/config/src/testonly.rs

@@ -270,6 +270,7 @@ impl Distribution<configs::ContractsConfig> for EncodeDist {
             l1_base_token_asset_id: self.sample_opt(|| rng.gen()),
             chain_admin_addr: self.sample_opt(|| rng.gen()),
             l2_da_validator_addr: self.sample_opt(|| rng.gen()),
+            no_da_validium_l1_validator_addr: self.sample_opt(|| rng.gen()),
         }
     }
 }
@@ -981,6 +982,7 @@ impl Distribution<configs::da_dispatcher::DADispatcherConfig> for EncodeDist {
             max_rows_to_dispatch: self.sample(rng),
             max_retries: self.sample(rng),
             use_dummy_inclusion_data: self.sample(rng),
+            inclusion_verification_transition_enabled: self.sample(rng),
         }
     }
 }

core/lib/constants/src/system_logs.rs

@@ -6,3 +6,6 @@ pub const STATE_DIFF_HASH_KEY_PRE_GATEWAY: u32 = 2;
 
 /// The key of the system log with value of the first blob linear hash for pre-gateway protocol versions
 pub const BLOB1_LINEAR_HASH_KEY_PRE_GATEWAY: u32 = 7;
+
+/// The key of the system log with value of the L2 DA validator address
+pub const L2_DA_VALIDATOR_OUTPUT_HASH_KEY: u32 = 6;

core/lib/dal/migrations/20250130233500_data_availability_add_l2_validator.down.sql

@@ -0,0 +1 @@
+ALTER TABLE data_availability DROP COLUMN l2_da_validator_address IF EXISTS ;

core/lib/dal/migrations/20250130233500_data_availability_add_l2_validator.up.sql

@@ -0,0 +1 @@
+ALTER TABLE data_availability ADD COLUMN l2_da_validator_address BYTEA;

core/lib/dal/src/data_availability_dal.rs

@@ -3,7 +3,9 @@ use zksync_db_connection::{
     error::DalResult,
     instrument::{InstrumentExt, Instrumented},
 };
-use zksync_types::{pubdata_da::DataAvailabilityBlob, L1BatchNumber};
+use zksync_types::{
+    l2_to_l1_log::L2ToL1Log, pubdata_da::DataAvailabilityBlob, Address, L1BatchNumber,
+};
 
 use crate::{
     models::storage_data_availability::{L1BatchDA, StorageDABlob},
@@ -24,17 +26,26 @@ impl DataAvailabilityDal<'_, '_> {
         number: L1BatchNumber,
         blob_id: &str,
         sent_at: chrono::NaiveDateTime,
+        l2_validator_address: Address,
     ) -> DalResult<()> {
         let update_result = sqlx::query!(
             r#"
             INSERT INTO
-            data_availability (l1_batch_number, blob_id, sent_at, created_at, updated_at)
+            data_availability (
+                l1_batch_number,
+                blob_id,
+                l2_da_validator_address,
+                sent_at,
+                created_at,
+                updated_at
+            )
             VALUES
-            ($1, $2, $3, NOW(), NOW())
+            ($1, $2, $3, $4, NOW(), NOW())
             ON CONFLICT DO NOTHING
             "#,
             i64::from(number.0),
             blob_id,
+            l2_validator_address.as_bytes(),
             sent_at,
         )
         .instrument("insert_l1_batch_da")
@@ -185,6 +196,7 @@ impl DataAvailabilityDal<'_, '_> {
             SELECT
                 number,
                 pubdata_input,
+                system_logs,
                 sealed_at
             FROM
                 l1_batches
@@ -216,7 +228,37 @@ impl DataAvailabilityDal<'_, '_> {
                 pubdata: row.pubdata_input.unwrap(),
                 l1_batch_number: L1BatchNumber(row.number as u32),
                 sealed_at: row.sealed_at.unwrap().and_utc(),
+                system_logs: row
+                    .system_logs
+                    .into_iter()
+                    .map(|raw_log| L2ToL1Log::from_slice(&raw_log))
+                    .collect(),
             })
             .collect())
     }
+
+    pub async fn set_dummy_inclusion_data_for_old_batches(
+        &mut self,
+        current_l2_da_validator: Address,
+    ) -> DalResult<()> {
+        sqlx::query!(
+            r#"
+            UPDATE data_availability
+            SET
+                inclusion_data = $1,
+                updated_at = NOW()
+            WHERE
+                inclusion_data IS NULL
+                AND l2_da_validator_address != $2
+            "#,
+            vec![],
+            current_l2_da_validator.as_bytes(),
+        )
+        .instrument("set_dummy_inclusion_data_for_old_batches")
+        .report_latency()
+        .execute(self.storage)
+        .await?;
+
+        Ok(())
+    }
 }

core/lib/dal/src/models/storage_data_availability.rs

@@ -1,5 +1,7 @@
 use chrono::{DateTime, NaiveDateTime, Utc};
-use zksync_types::{pubdata_da::DataAvailabilityBlob, L1BatchNumber};
+use zksync_types::{
+    l2_to_l1_log::L2ToL1Log, pubdata_da::DataAvailabilityBlob, Address, L1BatchNumber, H256,
+};
 
 /// Represents a blob in the data availability layer.
 #[derive(Debug, Clone)]
@@ -26,5 +28,24 @@ impl From<StorageDABlob> for DataAvailabilityBlob {
 pub struct L1BatchDA {
     pub pubdata: Vec<u8>,
     pub l1_batch_number: L1BatchNumber,
+    pub system_logs: Vec<L2ToL1Log>,
     pub sealed_at: DateTime<Utc>,
 }
+
+impl L1BatchDA {
+    pub fn l2_da_validator_address(&self) -> Address {
+        let value = self
+            .system_logs
+            .iter()
+            .find(|log| {
+                log.key
+                    == H256::from_low_u64_be(u64::from(
+                        zksync_system_constants::L2_DA_VALIDATOR_OUTPUT_HASH_KEY,
+                    ))
+            })
+            .unwrap() // The log is guaranteed to be present
+            .value;
+
+        value.into()
+    }
+}

core/lib/env_config/src/contracts.rs

@@ -98,6 +98,9 @@ mod tests {
             chain_admin_addr: Some(addr("0xdd6fa5c14e7550b4caf2aa2818d24c69cbc347ff")),
             l2_da_validator_addr: Some(addr("0xed6fa5c14e7550b4caf2aa2818d24c69cbc347ff")),
             l2_timestamp_asserter_addr: Some(addr("0x0000000000000000000000000000000000000002")),
+            no_da_validium_l1_validator_addr: Some(addr(
+                "0xbe8381498ED34E9c2EdB51Ecd778d71B225E26fb",
+            )),
         }
     }
 
@@ -132,6 +135,7 @@ CONTRACTS_PREDEPLOYED_L2_WRAPPED_BASE_TOKEN_ADDRESS="0x35ea7f92f4c5f433efe15284e
 CONTRACTS_CHAIN_ADMIN_ADDR="0xdd6fa5c14e7550b4caf2aa2818d24c69cbc347ff"
 CONTRACTS_L2_DA_VALIDATOR_ADDR="0xed6fa5c14e7550b4caf2aa2818d24c69cbc347ff"
 CONTRACTS_L2_TIMESTAMP_ASSERTER_ADDR="0x0000000000000000000000000000000000000002"
+CONTRACTS_NO_DA_VALIDIUM_L1_VALIDATOR_ADDR="0xbe8381498ED34E9c2EdB51Ecd778d71B225E26fb"
         "#;
         lock.set_env(config);
 

core/lib/env_config/src/da_dispatcher.rs

@@ -27,6 +27,7 @@ mod tests {
             max_rows_to_dispatch: Some(rows_limit),
             max_retries: Some(max_retries),
             use_dummy_inclusion_data: Some(true),
+            inclusion_verification_transition_enabled: None,
         }
     }
 

core/lib/protobuf_config/src/contracts.rs

@@ -139,6 +139,10 @@ impl ProtoRepr for proto::Contracts {
                 .map(|x| parse_h160(x))
                 .transpose()
                 .context("l2_da_validator_addr")?,
+            no_da_validium_l1_validator_addr: l1
+                .no_da_validium_l1_validator_addr
+                .as_ref()
+                .map(|x| parse_h160(x).expect("Invalid address")),
         })
     }
 
@@ -178,6 +182,9 @@ impl ProtoRepr for proto::Contracts {
                 base_token_addr: this.base_token_addr.map(|a| format!("{:?}", a)),
                 base_token_asset_id: this.l1_base_token_asset_id.map(|x| format!("{:?}", x)),
                 chain_admin_addr: this.chain_admin_addr.map(|a| format!("{:?}", a)),
+                no_da_validium_l1_validator_addr: this
+                    .no_da_validium_l1_validator_addr
+                    .map(|a| format!("{:?}", a)),
             }),
             l2: Some(proto::L2 {
                 testnet_paymaster_addr: this.l2_testnet_paymaster_addr.map(|a| format!("{:?}", a)),