feat(vm-runner): add protective reads persistence flag for state keep…

…er (#2307)

## What ❔

Makes `protective-reads-writer` actually write the data when it hasn't
already been written by state keeper.

## Why ❔

Next iteration of VM runner

## Checklist

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

- [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`.

core/bin/zksync_server/src/node_builder.rs

@@ -199,7 +199,8 @@ impl MainNodeBuilder {
                 .l2_shared_bridge_addr
                 .context("L2 shared bridge address")?,
             sk_config.l2_block_seal_queue_capacity,
-        );
+        )
+        .with_protective_reads_persistence_enabled(sk_config.protective_reads_persistence_enabled);
         let mempool_io_layer = MempoolIOLayer::new(
             self.genesis_config.l2_chain_id,
             sk_config.clone(),

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

@@ -120,6 +120,12 @@ pub struct StateKeeperConfig {
     /// the recursion layers' circuits.
     pub max_circuits_per_batch: usize,
 
+    /// Configures whether to persist protective reads when persisting L1 batches in the state keeper.
+    /// Protective reads can be written asynchronously in VM runner instead.
+    /// By default, set to `true` as a temporary safety measure.
+    #[serde(default = "StateKeeperConfig::default_protective_reads_persistence_enabled")]
+    pub protective_reads_persistence_enabled: bool,
+
     // Base system contract hashes, required only for generating genesis config.
     // #PLA-811
     #[deprecated(note = "Use GenesisConfig::bootloader_hash instead")]
@@ -132,6 +138,10 @@ pub struct StateKeeperConfig {
 }
 
 impl StateKeeperConfig {
+    fn default_protective_reads_persistence_enabled() -> bool {
+        true
+    }
+
     /// Creates a config object suitable for use in unit tests.
     /// Values mostly repeat the values used in the localhost environment.
     pub fn for_tests() -> Self {
@@ -163,6 +173,7 @@ impl StateKeeperConfig {
             validation_computational_gas_limit: 300000,
             save_call_traces: true,
             max_circuits_per_batch: 24100,
+            protective_reads_persistence_enabled: true,
             bootloader_hash: None,
             default_aa_hash: None,
             l1_batch_commit_data_generator_mode: L1BatchCommitmentMode::Rollup,

core/lib/config/src/testonly.rs

@@ -175,6 +175,7 @@ impl Distribution<configs::chain::StateKeeperConfig> for EncodeDist {
             validation_computational_gas_limit: self.sample(rng),
             save_call_traces: self.sample(rng),
             max_circuits_per_batch: self.sample(rng),
+            protective_reads_persistence_enabled: self.sample(rng),
             // These values are not involved into files serialization skip them
             fee_account_addr: None,
             bootloader_hash: None,

core/lib/env_config/src/chain.rs

@@ -104,6 +104,7 @@ mod tests {
             )),
             l1_batch_commit_data_generator_mode,
             max_circuits_per_batch: 24100,
+            protective_reads_persistence_enabled: true,
         }
     }
 

core/lib/protobuf_config/src/chain.rs

@@ -78,6 +78,10 @@ impl ProtoRepr for proto::StateKeeper {
             max_circuits_per_batch: required(&self.max_circuits_per_batch)
                 .and_then(|x| Ok((*x).try_into()?))
                 .context("max_circuits_per_batch")?,
+            protective_reads_persistence_enabled: *required(
+                &self.protective_reads_persistence_enabled,
+            )
+            .context("protective_reads_persistence_enabled")?,
 
             // We need these values only for instantiating configs from environmental variables, so it's not
             // needed during the initialization from files
@@ -115,6 +119,7 @@ impl ProtoRepr for proto::StateKeeper {
             validation_computational_gas_limit: Some(this.validation_computational_gas_limit),
             save_call_traces: Some(this.save_call_traces),
             max_circuits_per_batch: Some(this.max_circuits_per_batch.try_into().unwrap()),
+            protective_reads_persistence_enabled: Some(this.protective_reads_persistence_enabled),
         }
     }
 }

core/lib/protobuf_config/src/proto/config/chain.proto

@@ -33,6 +33,7 @@ message StateKeeper {
   optional bool save_call_traces = 22; // required
   optional uint64 max_circuits_per_batch = 27; // required
   optional uint64 miniblock_max_payload_size = 28; // required
+  optional bool protective_reads_persistence_enabled = 29; // optional
   reserved 23; reserved "virtual_blocks_interval";
   reserved 24; reserved "virtual_blocks_per_miniblock";
   reserved 26; reserved "enum_index_migration_chunk_size";

core/node/node_framework/src/implementations/layers/state_keeper/output_handler.rs

@@ -87,7 +87,7 @@ impl WiringLayer for OutputHandlerLayer {
         }
         if !self.protective_reads_persistence_enabled {
             // **Important:** Disabling protective reads persistence is only sound if the node will never
-            // run a full Merkle tree.
+            // run a full Merkle tree OR an accompanying protective-reads-writer is being run.
             tracing::warn!("Disabling persisting protective reads; this should be safe, but is considered an experimental option at the moment");
             persistence = persistence.without_protective_reads();
         }

core/node/vm_runner/src/impls/protective_reads.rs

@@ -139,7 +139,7 @@ impl StateKeeperOutputHandler for ProtectiveReadsOutputHandler {
             .finished
             .as_ref()
             .context("L1 batch is not actually finished")?;
-        let (_, protective_reads): (Vec<StorageLog>, Vec<StorageLog>) = finished_batch
+        let (_, computed_protective_reads): (Vec<StorageLog>, Vec<StorageLog>) = finished_batch
             .final_execution_state
             .deduplicated_storage_logs
             .iter()
@@ -149,30 +149,48 @@ impl StateKeeperOutputHandler for ProtectiveReadsOutputHandler {
             .pool
             .connection_tagged("protective_reads_writer")
             .await?;
-        let mut expected_protective_reads = connection
+        let mut written_protective_reads = connection
             .storage_logs_dedup_dal()
             .get_protective_reads_for_l1_batch(updates_manager.l1_batch.number)
             .await?;
 
-        for protective_read in protective_reads {
-            let address = protective_read.key.address();
-            let key = protective_read.key.key();
-            if !expected_protective_reads.remove(&protective_read.key) {
+        if !written_protective_reads.is_empty() {
+            tracing::debug!(
+                l1_batch_number = %updates_manager.l1_batch.number,
+                "Protective reads have already been written, validating"
+            );
+            for protective_read in computed_protective_reads {
+                let address = protective_read.key.address();
+                let key = protective_read.key.key();
+                if !written_protective_reads.remove(&protective_read.key) {
+                    tracing::error!(
+                        l1_batch_number = %updates_manager.l1_batch.number,
+                        address = %address,
+                        key = %key,
+                        "VM runner produced a protective read that did not happen in state keeper"
+                    );
+                }
+            }
+            for remaining_read in written_protective_reads {
                 tracing::error!(
                     l1_batch_number = %updates_manager.l1_batch.number,
-                    address = %address,
-                    key = %key,
-                    "VM runner produced a protective read that did not happen in state keeper"
+                    address = %remaining_read.address(),
+                    key = %remaining_read.key(),
+                    "State keeper produced a protective read that did not happen in VM runner"
                 );
             }
-        }
-        for remaining_read in expected_protective_reads {
-            tracing::error!(
+        } else {
+            tracing::debug!(
                 l1_batch_number = %updates_manager.l1_batch.number,
-                address = %remaining_read.address(),
-                key = %remaining_read.key(),
-                "State keeper produced a protective read that did not happen in VM runner"
+                "Protective reads have not been written, writing"
             );
+            connection
+                .storage_logs_dedup_dal()
+                .insert_protective_reads(
+                    updates_manager.l1_batch.number,
+                    &computed_protective_reads,
+                )
+                .await?;
         }
 
         Ok(())

etc/env/file_based/general.yaml

@@ -91,6 +91,7 @@ state_keeper:
   validation_computational_gas_limit: 300000
   save_call_traces: true
   max_circuits_per_batch: 24100
+  protective_reads_persistence_enabled: true
 mempool:
   delay_interval: 100
   sync_interval_ms: 10