fix(prover): Fixing sequence_number assigning (#4319)

## What ❔
1. `sequence_number` is now not continuous across all the BWG output
artifacts but instead reflects a sequence number within the same circuit
type.
2. Also it is guaranteed that urls match (sequence_number, circuit_id)
pair in DB.
<!-- 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 ❔
Circuit url includes a sequence_number which is a continuous numbering
across all the circuits. It turned out that the ordering between
different circuit types is not strict. e.g. we have circuit_10 A, B, C
and circuit_2 X, Y, Z. They can be pushed to vec in different order like
[A,B,C,X,Y,Z], [A,X,B,Y,C,Z], [A,X,Y,B,C,Z] etc. And then
sequence_number is assigned like [0,1,2,3,4,5].

That can cause creating two identical jobs with different url and
following proving failure.
<!-- Why are these changes done? What goal do they contribute to? What
are the principles behind them? -->
<!-- The `Why` has to be clear to non-Matter Labs entities running their
own ZK Chain -->
<!-- Example: PR templates ensure PR reviewers, observers, and future
iterators are in context about the evolution of repos. -->

## Is this a breaking change?
- [ ] Yes
- [ ] No

## Operational changes
<!-- Any config changes? Any new flags? Any changes to any scripts? -->
<!-- Please add anything that non-Matter Labs entities running their own
ZK Chain may need to know -->

## 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 `zkstack dev fmt` and `zkstack dev
lint`.

Author: AlexOnTheStorm

prover/crates/lib/prover_dal/src/fri_prover_dal.rs

@@ -70,21 +70,18 @@ impl FriProverDal<'_, '_> {
     pub async fn insert_prover_jobs(
         &mut self,
         batch_id: L1BatchId,
-        circuit_ids_and_urls: Vec<(u8, String)>,
+        circuit_ids_sequence_numbers_and_urls: Vec<(u8, usize, String)>,
         aggregation_round: AggregationRound,
         depth: u16,
         protocol_version_id: ProtocolSemanticVersion,
         batch_sealed_at: DateTime<Utc>,
     ) {
         let _latency = MethodLatency::new("save_fri_prover_jobs");
-        if circuit_ids_and_urls.is_empty() {
+        if circuit_ids_sequence_numbers_and_urls.is_empty() {
             return;
         }
 
-        for (chunk_index, chunk) in circuit_ids_and_urls
-            .chunks(Self::INSERT_JOBS_CHUNK_SIZE)
-            .enumerate()
-        {
+        for chunk in circuit_ids_sequence_numbers_and_urls.chunks(Self::INSERT_JOBS_CHUNK_SIZE) {
             // Build multi-row INSERT for the current chunk
             let mut query_builder = QueryBuilder::new(
                 r#"
@@ -108,14 +105,14 @@ impl FriProverDal<'_, '_> {
             );
 
             query_builder.push_values(
-                chunk.iter().enumerate(),
-                |mut row, (i, (circuit_id, circuit_blob_url))| {
+                chunk.iter(),
+                |mut row, (circuit_id, sequence_number, circuit_blob_url)| {
                     row.push_bind(batch_id.batch_number().0 as i64)
                         .push_bind(batch_id.chain_id().inner() as i64)
                         .push_bind(*circuit_id as i16)
                         .push_bind(circuit_blob_url)
                         .push_bind(aggregation_round as i64)
-                        .push_bind((chunk_index * Self::INSERT_JOBS_CHUNK_SIZE + i) as i64) // sequence_number
+                        .push_bind(*sequence_number as i64) // sequence_number
                         .push_bind(depth as i32)
                         .push_bind(false) // is_node_final_proof
                         .push_bind(protocol_version_id.minor as i32)
@@ -973,9 +970,9 @@ mod tests {
     use super::*;
     use crate::ProverDal;
 
-    fn mock_circuit_ids_and_urls(num_circuits: usize) -> Vec<(u8, String)> {
+    fn mock_circuit_ids_and_urls(num_circuits: usize) -> Vec<(u8, usize, String)> {
         (0..num_circuits)
-            .map(|i| (i as u8, format!("circuit{}", i)))
+            .map(|i| (i as u8, i, format!("circuit{}", i)))
             .collect()
     }
 

prover/crates/lib/witness_generator_service/src/artifact_manager.rs

@@ -8,7 +8,7 @@ use zksync_types::{L1BatchId, L1BatchNumber, L2ChainId};
 #[derive(Debug)]
 pub struct AggregationBlobUrls {
     pub aggregation_urls: String,
-    pub circuit_ids_and_urls: Vec<(u8, String)>,
+    pub circuit_ids_sequence_numbers_and_urls: Vec<(u8, usize, String)>,
 }
 
 #[derive(Debug, Clone, Copy)]

prover/crates/lib/witness_generator_service/src/rounds/basic_circuits/artifacts.rs

@@ -87,7 +87,7 @@ impl ArtifactsManager for BasicCircuits {
             .fri_prover_jobs_dal()
             .insert_prover_jobs(
                 job_id.into(),
-                artifacts.circuit_urls,
+                artifacts.circuit_ids_sequence_numbers_and_urls,
                 AggregationRound::BasicCircuits,
                 0,
                 protocol_version_id,

prover/crates/lib/witness_generator_service/src/rounds/basic_circuits/mod.rs

@@ -28,7 +28,7 @@ mod utils;
 
 #[derive(Clone)]
 pub struct BasicCircuitArtifacts {
-    pub(super) circuit_urls: Vec<(u8, String)>,
+    pub(super) circuit_ids_sequence_numbers_and_urls: Vec<(u8, usize, String)>,
     pub(super) queue_urls: Vec<(u8, String, usize)>,
     pub(super) scheduler_witness: SchedulerCircuitInstanceWitness<
         GoldilocksField,
@@ -45,7 +45,7 @@ pub struct BasicWitnessGeneratorJob {
 }
 
 type Witness = (
-    Vec<(u8, String)>,
+    Vec<(u8, usize, String)>,
     Vec<(u8, String, usize)>,
     SchedulerCircuitInstanceWitness<
         GoldilocksField,
@@ -75,11 +75,15 @@ impl JobManager for BasicCircuits {
             data: job,
         } = job;
 
-        let (circuit_urls, queue_urls, scheduler_witness, aux_output_witness) =
-            generate_witness(batch_id, object_store, job, max_circuits_in_flight).await;
+        let (
+            circuit_ids_sequence_numbers_and_urls,
+            queue_urls,
+            scheduler_witness,
+            aux_output_witness,
+        ) = generate_witness(batch_id, object_store, job, max_circuits_in_flight).await;
 
         Ok(BasicCircuitArtifacts {
-            circuit_urls,
+            circuit_ids_sequence_numbers_and_urls,
             queue_urls,
             scheduler_witness,
             aux_output_witness,

prover/crates/lib/witness_generator_service/src/rounds/basic_circuits/utils.rs

@@ -1,5 +1,5 @@
 use std::{
-    collections::HashSet,
+    collections::{HashMap, HashSet},
     hash::{DefaultHasher, Hash, Hasher},
     sync::Arc,
 };
@@ -155,15 +155,19 @@ pub(super) async fn generate_witness(
         // Ordering determines how we compose the circuit proofs in Leaf Aggregation Round.
         // Sequence is used to determine circuit ordering (the sequencing of instructions) .
         // If the order is tampered with, proving will fail (as the proof would be computed for a different sequence of instruction).
-        let mut circuit_sequence = 0;
+        let mut circuit_sequence_numbers = HashMap::new();
 
         while let Some(circuit) = circuit_receiver
             .recv()
             .instrument(tracing::info_span!("wait_for_circuit"))
             .await
         {
-            let sequence = circuit_sequence;
-            circuit_sequence += 1;
+            let circuit_id = circuit.numeric_circuit_type();
+            let sequence = circuit_sequence_numbers
+                .get(&circuit_id)
+                .cloned()
+                .unwrap_or(0);
+            circuit_sequence_numbers.insert(circuit_id, sequence + 1);
             let object_store = object_store.clone();
             let semaphore = semaphore.clone();
             let permit = semaphore
@@ -175,7 +179,7 @@ pub(super) async fn generate_witness(
                 let (circuit_id, circuit_url) =
                     save_circuit(batch_id, circuit, sequence, object_store).await;
                 drop(permit);
-                (circuit_id, circuit_url)
+                (circuit_id, sequence, circuit_url)
             }));
         }
     }
@@ -217,13 +221,14 @@ pub(super) async fn generate_witness(
     // `circuits_present` stores which circuits exist and is used to filter queues in `recursion_urls` later.
     let mut circuits_present = HashSet::<u8>::new();
 
-    let circuit_urls = futures::future::join_all(save_circuit_handles)
+    let circuit_ids_sequence_numbers_and_urls = futures::future::join_all(save_circuit_handles)
         .await
         .into_iter()
         .map(|result| {
-            let (circuit_id, circuit_url) = result.expect("failed to save circuit");
+            let (circuit_id, sequence_number, circuit_url) =
+                result.expect("failed to save circuit");
             circuits_present.insert(circuit_id);
-            (circuit_id, circuit_url)
+            (circuit_id, sequence_number, circuit_url)
         })
         .collect();
 
@@ -240,7 +245,7 @@ pub(super) async fn generate_witness(
     scheduler_witness.previous_block_aux_hash = input.previous_batch_metadata.aux_hash.0;
 
     (
-        circuit_urls,
+        circuit_ids_sequence_numbers_and_urls,
         recursion_urls,
         scheduler_witness,
         block_aux_witness,

prover/crates/lib/witness_generator_service/src/rounds/leaf_aggregation/artifacts.rs

@@ -58,7 +58,7 @@ impl ArtifactsManager for LeafAggregation {
 
         AggregationBlobUrls {
             aggregation_urls,
-            circuit_ids_and_urls: artifacts.circuit_ids_and_urls,
+            circuit_ids_sequence_numbers_and_urls: artifacts.circuit_ids_sequence_numbers_and_urls,
         }
     }
 
@@ -82,7 +82,7 @@ impl ArtifactsManager for LeafAggregation {
 
         let mut prover_connection = connection_pool.connection().await.unwrap();
         let mut transaction = prover_connection.start_transaction().await.unwrap();
-        let number_of_dependent_jobs = blob_urls.circuit_ids_and_urls.len();
+        let number_of_dependent_jobs = blob_urls.circuit_ids_sequence_numbers_and_urls.len();
         let protocol_version_id = transaction
             .fri_basic_witness_generator_dal()
             .protocol_version_for_l1_batch(artifacts.batch_id)
@@ -95,7 +95,7 @@ impl ArtifactsManager for LeafAggregation {
 
         tracing::info!(
             "Inserting {} prover jobs for job_id {}, block {} with circuit id {}",
-            blob_urls.circuit_ids_and_urls.len(),
+            blob_urls.circuit_ids_sequence_numbers_and_urls.len(),
             job_id,
             artifacts.batch_id,
             artifacts.circuit_id,
@@ -104,7 +104,7 @@ impl ArtifactsManager for LeafAggregation {
             .fri_prover_jobs_dal()
             .insert_prover_jobs(
                 artifacts.batch_id,
-                blob_urls.circuit_ids_and_urls,
+                blob_urls.circuit_ids_sequence_numbers_and_urls,
                 AggregationRound::LeafAggregation,
                 0,
                 protocol_version_id,

prover/crates/lib/witness_generator_service/src/rounds/leaf_aggregation/mod.rs

@@ -55,7 +55,7 @@ pub struct LeafAggregationArtifacts {
     circuit_id: u8,
     batch_id: L1BatchId,
     pub aggregations: Vec<(u64, RecursionQueueSimulator<GoldilocksField>)>,
-    pub circuit_ids_and_urls: Vec<(u8, String)>,
+    pub circuit_ids_sequence_numbers_and_urls: Vec<(u8, usize, String)>,
     #[allow(dead_code)]
     closed_form_inputs: Vec<ZkSyncBaseLayerClosedFormInput<GoldilocksField>>,
 }
@@ -167,8 +167,9 @@ impl JobManager for LeafAggregation {
             handles.push(handle);
         }
 
-        let circuit_ids_and_urls_results = futures::future::join_all(handles).await;
-        let circuit_ids_and_urls = circuit_ids_and_urls_results
+        let circuit_ids_sequence_numbers_and_urls_results =
+            futures::future::join_all(handles).await;
+        let circuit_ids_sequence_numbers_and_urls = circuit_ids_sequence_numbers_and_urls_results
             .into_iter()
             .flat_map(|x| x.unwrap())
             .collect();
@@ -184,7 +185,7 @@ impl JobManager for LeafAggregation {
             circuit_id,
             batch_id: job.batch_id,
             aggregations,
-            circuit_ids_and_urls,
+            circuit_ids_sequence_numbers_and_urls,
             closed_form_inputs: job.closed_form_inputs.0,
         })
     }

prover/crates/lib/witness_generator_service/src/rounds/node_aggregation/artifacts.rs

@@ -63,7 +63,8 @@ impl ArtifactsManager for NodeAggregation {
 
         AggregationBlobUrls {
             aggregation_urls,
-            circuit_ids_and_urls: artifacts.recursive_circuit_ids_and_urls,
+            circuit_ids_sequence_numbers_and_urls: artifacts
+                .recursive_circuit_ids_sequence_numbers_and_urls,
         }
     }
 
@@ -80,7 +81,7 @@ impl ArtifactsManager for NodeAggregation {
     ) -> anyhow::Result<()> {
         let mut prover_connection = connection_pool.connection().await.unwrap();
         let mut transaction = prover_connection.start_transaction().await.unwrap();
-        let dependent_jobs = blob_urls.circuit_ids_and_urls.len();
+        let dependent_jobs = blob_urls.circuit_ids_sequence_numbers_and_urls.len();
         let protocol_version_id = transaction
             .fri_basic_witness_generator_dal()
             .protocol_version_for_l1_batch(artifacts.batch_id)
@@ -97,7 +98,7 @@ impl ArtifactsManager for NodeAggregation {
                     .fri_prover_jobs_dal()
                     .insert_prover_jobs(
                         artifacts.batch_id,
-                        blob_urls.circuit_ids_and_urls,
+                        blob_urls.circuit_ids_sequence_numbers_and_urls,
                         AggregationRound::NodeAggregation,
                         artifacts.depth,
                         protocol_version_id,
@@ -118,7 +119,7 @@ impl ArtifactsManager for NodeAggregation {
                     .await;
             }
             false => {
-                let (_, blob_url) = blob_urls.circuit_ids_and_urls[0].clone();
+                let (_, _, blob_url) = blob_urls.circuit_ids_sequence_numbers_and_urls[0].clone();
                 transaction
                     .fri_prover_jobs_dal()
                     .insert_prover_job(

prover/crates/lib/witness_generator_service/src/rounds/node_aggregation/mod.rs

@@ -40,7 +40,7 @@ pub struct NodeAggregationArtifacts {
     batch_id: L1BatchId,
     depth: u16,
     pub next_aggregations: Vec<(u64, RecursionQueueSimulator<GoldilocksField>)>,
-    pub recursive_circuit_ids_and_urls: Vec<(u8, String)>,
+    pub recursive_circuit_ids_sequence_numbers_and_urls: Vec<(u8, usize, String)>,
 }
 
 #[derive(Clone)]
@@ -150,33 +150,36 @@ impl JobManager for NodeAggregation {
                     &all_leafs_layer_params,
                 );
 
-                let recursive_circuit_id_and_url = save_recursive_layer_prover_input_artifacts(
-                    job.batch_id,
-                    circuit_idx,
-                    vec![recursive_circuit],
-                    AggregationRound::NodeAggregation,
-                    job.depth + 1,
-                    &*object_store,
-                    Some(job.circuit_id),
-                )
-                .await;
+                let recursive_circuit_id_sequence_number_and_url =
+                    save_recursive_layer_prover_input_artifacts(
+                        job.batch_id,
+                        circuit_idx,
+                        vec![recursive_circuit],
+                        AggregationRound::NodeAggregation,
+                        job.depth + 1,
+                        &*object_store,
+                        Some(job.circuit_id),
+                    )
+                    .await;
 
                 (
                     (result_circuit_id, input_queue),
-                    recursive_circuit_id_and_url,
+                    recursive_circuit_id_sequence_number_and_url,
                 )
             });
 
             handles.push(handle);
         }
 
         let mut next_aggregations = vec![];
-        let mut recursive_circuit_ids_and_urls = vec![];
+        let mut recursive_circuit_ids_sequence_numbers_and_urls = vec![];
         for handle in handles {
-            let (next_aggregation, recursive_circuit_id_and_url) = handle.await.unwrap();
+            let (next_aggregation, recursive_circuit_id_sequence_number_and_url) =
+                handle.await.unwrap();
 
             next_aggregations.push(next_aggregation);
-            recursive_circuit_ids_and_urls.extend(recursive_circuit_id_and_url);
+            recursive_circuit_ids_sequence_numbers_and_urls
+                .extend(recursive_circuit_id_sequence_number_and_url);
         }
 
         tracing::info!(
@@ -193,7 +196,7 @@ impl JobManager for NodeAggregation {
             batch_id: job.batch_id,
             depth: job.depth + 1,
             next_aggregations,
-            recursive_circuit_ids_and_urls,
+            recursive_circuit_ids_sequence_numbers_and_urls,
         })
     }
 

prover/crates/lib/witness_generator_service/src/utils.rs

@@ -192,13 +192,14 @@ pub async fn save_recursive_layer_prover_input_artifacts(
     depth: u16,
     object_store: &dyn ObjectStore,
     base_layer_circuit_id: Option<u8>,
-) -> Vec<(u8, String)> {
-    let mut ids_and_urls = Vec::with_capacity(recursive_circuits.len());
+) -> Vec<(u8, usize, String)> {
+    let mut ids_sequence_numbers_and_urls = Vec::with_capacity(recursive_circuits.len());
     for (sequence_number, circuit) in recursive_circuits.into_iter().enumerate() {
         let circuit_id = base_layer_circuit_id.unwrap_or_else(|| circuit.numeric_circuit_type());
+        let sequence_number = sequence_number_offset + sequence_number;
         let circuit_key = FriCircuitKey {
             batch_id,
-            sequence_number: sequence_number_offset + sequence_number,
+            sequence_number,
             circuit_id,
             aggregation_round,
             depth,
@@ -207,9 +208,9 @@ pub async fn save_recursive_layer_prover_input_artifacts(
             .put(circuit_key, &CircuitWrapper::Recursive(circuit))
             .await
             .unwrap();
-        ids_and_urls.push((circuit_id, blob_url));
+        ids_sequence_numbers_and_urls.push((circuit_id, sequence_number, blob_url));
     }
-    ids_and_urls
+    ids_sequence_numbers_and_urls
 }
 
 #[tracing::instrument(skip_all)]