feat(state-keeper): implement asynchronous RocksDB cache

checks-config/era.dic

@@ -122,6 +122,7 @@ stdout
 GCS
 websocket
 struct
+struct's
 localhost
 TOML
 config

core/lib/dal/src/blocks_dal.rs

@@ -1850,6 +1850,24 @@ impl BlocksDal<'_, '_> {
         Ok(())
     }
 
+    async fn delete_logs_inner(
+        &mut self,
+        last_miniblock_to_keep: Option<MiniblockNumber>,
+    ) -> sqlx::Result<()> {
+        let block_number = last_miniblock_to_keep.map_or(-1, |number| number.0 as i64);
+        sqlx::query!(
+            r#"
+            DELETE FROM storage_logs
+            WHERE
+                miniblock_number > $1
+            "#,
+            block_number
+        )
+        .execute(self.storage.conn())
+        .await?;
+        Ok(())
+    }
+
     /// Returns sum of predicted gas costs on the given L1 batch range.
     /// Panics if the sum doesn't fit into `u32`.
     pub async fn get_l1_batches_predicted_gas(
@@ -2321,6 +2339,7 @@ impl BlocksDal<'_, '_> {
         self.delete_initial_writes_inner(None)
             .await
             .context("delete_initial_writes_inner()")?;
+        self.delete_logs_inner(None).await.context("no")?;
         Ok(())
     }
 }

core/lib/multivm/src/versions/vm_latest/types/internals/vm_state.rs

@@ -68,6 +68,7 @@ pub(crate) fn new_vm_state<S: WriteStorage, H: HistoryMode>(
     } else {
         // This is the scenario of either the first L2 block ever or
         // the first block after the upgrade for support of L2 blocks.
+        tracing::info!("Could not locate the last L2 block, presuming this is the first one");
         L2Block {
             number: l1_batch_env.first_l2_block.number.saturating_sub(1),
             timestamp: 0,

core/lib/multivm/src/versions/vm_latest/utils/l2_blocks.rs

@@ -62,6 +62,7 @@ pub fn load_last_l2_block<S: ReadStorage>(storage: StoragePtr<S>) -> Option<L2Bl
     let mut storage_ptr = storage.borrow_mut();
     let current_l2_block_info = storage_ptr.read_value(&current_l2_block_info_key);
     let (block_number, block_timestamp) = unpack_block_info(h256_to_u256(current_l2_block_info));
+    tracing::info!(block_number, block_timestamp, "Loaded last L2 block");
     let block_number = block_number as u32;
     if block_number == 0 {
         // The block does not exist yet

core/lib/state/src/postgres/mod.rs

@@ -382,6 +382,7 @@ impl<'a> PostgresStorage<'a> {
             .with_context(|| {
                 format!("failed resolving L1 batch number for miniblock #{block_number}")
             })?;
+        tracing::info!(?resolved, "Resolved L1 batch for Postgres storage");
         Ok(Self {
             rt_handle,
             connection,

core/lib/state/src/rocksdb/mod.rs

@@ -599,7 +599,7 @@ impl RocksdbStorage {
     /// # Panics
     ///
     /// Panics on RocksDB errors.
-    async fn l1_batch_number(&self) -> Option<L1BatchNumber> {
+    pub async fn l1_batch_number(&self) -> Option<L1BatchNumber> {
         let cf = StateKeeperColumnFamily::State;
         let db = self.db.clone();
         let number_bytes =

core/lib/zksync_core/src/state_keeper/batch_executor/main_executor.rs

@@ -11,16 +11,20 @@ use multivm::{
     MultiVMTracer, VmInstance,
 };
 use once_cell::sync::OnceCell;
-use tokio::sync::{mpsc, watch};
+use tokio::{
+    runtime::Handle,
+    sync::{mpsc, watch, RwLock},
+};
 use zksync_dal::ConnectionPool;
-use zksync_state::{RocksdbStorage, StorageView, WriteStorage};
+use zksync_state::{ReadStorage, StorageView, WriteStorage};
 use zksync_types::{vm_trace::Call, Transaction, U256};
 use zksync_utils::bytecode::CompressedBytecodeInfo;
 
 use super::{BatchExecutor, BatchExecutorHandle, Command, TxExecutionResult};
 use crate::{
     metrics::{InteractionType, TxStage, APP_METRICS},
     state_keeper::{
+        cached_storage::CachedStorage,
         metrics::{TxExecutionStage, BATCH_TIP_METRICS, EXECUTOR_METRICS, KEEPER_METRICS},
         types::ExecutionMetricsForCriteria,
     },
@@ -30,13 +34,11 @@ use crate::{
 /// Creates a "real" batch executor which maintains the VM (as opposed to the test builder which doesn't use the VM).
 #[derive(Debug, Clone)]
 pub struct MainBatchExecutor {
-    state_keeper_db_path: String,
-    pool: ConnectionPool,
     save_call_traces: bool,
     max_allowed_tx_gas_limit: U256,
     upload_witness_inputs_to_gcs: bool,
-    enum_index_migration_chunk_size: usize,
     optional_bytecode_compression: bool,
+    cached_storage: Arc<RwLock<CachedStorage>>,
 }
 
 impl MainBatchExecutor {
@@ -50,13 +52,15 @@ impl MainBatchExecutor {
         optional_bytecode_compression: bool,
     ) -> Self {
         Self {
-            state_keeper_db_path,
-            pool,
             save_call_traces,
             max_allowed_tx_gas_limit,
             upload_witness_inputs_to_gcs,
-            enum_index_migration_chunk_size,
             optional_bytecode_compression,
+            cached_storage: Arc::new(RwLock::new(CachedStorage::new(
+                pool,
+                state_keeper_db_path,
+                enum_index_migration_chunk_size,
+            ))),
         }
     }
 }
@@ -69,20 +73,6 @@ impl BatchExecutor for MainBatchExecutor {
         system_env: SystemEnv,
         stop_receiver: &watch::Receiver<bool>,
     ) -> Option<BatchExecutorHandle> {
-        let mut secondary_storage = RocksdbStorage::builder(self.state_keeper_db_path.as_ref())
-            .await
-            .expect("Failed initializing state keeper storage");
-        secondary_storage.enable_enum_index_migration(self.enum_index_migration_chunk_size);
-        let mut conn = self
-            .pool
-            .access_storage_tagged("state_keeper")
-            .await
-            .unwrap();
-        let secondary_storage = secondary_storage
-            .synchronize(&mut conn, stop_receiver)
-            .await
-            .expect("Failed synchronizing secondary state keeper storage")?;
-
         // Since we process `BatchExecutor` commands one-by-one (the next command is never enqueued
         // until a previous command is processed), capacity 1 is enough for the commands channel.
         let (commands_sender, commands_receiver) = mpsc::channel(1);
@@ -94,13 +84,22 @@ impl BatchExecutor for MainBatchExecutor {
         };
         let upload_witness_inputs_to_gcs = self.upload_witness_inputs_to_gcs;
 
+        let cached_storage = self.cached_storage.clone();
+        let stop_receiver = stop_receiver.clone();
         let handle = tokio::task::spawn_blocking(move || {
-            executor.run(
-                secondary_storage,
-                l1_batch_params,
-                system_env,
-                upload_witness_inputs_to_gcs,
-            )
+            let rt_handle = Handle::current();
+            let mut cached_storage = rt_handle.block_on(cached_storage.write());
+            if let Some(storage) = rt_handle
+                .block_on(cached_storage.access_storage(rt_handle.clone(), stop_receiver))
+                .unwrap()
+            {
+                executor.run(
+                    storage,
+                    l1_batch_params,
+                    system_env,
+                    upload_witness_inputs_to_gcs,
+                );
+            };
         });
         Some(BatchExecutorHandle {
             handle,
@@ -124,14 +123,18 @@ struct CommandReceiver {
 }
 
 impl CommandReceiver {
-    pub(super) fn run(
+    pub(super) fn run<S: ReadStorage>(
         mut self,
-        secondary_storage: RocksdbStorage,
+        secondary_storage: S,
         l1_batch_params: L1BatchEnv,
         system_env: SystemEnv,
         upload_witness_inputs_to_gcs: bool,
     ) {
-        tracing::info!("Starting executing batch #{:?}", &l1_batch_params.number);
+        tracing::info!(
+            first_l2_block = ?l1_batch_params.first_l2_block,
+            l1_batch_number = ?l1_batch_params.number,
+            "Starting executing batch",
+        );
 
         let storage_view = StorageView::new(secondary_storage).to_rc_ptr();