fix(aggregator): correct order of processing of prove transactions (#…

…1333)

## What ❔

<!-- What are the changes this PR brings about? -->
<!-- Example: This PR adds a PR template to the repo. -->
<!-- (For bigger PRs adding more context is appreciated) -->

## Why ❔

<!-- Why are these changes done? What goal do they contribute to? What
are the principles behind them? -->
<!-- Example: PR templates ensure PR reviewers, observers, and future
iterators are in context about the evolution of repos. -->

## Checklist

<!-- Check your PR fulfills the following items. -->
<!-- For draft PRs check the boxes as you complete them. -->

- [ ] PR title corresponds to the body of PR (we generate changelog
entries from PRs).
- [ ] Tests for the changes have been added / updated.
- [ ] Documentation comments have been added / updated.
- [ ] Code has been formatted via `zk fmt` and `zk lint`.
- [ ] Spellcheck has been run via `zk spellcheck`.
- [ ] Linkcheck has been run via `zk linkcheck`.

core/lib/dal/src/blocks_dal.rs

@@ -1096,19 +1096,10 @@ impl BlocksDal<'_, '_> {
     ///   respective commit transactions have been confirmed by the network.
     pub async fn get_ready_for_dummy_proof_l1_batches(
         &mut self,
-        only_commited_batches: bool,
         limit: usize,
     ) -> anyhow::Result<Vec<L1BatchWithMetadata>> {
-        let (confirmed_at_not_null, join_on_eth_tx_history) = if only_commited_batches {
-            (
-                "AND confirmed_at IS NOT NULL",
-                "JOIN eth_txs_history ON eth_commit_tx_id = eth_tx_id",
-            )
-        } else {
-            ("", "")
-        };
-
-        let query = format!(
+        let raw_batches = sqlx::query_as!(
+            StorageL1Batch,
             r#"
             SELECT
                 number,
@@ -1145,27 +1136,20 @@ impl BlocksDal<'_, '_> {
             FROM
                 l1_batches
                 LEFT JOIN commitments ON commitments.l1_batch_number = l1_batches.number
-                {join_on_eth_tx_history}
             WHERE
                 eth_commit_tx_id IS NOT NULL
                 AND eth_prove_tx_id IS NULL
-                {confirmed_at_not_null}
             ORDER BY
                 number
             LIMIT
                 $1
             "#,
-        );
-
-        let mut query = sqlx::query_as(&query);
-
-        query = query.bind(limit as i32);
-
-        let raw_batches: Vec<StorageL1Batch> = query
-            .instrument("get_ready_for_dummy_proof_l1_batches")
-            .with_arg("limit", &limit)
-            .fetch_all(self.storage)
-            .await?;
+            limit as i32
+        )
+        .instrument("get_ready_for_dummy_proof_l1_batches")
+        .with_arg("limit", &limit)
+        .fetch_all(self.storage)
+        .await?;
 
         self.map_l1_batches(raw_batches)
             .await

core/lib/zksync_core/src/eth_sender/aggregator.rs

@@ -248,6 +248,50 @@ impl Aggregator {
         })
     }
 
+    async fn load_dummy_proof_operations(
+        storage: &mut StorageProcessor<'_>,
+        limit: usize,
+        is_4844_mode: bool,
+    ) -> Vec<L1BatchWithMetadata> {
+        let mut ready_for_proof_l1_batches = storage
+            .blocks_dal()
+            .get_ready_for_dummy_proof_l1_batches(limit)
+            .await
+            .unwrap();
+
+        // need to find first batch with an unconfirmed commit transaction
+        // and discard it and all the following ones.
+        if is_4844_mode {
+            let mut committed_batches = vec![];
+
+            for batch in ready_for_proof_l1_batches.into_iter() {
+                let Some(commit_tx_id) = storage
+                    .blocks_dal()
+                    .get_eth_commit_tx_id(batch.header.number)
+                    .await
+                    .unwrap()
+                else {
+                    break;
+                };
+
+                if storage
+                    .eth_sender_dal()
+                    .get_confirmed_tx_hash_by_eth_tx_id(commit_tx_id as u32)
+                    .await
+                    .unwrap()
+                    .is_none()
+                {
+                    break;
+                }
+                committed_batches.push(batch);
+            }
+
+            ready_for_proof_l1_batches = committed_batches;
+        }
+
+        ready_for_proof_l1_batches
+    }
+
     async fn load_real_proof_operation(
         storage: &mut StorageProcessor<'_>,
         l1_verifier_config: L1VerifierConfig,
@@ -390,11 +434,8 @@ impl Aggregator {
             }
 
             ProofSendingMode::SkipEveryProof => {
-                let ready_for_proof_l1_batches = storage
-                    .blocks_dal()
-                    .get_ready_for_dummy_proof_l1_batches(self.operate_4844_mode, limit)
-                    .await
-                    .unwrap();
+                let ready_for_proof_l1_batches =
+                    Self::load_dummy_proof_operations(storage, limit, self.operate_4844_mode).await;
                 self.prepare_dummy_proof_operation(
                     storage,
                     ready_for_proof_l1_batches,