feat(en): Fix operator address assignment for ENs (#688)

## What ❔

- Moves `fee_account_address` from the `l1_batches` table to the
`miniblocks` table using a combined SQL and programmatic migration.
- Replaces uses of `l1_gas_price`, `l2_fair_gas_price` and
`base_fee_per_gas` from `l1_batches` to the corresponding `miniblocks`
fields. Removes uses of `l1_batches.is_finished`.
- Updates `L1BatchHeader` / `MiniblockHeader` correspondingly.

## Why ❔

Currently, operator address is not persisted for miniblocks, but only
for L1 batches. This has several negative implications:

- ENs cannot be used to synchronize other ENs since
SyncBlock.operator_address they return may be bogus.
- We have nowhere to get the real operator address for gossip-based
syncing as well.

## Checklist

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

core/bin/external_node/src/config/mod.rs

@@ -416,10 +416,6 @@ impl PostgresConfig {
     }
 }
 
-fn read_operator_address() -> anyhow::Result<Address> {
-    Ok(std::env::var("EN_OPERATOR_ADDR")?.parse()?)
-}
-
 pub(crate) fn read_consensus_config() -> anyhow::Result<consensus::FetcherConfig> {
     let path = std::env::var("EN_CONSENSUS_CONFIG_PATH")
         .context("EN_CONSENSUS_CONFIG_PATH env variable is not set")?;
@@ -429,7 +425,6 @@ pub(crate) fn read_consensus_config() -> anyhow::Result<consensus::FetcherConfig
     let node_key: node::SecretKey = consensus::config::read_secret("EN_CONSENSUS_NODE_KEY")?;
     Ok(consensus::FetcherConfig {
         executor: cfg.executor_config(node_key),
-        operator_address: read_operator_address().context("read_operator_address()")?,
     })
 }
 

core/bin/external_node/src/main.rs

@@ -270,6 +270,8 @@ async fn init_tasks(
     let consistency_checker_handle = tokio::spawn(consistency_checker.run(stop_receiver.clone()));
 
     let updater_handle = task::spawn(batch_status_updater.run(stop_receiver.clone()));
+    let fee_address_migration_handle =
+        task::spawn(state_keeper.run_fee_address_migration(connection_pool.clone()));
     let sk_handle = task::spawn(state_keeper.run());
     let fee_params_fetcher_handle =
         tokio::spawn(fee_params_fetcher.clone().run(stop_receiver.clone()));
@@ -355,12 +357,13 @@ async fn init_tasks(
     task_handles.extend(cache_update_handle);
     task_handles.extend([
         sk_handle,
+        fee_address_migration_handle,
         fetcher_handle,
         updater_handle,
         tree_handle,
+        consistency_checker_handle,
         fee_params_fetcher_handle,
     ]);
-    task_handles.push(consistency_checker_handle);
 
     Ok((task_handles, stop_sender, healthcheck_handle, stop_receiver))
 }

core/bin/snapshots_creator/src/tests.rs

@@ -142,6 +142,7 @@ async fn create_miniblock(
         hash: H256::from_low_u64_be(u64::from(miniblock_number.0)),
         l1_tx_count: 0,
         l2_tx_count: 0,
+        fee_account_address: Address::repeat_byte(1),
         base_fee_per_gas: 0,
         gas_per_pubdata_limit: 0,
         batch_fee_input: Default::default(),
@@ -164,14 +165,7 @@ async fn create_l1_batch(
     l1_batch_number: L1BatchNumber,
     logs_for_initial_writes: &[StorageLog],
 ) {
-    let mut header = L1BatchHeader::new(
-        l1_batch_number,
-        0,
-        Address::default(),
-        Default::default(),
-        Default::default(),
-    );
-    header.is_finished = true;
+    let header = L1BatchHeader::new(l1_batch_number, 0, Default::default(), Default::default());
     conn.blocks_dal()
         .insert_mock_l1_batch(&header)
         .await

core/bin/zksync_server/src/config.rs

@@ -1,11 +1,6 @@
 use anyhow::Context as _;
 use zksync_consensus_roles::{node, validator};
 use zksync_core::consensus;
-use zksync_types::Address;
-
-fn read_operator_address() -> anyhow::Result<Address> {
-    Ok(std::env::var("CHAIN_STATE_KEEPER_FEE_ACCOUNT_ADDR")?.parse()?)
-}
 
 pub(crate) fn read_consensus_config() -> anyhow::Result<consensus::MainNodeConfig> {
     let path = std::env::var("CONSENSUS_CONFIG_PATH").context("CONSENSUS_CONFIG_PATH")?;
@@ -18,6 +13,5 @@ pub(crate) fn read_consensus_config() -> anyhow::Result<consensus::MainNodeConfi
     Ok(consensus::MainNodeConfig {
         executor: cfg.executor_config(node_key),
         validator: cfg.validator_config(validator_key),
-        operator_address: read_operator_address().context("read_operator_address()")?,
     })
 }