Add blob database verifier

Author: macladson

beacon_node/http_api/src/lib.rs

@@ -79,6 +79,7 @@ use std::path::PathBuf;
 use std::pin::Pin;
 use std::str::FromStr;
 use std::sync::Arc;
+use store::BlobSidecarListFromRoot;
 use sysinfo::{System, SystemExt};
 use system_health::{observe_nat, observe_system_health_bn};
 use task_spawner::{Priority, TaskSpawner};
@@ -4747,7 +4748,6 @@ pub fn serve<T: BeaconChainTypes>(
              task_spawner: TaskSpawner<T::EthSpec>,
              chain: Arc<BeaconChain<T>>| {
                 task_spawner.spawn_async_with_rejection(Priority::P1, async move {
-                    let mut results = Vec::new();
                     let deneb_start_slot =
                         if let Some(deneb_fork_epoch) = chain.spec.deneb_fork_epoch {
                             deneb_fork_epoch.start_slot(T::EthSpec::slots_per_epoch())
@@ -4756,51 +4756,77 @@ pub fn serve<T: BeaconChainTypes>(
                                 "No Deneb fork scheduled".to_string(),
                             ));
                         };
-                    let start_slot = query.start_slot.unwrap_or(deneb_start_slot);
+                    let start_slot = query.start_slot;
+
                     if start_slot < deneb_start_slot {
                         return Err(warp_utils::reject::custom_bad_request(format!(
                             "start_slot ({}) must be >= deneb fork slot ({})",
                             start_slot, deneb_start_slot
                         )));
                     }
-                    // Maybe use chain.canonical_head.cached_head().head_slot()??
-                    let Ok(current_head_slot) = chain.slot() else {
-                        return Err(warp_utils::reject::custom_bad_request(
-                            "Failed to get current head slot".to_string(),
-                        ));
-                    };
 
-                    let end_slot = query.end_slot.unwrap_or(current_head_slot);
+                    let end_slot = query.end_slot;
                     if end_slot < start_slot {
                         return Err(warp_utils::reject::custom_bad_request(format!(
                             "end_slot ({}) must be >= start_slot ({})",
                             end_slot, start_slot
                         )));
                     }
 
+                    let verify = query.verify.unwrap_or(true);
+
+                    let mut blob_count = 0;
+                    let mut blobs_missing: Vec<u64> = Vec::new();
+                    let mut blobs_invalid: Vec<u64> = Vec::new();
+
                     for slot in start_slot.as_u64()..=end_slot.as_u64() {
                         if let Ok((root, _, _)) = BlockId::from_slot(Slot::from(slot)).root(&chain)
                         {
-                            if let Ok(blob_list_res) = chain.store.get_blobs(&root) {
-                                if let Some(blob_list) = blob_list_res.blobs() {
-                                    if let Err(_e) =
-                                        verify_kzg_for_blob_list(blob_list.iter(), &chain.kzg)
-                                    {
-                                        results.push(BlobsVerificationData {
-                                            block_root: root,
-                                            slot: slot.into(),
-                                            blobs_exist: true,
-                                            blobs_stored: false,
-                                            blobs_verified: true,
-                                        });
+                            match chain.store.get_blobs(&root) {
+                                Ok(blob_list) => {
+                                    match blob_list {
+                                        BlobSidecarListFromRoot::NoBlobs => {
+                                            // This means that no blobs exist for this slot.
+                                            continue;
+                                        }
+                                        BlobSidecarListFromRoot::NoRoot => {
+                                            // This means that there are blobs missing for this slot.
+                                            blobs_missing.push(slot);
+                                        }
+                                        BlobSidecarListFromRoot::Blobs(blob_list) => {
+                                            blob_count += blob_list.len();
+                                            // Optionally verify each blob_list.
+                                            if verify
+                                                && verify_kzg_for_blob_list(
+                                                    blob_list.iter(),
+                                                    &chain.kzg,
+                                                )
+                                                .is_err()
+                                            {
+                                                blobs_invalid.push(slot);
+                                            }
+                                        }
                                     }
                                 }
+                                Err(_) => {
+                                    // An error here means that we could not decode the blob list.
+                                    // This likely means a corrupted database.
+                                    blobs_invalid.push(slot);
+                                }
                             }
                         }
+                        // An Err here means the block does not exist. This is fine assuming the node is synced.
                     }
+
                     Ok::<_, warp::reject::Rejection>(
-                        warp::reply::json(&api_types::GenericResponse::from(results))
-                            .into_response(),
+                        warp::reply::json(&api_types::GenericResponse::from(
+                            BlobsVerificationData {
+                                blob_count,
+                                blobs_missing,
+                                blobs_invalid,
+                            },
+                        ))
+                        .into_response(),
                     )
                 })
             },

common/eth2/src/lighthouse.rs

@@ -395,12 +395,13 @@ impl BeaconNodeHttpClient {
         self.post_generic_with_ssz_body(path, blobs, None).await
     }
 
-    /// `POST lighthouse/database/verify_blobs`
+    /// `GET lighthouse/database/verify_blobs`
     pub async fn get_lighthouse_database_verify_blobs(
         &self,
-        start_slot: Option<Slot>,
-        end_slot: Option<Slot>,
-    ) -> Result<Vec<BlobsVerificationData>, Error> {
+        start_slot: Slot,
+        end_slot: Slot,
+        verify: Option<bool>,
+    ) -> Result<BlobsVerificationData, Error> {
         let mut path = self.server.full.clone();
 
         path.path_segments_mut()
@@ -409,14 +410,15 @@ impl BeaconNodeHttpClient {
             .push("database")
             .push("verify_blobs");
 
-        if let Some(start_slot) = start_slot {
-            path.query_pairs_mut()
-                .append_pair("start_slot", &start_slot.to_string());
-        }
+        path.query_pairs_mut()
+            .append_pair("start_slot", &start_slot.to_string());
 
-        if let Some(end_slot) = end_slot {
+        path.query_pairs_mut()
+            .append_pair("end_slot", &end_slot.to_string());
+
+        if let Some(verify) = verify {
             path.query_pairs_mut()
-                .append_pair("end_slot", &end_slot.to_string());
+                .append_pair("verify", &verify.to_string());
         }
 
         self.get(path).await

common/eth2/src/lighthouse/blobs_verification.rs

@@ -1,11 +1,8 @@
 use serde::{Deserialize, Serialize};
-use types::{Hash256, Slot};
 
 #[derive(Debug, Default, PartialEq, Clone, Serialize, Deserialize)]
 pub struct BlobsVerificationData {
-    pub block_root: Hash256,
-    pub slot: Slot,
-    pub blobs_exist: bool,
-    pub blobs_stored: bool,
-    pub blobs_verified: bool,
+    pub blob_count: usize,
+    pub blobs_missing: Vec<u64>,
+    pub blobs_invalid: Vec<u64>,
 }

common/eth2/src/types.rs

@@ -821,8 +821,9 @@ pub struct ImportBlobsQuery {
 
 #[derive(Debug, Deserialize)]
 pub struct VerifyBlobsQuery {
-    pub start_slot: Option<Slot>,
-    pub end_slot: Option<Slot>,
+    pub start_slot: Slot,
+    pub end_slot: Slot,
+    pub verify: Option<bool>,
 }
 
 #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, Hash)]

database_manager/src/blobs_manager/cli.rs

@@ -20,27 +20,38 @@ pub struct VerifyBlobs {
         display_order = 0
     )]
     pub beacon_node: Option<String>,
+
     #[clap(
         long,
         value_name = "SLOT",
         help = "The slot at which to begin blob verification. Defaults to the Deneb start slot",
         display_order = 0
     )]
     pub start_slot: Option<u64>,
+
     #[clap(
         long,
         value_name = "SLOT",
         help = "The slot at which to stop blob verification. Defaults to the latest slot",
         display_order = 0
     )]
     pub end_slot: Option<u64>,
+
     #[clap(
         long,
-        help = "Perform verification even if the beacon node is not synced",
+        help = "Perform checks even if the beacon node is not synced",
         display_order = 0,
         default_value = "false"
     )]
     pub allow_unsynced: bool,
+
+    #[clap(
+        long,
+        help = "Skip KZG verification and only perform blob availability checks",
+        display_order = 0,
+        default_value = "false"
+    )]
+    pub skip_verification: bool,
 }
 
 #[derive(Parser, Clone, Deserialize, Serialize, Debug)]
@@ -68,6 +79,7 @@ pub struct ImportBlobs {
         default_value = "false"
     )]
     pub skip_verification: bool,
+
     #[clap(
         long,
         help = "Attempt import even if the beacon node is not synced",
@@ -95,6 +107,7 @@ pub struct ExportBlobs {
         display_order = 0
     )]
     pub output_dir: PathBuf,
+
     #[clap(
         long,
         value_name = "SLOT",