async fn process_multiple_batches(
        &mut self,
        pool: &ConnectionPool<Core>,
        l1_batch_numbers: ops::RangeInclusive<L1BatchNumber>,
    ) -> anyhow::Result<L1BatchNumber> {
        let start = Instant::now();
        tracing::info!("Processing L1 batches #{l1_batch_numbers:?}");
        let mut storage = pool.connection_tagged("tree_updater").await?;
        let first_l1_batch_number = *l1_batch_numbers.start();
        let last_l1_batch_number = *l1_batch_numbers.end();
        let mut l1_batch_data = L1BatchWithLogs::new(&mut storage, first_l1_batch_number)
            .await
            .with_context(|| {
                format!("failed fetching tree input for L1 batch #{first_l1_batch_number}")
            })?;
        drop(storage);

        let l1_batch_numbers =
            (first_l1_batch_number.0..=last_l1_batch_number.0).map(L1BatchNumber);
        let mut total_logs = 0;
        let mut updated_batch_stats = vec![];
        for l1_batch_number in l1_batch_numbers {
            let mut storage = pool.connection_tagged("tree_updater").await?;
            let Some(current_l1_batch_data) = l1_batch_data else {
                // TODO: ensure that l1_batch_number is not pruned
                return Ok(l1_batch_number);
            };
            total_logs += current_l1_batch_data.tree_logs.len();

            let process_l1_batch_task = self.process_l1_batch(current_l1_batch_data);
            let load_next_l1_batch_task = async {
                if l1_batch_number < last_l1_batch_number {
                    let next_l1_batch_number = l1_batch_number + 1;
                    let batch_result =
                        L1BatchWithLogs::new(&mut storage, next_l1_batch_number).await;
                    // Drop storage at the earliest possible moment so that it doesn't block logic running concurrently,
                    // such as tree pruning.
                    drop(storage);
                    batch_result.with_context(|| {
                        format!("failed fetching tree input for L1 batch #{next_l1_batch_number}")
                    })
                } else {
                    Ok(None) // Don't need to load the next L1 batch after the last one we're processing.
                }
            };
            let ((batch_stats, tree_data), next_l1_batch_data) =
                futures::future::try_join(process_l1_batch_task, load_next_l1_batch_task).await?;

            let save_postgres_latency = METRICS.start_stage(TreeUpdateStage::SavePostgres);

            let mut storage = pool.connection_tagged("tree_updater").await?;
            storage
                .blocks_dal()
                .save_l1_batch_tree_data(l1_batch_number, &tree_data)
                .await?;
            // ^ Note that `save_l1_batch_tree_data()` will not blindly overwrite changes if L1 batch
            // metadata already exists; instead, it'll check that the old and new metadata match.
            // That is, if we run multiple tree instances, we'll get metadata correspondence
            // right away without having to implement dedicated code.
            drop(storage);
            save_postgres_latency.observe();
            tracing::info!("Updated metadata for L1 batch #{l1_batch_number} in Postgres");

            updated_batch_stats.push(batch_stats);
            l1_batch_data = next_l1_batch_data;
        }

        let save_rocksdb_latency = METRICS.start_stage(TreeUpdateStage::SaveRocksdb);
        self.tree.save().await?;
        save_rocksdb_latency.observe();
        update_tree_metrics(&updated_batch_stats, total_logs, start);

        Ok(last_l1_batch_number + 1)
    }
Processes a range of L1 batches with a single flushing of the tree updates to RocksDB at the end.
This allows to save on RocksDB I/O ops.

Returns the number of the next L1 batch to be processed by the tree.

# Implementation details

We load L1 batch data from Postgres in parallel with updating the tree. (Naturally, we need to load
the first L1 batch data beforehand.) This allows saving some time if we actually process
multiple L1 batches at once (e.g., during the initial tree syncing), and if loading data from Postgres
is slow for whatever reason.
core/node/zk_os_tree_manager/src/updater.rs