pub async fn run(self, stop_receiver: &watch::Receiver<bool>) -> anyhow::Result<()> {
        // Join handles for asynchronous tasks that are being run in the background
        let mut task_handles: Vec<(L1BatchNumber, JoinHandle<anyhow::Result<()>>)> = Vec::new();
        let mut next_batch = self
            .io
            .latest_processed_batch(&mut self.pool.connection_tagged("vm_runner").await?)
            .await?
            + 1;
        loop {
            if *stop_receiver.borrow() {
                tracing::info!("VM runner was interrupted");
                return Ok(());
            }

            // Traverse all handles and filter out tasks that have been finished. Also propagates
            // any panic/error that might have happened during the task's execution.
            let mut retained_handles = Vec::new();
            for (l1_batch_number, handle) in task_handles {
                if handle.is_finished() {
                    handle
                        .await
                        .with_context(|| format!("Processing batch #{} panicked", l1_batch_number))?
                        .with_context(|| format!("Failed to process batch #{}", l1_batch_number))?;
                } else {
                    retained_handles.push((l1_batch_number, handle));
                }
            }
            task_handles = retained_handles;
            METRICS
                .in_progress_l1_batches
                .set(task_handles.len() as u64);

            let last_ready_batch = self
                .io
                .last_ready_to_be_loaded_batch(&mut self.pool.connection_tagged("vm_runner").await?)
                .await?;
            METRICS.last_ready_batch.set(last_ready_batch.0.into());
            if next_batch > last_ready_batch {
                // Next batch is not ready to be processed yet
                tokio::time::sleep(SLEEP_INTERVAL).await;
                continue;
            }
            let handle = tokio::spawn(self.clone().process_batch(next_batch));
            task_handles.push((next_batch, handle));
            next_batch += 1;
        }
    }
Consumes VM runner to execute a loop that continuously pulls data from [`VmRunnerIo`] and
processes it.
core/node/vm_runner/src/process.rs