async fn run_inner(
        &mut self,
        mut stop_receiver: watch::Receiver<bool>,
    ) -> Result<Infallible, OrStopped> {
        let (cursor, pending_batch_params) = self.io.initialize().await?;
        self.output_handler.initialize(&cursor).await?;
        self.health_updater
            .update(StateKeeperHealthDetails::from(&cursor).into());
        tracing::info!(
            "Starting state keeper. Next l1 batch to seal: {}, next L2 block to seal: {}",
            cursor.l1_batch,
            cursor.next_l2_block
        );

        // Re-execute pending batch if it exists. Otherwise, initialize a new batch.
        let PendingBatchData {
            mut l1_batch_env,
            mut system_env,
            mut pubdata_params,
            pending_l2_blocks,
        } = match pending_batch_params {
            Some(params) => {
                tracing::info!(
                    "There exists a pending batch consisting of {} L2 blocks, the first one is {}",
                    params.pending_l2_blocks.len(),
                    params
                        .pending_l2_blocks
                        .first()
                        .context("expected at least one pending L2 block")?
                        .number
                );
                params
            }
            None => {
                tracing::info!("There is no open pending batch, starting a new empty batch");
                let (system_env, l1_batch_env, pubdata_params) = self
                    .wait_for_new_batch_env(&cursor, &mut stop_receiver)
                    .await
                    .stop_context("failed getting new batch params")?;
                PendingBatchData {
                    l1_batch_env,
                    pending_l2_blocks: Vec::new(),
                    system_env,
                    pubdata_params,
                }
            }
        };

        let protocol_version = system_env.version;
        let previous_batch_protocol_version = self
            .io
            .load_batch_version_id(l1_batch_env.number - 1)
            .await?;
        let mut updates_manager = UpdatesManager::new(
            &l1_batch_env,
            &system_env,
            pubdata_params,
            previous_batch_protocol_version,
        );
        let mut protocol_upgrade_tx: Option<ProtocolUpgradeTx> = self
            .load_protocol_upgrade_tx(&pending_l2_blocks, protocol_version, l1_batch_env.number)
            .await?;

        let mut batch_executor = self
            .create_batch_executor(
                l1_batch_env.clone(),
                system_env.clone(),
                pubdata_params,
                &stop_receiver,
            )
            .await?;
        self.restore_state(
            &mut *batch_executor,
            &mut updates_manager,
            pending_l2_blocks,
            &stop_receiver,
        )
        .await?;

        let mut l1_batch_seal_delta: Option<Instant> = None;
        while !is_canceled(&stop_receiver) {
            // This function will run until the batch can be sealed.
            self.process_l1_batch(
                &mut *batch_executor,
                &mut updates_manager,
                protocol_upgrade_tx,
                &stop_receiver,
            )
            .await?;
            assert!(!updates_manager.has_next_block_params());

            // Finish current batch with an empty block.
            if !updates_manager.l2_block.executed_transactions.is_empty() {
                self.seal_l2_block(&updates_manager).await?;
                // We've sealed the L2 block that we had, but we still need to set up the timestamp
                // for the fictive L2 block.
                let new_l2_block_params = self
                    .wait_for_new_l2_block_params(&updates_manager, &stop_receiver)
                    .await?;
                Self::set_l2_block_params(&mut updates_manager, new_l2_block_params);
                Self::start_next_l2_block(&mut updates_manager, &mut *batch_executor).await?;
            }

            let (finished_batch, _) = batch_executor.finish_batch().await?;
            let sealed_batch_protocol_version = updates_manager.protocol_version();
            updates_manager.finish_batch(finished_batch);
            let mut next_cursor = updates_manager.io_cursor();
            let previous_batch_protocol_version = updates_manager.protocol_version();
            self.output_handler
                .handle_l1_batch(Arc::new(updates_manager))
                .await
                .with_context(|| format!("failed sealing L1 batch {l1_batch_env:?}"))?;

            if let Some(delta) = l1_batch_seal_delta {
                L1_BATCH_METRICS.seal_delta.observe(delta.elapsed());
            }
            l1_batch_seal_delta = Some(Instant::now());

            // Start the new batch.
            next_cursor.l1_batch += 1;
            (system_env, l1_batch_env, pubdata_params) = self
                .wait_for_new_batch_env(&next_cursor, &mut stop_receiver)
                .await?;
            updates_manager = UpdatesManager::new(
                &l1_batch_env,
                &system_env,
                pubdata_params,
                previous_batch_protocol_version,
            );
            batch_executor = self
                .create_batch_executor(
                    l1_batch_env.clone(),
                    system_env.clone(),
                    pubdata_params,
                    &stop_receiver,
                )
                .await?;

            let version_changed = system_env.version != sealed_batch_protocol_version;
            protocol_upgrade_tx = if version_changed {
                self.load_upgrade_tx(system_env.version).await?
            } else {
                None
            };
        }
        Err(OrStopped::Stopped)
    }
Fallible version of `run` routine that allows to easily exit upon cancellation.
core/node/state_keeper/src/keeper.rs