pub async fn new(
        pool: ConnectionPool<Core>,
        rocksdb_path: PathBuf,
        chain_id: L2ChainId,
        first_processed_batch: L1BatchNumber,
        window_size: u32,
    ) -> anyhow::Result<(Self, ProtectiveReadsWriterTasks)> {
        let io = ProtectiveReadsIo {
            first_processed_batch,
            window_size,
        };
        let (loader, loader_task) =
            VmRunnerStorage::new(pool.clone(), rocksdb_path, io.clone(), chain_id).await?;
        let output_handler_factory = ProtectiveReadsOutputHandlerFactory { pool: pool.clone() };
        let (output_handler_factory, output_handler_factory_task) =
            ConcurrentOutputHandlerFactory::new(pool.clone(), io.clone(), output_handler_factory);
        let batch_processor = MainBatchExecutorFactory::<()>::new(false);
        let vm_runner = VmRunner::new(
            pool,
            Arc::new(io),
            Arc::new(loader),
            Arc::new(output_handler_factory),
            Box::new(batch_processor),
        );
        Ok((
            Self { vm_runner },
            ProtectiveReadsWriterTasks {
                loader_task,
                output_handler_factory_task,
            },
        ))
    }
Create a new protective reads writer from the provided DB parameters and window size which
regulates how many batches this component can handle at the same time.
core/node/vm_runner/src/impls/protective_reads.rs