Integrate engine_getBlobsV1 with PeerDAS

* Republish data column sidecars only once
hangleang · Feb 5, 2025 · 8c8dc94 · 8c8dc94
1 parent 0c257e4
commit 8c8dc94
Show file tree

Hide file tree

Showing 8 changed files with 272 additions and 159 deletions.
diff --git a/Cargo.lock b/Cargo.lock
diff --git a/eth1_api/Cargo.toml b/eth1_api/Cargo.toml
@@ -12,6 +12,7 @@ arc-swap = { workspace = true }
 bls = { workspace = true }
 dedicated_executor = { workspace = true }
 derive_more = { workspace = true }
+eip_7594 = { workspace = true }
 either = { workspace = true }
 enum-iterator = { workspace = true }
 ethereum-types = { workspace = true }

diff --git a/eth1_api/src/tasks.rs b/eth1_api/src/tasks.rs
@@ -49,51 +49,102 @@ async fn download_blobs<P: Preset, W: Wait>(
     blob_indices: Vec<BlobIndex>,
 ) {
     if let Some(body) = block.message().body().post_deneb() {
-        let kzg_commitments = body
-            .blob_kzg_commitments()
-            .iter()
-            .zip(0..)
-            .filter(|(_, index)| blob_indices.contains(index))
-            .collect::<Vec<_>>();
-
-        let versioned_hashes = kzg_commitments
-            .iter()
-            .copied()
-            .map(|(commitment, _)| misc::kzg_commitment_to_versioned_hash(*commitment))
-            .collect();
-
-        match eth1_api.get_blobs::<P>(versioned_hashes).await {
-            Ok(blobs_and_proofs) => {
-                let block_header = block.to_header();
-
-                for (blob_and_proof, kzg_commitment, index) in blobs_and_proofs
-                    .into_iter()
-                    .zip(kzg_commitments.into_iter())
-                    .filter_map(|(blob_and_proof, (kzg_commitment, index))| {
-                        blob_and_proof.map(|blob_and_proof| (blob_and_proof, kzg_commitment, index))
-                    })
-                {
-                    let BlobAndProofV1 { blob, proof } = blob_and_proof;
-
-                    match misc::construct_blob_sidecar(
-                        &block,
-                        block_header,
-                        index,
-                        blob,
-                        *kzg_commitment,
-                        proof,
-                    ) {
-                        Ok(blob_sidecar) => {
-                            controller.on_el_blob_sidecar(Arc::new(blob_sidecar));
+        if block.phase().is_peerdas_activated() {
+            let versioned_hashes = body
+                .blob_kzg_commitments()
+                .iter()
+                .copied()
+                .map(misc::kzg_commitment_to_versioned_hash)
+                .collect::<Vec<_>>();
+            let blob_count = versioned_hashes.len();
+
+            match eth1_api.get_blobs::<P>(versioned_hashes).await {
+                Ok(blobs_and_proofs) => {
+                    let blobs = blobs_and_proofs
+                        .into_iter()
+                        .filter_map(|blob_and_proof| {
+                            blob_and_proof.map(|blob_and_proof| blob_and_proof.blob)
+                        })
+                        .collect::<Vec<_>>();
+
+                    if blobs.len() == blob_count {
+                        match eip_7594::try_convert_to_cells_and_kzg_proofs::<P>(blobs.into_iter())
+                        {
+                            Ok(cells_and_kzg_proofs) => {
+                                match eip_7594::construct_data_column_sidecars(
+                                    &block,
+                                    &cells_and_kzg_proofs,
+                                    controller.chain_config(),
+                                ) {
+                                    Ok(data_column_sidecars) => {
+                                        for data_column_sidecar in data_column_sidecars {
+                                            controller.on_el_data_column_sidecar(Arc::new(
+                                                data_column_sidecar,
+                                            ));
+                                        }
+                                    }
+                                    Err(error) => warn!(
+                                        "failed to construct data column sidecars with \
+                                        cells and kzg proofs: {error:?}"
+                                    ),
+                                }
+                            }
+                            Err(error) => warn!(
+                                "failed to convert blobs received from execution layer \
+                                into cells and kzg proofs: {error:?}"
+                            ),
                         }
-                        Err(error) => warn!(
-                            "failed to construct blob sidecar with blob and proof \
-                            received from execution layer: {error:?}"
-                        ),
                     }
                 }
+                Err(error) => warn!("engine_getBlobsV1 call failed: {error}"),
+            }
+        } else {
+            let kzg_commitments = body
+                .blob_kzg_commitments()
+                .iter()
+                .zip(0..)
+                .filter(|(_, index)| blob_indices.contains(index))
+                .collect::<Vec<_>>();
+            let versioned_hashes = kzg_commitments
+                .iter()
+                .copied()
+                .map(|(commitment, _)| misc::kzg_commitment_to_versioned_hash(*commitment))
+                .collect();
+
+            match eth1_api.get_blobs::<P>(versioned_hashes).await {
+                Ok(blobs_and_proofs) => {
+                    let block_header = block.to_header();
+
+                    for (blob_and_proof, kzg_commitment, index) in blobs_and_proofs
+                        .into_iter()
+                        .zip(kzg_commitments.into_iter())
+                        .filter_map(|(blob_and_proof, (kzg_commitment, index))| {
+                            blob_and_proof
+                                .map(|blob_and_proof| (blob_and_proof, kzg_commitment, index))
+                        })
+                    {
+                        let BlobAndProofV1 { blob, proof } = blob_and_proof;
+
+                        match misc::construct_blob_sidecar(
+                            &block,
+                            block_header,
+                            index,
+                            blob,
+                            *kzg_commitment,
+                            proof,
+                        ) {
+                            Ok(blob_sidecar) => {
+                                controller.on_el_blob_sidecar(Arc::new(blob_sidecar));
+                            }
+                            Err(error) => warn!(
+                                "failed to construct blob sidecar with blob and proof \
+                                received from execution layer: {error:?}"
+                            ),
+                        }
+                    }
+                }
+                Err(error) => warn!("engine_getBlobsV1 call failed: {error}"),
             }
-            Err(error) => warn!("engine_getBlobsV1 call failed: {error}"),
         }
     }
 }

diff --git a/fork_choice_control/src/controller.rs b/fork_choice_control/src/controller.rs
@@ -446,6 +446,14 @@ where
         self.spawn_blob_sidecar_task(blob_sidecar, true, BlobSidecarOrigin::ExecutionLayer)
     }
 
+    pub fn on_el_data_column_sidecar(&self, data_column_sidecar: Arc<DataColumnSidecar<P>>) {
+        self.spawn_data_column_sidecar_task(
+            data_column_sidecar,
+            true,
+            DataColumnSidecarOrigin::ExecutionLayer,
+        )
+    }
+
     pub fn on_reconstruct_data_column_sidecar(
         &self,
         data_column_sidecar: Arc<DataColumnSidecar<P>>,