feat(prover)!: Add EIP4844 support for provers subsystem

core/lib/dal/migrations/20240226120530_add_support_for_eip4844.down.sql

@@ -0,0 +1,4 @@
+ALTER TABLE scheduler_dependency_tracker_fri DROP COLUMN IF EXISTS circuit_255_final_prover_job_id_1;
+ALTER TABLE scheduler_dependency_tracker_fri DROP COLUMN IF EXISTS circuit_255_final_prover_job_id_0;
+
+ALTER TABLE witness_inputs_fri DROP COLUMN IF EXISTS eip_4844_blobs;

core/lib/dal/migrations/20240226120530_add_support_for_eip4844.up.sql

@@ -0,0 +1,4 @@
+ALTER TABLE witness_inputs_fri ADD COLUMN eip_4844_blobs BYTEA;
+
+ALTER TABLE scheduler_dependency_tracker_fri ADD COLUMN IF NOT EXISTS circuit_255_final_prover_job_id_0 BIGINT DEFAULT NULL;
+ALTER TABLE scheduler_dependency_tracker_fri ADD COLUMN IF NOT EXISTS circuit_255_final_prover_job_id_1 BIGINT DEFAULT NULL;

core/lib/dal/src/fri_prover_dal.rs

@@ -23,6 +23,10 @@ pub mod types {
     use sqlx::types::chrono::{DateTime, Utc};
     use zksync_types::{basic_fri_types::AggregationRound, L1BatchNumber};
 
+    // This currently lives in zksync_prover_types -- we don't want a dependency between prover types (zkevm_test_harness) and DAL.
+    // This will be gone as part of 1.5.0, when EIP_4844 becomes normal jobs, rather than special cased ones.
+    pub(crate) const EIP_4844_CIRCUIT_ID: u8 = 255;
+
     #[derive(Debug, Clone)]
     pub struct FriProverJobMetadata {
         pub id: u32,
@@ -249,14 +253,15 @@ impl FriProverDal<'_, '_> {
         for (sequence_number, (circuit_id, circuit_blob_url)) in
             circuit_ids_and_urls.iter().enumerate()
         {
+            let is_final = *circuit_id == types::EIP_4844_CIRCUIT_ID;
             self.insert_prover_job(
                 l1_batch_number,
                 *circuit_id,
                 depth,
                 sequence_number,
                 aggregation_round,
                 circuit_blob_url,
-                false,
+                is_final,
                 protocol_version_id,
             )
             .await;

core/lib/dal/src/fri_scheduler_dependency_tracker_dal.rs

@@ -1,6 +1,6 @@
 use zksync_types::L1BatchNumber;
 
-use crate::StorageProcessor;
+use crate::{fri_prover_dal::types, StorageProcessor};
 
 #[derive(Debug)]
 pub struct FriSchedulerDependencyTrackerDal<'a, 'c> {
@@ -35,6 +35,8 @@ impl FriSchedulerDependencyTrackerDal<'_, '_> {
                         AND circuit_11_final_prover_job_id IS NOT NULL
                         AND circuit_12_final_prover_job_id IS NOT NULL
                         AND circuit_13_final_prover_job_id IS NOT NULL
+                        AND circuit_255_final_prover_job_id_0 IS NOT NULL
+                        AND circuit_255_final_prover_job_id_1 IS NOT NULL
                 )
             RETURNING
                 l1_batch_number;
@@ -69,15 +71,27 @@ impl FriSchedulerDependencyTrackerDal<'_, '_> {
         circuit_id: u8,
         final_prover_job_id: u32,
         l1_batch_number: L1BatchNumber,
+        sequence: u16,
     ) {
-        let query = format!(
-            r#"
+        let query = if circuit_id != types::EIP_4844_CIRCUIT_ID {
+            format!(
+                r#"
                 UPDATE scheduler_dependency_tracker_fri
                 SET circuit_{}_final_prover_job_id = $1
                 WHERE l1_batch_number = $2
             "#,
-            circuit_id
-        );
+                circuit_id
+            )
+        } else {
+            format!(
+                r#"
+                    UPDATE scheduler_dependency_tracker_fri
+                    SET circuit_{}_final_prover_job_id_{} = $1
+                    WHERE l1_batch_number = $2
+                "#,
+                circuit_id, sequence,
+            )
+        };
         sqlx::query(&query)
             .bind(final_prover_job_id as i64)
             .bind(l1_batch_number.0 as i64)
@@ -89,7 +103,7 @@ impl FriSchedulerDependencyTrackerDal<'_, '_> {
     pub async fn get_final_prover_job_ids_for(
         &mut self,
         l1_batch_number: L1BatchNumber,
-    ) -> [u32; 13] {
+    ) -> ([u32; 13], [u32; 2]) {
         sqlx::query!(
             r#"
             SELECT
@@ -107,21 +121,27 @@ impl FriSchedulerDependencyTrackerDal<'_, '_> {
         .into_iter()
         .next()
         .map(|row| {
-            [
-                row.circuit_1_final_prover_job_id.unwrap() as u32,
-                row.circuit_2_final_prover_job_id.unwrap() as u32,
-                row.circuit_3_final_prover_job_id.unwrap() as u32,
-                row.circuit_4_final_prover_job_id.unwrap() as u32,
-                row.circuit_5_final_prover_job_id.unwrap() as u32,
-                row.circuit_6_final_prover_job_id.unwrap() as u32,
-                row.circuit_7_final_prover_job_id.unwrap() as u32,
-                row.circuit_8_final_prover_job_id.unwrap() as u32,
-                row.circuit_9_final_prover_job_id.unwrap() as u32,
-                row.circuit_10_final_prover_job_id.unwrap() as u32,
-                row.circuit_11_final_prover_job_id.unwrap() as u32,
-                row.circuit_12_final_prover_job_id.unwrap() as u32,
-                row.circuit_13_final_prover_job_id.unwrap() as u32,
-            ]
+            (
+                [
+                    row.circuit_1_final_prover_job_id.unwrap() as u32,
+                    row.circuit_2_final_prover_job_id.unwrap() as u32,
+                    row.circuit_3_final_prover_job_id.unwrap() as u32,
+                    row.circuit_4_final_prover_job_id.unwrap() as u32,
+                    row.circuit_5_final_prover_job_id.unwrap() as u32,
+                    row.circuit_6_final_prover_job_id.unwrap() as u32,
+                    row.circuit_7_final_prover_job_id.unwrap() as u32,
+                    row.circuit_8_final_prover_job_id.unwrap() as u32,
+                    row.circuit_9_final_prover_job_id.unwrap() as u32,
+                    row.circuit_10_final_prover_job_id.unwrap() as u32,
+                    row.circuit_11_final_prover_job_id.unwrap() as u32,
+                    row.circuit_12_final_prover_job_id.unwrap() as u32,
+                    row.circuit_13_final_prover_job_id.unwrap() as u32,
+                ],
+                [
+                    row.circuit_255_final_prover_job_id_0.unwrap() as u32,
+                    row.circuit_255_final_prover_job_id_1.unwrap() as u32,
+                ],
+            )
         })
         .unwrap()
     }

core/lib/dal/src/fri_witness_generator_dal.rs

@@ -39,6 +39,7 @@ impl FriWitnessGeneratorDal<'_, '_> {
         block_number: L1BatchNumber,
         object_key: &str,
         protocol_version_id: FriProtocolVersionId,
+        eip_4844_blobs: Vec<u8>,
     ) {
         sqlx::query!(
             r#"
@@ -47,17 +48,19 @@ impl FriWitnessGeneratorDal<'_, '_> {
                     l1_batch_number,
                     merkle_tree_paths_blob_url,
                     protocol_version,
+                    eip_4844_blobs,
                     status,
                     created_at,
                     updated_at
                 )
             VALUES
-                ($1, $2, $3, 'queued', NOW(), NOW())
+                ($1, $2, $3, $4, 'queued', NOW(), NOW())
             ON CONFLICT (l1_batch_number) DO NOTHING
             "#,
             block_number.0 as i64,
             object_key,
             protocol_version_id as i32,
+            bincode::serialize(&eip_4844_blobs).expect("Failed to serialize 4844 blobs to bytes"),
         )
         .fetch_optional(self.storage.conn())
         .await
@@ -69,7 +72,7 @@ impl FriWitnessGeneratorDal<'_, '_> {
         last_l1_batch_to_process: u32,
         protocol_versions: &[FriProtocolVersionId],
         picked_by: &str,
-    ) -> Option<L1BatchNumber> {
+    ) -> Option<(L1BatchNumber, Vec<u8>)> {
         let protocol_versions: Vec<i32> = protocol_versions.iter().map(|&id| id as i32).collect();
         sqlx::query!(
             r#"
@@ -107,7 +110,13 @@ impl FriWitnessGeneratorDal<'_, '_> {
         .fetch_optional(self.storage.conn())
         .await
         .unwrap()
-        .map(|row| L1BatchNumber(row.l1_batch_number as u32))
+        .map(|row| {
+            (
+                L1BatchNumber(row.l1_batch_number as u32),
+                bincode::deserialize(&row.eip_4844_blobs.unwrap())
+                    .expect("Failed to serialize 4844 blobs to bytes"),
+            )
+        })
     }
 
     pub async fn get_basic_circuit_witness_job_attempts(

core/lib/prover_interface/src/api.rs

@@ -9,12 +9,44 @@ use zksync_types::{
 
 use crate::{inputs::PrepareBasicCircuitsJob, outputs::L1BatchProofForL1};
 
+const BLOB_CHUNK_SIZE: usize = 31;
+const ELEMENTS_PER_4844_BLOCK: usize = 4096;
+
+pub const EIP_4844_BLOB_SIZE: usize = BLOB_CHUNK_SIZE * ELEMENTS_PER_4844_BLOCK;
+
+pub type Blob = Vec<u8>;
+
+#[derive(Debug, Serialize, Deserialize, Clone)]
+pub struct Eip4844Blobs {
+    pub blobs: Vec<Blob>,
+}
+
+impl From<Vec<u8>> for Eip4844Blobs {
+    fn from(mut blobs: Vec<u8>) -> Self {
+        let blob_size = blobs.len() / EIP_4844_BLOB_SIZE * EIP_4844_BLOB_SIZE;
+        blobs.resize(blob_size, 0u8);
+        Self {
+            blobs: blobs
+                .chunks(EIP_4844_BLOB_SIZE)
+                .map(|chunk| chunk.into())
+                .collect(),
+        }
+    }
+}
+
+impl From<Eip4844Blobs> for Vec<u8> {
+    fn from(eip_4844_blobs: Eip4844Blobs) -> Self {
+        eip_4844_blobs.blobs.iter().flatten().copied().collect()
+    }
+}
+
 #[derive(Debug, Serialize, Deserialize)]
 pub struct ProofGenerationData {
     pub l1_batch_number: L1BatchNumber,
     pub data: PrepareBasicCircuitsJob,
     pub fri_protocol_version_id: FriProtocolVersionId,
     pub l1_verifier_config: L1VerifierConfig,
+    pub eip_4844_blobs: Eip4844Blobs,
 }
 
 #[derive(Debug, Serialize, Deserialize)]

core/lib/zksync_core/src/proof_data_handler/request_processor.rs

@@ -12,7 +12,7 @@ use zksync_config::configs::{
 use zksync_dal::{ConnectionPool, SqlxError};
 use zksync_object_store::{ObjectStore, ObjectStoreError};
 use zksync_prover_interface::api::{
-    ProofGenerationData, ProofGenerationDataRequest, ProofGenerationDataResponse,
+    Eip4844Blobs, ProofGenerationData, ProofGenerationDataRequest, ProofGenerationDataResponse,
     SubmitProofRequest, SubmitProofResponse,
 };
 use zksync_types::{
@@ -118,13 +118,31 @@ impl RequestProcessor {
             }
         };
 
+        let storage_batch = self
+            .pool
+            .access_storage()
+            .await
+            .unwrap()
+            .blocks_dal()
+            .get_storage_l1_batch(l1_batch_number)
+            .await
+            .unwrap()
+            .unwrap();
+
+        let eip_4844_blobs: Eip4844Blobs = storage_batch
+            .pubdata_input
+            .expect(&format!(
+                "expected pubdata, but it is not available for batch {l1_batch_number:?}"
+            ))
+            .into();
+
         let proof_gen_data = ProofGenerationData {
             l1_batch_number,
             data: blob,
             fri_protocol_version_id,
             l1_verifier_config,
+            eip_4844_blobs,
         };
-
         Ok(Json(ProofGenerationDataResponse::Success(Some(
             proof_gen_data,
         ))))