pub(super) async fn seal_l1_batch(
        &self,
        pool: ConnectionPool<Core>,
        l2_legacy_shared_bridge_addr: Option<Address>,
        insert_protective_reads: bool,
    ) -> anyhow::Result<()> {
        let started_at = Instant::now();
        let finished_batch = self
            .l1_batch
            .finished
            .as_ref()
            .context("L1 batch is not actually finished")?;

        let progress = L1_BATCH_METRICS.start(L1BatchSealStage::FictiveL2Block);
        // Seal fictive L2 block with last events and storage logs.
        let l2_block_command = self.seal_l2_block_command(
            l2_legacy_shared_bridge_addr,
            false, // fictive L2 blocks don't have txs, so it's fine to pass `false` here.
        );

        let mut connection = pool.connection_tagged("state_keeper").await?;
        let transaction = connection.start_transaction().await?;

        // We rely on the fact that fictive L2 block and L1 batch data is saved in the same transaction.
        let mut strategy = SealStrategy::Sequential(transaction);
        l2_block_command
            .seal_inner(&mut strategy, true)
            .await
            .context("failed persisting fictive L2 block")?;
        let SealStrategy::Sequential(mut transaction) = strategy else {
            panic!("Sealing L2 block should not mutate type of strategy");
        };
        progress.observe(None);

        let progress = L1_BATCH_METRICS.start(L1BatchSealStage::LogDeduplication);

        progress.observe(
            finished_batch
                .final_execution_state
                .deduplicated_storage_logs
                .len(),
        );

        let l1_tx_count = self.l1_batch.l1_tx_count;
        let l2_tx_count = self.l1_batch.executed_transaction_hashes.len() - l1_tx_count;
        let (dedup_writes_count, dedup_reads_count) = log_query_write_read_counts(
            finished_batch
                .final_execution_state
                .deduplicated_storage_logs
                .iter(),
        );

        tracing::info!(
            "Sealing L1 batch {current_l1_batch_number} with timestamp {ts}, {total_tx_count} \
             ({l2_tx_count} L2 + {l1_tx_count} L1) txs, {l2_to_l1_log_count} l2_l1_logs, \
             {event_count} events, {dedup_reads_count} deduped reads, \
             {dedup_writes_count} deduped writes",
            ts = display_timestamp(self.batch_timestamp()),
            total_tx_count = l1_tx_count + l2_tx_count,
            l2_to_l1_log_count = finished_batch
                .final_execution_state
                .user_l2_to_l1_logs
                .len(),
            event_count = finished_batch.final_execution_state.events.len(),
            current_l1_batch_number = self.l1_batch.number
        );

        let progress = L1_BATCH_METRICS.start(L1BatchSealStage::InsertL1BatchHeader);
        let l2_to_l1_messages =
            VmEvent::extract_long_l2_to_l1_messages(&finished_batch.final_execution_state.events);
        let l1_batch = L1BatchHeader {
            number: self.l1_batch.number,
            timestamp: self.batch_timestamp(),
            priority_ops_onchain_data: self.l1_batch.priority_ops_onchain_data.clone(),
            l1_tx_count: l1_tx_count as u16,
            l2_tx_count: l2_tx_count as u16,
            l2_to_l1_logs: finished_batch
                .final_execution_state
                .user_l2_to_l1_logs
                .clone(),
            l2_to_l1_messages,
            bloom: Default::default(),
            used_contract_hashes: finished_batch
                .final_execution_state
                .used_contract_hashes
                .clone(),
            base_system_contracts_hashes: self.base_system_contract_hashes(),
            protocol_version: Some(self.protocol_version()),
            system_logs: finished_batch.final_execution_state.system_logs.clone(),
            pubdata_input: finished_batch.pubdata_input.clone(),
            fee_address: self.fee_account_address,
            batch_fee_input: self.batch_fee_input,
        };

        let final_bootloader_memory = finished_batch
            .final_bootloader_memory
            .clone()
            .unwrap_or_default();

        transaction
            .blocks_dal()
            .mark_l1_batch_as_sealed(
                &l1_batch,
                &final_bootloader_memory,
                &finished_batch.final_execution_state.storage_refunds,
                &finished_batch.final_execution_state.pubdata_costs,
                self.pending_execution_metrics().circuit_statistic,
            )
            .await?;
        progress.observe(None);

        let progress = L1_BATCH_METRICS.start(L1BatchSealStage::SetL1BatchNumberForL2Blocks);
        transaction
            .blocks_dal()
            .mark_l2_blocks_as_executed_in_l1_batch(self.l1_batch.number)
            .await?;
        progress.observe(None);

        let progress = L1_BATCH_METRICS.start(L1BatchSealStage::MarkTxsAsExecutedInL1Batch);
        transaction
            .transactions_dal()
            .mark_txs_as_executed_in_l1_batch(
                self.l1_batch.number,
                &self.l1_batch.executed_transaction_hashes,
            )
            .await?;
        progress.observe(None);

        if insert_protective_reads {
            let progress = L1_BATCH_METRICS.start(L1BatchSealStage::InsertProtectiveReads);
            let protective_reads: Vec<_> = finished_batch
                .final_execution_state
                .deduplicated_storage_logs
                .iter()
                .filter(|log_query| !log_query.is_write())
                .copied()
                .collect();
            transaction
                .storage_logs_dedup_dal()
                .insert_protective_reads(self.l1_batch.number, &protective_reads)
                .await?;
            progress.observe(protective_reads.len());
        }

        let progress = L1_BATCH_METRICS.start(L1BatchSealStage::FilterWrittenSlots);
        let (initial_writes, all_writes_len): (Vec<_>, usize) =
            if let Some(state_diffs) = &finished_batch.state_diffs {
                let all_writes_len = state_diffs.len();

                (
                    state_diffs
                        .iter()
                        .filter(|diff| diff.is_write_initial())
                        .map(|diff| {
                            H256(StorageKey::raw_hashed_key(
                                &diff.address,
                                &u256_to_h256(diff.key),
                            ))
                        })
                        .collect(),
                    all_writes_len,
                )
            } else {
                let deduplicated_writes_hashed_keys_iter = finished_batch
                    .final_execution_state
                    .deduplicated_storage_logs
                    .iter()
                    .filter(|log| log.is_write())
                    .map(|log| log.key.hashed_key());

                let deduplicated_writes_hashed_keys: Vec<_> =
                    deduplicated_writes_hashed_keys_iter.clone().collect();
                let all_writes_len = deduplicated_writes_hashed_keys.len();
                let non_initial_writes = transaction
                    .storage_logs_dedup_dal()
                    .filter_written_slots(&deduplicated_writes_hashed_keys)
                    .await?;

                (
                    deduplicated_writes_hashed_keys_iter
                        .filter(|hashed_key| !non_initial_writes.contains(hashed_key))
                        .collect(),
                    all_writes_len,
                )
            };
        progress.observe(all_writes_len);

        let progress = L1_BATCH_METRICS.start(L1BatchSealStage::InsertInitialWrites);
        transaction
            .storage_logs_dedup_dal()
            .insert_initial_writes(self.l1_batch.number, &initial_writes)
            .await?;
        progress.observe(initial_writes.len());

        let progress = L1_BATCH_METRICS.start(L1BatchSealStage::CommitL1Batch);
        transaction.commit().await?;
        progress.observe(None);

        let writes_metrics = self.storage_writes_deduplicator.metrics();
        // Sanity check metrics.
        anyhow::ensure!(
            all_writes_len
                == writes_metrics.initial_storage_writes + writes_metrics.repeated_storage_writes,
            "Results of in-flight and common deduplications are mismatched"
        );

        self.report_l1_batch_metrics(started_at, &writes_metrics);
        Ok(())
    }
Persists an L1 batch in the storage.
This action includes a creation of an empty "fictive" L2 block that contains
the events generated during the bootloader "tip phase". Returns updates for this fictive L2 block.
core/node/state_keeper/src/io/seal_logic/mod.rs