async fn dispatch(&self) -> anyhow::Result<()> {
        let mut conn = self.pool.connection_tagged("da_dispatcher").await?;
        let batches = conn
            .data_availability_dal()
            .get_ready_for_da_dispatch_l1_batches(self.config.max_rows_to_dispatch as usize)
            .await?;
        drop(conn);

        for batch in &batches {
            let dispatch_latency = METRICS.blob_dispatch_latency.start();
            let dispatch_response = retry(
                self.config.max_retries,
                batch.l1_batch_number,
                "DA dispatch",
                || {
                    self.client
                        .dispatch_blob(batch.l1_batch_number.0, batch.pubdata.clone())
                },
            )
            .await
            .with_context(|| {
                format!(
                    "failed to dispatch a blob with batch_number: {}, pubdata_len: {}",
                    batch.l1_batch_number,
                    batch.pubdata.len()
                )
            })?;
            let dispatch_latency_duration = dispatch_latency.observe();

            let sent_at = Utc::now();

            let mut conn = self.pool.connection_tagged("da_dispatcher").await?;
            conn.data_availability_dal()
                .insert_l1_batch_da_request_id(
                    batch.l1_batch_number,
                    dispatch_response.request_id.as_str(),
                    sent_at.naive_utc(),
                    self.client.client_type().into_pubdata_type(),
                    Some(find_l2_da_validator_address(batch.system_logs.as_slice())?),
                )
                .await?;
            drop(conn);

            METRICS
                .last_dispatched_l1_batch
                .set(batch.l1_batch_number.0 as usize);
            METRICS.blob_size.observe(batch.pubdata.len());
            METRICS.sealed_to_dispatched_lag.observe(
                sent_at
                    .signed_duration_since(batch.sealed_at)
                    .to_std()
                    .context("sent_at has to be higher than sealed_at")?,
            );
            tracing::info!(
                "Dispatched a DA for batch_number: {}, pubdata_size: {}, dispatch_latency: {dispatch_latency_duration:?}",
                batch.l1_batch_number,
                batch.pubdata.len(),
            );
        }

        // We don't need to report this metric every iteration, only once when the balance is changed
        if !batches.is_empty() {
            let client_arc = Arc::new(self.client.clone_boxed());

            tokio::spawn(async move {
                let balance = client_arc
                    .balance()
                    .await
                    .context("Unable to retrieve DA operator balance");

                match balance {
                    Ok(balance) => {
                        METRICS.operator_balance.set(balance);
                    }
                    Err(err) => {
                        tracing::error!("{err}")
                    }
                }
            });
        }

        Ok(())
    }
Dispatches the blobs to the data availability layer, and saves the dispatch_request_id in the database.
core/node/da_dispatcher/src/da_dispatcher.rs