pub async fn run(self, stop_receiver: watch::Receiver<bool>) -> anyhow::Result<()> {
        const SLEEP_INTERVAL: Duration = Duration::from_millis(50);

        self.catchup_task.run(stop_receiver.clone()).await?;
        let mut rocksdb = try_stoppable!(self.rocksdb_cell.wait().await);
        loop {
            if *stop_receiver.borrow() {
                tracing::info!("`StorageSyncTask` was interrupted");
                return Ok(());
            }
            let mut conn = self.pool.connection_tagged(self.io.name()).await?;
            let latest_processed_batch = self.io.latest_processed_batch(&mut conn).await?;
            if rocksdb.next_l1_batch_number().await == latest_processed_batch + 1 {
                // RocksDB is already caught up, we might not need to do anything.
                // Just need to check that the memory diff is up-to-date in case this is a fresh start.
                let last_ready_batch = self.io.last_ready_to_be_loaded_batch(&mut conn).await?;
                let state = self.state.read().await;
                if last_ready_batch == latest_processed_batch
                    || state.storage.contains_key(&last_ready_batch)
                {
                    // No need to do anything, killing time until last processed batch is updated.
                    drop(conn);
                    drop(state);
                    tokio::time::sleep(SLEEP_INTERVAL).await;
                    continue;
                }
            }
            // We rely on the assumption that no one is holding storage access to a batch with
            // number less than `latest_processed_batch`. If they do, RocksDB synchronization below
            // will cause them to have an inconsistent view on DB which we consider to be an
            // undefined behavior.
            rocksdb = try_stoppable!(
                rocksdb
                    .synchronize(&mut conn, &stop_receiver, Some(latest_processed_batch))
                    .await
            );

            let mut state = self.state.write().await;
            state.rocksdb = Some(rocksdb.clone());
            state.l1_batch_number = latest_processed_batch;
            state
                .storage
                .retain(|l1_batch_number, _| l1_batch_number > &latest_processed_batch);
            let max_present = state
                .storage
                .last_entry()
                .map(|e| *e.key())
                .unwrap_or(latest_processed_batch);
            drop(state);
            let max_desired = self.io.last_ready_to_be_loaded_batch(&mut conn).await?;
            for l1_batch_number in max_present.0 + 1..=max_desired.0 {
                let latency = METRICS.storage_load_time.start();
                let l1_batch_number = L1BatchNumber(l1_batch_number);
                let Some(execute_data) = load_batch_execute_data(
                    &mut conn,
                    l1_batch_number,
                    &self.l1_batch_params_provider,
                    self.chain_id,
                )
                .await?
                else {
                    break;
                };
                let state_diff = conn
                    .storage_logs_dal()
                    .get_touched_slots_for_l1_batch(l1_batch_number)
                    .await?;
                let enum_index_diff = conn
                    .storage_logs_dedup_dal()
                    .initial_writes_for_batch(l1_batch_number)
                    .await?
                    .into_iter()
                    .collect::<HashMap<_, _>>();
                let factory_dep_diff = conn
                    .blocks_dal()
                    .get_l1_batch_factory_deps(l1_batch_number)
                    .await?;
                let diff = BatchDiff {
                    state_diff,
                    enum_index_diff,
                    factory_dep_diff,
                };

                let mut state = self.state.write().await;
                state
                    .storage
                    .insert(l1_batch_number, BatchData { execute_data, diff });
                drop(state);
                latency.observe();
            }
            drop(conn);
        }
    }
Block until RocksDB cache instance is caught up with Postgres and then continuously makes
sure that the new ready batches are loaded into the cache.

# Errors

Propagates RocksDB and Postgres errors.
core/node/vm_runner/src/storage.rs